Create-knowledge-find-decode/creation_knowledge/cli.py

"""命令行入口：python -m creation_knowledge.cli run --urls <...>"""
from __future__ import annotations

import argparse
import json

from creation_knowledge.pipeline import run_pipeline


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="creation_knowledge")
    sub = parser.add_subparsers(dest="command", required=True)
    run = sub.add_parser("run", help="跑流水线：拉取->提取->筛选->拆分->解构->组装[->入库]")
    run.add_argument("--urls", nargs="+", required=True, help="帖子链接或 content_id")
    run.add_argument("--env-file", default=".env", help="环境变量文件路径")
    run.add_argument("--ingest", action="store_true",
                     help="开启真实入库（默认关闭，只组装+存库）")
    run.add_argument("--run-id", default=None,
                     help="运行批次 id（媒体落盘目录 data/<run_id>/...，默认时间戳）")
    return parser


def main() -> int:
    args = build_parser().parse_args()
    if args.command == "run":
        results = run_pipeline(
            args.urls,
            env_file=args.env_file,
            ingest_enabled=True if args.ingest else False,
            run_id=args.run_id,
        )
        for r in results:
            print(json.dumps(r, ensure_ascii=False))
        ok = sum(1 for r in results if r["status"] == "done")
        rej = sum(1 for r in results if r["status"] == "rejected")
        skip = sum(1 for r in results if r["status"] == "skipped")
        fail = len(results) - ok - rej - skip
        print(f"\n汇总：done={ok} rejected={rej} skipped={skip} failed={fail} "
              f"/ 共 {len(results)}")
    return 0


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