"""Phase 2 TaskSession: in-memory + JSON file. Not a TaskInstance."""

from __future__ import annotations

import asyncio
import json
import os
import time
import uuid
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List, Optional, TYPE_CHECKING

from config.config_loader import get_project_dir
from config.logger import setup_logging

if TYPE_CHECKING:
    from core.connection import ConnectionHandler

TAG = __name__
logger = setup_logging()

SKILL_EVENT_PREFIX = "[skill_event]"
DEFER_ASR_REPORT_MODES = ("PLAY_AUDIO", "SHADOWING")


def _now_ms() -> int:
    return int(time.time() * 1000)


def session_dir() -> str:
    return os.path.join(get_project_dir(), "data", "task_sessions")


def parse_skill_event(text: str) -> Optional[Dict[str, str]]:
    """Parse `[skill_event]audio.completed|app_id|task_id` into a dict."""
    if not text:
        return None
    raw = str(text).strip()
    if not raw.startswith(SKILL_EVENT_PREFIX):
        return None
    payload = raw[len(SKILL_EVENT_PREFIX) :].strip()
    parts = [p.strip() for p in payload.split("|")]
    name = parts[0] if parts else ""
    if not name:
        return None
    return {
        "name": name,
        "app_id": parts[1] if len(parts) > 1 else "",
        "task_id": parts[2] if len(parts) > 2 else "",
        "duration_ms": parts[3] if len(parts) > 3 else "",
    }


def should_report_asr_now(conn: Any) -> bool:
    """PLAY_AUDIO / SHADOWING defer report so echo and ignored uplink are not archived."""
    mode = getattr(conn, "interaction_mode", "CHAT")
    return mode not in DEFER_ASR_REPORT_MODES


def stash_asr_audio(conn: Any, audio_snapshot: Any) -> None:
    conn.last_asr_audio_snapshot = audio_snapshot


