"""
Web 控制台进程后端：优先 PM2（Linux/Windows 通用），不可用时回退为本地 PID 注册表。
"""
from __future__ import annotations

import asyncio
import json
import os
import re
import signal
import subprocess
import sys
import time
from pathlib import Path
from typing import Any, Optional

BACKEND_STATE_VERSION = 1


def strip_ansi_codes(text: str) -> str:
    return re.sub(r"\x1b\[[0-9;]*m", "", text)


async def _shell_out(cmd: str, timeout: float = 15.0) -> tuple[int, str]:
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        return 124, "ERROR: 命令超时"
    out = (stdout or b"").decode(errors="ignore").strip()
    err = (stderr or b"").decode(errors="ignore").strip()
    text = out if out else err
    return proc.returncode or 0, text if text else "命令执行完成（无输出）"


async def pm2_available() -> bool:
    code, _ = await _shell_out("pm2 -v", timeout=5.0)
    return code == 0


async def run_pm2(cmd: str, timeout: float = 15.0) -> str:
    _, text = await _shell_out(f"pm2 {cmd}", timeout=timeout)
    return text


async def force_kill_ffmpeg(process_name: str) -> None:
    """尽量结束与本业务相关的 ffmpeg（Linux 精准 + 兜底；Windows 与 Linux 兜底策略一致：结束 ffmpeg 进程）。"""
    if sys.platform == "win32":
        _ = process_name  # 保留参数供将来按命令行筛选；当前与 Linux killall 行为对齐
        proc2 = await asyncio.create_subprocess_shell(
            "taskkill /IM ffmpeg.exe /F /T 2>nul || exit /b 0",
            stdout=asyncio.subprocess.DEVNULL,
            stderr=asyncio.subprocess.DEVNULL,
        )
        await proc2.wait()
        return

    for cmd in (
        f"pkill -f 'ffmpeg.*{process_name}' || true",
        "killall -9 ffmpeg 2>/dev/null || true",
    ):
        proc = await asyncio.create_subprocess_shell(
            cmd,
            stdout=asyncio.subprocess.DEVNULL,
            stderr=asyncio.subprocess.DEVNULL,
        )
        await proc.wait()
        await asyncio.sleep(0.4)


