Create-knowledge-find-decode/scripts/inspect_multi.py

"""读回多平台跑批结果：确认平台/阶段/卡片类型/段卡时间区间/知识点溯源。"""
from __future__ import annotations

import json
import sys

from creation_knowledge.config import PgConfig
from creation_knowledge.integrations.db import CkStore


def main() -> int:
    store = CkStore(PgConfig.from_env(".env"))
    ids = sys.argv[1:] or [
        "xhs_67e4bdf50000000006028a59",
        "dy_7612631899479648866",
        "ks_3xepqgwddgbickc",
    ]
    for pid in ids:
        p = store.read_post(pid)
        if not p:
            print(f"{pid}: NOT FOUND")
            continue
        cards = p.get("cards") or []
        kinds: dict = {}
        for c in cards:
            kinds[c.get("kind")] = kinds.get(c.get("kind"), 0) + 1
        seg = [c for c in cards if c.get("kind") == "segment"]
        seg_times = ["{}-{}".format(c.get("start"), c.get("end")) for c in seg[:4]]
        items = store.read_items(pid)
        srcs = []
        for it in items[:4]:
            d = it["item"] if isinstance(it["item"], dict) else json.loads(it["item"])
            srcs.append({"title": (d.get("title") or "")[:18],
                         "source_cards": d.get("source_cards")})
        print(f"\n{pid}")
        print(f"  platform={p.get('platform')} stage={p.get('stage')} "
              f"url={(p.get('url') or '')[:50]}")
        print(f"  cards={kinds} seg_times={seg_times}")
        print(f"  items={len(items)}")
        for s in srcs:
            print(f"    - {s['title']!r} source_cards={s['source_cards']}")
    return 0


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