@dataclass
class TaskSession:
    task_session_id: str
    ws_session_id: str = ""
    device_id: str = ""
    agent_id: str = ""
    app_id: str = ""
    task_id: str = ""
    source: str = "voice"
    interaction_mode: str = ""
    started_at: int = 0
    ended_at: int = 0
    status: str = "running"
    end_reason: str = ""
    turns: List[Dict[str, Any]] = field(default_factory=list)
    events: List[Dict[str, Any]] = field(default_factory=list)

    def find_turn(self, index: int) -> Optional[Dict[str, Any]]:
        for turn in self.turns:
            if turn.get("index") == index:
                return turn
        return None

    def upsert_turn(
        self,
        index: int,
        prompt_text: str = "",
        asr_text: Optional[str] = None,
        echo_rejected: bool = False,
        expected_text: Optional[str] = None,
        metrics: Optional[List[Dict[str, Any]]] = None,
        assessed_at: Optional[int] = None,
        audio_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Same step_index stays one row; last ASR wins. Empty metrics do not wipe old."""
        existing = self.find_turn(index)
        if existing is None:
            turn: Dict[str, Any] = {
                "index": index,
                "prompt_text": prompt_text or "",
                "asr_text": asr_text or "",
                "echo_rejected": bool(echo_rejected),
                "recorded_at": _now_ms(),
            }
            if expected_text:
                turn["expected_text"] = expected_text
            if metrics:
                turn["metrics"] = metrics
            if assessed_at is not None:
                turn["assessed_at"] = assessed_at
            if audio_id:
                turn["audio_id"] = audio_id
            self.turns.append(turn)
            return turn
        if prompt_text:
            existing["prompt_text"] = prompt_text
        if asr_text is not None:
            existing["asr_text"] = asr_text
            existing["echo_rejected"] = bool(echo_rejected)
            existing["recorded_at"] = _now_ms()
        if expected_text:
            existing["expected_text"] = expected_text
        if metrics:
            existing["metrics"] = metrics
        if assessed_at is not None:
            existing["assessed_at"] = assessed_at
        if audio_id:
            existing["audio_id"] = audio_id
        return existing

    def add_turn(
        self,
        index: int,
        prompt_text: str,
        asr_text: str,
        echo_rejected: bool = False,
    ) -> None:
        self.upsert_turn(
            index=index,
            prompt_text=prompt_text,
            asr_text=asr_text,
            echo_rejected=echo_rejected,
        )

    def add_event(self, name: str, app_id: str = "", task_id: str = "") -> None:
        self.events.append(
            {
                "name": name,
                "app_id": app_id or self.app_id,
                "task_id": task_id or self.task_id,
                "at": _now_ms(),
            }
        )

    def finish(self, reason: str) -> None:
        if self.status != "running":
            return
        self.status = "completed" if reason in ("finished", "play_completed") else "cancelled"
        self.end_reason = reason
        self.ended_at = _now_ms()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

    def save(self) -> None:
        try:
            directory = session_dir()
            os.makedirs(directory, exist_ok=True)
            path = os.path.join(directory, f"{self.task_session_id}.json")
            with open(path, "w", encoding="utf-8") as handle:
                json.dump(self.to_dict(), handle, ensure_ascii=False, indent=2)
        except Exception as exc:
            logger.bind(tag=TAG).warning(f"保存 TaskSession 失败: {exc}")
            return
        if getattr(self, "_skip_upsert", False):
            return
        try:
            _schedule_upsert(self.to_dict(), self)
        except Exception as exc:
            logger.bind(tag=TAG).warning(f"TaskInstance 上报调度失败: {exc}")

    @classmethod
    def load(cls, task_session_id: str) -> Optional["TaskSession"]:
        path = os.path.join(session_dir(), f"{task_session_id}.json")
        if not os.path.isfile(path):
            return None
        with open(path, "r", encoding="utf-8") as handle:
            data = json.load(handle)
        return cls(
            task_session_id=data.get("task_session_id") or task_session_id,
            ws_session_id=data.get("ws_session_id") or "",
            device_id=data.get("device_id") or "",
            agent_id=data.get("agent_id") or "",
            app_id=data.get("app_id") or "",
            task_id=data.get("task_id") or "",
            source=data.get("source") or "voice",
            interaction_mode=data.get("interaction_mode") or "",
            started_at=int(data.get("started_at") or 0),
            ended_at=int(data.get("ended_at") or 0),
            status=data.get("status") or "running",
            end_reason=data.get("end_reason") or "",
            turns=list(data.get("turns") or []),
            events=list(data.get("events") or []),
        )


def begin_session(conn: "ConnectionHandler", app_id: str, task_id: str, source: str, mode: str) -> TaskSession:
    existing = getattr(conn, "task_session", None)
    if existing and getattr(existing, "status", "") == "running":
        end_session(conn, "preempted")
    session = TaskSession(
        task_session_id=uuid.uuid4().hex,
        ws_session_id=str(getattr(conn, "session_id", "") or ""),
        device_id=str(getattr(conn, "device_id", "") or ""),
        agent_id=str(getattr(conn, "agent_id", "") or ""),
        app_id=app_id,
        task_id=task_id,
        source=source or "voice",
        interaction_mode=mode or "",
        started_at=_now_ms(),
        status="running",
    )
    conn.task_session = session
    session.save()
    logger.bind(tag=TAG).info(
        f"TaskSession 开始 {session.task_session_id} {app_id}/{task_id} mode={mode}"
    )
    return session


def end_session(conn: "ConnectionHandler", reason: str) -> None:
    session = getattr(conn, "task_session", None)
    if not session or getattr(session, "status", "") != "running":
        return
    session.finish(reason)
    session.save()
    logger.bind(tag=TAG).info(
        f"TaskSession 结束 {session.task_session_id} status={session.status} reason={reason}"
    )
    _schedule_close_parent_matching(conn, session, reason)


def current_session(conn: "ConnectionHandler") -> Optional[TaskSession]:
    session = getattr(conn, "task_session", None)
    if session and getattr(session, "status", "") == "running":
        return session
    return None


def schedule_close_parent_except(
    conn: "ConnectionHandler",
    reason: str,
    except_app: str = "",
    except_task: str = "",
) -> None:
    device_id = str(getattr(conn, "device_id", "") or "")
    if not device_id:
        return
    payload: Dict[str, Any] = {
        "device_id": device_id,
        "reason": reason or "server",
        "except_app_id": except_app or "",
        "except_task_id": except_task or "",
    }
    _schedule_close_parent(payload)


def _schedule_close_parent_matching(
    conn: "ConnectionHandler",
    session: TaskSession,
    reason: str,
) -> None:
    device_id = str(getattr(conn, "device_id", "") or getattr(session, "device_id", "") or "")
    if not device_id:
        return
    status = "done" if getattr(session, "status", "") == "completed" else "cancelled"
    payload: Dict[str, Any] = {
        "device_id": device_id,
        "reason": reason or "",
        "status": status,
        "match_app_id": session.app_id or "",
        "match_task_id": session.task_id or "",
    }
    _schedule_close_parent(payload)


def _schedule_close_parent(payload: Dict[str, Any]) -> None:
    async def _run():
        try:
            from config.manage_api_client import close_parent_immediate_tasks

            await close_parent_immediate_tasks(payload)
        except Exception as exc:
            logger.bind(tag=TAG).warning(f"关闭家长即时任务失败: {exc}")

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return
    loop.create_task(_run())


async def _upsert_task_instance(payload: Dict[str, Any]) -> bool:
    from config.manage_api_client import upsert_task_instance

    data = await upsert_task_instance(payload)
    return data is not None


def _schedule_upsert(payload: Dict[str, Any], session: Optional["TaskSession"] = None) -> None:
    async def _run():
        try:
            ok = await _upsert_task_instance(payload)
            if not ok and session is not None:
                session._skip_upsert = True
                logger.bind(tag=TAG).warning("TaskInstance 上报失败，本会话不再重复上报")
        except Exception as exc:
            if session is not None:
                session._skip_upsert = True
            logger.bind(tag=TAG).warning(f"TaskInstance 上报失败: {exc}")

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return
    loop.create_task(_run())