"""站点发现、能力探测与状态读取(纯函数,无副作用)。""" from __future__ import annotations import json import mimetypes import re import shutil from dataclasses import dataclass, field from pathlib import Path from typing import Any from deerflow.config import get_app_config from deerflow.config.runtime_paths import resolve_path SITE_MANIFEST_NAME = "site.json" BUILD_RESULT_NAME = ".build-result.json" # 站点目录名约束(与 build_site.py 的 SITE_NAME_RE 一致) SITE_NAME_RE_OK = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") # dist/ 静态托管的 MIME 修正表 _EXTRA_MIME_TYPES = { ".js": "text/javascript; charset=utf-8", ".mjs": "text/javascript; charset=utf-8", ".css": "text/css; charset=utf-8", ".json": "application/json; charset=utf-8", ".geojson": "application/json; charset=utf-8", ".wasm": "application/wasm", ".html": "text/html; charset=utf-8", ".htm": "text/html; charset=utf-8", ".svg": "image/svg+xml", ".woff": "font/woff", ".woff2": "font/woff2", } @dataclass class WebAppCapability: """能力探测结果:任何一项缺失都只是让功能优雅不可用。""" available: bool = False enabled: bool = False checks: dict[str, dict[str, Any]] = field(default_factory=dict) template_dir: str = "" node_bin: str = "" def resolve_template_dir() -> Path: """解析站点模板目录(相对 project root 或绝对路径)。""" config = get_app_config() return resolve_path(config.web_app.template_dir) def capability_status() -> WebAppCapability: """检查配置开关、模板、预置依赖与 node 二进制。永不抛异常。""" checks: dict[str, dict[str, Any]] = {} try: config = get_app_config() enabled = bool(config.web_app.enabled) checks["enabled"] = {"ok": enabled, "detail": "config.yaml web_app.enabled"} template_dir = resolve_template_dir() template_ok = template_dir.is_dir() and (template_dir / SITE_MANIFEST_NAME).is_file() checks["template"] = {"ok": template_ok, "detail": str(template_dir)} vite_bin = template_dir / "node_modules" / "vite" / "bin" / "vite.js" deps_ok = vite_bin.is_file() checks["node_modules"] = {"ok": deps_ok, "detail": str(vite_bin)} node_path = config.web_app.node_path.strip() node_bin = node_path if node_path and Path(node_path).exists() else (shutil.which("node") or "") checks["node"] = {"ok": bool(node_bin), "detail": node_bin} available = all(item["ok"] for item in checks.values()) return WebAppCapability( available=available, enabled=enabled, checks=checks, template_dir=str(template_dir), node_bin=node_bin, ) except Exception as exc: # noqa: BLE001 - 能力探测永不抛出 checks["internal"] = {"ok": False, "detail": str(exc)} return WebAppCapability(available=False, enabled=False, checks=checks) def read_site_manifest(site_dir: Path) -> dict[str, Any]: """读取 site.json(损坏时返回空 dict,由调用方决定如何呈现)。""" manifest_path = site_dir / SITE_MANIFEST_NAME try: data = json.loads(manifest_path.read_text(encoding="utf-8")) return data if isinstance(data, dict) else {} except (OSError, ValueError): return {} def _dist_info(site_dir: Path) -> dict[str, Any]: dist = site_dir / "dist" if not dist.is_dir(): return {"exists": False} try: entries = [p for p in dist.rglob("*") if p.is_file()] latest = max((p.stat().st_mtime for p in entries), default=None) return { "exists": True, "file_count": len(entries), "total_bytes": sum(p.stat().st_size for p in entries), "mtime": latest, } except OSError: return {"exists": True, "file_count": 0, "total_bytes": 0, "mtime": None} def _last_build(site_dir: Path) -> dict[str, Any] | None: result_path = site_dir / BUILD_RESULT_NAME try: data = json.loads(result_path.read_text(encoding="utf-8")) return data if isinstance(data, dict) else None except (OSError, ValueError): return None def find_sites(outputs_dir: Path) -> list[dict[str, Any]]: """扫描 outputs/ 一级目录,凡带 site.json 的即站点。永不抛异常。""" sites: list[dict[str, Any]] = [] try: candidates = sorted(outputs_dir.iterdir(), key=lambda p: (p.name.lower())) except OSError: return sites for entry in candidates: if not entry.is_dir() or not SITE_NAME_RE_OK.match(entry.name): continue if not (entry / SITE_MANIFEST_NAME).is_file(): continue manifest = read_site_manifest(entry) try: mtime = (entry / SITE_MANIFEST_NAME).stat().st_mtime except OSError: mtime = None sites.append( { "name": entry.name, "title": str(manifest.get("title") or entry.name), "description": str(manifest.get("description") or ""), "nav": manifest.get("nav") if isinstance(manifest.get("nav"), list) else [], "map": manifest.get("map") if isinstance(manifest.get("map"), dict) else {}, "manifest_mtime": mtime, "dist": _dist_info(entry), "last_build": _last_build(entry), } ) sites.sort(key=lambda s: (-(s["manifest_mtime"] or 0), str(s["name"]))) return sites def guess_site_mime(path: Path) -> str: """dist/ 静态文件 MIME(带编码修正;未知扩展名回退 octet-stream)。""" suffix = path.suffix.lower() if suffix in _EXTRA_MIME_TYPES: return _EXTRA_MIME_TYPES[suffix] guessed, _ = mimetypes.guess_type(path.name) return guessed or "application/octet-stream"