"""站点构建执行器：子进程隔离 + 每站点并发锁 + 任务注册表。

构建失败/超时只影响本次任务状态，绝不影响 gateway 主链路。
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import subprocess
import sys
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from deerflow.config import get_app_config
from deerflow.webapp.engine import resolve_template_dir

logger = logging.getLogger(__name__)

MAX_OUTPUT_CHARS = 64_000


@dataclass
class BuildJob:
    job_id: str
    thread_id: str
    site_name: str
    status: str  # running | success | failed
    started_at: float = field(default_factory=time.time)
    finished_at: float | None = None
    result: dict[str, Any] = field(default_factory=dict)
    error: str = ""


class SiteBuildRunner:
    """每个 (thread_id, site_name) 同时只允许一个构建任务。"""

    def __init__(self) -> None:
        self._jobs: dict[tuple[str, str], BuildJob] = {}
        self._locks: dict[tuple[str, str], asyncio.Lock] = {}

    def get_job(self, thread_id: str, site_name: str) -> BuildJob | None:
        return self._jobs.get((thread_id, site_name))

    async def start_build(self, thread_id: str, site_name: str, site_dir: Path) -> BuildJob:
        """启动构建；同站点已有运行中任务时直接返回该任务（幂等）。"""
        key = (thread_id, site_name)
        lock = self._locks.setdefault(key, asyncio.Lock())
        if lock.locked():
            existing = self._jobs.get(key)
            if existing is not None and existing.status == "running":
                return existing
        await lock.acquire()
        job = BuildJob(
            job_id=uuid.uuid4().hex[:12],
            thread_id=thread_id,
            site_name=site_name,
            status="running",
        )
        self._jobs[key] = job
        asyncio.create_task(self._run_build(job, site_dir, lock))
        return job

    async def _run_build(self, job: BuildJob, site_dir: Path, lock: asyncio.Lock) -> None:
        try:
            result = await asyncio.to_thread(self._run_build_sync, job, site_dir)
            job.result = result
            job.status = "success" if result.get("success") else "failed"
            if not result.get("success"):
                job.error = str(result.get("error") or result.get("stderr_tail") or "构建失败")
        except Exception as exc:  # noqa: BLE001 - 构建异常全部收进任务状态
            logger.warning("web-app build job %s crashed: %s", job.job_id, exc)
            job.status = "failed"
            job.error = f"构建执行异常: {exc}"
            job.result = {"success": False, "error": job.error}
        finally:
            job.finished_at = time.time()
            lock.release()

    def _run_build_sync(self, _job: BuildJob, site_dir: Path) -> dict[str, Any]:
        config = get_app_config()
        template_dir = resolve_template_dir()
        script = template_dir / "scripts" / "build_site.py"
        if not script.is_file():
            return {"success": False, "error": f"构建脚本不存在: {script}"}
        env = dict(os.environ)
        env["DEERFLOW_WEBAPP_TEMPLATE"] = str(template_dir)
        env["DEERFLOW_WEBAPP_TIMEOUT"] = str(config.web_app.build_timeout_seconds)
        node_override = config.web_app.node_path.strip()
        if node_override:
            env["DEERFLOW_WEBAPP_NODE"] = node_override
        timeout = config.web_app.build_timeout_seconds + 30  # 脚本自身超时 + 通信余量
        try:
            proc = subprocess.run(
                [sys.executable, str(script), "build", str(site_dir)],
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=timeout,
                env=env,
                cwd=str(site_dir.parent),
            )
        except subprocess.TimeoutExpired:
            return {"success": False, "error": f"构建超时（>{timeout}s）已终止"}
        payload: dict[str, Any] = {"success": False, "error": f"构建脚本退出码 {proc.returncode}"}
        # stdout 是一行 JSON（取最后一个非空行，容忍脚本前置告警输出）
        for line in reversed((proc.stdout or "").strip().splitlines()):
            line = line.strip()
            if line.startswith("{"):
                try:
                    parsed = json.loads(line)
                    if isinstance(parsed, dict):
                        payload = parsed
                        break
                except ValueError:
                    continue
        payload.setdefault("success", False)
        payload["stdout_tail"] = (proc.stdout or "")[-MAX_OUTPUT_CHARS:]
        payload["stderr_tail"] = (proc.stderr or "")[-MAX_OUTPUT_CHARS:]
        return payload


_runner: SiteBuildRunner | None = None


def get_site_build_runner() -> SiteBuildRunner:
    global _runner
    if _runner is None:
        _runner = SiteBuildRunner()
    return _runner