scripts/local_ingest.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
local_ingest.py
將本地 markdown / txt 檔案（或整個目錄）轉成 XKB 知識卡片，
加入 search_index.json，供後續 sync_cards_to_wiki 使用。
這是 quickstart 和 demo mode 的共用 ingest 底層。
Usage:
python3 scripts/local_ingest.py path/to/notes/ # 整個目錄
python3 scripts/local_ingest.py path/to/file.md # 單一檔案
python3 scripts/local_ingest.py notes/ --dry-run # 預覽不寫入
python3 scripts/local_ingest.py notes/ --tag personal --category learning
python3 scripts/local_ingest.py notes/ --limit 20 # 最多 20 個檔案
Exit codes:
0 — success, 0 new cards
1 — error
2 — success, >= 1 new cards added
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from gbrain_publish import PublicationBatch
import xkb_frontmatter
from _card_prompt import gbrain_put as _gbrain_put
from _card_prompt import (
build_prompt, condense_long_content, extract_summary, find_related_context,
llm_call,
)
from xkb_frontmatter import parse as extract_frontmatter
from category_classifier import classify_content, apply_category
import xkb_paths
# ── Paths ──────────────────────────────────────────────────────────────────────
CARDS_DIR = xkb_paths.CARDS_DIR
INDEX_FILE = xkb_paths.INDEX_FILE
SUPPORTED_EXTS = {".md", ".txt", ".markdown"}
def load_index() -> dict:
if INDEX_FILE.exists():
return json.loads(INDEX_FILE.read_text(encoding="utf-8"))
return {"version": "1.1", "items": []}
def file_hash(path: Path) -> str:
"""Short hash to detect duplicate ingest of same file."""
content = path.read_bytes()
return hashlib.md5(content).hexdigest()[:8]
def card_id_for_file(path: Path) -> str:
stem = re.sub(r"[^a-zA-Z0-9\-]", "-", path.stem.lower())
stem = re.sub(r"-+", "-", stem).strip("-")[:40]
h = file_hash(path)
return f"local-{stem}-{h}"
def pmc_url_from_filename(filename: str) -> str:
"""Extract PMC ID from filename and return NCBI URL."""
m = re.match(r"(PMC\d+)", filename, re.IGNORECASE)
if m:
return f"https://www.ncbi.nlm.nih.gov/pmc/articles/{m.group(1)}/"
return ""
def collect_files(input_path: Path) -> list[Path]:
if input_path.is_file():
if input_path.suffix.lower() in SUPPORTED_EXTS:
return [input_path]
print(f"[WARN] 不支援的檔案類型：{input_path.suffix}", file=sys.stderr)
return []
if input_path.is_dir():
files = []
for ext in SUPPORTED_EXTS:
files.extend(input_path.rglob(f"*{ext}"))
return sorted(files)
print(f"[ERROR] 路徑不存在：{input_path}", file=sys.stderr)
return []
def process_file(
path: Path,
api_key: str,
existing_keys: set,
existing_items: list[dict],
force_category: str | None,
extra_tags: list[str],
dry_run: bool,
publication=None,
) -> dict | None:
card_id = card_id_for_file(path)
dedup_key = f"local|{card_id}"
saved = CARDS_DIR / f"{card_id}.md"
if saved.exists() and not dry_run:
if publication is not None:
publication.existing(saved, card_id)
return None
_gbrain_put(saved, card_id)
text = saved.read_text(encoding="utf-8")
return {"title": extract_frontmatter(text).get("title", path.stem),
"tags": [], "summary": extract_summary(text)}
if dedup_key in existing_keys:
print(f" [SKIP] 已存在：{path.name}")
return None
content = path.read_text(encoding="utf-8", errors="ignore").strip()
if not content:
print(f" [SKIP] 空檔案：{path.name}")
return None
print(f" {path.name} ({len(content)} characters)")
if dry_run:
return None
truncated = condense_long_content(content, verbose=True)
source_url = pmc_url_from_filename(path.name)
category = force_category or classify_content(
content, source_type="local-paper", current_category="research"
)["category"]
related_ctx = find_related_context(content, existing_items)
prompt = build_prompt(
content=f"Filename: {path.name}\n\n{truncated}",
card_id=card_id,
source_type="local-paper",
source_url=source_url or str(path),
category=category,
related_context=related_ctx,
)
try:
card_content = llm_call(prompt, api_key)
except Exception as e:
print(f" ❌ LLM 失敗：{e}")
return None
# Ensure id in frontmatter
if "---\n" in card_content and "id:" not in card_content:
card_content = card_content.replace("---\n", f"---\nid: {card_id}\n", 1)
card_content = apply_category(card_content, category)
if extra_tags:
# Preserve the original YAML elements, including quoted commas. Only
# encode the new values; splitting/re-encoding would change existing tags.
tags_raw = (xkb_frontmatter.get(card_content, "tags") or "").strip()
if tags_raw.startswith("[") and tags_raw.endswith("]"):
tags_raw = tags_raw[1:-1].strip()
added = ", ".join(json.dumps(tag, ensure_ascii=False) for tag in dict.fromkeys(extra_tags))
separator = ", " if tags_raw else ""
card_content = xkb_frontmatter.set_field(
card_content, "tags", f"[{tags_raw}{separator}{added}]"
)
# Save card
CARDS_DIR.mkdir(parents=True, exist_ok=True)
card_path = CARDS_DIR / f"{card_id}.md"
card_path.write_text(card_content, encoding="utf-8")
if publication is None:
_gbrain_put(card_path, card_id)
else:
publication.publish(card_path, card_id, generated=True)
print(f" 💾 cards/{card_id}.md")
# Context for subsequent cards; the search index is derived from the files.
fm = extract_frontmatter(card_content)
return {
"title": fm.get("title", path.stem),
"tags": [t.strip().strip("\"'") for t in fm.get("tags", "").strip("[]").split(",") if t.strip()],
"summary": extract_summary(card_content),
}
def main() -> int:
parser = argparse.ArgumentParser(
description="Ingest local markdown/txt files into XKB knowledge cards")
parser.add_argument("path", help="檔案或目錄路徑")
parser.add_argument("--dry-run", action="store_true", help="只列出檔案，不生成卡片")
parser.add_argument("--limit", type=int, default=0, help="最多處理 N 個檔案（0=全部）")
parser.add_argument("--category", help="覆蓋 LLM 分類（強制指定）")
parser.add_argument("--tag", dest="tags", action="append", default=[],
help="額外標籤（可多次使用）")
args = parser.parse_args()
api_key = "" # auth handled by _llm.py via openclaw CLI
input_path = Path(args.path)
files = collect_files(input_path)
if not files:
print("沒有找到可匯入的檔案。")
return 0
print(f"📂 找到 {len(files)} 個檔案")
index_data = load_index()
existing_items = index_data.get("items", [])
existing_keys = {
f"local|{item.get('relative_path','').split('/')[-1].replace('.md','')}"
for item in existing_items
if item.get("source_type") in ("local", "local-paper")
}
batch = PublicationBatch(_gbrain_put)
if not args.dry_run and not batch.ready():
return 1
attempted = 0
new_items: list[dict] = []
for path in files:
if batch.blocked:
break
saved = CARDS_DIR / f"{card_id_for_file(path)}.md"
if not args.dry_run and batch.existing(saved, card_id_for_file(path)):
continue
if args.limit and attempted >= args.limit:
continue
if f"local|{card_id_for_file(path)}" in existing_keys:
continue
attempted += 1
result = process_file(
path, api_key, existing_keys, existing_items,
args.category, args.tags, args.dry_run, batch
)
if result:
new_items.append(result)
existing_items.append(result)
existing_keys.add(f"local|{card_id_for_file(path)}")
if not args.dry_run and not batch.finish():
return 1
if new_items and not args.dry_run:
print(f"\n✅ 完成：新增 {len(new_items)} 張知識卡片")
print("💡 下一步：python3 scripts/sync_cards_to_wiki.py --apply --limit 20")
elif args.dry_run:
print(f"\n（dry-run 模式，未寫入任何卡片）")
else:
print(f"\n✅ 完成：無新增卡片（全部已存在）")
return 2 if new_items and not args.dry_run else 0
import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py
if __name__ == "__main__":
xkb_usage.record(__file__)
raise SystemExit(main())