137 lines
5.0 KiB
Python
137 lines
5.0 KiB
Python
"""站点构建执行器：子进程隔离 + 每站点并发锁 + 任务注册表。
|
|
|
|
构建失败/超时只影响本次任务状态，绝不影响 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
|