4add7fa9d9
issue #2：--all-pending 長時間本地 Ollama 攝入全程靜默，使用者不確定有沒有在 跑、又不敢中斷。三處補強： - 進度可見：每份來源印 [i/N] 起始行、長文件逐分段印「分段 k/n」、每個 knowledge-item 印「↳ 新增/更新」、完成印「✓」，全部 flush=True 即時顯示。 最會靜默的分段迴圈正是重點回報處。 - Ctrl-C 安全：KeyboardInterrupt 不是 Exception 子類，原本會略過善後、把使用者 留在 ingest 分支且無指引。改為獨立攔截——commit 在迴圈之後，故 main 必然未受 影響——印出可照做的回復指令（git checkout main && git stash -u && git branch -D <branch>），以結束碼 130 收場。 selfcheck_ingest 新增：驗進度標記出現；驗中斷後 main HEAD 未動、停在 ingest 分支、且照文件回復指令實測能回到乾淨 main（驗結果可用，非字串存在，§13.4）。 README §7 補一列。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
327 lines
14 KiB
Python
327 lines
14 KiB
Python
"""攝入管線（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()
|