Jack
f1ce583fd1
同步家纺情报 v2 数据、59 个严格权威源、趋势图库、移动端交互、Google/Yandex 翻译门控及静态 SEO/GEO 构建。 补充面向 C# 研发的 README，明确当前无 LLM、多模态与生图模型，记录翻译请求参数、缓存与失败策略、数据处理边界、前端职责和无 GitHub Pages 部署流程。 新增按日期倒序维护的 CHANGELOG，记录影响、验证与代码/数据/dist 回滚方法。 验证：11 个 Python 契约测试、13 个 Node 前端测试通过；静态构建生成 1096 个 Sitemap URL。
135 lines
4.9 KiB
Python
135 lines
4.9 KiB
Python
#!/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())
|