| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162 |
- """Dump 5 棵作用域分类树（PG public.global_category）→ 本地 JSON。
- 产物（scope_trees/）：
- - trees.json 整个分类树，按 source_type 分组（含已退役节点，live 标记）
- - trees_index.json 仅 live 节点（scope-link 定位目标），与 trees_embeddings.npy 行对齐
- 须在能连 RDS 的环境跑（云端）。用法：python scripts/dump_trees.py [env_file]
- """
- from __future__ import annotations
- import json
- import sys
- from collections import Counter
- from pathlib import Path
- from core.config import PgConfig
- from core.db import fetch_all
- OUT = Path("scope_trees")
- def fetch_nodes(cfg: PgConfig) -> list[dict]:
- return fetch_all(
- cfg,
- "SELECT stable_id, name, source_type, path, level, parent_stable_id, "
- "(retired_at_execution_id IS NULL) AS live "
- "FROM global_category ORDER BY source_type, level, stable_id",
- schema="public",
- )
- def main(env_file: str = ".env") -> None:
- cfg = PgConfig.from_env(env_file)
- rows = fetch_nodes(cfg)
- OUT.mkdir(exist_ok=True)
- trees: dict[str, list[dict]] = {}
- for r in rows:
- trees.setdefault(r["source_type"], []).append({
- "stable_id": r["stable_id"], "name": r["name"], "path": r["path"],
- "level": r["level"], "parent_stable_id": r["parent_stable_id"],
- "live": r["live"],
- })
- (OUT / "trees.json").write_text(
- json.dumps(trees, ensure_ascii=False, indent=2), encoding="utf-8")
- live_rows = [r for r in rows if r["live"]]
- index = [{
- "row": i, "stable_id": r["stable_id"], "source_type": r["source_type"],
- "name": r["name"], "path": r["path"],
- } for i, r in enumerate(live_rows)]
- (OUT / "trees_index.json").write_text(
- json.dumps(index, ensure_ascii=False, indent=2), encoding="utf-8")
- total = Counter(r["source_type"] for r in rows)
- live = Counter(r["source_type"] for r in live_rows)
- print("trees.json (全部):", dict(total), "总计", sum(total.values()))
- print("trees_index (live):", dict(live), "总计", len(index))
- if __name__ == "__main__":
- main(sys.argv[1] if len(sys.argv) > 1 else ".env")
|