class LocalProcessRegistry:
    """无 PM2 时：用 JSON 记录 pid，日志写入 .pm2/logs。"""

    def __init__(self, base_dir: Path, log_dir: Path):
        self.base_dir = base_dir.resolve()
        self.log_dir = log_dir
        self.state_path = self.base_dir / ".process_runner" / "state.json"
        self._lock = asyncio.Lock()

    def _read_state(self) -> dict[str, Any]:
        if not self.state_path.exists():
            return {"version": BACKEND_STATE_VERSION, "processes": {}}
        try:
            data = json.loads(self.state_path.read_text(encoding="utf-8"))
            if not isinstance(data.get("processes"), dict):
                data["processes"] = {}
            return data
        except Exception:
            return {"version": BACKEND_STATE_VERSION, "processes": {}}

    def _write_state(self, data: dict[str, Any]) -> None:
        self.state_path.parent.mkdir(parents=True, exist_ok=True)
        payload = json.dumps(data, indent=2, ensure_ascii=False)
        tmp = self.state_path.with_suffix(".json.tmp")
        tmp.write_text(payload, encoding="utf-8")
        os.replace(str(tmp), str(self.state_path))

    def _pid_alive(self, pid: int) -> bool:
        if pid <= 0:
            return False
        if sys.platform == "win32":
            try:
                out = subprocess.check_output(
                    ["tasklist", "/FI", f"PID eq {pid}", "/FO", "CSV", "/NH"],
                    stderr=subprocess.DEVNULL,
                    text=True,
                    encoding="utf-8",
                    errors="ignore",
                )
                return str(pid) in out
            except Exception:
                return False
        try:
            os.kill(pid, 0)
            return True
        except OSError:
            return False

    def _kill_tree(self, pid: int) -> None:
        if pid <= 0:
            return
        if sys.platform == "win32":
            subprocess.run(
                ["taskkill", "/PID", str(pid), "/T", "/F"],
                capture_output=True,
                text=True,
                timeout=30,
            )
            return
        try:
            os.killpg(os.getpgid(pid), signal.SIGTERM)
        except (ProcessLookupError, PermissionError, OSError):
            try:
                os.kill(pid, signal.SIGTERM)
            except ProcessLookupError:
                pass

    def build_command(self, script: str) -> Optional[list[str]]:
        path = self.base_dir / script
        if not path.is_file():
            return None
        if script.lower().endswith(".py"):
            return [sys.executable, str(path)]
        if script.lower().endswith(".sh"):
            bash = shutil_which("bash") or shutil_which("sh")
            if not bash:
                return None
            return [bash, str(path)]
        if script.lower().endswith((".bat", ".cmd")):
            return ["cmd.exe", "/c", str(path)]
        return None

    async def start(self, name: str, script: str) -> str:
        async with self._lock:
            data = self._read_state()
            procs = data["processes"]
            if name in procs:
                old_pid = int(procs[name].get("pid") or 0)
                if self._pid_alive(old_pid):
                    return f"进程 {name} 已在运行 (PID {old_pid})"

            cmd = self.build_command(script)
            if not cmd:
                return f"ERROR: 无法启动 {script}（文件不存在或缺少 bash/cmd）"

            self.log_dir.mkdir(parents=True, exist_ok=True)
            out_log = open(self.log_dir / f"{name}-out.log", "a", encoding="utf-8", errors="ignore")
            err_log = open(self.log_dir / f"{name}-error.log", "a", encoding="utf-8", errors="ignore")

            preexec_fn = os.setsid if sys.platform != "win32" else None
            creationflags = 0
            if sys.platform == "win32":
                creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)

            try:
                kw: dict[str, Any] = {
                    "cwd": str(self.base_dir),
                    "stdout": out_log,
                    "stderr": err_log,
                }
                if preexec_fn is not None:
                    kw["preexec_fn"] = preexec_fn
                if creationflags:
                    kw["creationflags"] = creationflags
                proc = await asyncio.create_subprocess_exec(*cmd, **kw)
            except Exception as e:
                out_log.close()
                err_log.close()
                return f"ERROR: 启动失败: {e}"

            out_log.close()
            err_log.close()

            procs[name] = {
                "pid": proc.pid,
                "script": script,
                "started_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
            }
            self._write_state(data)
            return f"已本地启动 {name} (PID {proc.pid})，命令: {' '.join(cmd)}"

    async def stop(self, name: str) -> str:
        async with self._lock:
            data = self._read_state()
            procs = data["processes"]
            if name not in procs:
                return f"进程 {name} 未在本地注册表中"
            pid = int(procs[name].get("pid") or 0)
            if self._pid_alive(pid):
                self._kill_tree(pid)
                await asyncio.sleep(0.5)
            procs.pop(name, None)
            self._write_state(data)
            return f"已停止 {name}（本地后端）"

    async def restart(self, name: str, script: str) -> str:
        await self.stop(name)
        await asyncio.sleep(0.3)
        return await self.start(name, script)

    def describe_status(self, name: str) -> dict[str, Any]:
        data = self._read_state()
        procs = data.get("processes") or {}
        out_path = str(self.log_dir / f"{name}-out.log")
        err_path = str(self.log_dir / f"{name}-error.log")
        if name not in procs:
            return {
                "process_status": "not_found",
                "out_path": out_path,
                "err_path": err_path,
            }
        pid = int(procs[name].get("pid") or 0)
        if self._pid_alive(pid):
            status = "online"
        else:
            status = "stopped"
        return {
            "process_status": status,
            "out_path": out_path,
            "err_path": err_path,
        }

    def status_table_text(self) -> str:
        data = self._read_state()
        lines = ["┌─────────────┬──────────┬──────────────────────────┐", "│ name        │ status   │ pid                      │", "├─────────────┼──────────┼──────────────────────────┤"]
        for name, info in sorted((data.get("processes") or {}).items()):
            pid = int(info.get("pid") or 0)
            st = "online" if self._pid_alive(pid) else "stopped"
            lines.append(f"│ {name:<11} │ {st:<8} │ {pid:<24} │")
        lines.append("└─────────────┴──────────┴──────────────────────────┘")
        lines.append("(本地进程后端，非 PM2)")
        return "\n".join(lines)


