scripts/sync_enriched_index.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
Sync enriched cards from memory/cards/ back into search_index.json.
For each enriched card found in memory/cards/:
- Update path to point to memory/cards/[id].md
- Keep relative_path stable and truly relative
- Extract and update summary from "## 1. 核心摘要" section
- Extract and update tags from frontmatter
- Extract and update title from "# Title" line
- Set enriched: true
- Only add orphan cards when they have valid, stable metadata
Usage:
python3 scripts/sync_enriched_index.py
python3 scripts/sync_enriched_index.py --dry-run
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
from pathlib import Path
from urllib.parse import urlparse
sys.path.insert(0, str(Path(__file__).resolve().parent))
import xkb_paths
import xkb_index
import xkb_frontmatter
WORKSPACE = xkb_paths.WORKSPACE
BOOKMARKS_DIR = xkb_paths.BOOKMARKS_DIR
CARDS_DIR = xkb_paths.CARDS_DIR
INDEX_FILE = xkb_paths.INDEX_FILE
def _extract_summary(text: str) -> str:
# ── New format: ## 7. 雙語摘要（搜尋索引用）with ZH:/EN: labels ──────────
bilingual = re.search(r"##\s+7\.\s*雙語摘要[^\n]*\n(.+?)(?=\n##|\Z)", text, re.DOTALL)
if bilingual:
block = bilingual.group(1)
zh_m = re.search(r"^(?:ZH|中文)[：:]\s*(.+)$", block, re.MULTILINE)
en_m = re.search(r"^(?:EN|英文)[：:]\s*(.+)$", block, re.MULTILINE)
parts = [m.group(1).strip() for m in [zh_m, en_m] if m and m.group(1).strip()]
if parts:
return " | ".join(parts)
# ── Legacy formats (old worker cards, github/local ingest cards) ──────────
patterns = [
r"##\s+1\.\s*核心摘要\s*\n(.+?)(?=\n##|\Z)",
r"##\s+1\.\s*English Summary\s*\n(.+?)(?=\n##|\Z)",
r"##\s*📝\s*English Summary\s*\n(.+?)(?=\n##|\Z)",
]
parts = []
for pat in patterns:
m = re.search(pat, text, re.DOTALL)
if m:
lines = [l.strip().lstrip("-").strip() for l in m.group(1).strip().splitlines() if l.strip()]
parts.append(" ".join(lines)[:300])
return " | ".join(parts) if parts else ""
def _extract_tags(text: str) -> list[str]:
m = re.search(r"^tags:\s*\[(.+?)\]", text, re.MULTILINE)
if m:
return [t.strip().strip('"').strip("'") for t in m.group(1).split(",") if t.strip()]
return []
def _extract_title(text: str) -> str:
m = re.search(r"^# (.+)$", text, re.MULTILINE)
return m.group(1).strip() if m else ""
def _extract_frontmatter_value(text: str, key: str) -> str:
m = re.search(r"^" + re.escape(key) + r":\s*[\"']?([^\"'\n]+)[\"']?\s*$", text, re.MULTILINE)
return m.group(1).strip() if m else ""
def _extract_status_id(value: str) -> str:
if not value:
return ""
m = re.search(r"/status/(\d{15,20})", value)
return m.group(1) if m else ""
def _looks_like_url(value: str) -> bool:
if not value:
return False
try:
parsed = urlparse(value)
return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
except Exception:
return False
def _placeholder_text(value: str) -> bool:
v = (value or "").strip().lower()
return v in {"", "待整理", "待補充", "tbd", "todo", "n/a", "untitled", "(untitled)"}
def _relative_from_workspace(path: Path) -> str:
return str(path.relative_to(WORKSPACE))
def _resolve_item_id(item: dict) -> str:
source_url = item.get("source_url", "")
source_id = _extract_status_id(source_url)
if source_id:
return source_id
for key in ("relative_path", "path"):
stem = Path(item.get(key, "")).stem
if re.fullmatch(r"\d{15,20}", stem):
return stem
m = re.match(r"^(\d{15,20})", stem)
if m:
return m.group(1)
# Non-numeric stems (github_star-xxx, github_fork-xxx, local cards) — use full stem as id
if stem:
return stem
return ""
def _resolve_card_id(text: str, stem: str) -> str:
# Non-numeric stems (github/local cards) — stem itself is the stable id
if stem and not re.match(r"\d", stem):
return stem
if re.fullmatch(r"\d{15,20}", stem):
return stem
m = re.match(r"^(\d{15,20})", stem)
if m:
return m.group(1)
for key in ("id", "tweet_id", "source_url"):
value = _extract_frontmatter_value(text, key)
resolved = _extract_status_id(value) if key == "source_url" else (value if re.fullmatch(r"\d{15,20}", value) else "")
if resolved:
return resolved
m = re.search(r"/status/(\d{15,20})", text)
if m:
return m.group(1)
return ""
def _is_valid_orphan_card(text: str, card_id: str, source_url: str, title: str, summary: str) -> bool:
_cleaned = re.sub(r"<think>[\s\S]*?</think>", "", text).strip()
if not _cleaned.startswith("---"):
return False
card_type = _extract_frontmatter_value(text, "type")
if card_type not in {"x-knowledge-card", "knowledge-card"}:
return False
source_type = _extract_frontmatter_value(text, "source_type")
if source_type not in {"x-bookmark", "youtube", "github_fork", "github_star", "local-paper", "pubmed"}:
return False
if not _looks_like_url(source_url):
return False
if source_type == "x-bookmark" and not _extract_status_id(source_url):
return False
if source_type in {"youtube", "github_fork", "github_star", "local-paper", "pubmed"} and not card_id:
return False
if _placeholder_text(title) and _placeholder_text(summary):
return False
return True
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()
if not CARDS_DIR.exists():
print(f"❌ Cards dir not found: {CARDS_DIR}")
sys.exit(1)
if not INDEX_FILE.exists():
print(f"❌ Index not found: {INDEX_FILE}")
sys.exit(1)
raw = json.loads(INDEX_FILE.read_text(encoding="utf-8"))
items = raw.get("items", raw) if isinstance(raw, dict) else raw
is_dict = isinstance(raw, dict)
id_to_idx: dict[str, int] = {}
for idx, item in enumerate(items):
item_id = _resolve_item_id(item)
if item_id:
id_to_idx[item_id] = idx
card_files = xkb_paths.card_files()
print(f"📁 {len(card_files)} enriched cards | {len(id_to_idx)} indexed items")
updated = 0
unexcluded = 0
not_found_ids: list[str] = []
for card_path in card_files:
raw_card_id = card_path.stem
text = card_path.read_text(encoding="utf-8", errors="ignore")
resolved_id = _resolve_card_id(text, raw_card_id)
if not resolved_id or resolved_id not in id_to_idx:
not_found_ids.append(raw_card_id)
continue
idx = id_to_idx[resolved_id]
item = items[idx]
summary = _extract_summary(text)
tags = _extract_tags(text)
title = _extract_title(text)
category = _extract_frontmatter_value(text, "category")
new_path = str(card_path)
new_relative_path = _relative_from_workspace(card_path)
changes: dict = {}
if item.get("path") != new_path:
changes["path"] = new_path
if item.get("relative_path") != new_relative_path:
changes["relative_path"] = new_relative_path
if summary and item.get("summary") != summary:
changes["summary"] = summary
if tags and item.get("tags") != tags:
changes["tags"] = tags
if title and item.get("title") != title:
changes["title"] = title
if category and item.get("category") != category:
changes["category"] = category
if not item.get("enriched"):
changes["enriched"] = True
# 充實之後要把過期的排除標記撤掉——而且要撤在檔案上。
#
# 這一段原本只改索引列。索引現在是衍生物，所以那是死碼：下一次重建會
# 從卡片檔案把 excluded: true 原封不動讀回來。結果是排除變成單行道——
# 一張被誤排除的卡（例如被那個「空欄位吞掉下一行」的解析 bug 害到的那
# 兩張），即使後來補上了標題與摘要，也沒有任何管線步驟能把它救回來，
# 只能手動改檔案。原本反而是安全的：那時候下一次重建就洗掉了。
#
# 只撤品質類的理由。duplicate_source_url 不是品質問題，撤掉它會把一份
# 重複的副本放回召回裡。
if item.get("excluded") and title and summary and not title.startswith("Tweet "):
reason = str(item.get("excluded_reason") or "")
if "duplicate_source_url" not in reason:
for path in xkb_index.files_for_item(item):
if xkb_frontmatter.unmark_excluded(path, dry_run=args.dry_run):
unexcluded += 1
if changes:
if not args.dry_run:
item.update(changes)
updated += 1
if args.dry_run:
print(f" [dry-run] {raw_card_id}: {list(changes.keys())}")
print(f"✅ Updated: {updated} | Not in index: {len(not_found_ids)}")
if unexcluded:
print(f"↩️ 撤掉 {unexcluded} 個過期的排除標記（卡片已補上標題與摘要）")
if not_found_ids:
print(f" (first 5 not found: {not_found_ids[:5]})")
added = 0
existing_ids = {_resolve_item_id(item) for item in items}
existing_urls = {(item.get("source_url") or "").strip() for item in items if (item.get("source_url") or "").strip()}
existing_paths = {(item.get("relative_path") or item.get("path") or "").strip() for item in items}
for raw_card_id in not_found_ids:
card_path = CARDS_DIR / f"{raw_card_id}.md"
if not card_path.exists():
continue
text = card_path.read_text(encoding="utf-8", errors="ignore")
title = _extract_title(text)
summary = _extract_summary(text)
tags = _extract_tags(text)
source_url = _extract_frontmatter_value(text, "source_url")
category = _extract_frontmatter_value(text, "category")
source_type = _extract_frontmatter_value(text, "source_type")
resolved_id = _resolve_card_id(text, raw_card_id) or raw_card_id
if not _is_valid_orphan_card(text, resolved_id, source_url, title, summary):
continue
card_abs = str(card_path)
card_rel = _relative_from_workspace(card_path)
if resolved_id in existing_ids or source_url in existing_urls or card_rel in existing_paths or card_abs in existing_paths:
continue
new_entry = {
"id": resolved_id,
"path": card_abs,
"relative_path": card_rel,
"title": title,
"summary": summary,
"tags": tags,
"source_url": source_url,
"category": category,
"source_type": source_type,
"enriched": True,
}
if not args.dry_run:
items.append(new_entry)
added += 1
if args.dry_run:
print(f" [dry-run] ADD {resolved_id}: {title[:60]}")
if added:
print(f"➕ Added {added} new index entries from orphaned cards")
# 這支腳本不再寫索引。
#
# 它做的事——把卡片的標題／摘要／標籤同步進索引列、把孤兒卡補進去——
# build_search_index.sh 從檔案重算時本來就全做了，而且做得更全：實測
# 一次乾淨重建找回三張這支腳本漏掉的卡（它只走 X 書籤 ID 那條路，
# github_star-* 與 legacy-* 那類卡片它掃不到）。
#
# 保留這個入口是為了不打斷既有的呼叫與習慣；真正做事的是 builder。
if not args.dry_run and (updated > 0 or added > 0 or not_found_ids):
xkb_index.rebuild()
import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py
if __name__ == "__main__":
xkb_usage.record(__file__)
main()