# -*- coding: utf-8 -*-
"""Launch the official DeepSeek Harness Web UI (`npx @deepseek-ai/dsh web`)."""

from __future__ import annotations

import os
import re
import shutil
import socket
import subprocess
import threading
import time
import urllib.error
import urllib.request
from pathlib import Path

from runtime_paths import application_data_dir

READY_RE = re.compile(r"dsh web:\s*(https?://127\.0\.0\.1:\d+)", re.I)
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 18765
START_TIMEOUT_SECONDS = 180.0
PACKAGE = "@deepseek-ai/dsh"


def parse_ready_url(text: str) -> str:
    match = READY_RE.search(str(text or ""))
    return match.group(1).rstrip("/") if match else ""


def workspace_dir() -> Path:
    path = application_data_dir() / "dsh_desk" / "workspace"
    path.mkdir(parents=True, exist_ok=True)
    return path


def _which(name: str) -> str:
    found = shutil.which(name)
    if found:
        return found
    if os.name == "nt":
        return shutil.which(f"{name}.cmd") or shutil.which(f"{name}.exe") or ""
    return ""


def launch_command(port: int = DEFAULT_PORT) -> list[str]:
    npx = _which("npx")
    if not npx:
        raise FileNotFoundError(
            "未找到 Node.js / npx。请安装 Node 22.19+ 或 24+ 后再启动官方 DeepSeek Harness。"
        )
    args = [
        npx,
        "--yes",
        PACKAGE,
        "web",
        "--host",
        DEFAULT_HOST,
        "--port",
        str(int(port)),
    ]
    if os.name == "nt" and npx.lower().endswith((".cmd", ".bat")):
        comspec = os.environ.get("COMSPEC") or "cmd.exe"
        args = [comspec, "/d", "/s", "/c"] + args
    return args


def port_open(port: int, host: str = DEFAULT_HOST) -> bool:
    try:
        with socket.create_connection((host, int(port)), timeout=0.4):
            return True
    except OSError:
        return False


def http_ready(url: str, timeout: float = 1.5) -> bool:
    try:
        with urllib.request.urlopen(url, timeout=timeout) as response:
            return int(getattr(response, "status", 200) or 200) < 500
    except (urllib.error.URLError, TimeoutError, OSError):
        return False


class DshWebServer:
    """One official `dsh web` process, reused for the embedded AI 客服 page."""

    def __init__(self, port: int = DEFAULT_PORT) -> None:
        self.port = int(port)
        self.url = f"http://{DEFAULT_HOST}:{self.port}"
        self._proc: subprocess.Popen[str] | None = None
        self._owned = False
        self._lock = threading.Lock()
        self._log: list[str] = []

    def start(self, timeout: float = START_TIMEOUT_SECONDS) -> str:
        with self._lock:
            if self._proc is not None and self._proc.poll() is None:
                return self.url
            if http_ready(self.url):
                self._owned = False
                return self.url
            args = launch_command(self.port)
            env = os.environ.copy()
            try:
                import ai_config

                key = str(getattr(ai_config, "AI_API_KEY", "") or "").strip()
                base = str(getattr(ai_config, "AI_API_BASE", "") or "").strip()
                if key:
                    env["DEEPSEEK_API_KEY"] = key
                if base:
                    env["DEEPSEEK_BASE_URL"] = base
            except Exception:
                pass
            cwd = str(workspace_dir())
            creationflags = 0
            startupinfo = None
            if os.name == "nt":
                creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
                startupinfo = subprocess.STARTUPINFO()
                startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
            self._proc = subprocess.Popen(
                args,
                cwd=cwd,
                env=env,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                encoding="utf-8",
                errors="replace",
                bufsize=1,
                creationflags=creationflags,
                startupinfo=startupinfo,
            )
            self._owned = True
            threading.Thread(
                target=self._pump_output,
                name="dsh-web-stdout",
                daemon=True,
            ).start()
        deadline = time.monotonic() + max(15.0, float(timeout))
        while time.monotonic() < deadline:
            proc = self._proc
            if proc is not None and proc.poll() is not None:
                raise RuntimeError(self._fail_message(f"进程已退出，代码 {proc.returncode}"))
            snapshot = "\n".join(self._log[-40:])
            parsed = parse_ready_url(snapshot)
            if parsed:
                self.url = parsed
            if http_ready(self.url):
                return self.url
            time.sleep(0.4)
        raise RuntimeError(self._fail_message("启动超时"))

    def stop(self) -> None:
        with self._lock:
            proc = self._proc
            self._proc = None
            owned = self._owned
            self._owned = False
        if not owned or proc is None:
            return
        if proc.poll() is not None:
            return
        if os.name == "nt":
            subprocess.run(
                ["taskkill", "/F", "/T", "/PID", str(proc.pid)],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            )
            return
        proc.terminate()
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()

    def _pump_output(self) -> None:
        proc = self._proc
        if proc is None or proc.stdout is None:
            return
        for line in proc.stdout:
            text = line.rstrip()
            if text:
                self._log.append(text)
                if len(self._log) > 400:
                    del self._log[:200]

    def _fail_message(self, reason: str) -> str:
        tail = "\n".join(self._log[-12:])
        detail = f"{reason}。{tail}" if tail else reason
        return f"官方 DeepSeek Harness Web UI 未能启动：{detail}"


_server: DshWebServer | None = None
_server_lock = threading.Lock()


def get_server() -> DshWebServer:
    global _server
    with _server_lock:
        if _server is None:
            _server = DshWebServer()
        return _server


def stop_server() -> None:
    global _server
    with _server_lock:
        server = _server
        _server = None
    if server is not None:
        server.stop()