from __future__ import annotations

import json
import re
import subprocess
import sys
import threading
import time
from collections.abc import Callable
from pathlib import Path
from typing import Any

from xhs_bridge.config import AppConfig
from xhs_bridge.exceptions import SkillsError
from xhs_bridge.human_behavior import human_behavior_json_path
from xhs_bridge.subprocess_win import popen_kwargs_hide_console
from xhs_bridge.tools_catalog import XHS_TOOLS
from xhs_bridge.skills_tool_adapter import build_cli_argv, format_tool_result_text

ProgressFn = Callable[[str], None] | None


class SkillsClient:
    """通过子进程调用内嵌 xiaohongshu-skills CLI。"""

    _CANCELLABLE_COMMANDS = frozenset({
        "list-feeds",
        "search-feeds",
        "my-account-feeds",
    })

    def __init__(self, config: AppConfig) -> None:
        self.config = config
        self._active_proc: subprocess.Popen[str] | None = None
        self._cancel_requested = False
        self._browse_targets: dict[str, float | int] | None = None

    def _cli_script(self) -> Path:
        return self.config.skills.cli_script()

    def _browse_cancel_file(self) -> Path:
        return self.config.state_dir / "browse.cancel"

    def _browse_partial_file(self) -> Path:
        return self.config.state_dir / "browse.partial.json"

    def begin_browse_task(
        self, *, browse_targets: dict[str, float | int] | None = None
    ) -> None:
        """新一次首页/搜索采集开始前调用，清除上一轮取消状态。"""
        self._browse_targets = browse_targets
        self._cancel_requested = False
        self.config.state_dir.mkdir(parents=True, exist_ok=True)
        cancel = self._browse_cancel_file()
        if cancel.exists():
            cancel.unlink()
        partial = self._browse_partial_file()
        partial.write_text(
            json.dumps({"feeds": []}, ensure_ascii=False),
            encoding="utf-8",
        )

    def is_browse_cancel_requested(self) -> bool:
        return self._cancel_requested

    def browse_cancelled_payload(self) -> dict[str, Any]:
        """用户停止或子进程被中断时返回的最小结果。"""
        partial = self._load_partial_payload()
        if partial:
            return partial
        return {"feeds": [], "count": 0, "cancelled": True}

    def _prepare_browse_session(self, env: dict[str, str]) -> None:
        self.config.state_dir.mkdir(parents=True, exist_ok=True)
        cancel = self._browse_cancel_file()
        partial = self._browse_partial_file()
        if not self._cancel_requested and cancel.exists():
            cancel.unlink()
        if not partial.exists():
            partial.write_text(
                json.dumps({"feeds": []}, ensure_ascii=False),
                encoding="utf-8",
            )
        env["XHS_CANCEL_FILE"] = str(cancel)
        env["XHS_PARTIAL_FEEDS_FILE"] = str(partial)
        env["XHS_STATE_DIR"] = str(self.config.state_dir)
        targets = getattr(self, "_browse_targets", None)
        if targets:
            env["XHS_BROWSE_TARGETS"] = json.dumps(targets, ensure_ascii=False)

    def _load_partial_payload(self) -> dict[str, Any] | None:
        partial = self._browse_partial_file()
        if not partial.is_file():
            return None
        try:
            data = json.loads(partial.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            return None
        if not isinstance(data, dict):
            return None
        feeds = data.get("feeds")
        if not isinstance(feeds, list) or not feeds:
            return None
        return {"feeds": feeds, "count": len(feeds), "cancelled": True}

    def request_cancel_browse(self) -> None:
        """停止当前 list-feeds / search-feeds 子进程并保留已落盘数据。"""
        self._cancel_requested = True
        cancel = self._browse_cancel_file()
        cancel.parent.mkdir(parents=True, exist_ok=True)
        cancel.write_text("1", encoding="utf-8")
        proc = self._active_proc
        if proc is not None and proc.poll() is None:
            proc.kill()

    def _run_cli(
        self,
        argv: list[str],
        *,
        timeout: int = 300,
        on_progress: ProgressFn = None,
    ) -> tuple[dict[str, Any], int]:
        env = dict(**__import__("os").environ)
        if not getattr(sys, "frozen", False):
            cli = self._cli_script()
            if not cli.is_file():
                raise SkillsError(f"CLI 脚本不存在: {cli}")
            env["PYTHONPATH"] = str(self.config.skills.scripts_dir())
        env["PYTHONUTF8"] = "1"
        hb_json = human_behavior_json_path(self.config.state_dir)
        if hb_json.is_file():
            env["XHS_HUMAN_BEHAVIOR_FILE"] = str(hb_json)
        cancellable = bool(argv) and argv[0] in self._CANCELLABLE_COMMANDS
        if cancellable:
            self._prepare_browse_session(env)
            if self._cancel_requested:
                return self.browse_cancelled_payload(), 0

        cmd = self.config.skills.cli_launcher(argv)
        if cmd and cmd[0].endswith(".py") and not Path(cmd[0]).is_file():
            raise SkillsError(f"CLI 脚本不存在: {cmd[0]}")
        if cmd and cmd[0].endswith(".exe") and not Path(cmd[0]).is_file():
            raise SkillsError(f"CLI 引擎不存在: {cmd[0]}")
        try:
            proc = subprocess.Popen(
                cmd,
                cwd=str(self.config.skills.project_dir),
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                encoding="utf-8",
                errors="replace",
                env=env,
                **popen_kwargs_hide_console(),
            )
        except OSError as exc:
            raise SkillsError(f"无法启动 CLI: {exc}") from exc

        stderr_chunks: list[str] = []
        stdout_chunks: list[str] = []

        def _read_stderr() -> None:
            stream = proc.stderr
            if stream is None:
                return
            try:
                for line in stream:
                    stderr_chunks.append(line)
                    stripped = line.strip()
                    if stripped.startswith("@@PROGRESS@@") and on_progress:
                        on_progress(stripped[len("@@PROGRESS@@") :])
            except (ValueError, OSError):
                # communicate / wait 结束后管道可能已关闭
                pass

        def _read_stdout() -> None:
            stream = proc.stdout
            if stream is None:
                return
            try:
                stdout_chunks.append(stream.read())
            except (ValueError, OSError):
                pass

        stderr_reader = threading.Thread(target=_read_stderr, daemon=True)
        stdout_reader = threading.Thread(target=_read_stdout, daemon=True)
        stderr_reader.start()
        stdout_reader.start()
        self._active_proc = proc
        try:
            deadline = time.monotonic() + timeout
            while proc.poll() is None:
                if self._cancel_requested:
                    proc.kill()
                    break
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    proc.kill()
                    proc.wait(timeout=5)
                    raise SkillsError(f"CLI 执行超时（{timeout}s）: {' '.join(argv)}")
                try:
                    proc.wait(timeout=min(0.5, remaining))
                except subprocess.TimeoutExpired:
                    continue
        finally:
            self._active_proc = None
            stderr_reader.join(timeout=2)
            stdout_reader.join(timeout=2)

        stdout = stdout_chunks[0] if stdout_chunks else ""
        stderr = "".join(stderr_chunks)

        data = self._parse_json_output(stdout or "")
        if data is None and cancellable and (self._cancel_requested or proc.returncode != 0):
            partial = self._load_partial_payload()
            if partial:
                return partial, 0
            if self._cancel_requested:
                return self.browse_cancelled_payload(), 0
        if data is None and proc.returncode != 0:
            err = stderr.strip() or (stdout or "").strip() or f"退出码 {proc.returncode}"
            raise SkillsError(err)
        if data is None:
            data = {"success": proc.returncode == 0, "raw": (stdout or "").strip()}
        return data, proc.returncode

    @staticmethod
    def _parse_json_output(stdout: str) -> dict[str, Any] | None:
        text = stdout.strip()
        if not text:
            return None
        for candidate in (text, text.splitlines()[-1] if text else ""):
            candidate = candidate.strip()
            if not candidate.startswith("{"):
                continue
            try:
                parsed = json.loads(candidate)
                if isinstance(parsed, dict):
                    return parsed
            except json.JSONDecodeError:
                continue
        match = re.search(r"\{[\s\S]*\}\s*$", text)
        if match:
            try:
                parsed = json.loads(match.group(0))
                if isinstance(parsed, dict):
                    return parsed
            except json.JSONDecodeError:
                pass
        return None

    def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        timeout: int = 300,
        on_progress: ProgressFn = None,
    ) -> str:
        argv = build_cli_argv(name, arguments)
        data, _code = self._run_cli(argv, timeout=timeout, on_progress=on_progress)
        return format_tool_result_text(name, data)

    def call_tool_json(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        timeout: int = 300,
        on_progress: ProgressFn = None,
    ) -> dict[str, Any]:
        argv = build_cli_argv(name, arguments)
        data, _code = self._run_cli(argv, timeout=timeout, on_progress=on_progress)
        return data

    def list_tools(self) -> list[dict[str, Any]]:
        return [
            {
                "name": t["name"],
                "description": t.get("description", ""),
                "inputSchema": {"type": "object", "properties": {}},
            }
            for t in XHS_TOOLS
        ]

    @staticmethod
    def is_success_text(text: str) -> bool:
        lowered = text.lower()
        if "❌" in text or "失败" in text or "错误" in text:
            return False
        if "✅" in text or "成功" in text or "已登录" in text:
            return True
        return "未登录" not in text

    def check_login_status(self) -> str:
        return self.call_tool("check_login_status", timeout=90)

    def list_feeds(self, *, on_progress: ProgressFn = None) -> dict[str, Any]:
        return self.call_tool_json("list_feeds", timeout=300, on_progress=on_progress)

    def list_my_account_feeds(self, *, on_progress: ProgressFn = None) -> dict[str, Any]:
        return self.call_tool_json(
            "my_account_feeds", timeout=360, on_progress=on_progress
        )

    def search_feeds(
        self,
        keyword: str,
        filters: dict[str, Any] | None = None,
        *,
        on_progress: ProgressFn = None,
    ) -> dict[str, Any]:
        args: dict[str, Any] = {"keyword": keyword}
        if filters:
            args["filters"] = filters
        return self.call_tool_json("search_feeds", args, timeout=360, on_progress=on_progress)

    def feed_detail(
        self,
        payload: dict[str, Any],
        *,
        on_progress: ProgressFn = None,
    ) -> dict[str, Any]:
        return self.call_tool_json(
            "get_feed_detail",
            payload,
            timeout=360,
            on_progress=on_progress,
        )

    def login_status(self) -> dict[str, Any]:
        data = self.call_tool_json("check_login_status", timeout=90)
        logged_in = bool(data.get("logged_in"))
        return {
            "is_logged_in": logged_in,
            "logged_in": logged_in,
            "message": "已登录" if logged_in else (data.get("hint") or data.get("message") or "未登录"),
        }

    def browser_login(self, timeout: int = 180) -> dict[str, Any]:
        """在 Chrome 中打开小红书登录页并等待扫码（阻塞至完成或超时）。"""
        data, _code = self._run_cli(["login"], timeout=timeout)
        return data

    def delete_cookies(self) -> dict[str, Any]:
        return self.call_tool_json("delete_cookies", timeout=90)

    def close(self) -> None:
        if self._active_proc is not None and self._active_proc.poll() is None:
            self._active_proc.kill()
        self._active_proc = None