Sansheng-Intelligence/scripts/update_all.py

#!/usr/bin/env python3
"""抓取、校验、构建成功后再替换正式数据和站点。"""

from __future__ import annotations

import argparse
import gzip
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
DATA_FILES = (
    "textile-intelligence.json",
    "textile-trends.json",
    "textile-intelligence-source-status.json",
    "textile-zh-cache.json",
)
MAX_SNAPSHOTS = 14
MAX_SNAPSHOT_BYTES = 120 * 1024 * 1024


def run(*parts: str) -> None:
    subprocess.run(parts, cwd=ROOT, check=True)


def digest(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest() if path.exists() else ""


def validate(data_dir: Path) -> dict:
    payload = json.loads((data_dir / "textile-intelligence.json").read_text(encoding="utf-8"))
    if payload.get("source_target") != payload.get("source_success"):
        raise RuntimeError("权威源未全部通过，拒绝发布")
    for key in ("calendar", "exhibitions", "policies", "platforms", "activities"):
        if not payload.get(key):
            raise RuntimeError(f"核心栏目为空，拒绝发布：{key}")
    json.loads((data_dir / "textile-trends.json").read_text(encoding="utf-8"))
    return payload


def snapshot_current(data_dir: Path, snapshot_dir: Path) -> None:
    if not any((data_dir / name).exists() for name in DATA_FILES):
        return
    snapshot_dir.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
    target = snapshot_dir / f"{stamp}.json.gz"
    content = {
        name: json.loads((data_dir / name).read_text(encoding="utf-8"))
        for name in DATA_FILES
        if (data_dir / name).exists()
    }
    with gzip.open(target, "wt", encoding="utf-8") as handle:
        json.dump(content, handle, ensure_ascii=False, separators=(",", ":"))


def prune_snapshots(snapshot_dir: Path) -> None:
    files = sorted(snapshot_dir.glob("*.json.gz"), key=lambda path: path.stat().st_mtime, reverse=True)
    total = 0
    for index, path in enumerate(files):
        size = path.stat().st_size
        total += size
        if index >= MAX_SNAPSHOTS or total > MAX_SNAPSHOT_BYTES:
            path.unlink()


def publish_data(staged: Path, target: Path) -> list[str]:
    target.mkdir(parents=True, exist_ok=True)
    changed = [name for name in DATA_FILES if (staged / name).exists() and digest(staged / name) != digest(target / name)]
    if changed:
        snapshot_dir = ROOT / "runtime" / "snapshots"
        snapshot_current(target, snapshot_dir)
        for name in changed:
            temp = target / f".{name}.tmp"
            shutil.copy2(staged / name, temp)
            os.replace(temp, target / name)
        prune_snapshots(snapshot_dir)
    return changed


def main() -> int:
    parser = argparse.ArgumentParser(description="更新三笙AI情报局全部数据并构建站点")
    parser.add_argument("--site-url", default=os.environ.get("SANSHENG_SITE_URL", "http://127.0.0.1:8765"))
    parser.add_argument("--now", default="", help="验收日期 YYYY-MM-DD；生产运行留空")
    args = parser.parse_args()
    data_dir = ROOT / "data"
    runtime = ROOT / "runtime"
    runtime.mkdir(exist_ok=True)
    try:
        with tempfile.TemporaryDirectory(prefix="update-", dir=runtime) as temp_name:
            staged = Path(temp_name) / "data"
            staged.mkdir()
            for name in DATA_FILES:
                if (data_dir / name).exists():
                    shutil.copy2(data_dir / name, staged / name)
            run(sys.executable, "scripts/update_textile_trends.py", "--output-dir", str(staged))
            intelligence_command = [
                sys.executable, "scripts/update_textile_intelligence.py",
                "--output-dir", str(staged), "--strict",
            ]
            if args.now:
                intelligence_command.extend(("--now", args.now))
            run(*intelligence_command)
            validate(staged)
            preview = Path(temp_name) / "dist"
            run(
                sys.executable, "scripts/build_site.py",
                "--data-dir", str(staged),
                "--output-dir", str(preview),
                "--site-url", args.site_url.rstrip("/"),
            )
            changed = publish_data(staged, data_dir)
            final_stage = ROOT / "runtime" / ".dist-next"
            if final_stage.exists():
                shutil.rmtree(final_stage)
            shutil.move(str(preview), final_stage)
            from build_site import replace_directory
            replace_directory(final_stage, ROOT / "dist")
        print(f"[更新完成] 数据变更 {len(changed)} 个文件；站点已重新构建")
        return 0
    except Exception as exc:
        print(f"[更新失败] 正式数据未被不完整抓取覆盖：{exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())