"""攝入管線（AGENTS.md §6）：讀 raw/converted 的 Markdown，呼叫本地 Ollama 產出/更新 wiki 頁，重建 index.md、追加 log.md，並以 git branch 準備 PR。 用法： python tools/ingest.py raw/converted/xxx.md [...] # 攝入指定來源 python tools/ingest.py --all-pending # 攝入所有尚無 summary 的來源 共用選項：[--force] [--dry-run] [--root DIR] 注意：--all-pending 以「目前分支」的 wiki 為準；尚未合併的 ingest PR 不會被看到。 攝入腳本永遠不直接 commit 到 main（AGENTS.md §1.5）。 """ import argparse import datetime import json import pathlib import subprocess import sys sys.path.insert(0, str(pathlib.Path(__file__).parent)) import kb SUMMARY_PROMPT = """你是金融電子支付 QA 部門知識庫的攝入引擎。閱讀以下來源文件，只輸出一個 JSON 物件，不得有任何其他文字。 規則： - 文字內容用繁體中文；slug 用英文小寫 kebab-case。 - description 為一句話摘要；tags 為 2 至 6 個英文小寫 kebab-case 標籤。 - summary_markdown：文件重點摘要（目的、關鍵事實、結論、教訓）。不要逐步驟複製測試案例內容——單檔可 grep 的細節不進 wiki。 - knowledge_items：值得建頁或更新的項目。entity＝系統/模組/API/法規條目/專案；concept＝跨來源的缺陷模式/測試策略/法遵準則。沒有就給空陣列。 - facts_markdown：本文件中關於該項目的事實要點（markdown bullet），供合併至該項目頁面。 JSON 格式： {{"title": "...", "description": "...", "slug": "...", "tags": ["..."], "summary_markdown": "...", "knowledge_items": [{{"type": "entity 或 concept", "slug": "...", "title": "...", "description": "...", "tags": ["..."], "facts_markdown": "..."}}]}} 檔名：{filename} 文件內容： <<< {content} >>>""" CHUNK_PROMPT = """以下是一份長文件的第 {i}/{n} 段。請以繁體中文摘要此段重點（markdown bullet），只輸出 JSON：{{"summary_markdown": "..."}} <<< {content} >>>""" MERGE_PROMPT = """你是知識庫維護引擎。以下是既有 wiki 頁面內文，以及來自新來源的事實要點。 請把新事實整合進頁面（就地編輯：合併、去重；若與既有內容矛盾，兩種說法並陳並標註「⚠ 待查證」）。 只輸出 JSON：{{"updated_markdown": "...", "updated_description": "..."}} 規則：繁體中文；保留原有結構與仍然有效的內容。 頁面標題：{title} 既有內文： <<< {body} >>> 新事實（來源 {source_ref}）： <<< {facts} >>>""" def _s(o, k): return isinstance(o.get(k), str) and o[k].strip() def _tags_ok(o): return isinstance(o.get("tags"), list) and o["tags"] and all(isinstance(t, str) for t in o["tags"]) def validate_summary(obj): for k in ("title", "description", "slug", "summary_markdown"): if not _s(obj, k): return f"欄位 {k} 缺漏或非字串" if not _tags_ok(obj): return "tags 須為非空字串列表" if not isinstance(obj.get("knowledge_items"), list): return "knowledge_items 須為列表" for it in obj["knowledge_items"]: if not isinstance(it, dict) or it.get("type") not in ("entity", "concept"): return "knowledge_item 須為物件且 type 為 entity|concept" for k in ("slug", "title", "description", "facts_markdown"): if not _s(it, k): return f"knowledge_item 欄位 {k} 缺漏或非字串" if not _tags_ok(it): return "knowledge_item.tags 須為非空字串列表" return None def validate_chunk(obj): return None if _s(obj, "summary_markdown") else "欄位 summary_markdown 缺漏" def validate_merge(obj): for k in ("updated_markdown", "updated_description"): if not _s(obj, k): return f"欄位 {k} 缺漏或非字串" return None def split_chunks(text, limit): """依段落邊界切塊，每塊不超過 limit 字元（單一超長段落則硬切）。""" chunks, cur = [], "" for para in text.split("\n\n"): while len(para) > limit: # 單段超長：硬切 chunks.append(para[:limit]) para = para[limit:] if len(cur) + len(para) + 2 > limit and cur: chunks.append(cur) cur = para else: cur = f"{cur}\n\n{para}" if cur else para if cur: chunks.append(cur) return chunks def summarize(cfg, filename, text): limit = cfg["limits"]["ingest_chunk_max_chars"] if len(text) > limit: parts = split_chunks(text, limit) partials = [] for i, part in enumerate(parts, 1): # 長文件的分段迴圈是最容易「長時間靜默」的地方（issue #2）：逐段回報進度 print(f" 分段 {i}/{len(parts)} …", flush=True) obj, _ = kb.llm_json(cfg, "ingest", CHUNK_PROMPT.format(i=i, n=len(parts), content=part), validate_chunk) partials.append(f"### 分段 {i} 摘要\n{obj['summary_markdown']}") text = "（以下為長文件的分段摘要，請據此綜合）\n\n" + "\n\n".join(partials) return kb.llm_json(cfg, "ingest", SUMMARY_PROMPT.format(filename=filename, content=text), validate_summary) def _unique_path(d, slug, sha8): p = d / f"{slug}.md" return p if not p.exists() else d / f"{slug}-{sha8}.md" def write_summary_page(root, obj, source_ref, sha8, today): d = root / kb.PAGE_DIRS["summary"] p = _unique_path(d, kb.safe_slug(obj["slug"], f"doc-{sha8}"), sha8) meta = {"type": "summary", "title": obj["title"], "description": obj["description"], "tags": obj["tags"], "timestamp": today, "sources": [source_ref], "status": "draft"} p.write_text(kb.dump_page(meta, obj["summary_markdown"]), encoding="utf-8") return p def upsert_item(root, cfg, item, source_ref, today, changed): slug = kb.safe_slug(item["slug"], f"item-{source_ref[-8:]}") existing = None for t in ("entity", "concept"): # 既有頁優先（就地編輯），不管本次 LLM 判的型別 p = root / kb.PAGE_DIRS[t] / f"{slug}.md" if p.exists(): existing = p break if existing: meta, body = kb.parse_page(existing.read_text(encoding="utf-8")) obj, _ = kb.llm_json(cfg, "ingest", MERGE_PROMPT.format(title=meta["title"], body=body, source_ref=source_ref, facts=item["facts_markdown"]), validate_merge) meta["description"] = obj["updated_description"] meta["tags"] = sorted(set(meta["tags"]) | set(item["tags"])) meta["timestamp"] = today meta["status"] = "draft" # 內容變動須重新人審 if source_ref not in meta["sources"]: meta["sources"].append(source_ref) existing.write_text(kb.dump_page(meta, obj["updated_markdown"]), encoding="utf-8") msg = f"更新 {existing.relative_to(root).as_posix()}" else: p = root / kb.PAGE_DIRS[item["type"]] / f"{slug}.md" meta = {"type": item["type"], "title": item["title"], "description": item["description"], "tags": item["tags"], "timestamp": today, "sources": [source_ref], "status": "draft"} p.write_text(kb.dump_page(meta, item["facts_markdown"]), encoding="utf-8") msg = f"新增 {p.relative_to(root).as_posix()}" changed.append(msg) print(f" ↳ {msg}", flush=True) # ---------- git ---------- def run_git(root, *args, capture=False): return subprocess.run(["git", *args], cwd=root, check=True, text=True, encoding="utf-8", capture_output=capture) def git_preflight(root): branch = run_git(root, "branch", "--show-current", capture=True).stdout.strip() if branch != "main": raise SystemExit(f"須在 main 分支執行（目前：{branch}）") # git 預設 core.quotepath=true，含非 ASCII（中文檔名）或空格的路徑會被引號包住並轉義， # 故比對前先剝除開頭引號，否則 convert 產出的中文檔名會被誤判為 raw/ 以外的變更。 dirty = [l for l in run_git(root, "status", "--porcelain", capture=True) .stdout.splitlines() if l.strip() and not l[3:].lstrip('"').startswith("raw/")] if dirty: raise SystemExit("工作區有 raw/ 以外的未提交變更，請先處理：\n" + "\n".join(dirty)) def referenced_sha8s(root): refs = set() for p in kb.iter_pages(root): meta, _ = kb.parse_page(p.read_text(encoding="utf-8")) for s in meta.get("sources", []): refs.add(str(s).rsplit("#", 1)[-1]) return refs def main(argv=None): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("paths", nargs="*", help="raw/converted 下的 Markdown 路徑") ap.add_argument("--all-pending", action="store_true") ap.add_argument("--force", action="store_true", help="已攝入過（hash 相同）也重跑") ap.add_argument("--dry-run", action="store_true", help="只列計畫，不呼叫 LLM、不寫檔") ap.add_argument("--root", default=str(kb.ROOT)) a = ap.parse_args(argv) root = pathlib.Path(a.root).resolve() cfg = kb.load_config(root) manifest = json.loads((root / "raw" / "manifest.json").read_text(encoding="utf-8")) conv = {k: v for k, v in manifest["files"].items() if v["kind"] == "converted"} if a.paths: targets = [] for p in a.paths: rel = pathlib.Path(p) rel = (rel.relative_to(root) if rel.is_absolute() else rel).as_posix() if rel not in conv: raise SystemExit(f"{rel} 不在 manifest 的 converted 條目中，請先跑 convert.py") targets.append(rel) elif a.all_pending: done = referenced_sha8s(root) targets = [k for k, v in conv.items() if v["sha256"][:8] not in done] else: ap.error("請指定來源路徑或 --all-pending") if not a.force: done = referenced_sha8s(root) skipped = [t for t in targets if conv[t]["sha256"][:8] in done] targets = [t for t in targets if conv[t]["sha256"][:8] not in done] for t in skipped: print(f"skip: {t} 已有 summary 引用（--force 可重跑）", flush=True) if not targets: print("沒有待攝入的來源。") return today = datetime.date.today().isoformat() stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") branch = f"ingest/{stamp}-{kb.safe_slug(pathlib.Path(targets[0]).stem, 'batch')[:30]}" if a.dry_run: limit = cfg["limits"]["ingest_chunk_max_chars"] print(f"[dry-run] 分支：{branch}，模型：{cfg['tasks']['ingest']['model']}") for t in targets: n = len((root / t).read_text(encoding="utf-8")) print(f"[dry-run] {t}: {n} chars，{max(1, -(-n // limit))} 段") return git_preflight(root) run_git(root, "checkout", "-q", "-b", branch) ok, failed, changed = [], [], [] try: for i, t in enumerate(targets, 1): try: print(f"[{i}/{len(targets)}] 攝入 {t} …", flush=True) text = (root / t).read_text(encoding="utf-8") sha8 = conv[t]["sha256"][:8] source_ref = f"{t}#{sha8}" obj, model = summarize(cfg, pathlib.Path(conv[t]["original_path"]).name, text) p = write_summary_page(root, obj, source_ref, sha8, today) msg = f"新增 {p.relative_to(root).as_posix()}" changed.append(msg) print(f" ↳ {msg}", flush=True) for item in obj["knowledge_items"]: upsert_item(root, cfg, item, source_ref, today, changed) ok.append((t, model)) print(f" ✓ {t} 完成（model={model}）", flush=True) except Exception as e: failed.append((t, e)) print(f" ✗ {t}: {e}", file=sys.stderr, flush=True) if not ok: run_git(root, "checkout", "-q", "main") run_git(root, "branch", "-q", "-D", branch) raise SystemExit("全部來源攝入失敗，已還原至 main。") kb.rebuild_index(root) kb.append_log(root, "ingest", f"攝入 {len(ok)} 份來源（branch {branch}）", changed + [f"失敗：{t}（{e}）" for t, e in failed]) run_git(root, "add", "wiki", "index.md", "log.md", "raw") run_git(root, "commit", "-q", "-m", f"ingest: {len(ok)} 份來源\n\n" + "\n".join(f"- {c}" for c in changed)) try: run_git(root, "push", "-q", "-u", "origin", branch) remote = run_git(root, "remote", "get-url", "origin", capture=True).stdout.strip().removesuffix(".git") print(f"PR 建立網址：{remote}/compare/main...{branch}") except subprocess.CalledProcessError: print(f"push 失敗（無 remote 或離線）。分支 {branch} 保留於本地，請手動 push 後開 PR。") run_git(root, "checkout", "-q", "main") print(f"完成：{len(ok)} 成功、{len(failed)} 失敗。變更在分支 {branch}，經 PR 審核後合併。") except SystemExit: raise except KeyboardInterrupt: # KeyboardInterrupt 不是 Exception 子類，會略過下方善後——故獨立攔截。 # 中斷時尚未 commit（commit 在迴圈之後），main 必然未受影響；給一條可照做的回復路徑。 print(f"\n已中斷（Ctrl-C）。main 未受影響——本次半成品都在分支 {branch}、尚未 commit。\n" f"回到乾淨的 main：\n" f" git checkout main && git stash -u && git branch -D {branch}\n" f"（git stash -u 收起分支上未提交的半成品含未追蹤頁面；確定不要再 git stash drop）\n" f"續跑 python tools/ingest.py --all-pending 會自動從尚未攝入的來源接續。", file=sys.stderr, flush=True) raise SystemExit(130) except Exception: print(f"攝入中斷。目前在分支 {branch}，工作區可能有未提交變更，" "請人工檢查（不自動清除以免遺失資料）。", file=sys.stderr) raise if failed: raise SystemExit(1) if __name__ == "__main__": main()