def shutil_which(cmd: str) -> Optional[str]:
    from shutil import which

    return which(cmd)


class WebProcessBackend:
    def __init__(self, base_dir: Path, default_log_dir: Path):
        self.base_dir = base_dir
        self.default_log_dir = default_log_dir
        self._local = LocalProcessRegistry(base_dir, default_log_dir)
        self._use_pm2: Optional[bool] = None

    async def refresh_mode(self) -> None:
        self._use_pm2 = await pm2_available()

    @property
    def mode(self) -> str:
        if self._use_pm2 is None:
            return "unknown"
        return "pm2" if self._use_pm2 else "local"

    async def ensure_mode(self) -> None:
        if self._use_pm2 is None:
            await self.refresh_mode()

    async def start(self, process: str, script: str) -> str:
        await self.ensure_mode()
        if self._use_pm2:
            return await run_pm2(f"start {process}")
        return await self._local.start(process, script)

    async def stop(self, process: str) -> str:
        await self.ensure_mode()
        if self._use_pm2:
            return await run_pm2(f"stop {process}")
        return await self._local.stop(process)

    async def restart(self, process: str, script: str) -> str:
        await self.ensure_mode()
        if self._use_pm2:
            return await run_pm2(f"restart {process}")
        return await self._local.restart(process, script)

    async def describe_raw(self, process: str) -> str:
        await self.ensure_mode()
        if self._use_pm2:
            return await run_pm2(f"describe {process}", timeout=10.0)
        info = self._local.describe_status(process)
        if info["process_status"] == "not_found":
            return "No such process"
        return f"status │ {info['process_status']}\nout_log_path │ {info['out_path']}\nerr_log_path │ {info['err_path']}"

    async def full_status_raw(self) -> str:
        await self.ensure_mode()
        if self._use_pm2:
            return await run_pm2("status", timeout=8.0)
        return self._local.status_table_text()

    async def parse_pm2_describe(self, process: str) -> dict[str, Any]:
        await self.ensure_mode()
        if not self._use_pm2:
            d = self._local.describe_status(process)
            return {
                "process_status": d["process_status"],
                "out_path": d["out_path"],
                "err_path": d["err_path"],
            }

        describe_raw = await run_pm2(f"describe {process}", timeout=10.0)
        describe_clean = strip_ansi_codes(describe_raw)

        if any(
            x in describe_clean.lower()
            for x in ["no such process", "unknown process", "not found"]
        ) or not describe_clean.strip():
            return {
                "process_status": "not_found",
                "out_path": None,
                "err_path": None,
            }

        status = "unknown"
        out_path = None
        err_path = None

        for line in describe_raw.splitlines():
            line_clean = strip_ansi_codes(line)
            if line_clean.count("│") < 2:
                continue
            parts = [p.strip() for p in line_clean.split("│") if p.strip()]
            if len(parts) >= 2:
                key = parts[0].lower()
                value = parts[1]
                if key == "status":
                    status = value.lower()
                elif key == "out_log_path":
                    out_path = value
                elif key == "err_log_path":
                    err_path = value

        if not out_path:
            out_path = str(self.default_log_dir / f"{process}-out.log")
        if not err_path:
            err_path = str(self.default_log_dir / f"{process}-error.log")

        return {
            "process_status": status,
            "out_path": out_path,
            "err_path": err_path,
        }
