164 lines
5.8 KiB
Python
164 lines
5.8 KiB
Python
"""站点发现、能力探测与状态读取（纯函数，无副作用）。"""
|
|
|
|
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"
|