scripts/run_scan_worker.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
Scan-mode bookmark enrichment worker.
Instead of reading from tiege-queue.json, directly scans memory/bookmarks/
for files that don't yet have a corresponding enriched card in memory/cards/.
Handles:
- Named files (01-clawpal.md) that have tweet_id inside frontmatter
- Files with truncated/non-standard tweet IDs
- Any bookmark without a matching memory/cards/ entry
Usage:
python3 scripts/run_scan_worker.py --limit 20
python3 scripts/run_scan_worker.py --dry-run
python3 scripts/run_scan_worker.py --limit 50 --worker pipeline
python3 scripts/run_scan_worker.py --category 01-openclaw-workflows --limit 5
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
from pathlib import Path
# ── Shared card prompt module ─────────────────────────────────────────────────
sys.path.insert(0, str(Path(__file__).parent))
from _card_prompt import build_prompt, find_related_context, llm_call as _llm_call, gbrain_put as _gbrain_put
from category_classifier import apply_category, classify_content
sys.path.insert(0, str(Path(__file__).resolve().parent))
import xkb_paths
from gbrain_publish import PublicationBatch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
import xkb_failures
from runtime_config import runtime_env
WORKSPACE = xkb_paths.WORKSPACE
BOOKMARKS_DIR = xkb_paths.BOOKMARKS_DIR
CARDS_DIR = xkb_paths.CARDS_DIR
def _get_api_key() -> str:
return runtime_env().get("LLM_API_KEY", "")
def _get_index_items() -> list[dict]:
index_path = WORKSPACE / "memory" / "bookmarks" / "search_index.json"
if not index_path.exists():
return []
try:
data = json.loads(index_path.read_text(encoding="utf-8"))
return data.get("items", []) if isinstance(data, dict) else []
except Exception as err:
# 索引壞掉會被當成「索引是空的」，於是整批重做或跳過去重。
xkb_failures.note("search index", err, detail=str(index_path))
return []
def _call_llm(api_key: str, content: str, card_id: str, source_url: str, category: str,
related_context: str = "") -> str:
prompt = build_prompt(
content=f"Please process this bookmark. If content is low-value (login page/404/noise), output only: SKIPPED\n\n"
f"--- Raw content ---\n{content[:4000]}\n---",
card_id=card_id,
source_type="x-bookmark",
source_url=source_url,
category=category,
related_context=related_context or "（知識庫中尚無明顯相關的已存 card）",
)
return _llm_call(prompt, api_key, max_tokens=2500)
# ── Scan-specific helpers ──────────────────────────────────────────────────────
def _extract_frontmatter_value(text: str, key: str) -> str:
m = re.search(rf'^{re.escape(key)}:\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 _build_legacy_card_id(filepath: Path) -> str:
slug = re.sub(r"[^a-zA-Z0-9._-]+", "-", filepath.stem).strip("-").lower() or "untitled"
return f"legacy-{slug}"
def _get_card_id(filepath: Path, content: str) -> str:
"""Determine a stable card ID for this bookmark file."""
tweet_id = _extract_frontmatter_value(content, "tweet_id")
if tweet_id and re.fullmatch(r"\d{15,20}", tweet_id):
return tweet_id
for field in ("source_url", "source"):
url = _extract_frontmatter_value(content, field)
status_id = _extract_status_id(url)
if status_id:
return status_id
if re.fullmatch(r"\d{15,20}", filepath.stem):
return filepath.stem
m = re.match(r"^(\d{15,20})", filepath.stem)
if m:
return m.group(1)
return _build_legacy_card_id(filepath)
def _get_source_url(content: str, card_id: str) -> str:
for field in ("source_url", "source"):
url = _extract_frontmatter_value(content, field)
if url and url.startswith("http"):
return url
if re.fullmatch(r"\d{15,20}", card_id):
return f"https://x.com/i/status/{card_id}"
return ""
def _get_category(filepath: Path) -> str:
try:
parts = filepath.relative_to(BOOKMARKS_DIR).parts
if len(parts) >= 2:
return parts[0]
except Exception:
pass
return ""
# Known markers that indicate the page failed to load (X login wall, etc.)
_JUNK_PATTERNS = [
"Don't miss what's happening",
"People on X are the first to know",
"Log in to X",
"Sign in to X",
"JavaScript is not available",
]
# junk detection uses pattern matching only (no length threshold)
def _is_junk_content(content: str) -> bool:
"""Return True if bookmark content is too thin or is a known empty page."""
# Strip frontmatter
body = content.split("---", 2)[-1].strip() if content.startswith("---") else content.strip()
for pat in _JUNK_PATTERNS:
if pat in body:
return True
return False
def scan_missing(limit: int, category_filter: str = "") -> list[tuple[Path, str, str, str, str]]:
"""Return list of (filepath, content, card_id, source_url, category) for unenriched files."""
CARDS_DIR.mkdir(parents=True, exist_ok=True)
existing_card_ids = xkb_paths.card_ids()
results = []
skip_dirs = {"notebooklm_exports", "__pycache__", "youtube"}
for md_file in sorted(BOOKMARKS_DIR.rglob("*.md")):
if any(d in md_file.parts for d in skip_dirs):
continue
if md_file.name.startswith("."):
continue
content = md_file.read_text(encoding="utf-8", errors="ignore")
card_id = _get_card_id(md_file, content)
source_url = _get_source_url(content, card_id)
category = _get_category(md_file)
if category_filter and category_filter not in category:
continue
if card_id in existing_card_ids:
continue
if _is_junk_content(content):
print(f" 🗑️ Junk (no card, empty content): {md_file.name}, deleting")
md_file.unlink()
continue
results.append((md_file, content, card_id, source_url, category))
if len(results) >= limit:
break
return results
def main() -> int:
parser = argparse.ArgumentParser(description="Scan-mode bookmark enrichment worker")
parser.add_argument("--limit", type=int, default=20, help="Max items to process (default: 20)")
parser.add_argument("--worker", default="scan-worker", help="Worker name")
parser.add_argument("--dry-run", action="store_true", help="Simulate without API calls")
parser.add_argument("--local-only", action="store_true", help="Skip LLM enrichment — scan and list unenriched bookmarks without sending content to any API")
parser.add_argument("--category", default="", help="Filter by category slug")
args = parser.parse_args()
if args.local_only:
args.dry_run = True # local-only implies dry-run (no API calls)
api_key = "" if args.dry_run else _get_api_key()
if not api_key and not args.dry_run:
print("❌ LLM_API_KEY not found. Set process env or XKB_ENV_FILE.")
sys.exit(1)
batch = PublicationBatch(_gbrain_put)
if not args.dry_run:
if not batch.ready():
return 1
batch.recover_saved((path, path.stem) for path in xkb_paths.card_files())
if batch.blocked:
batch.finish()
return 1
missing = scan_missing(args.limit, args.category)
total_missing = len(scan_missing(9999, args.category))
if not missing:
print("✅ All bookmarks already enriched")
return 0 if args.dry_run or batch.finish() else 1
print(f"📋 Found {total_missing} unenriched bookmarks | Processing {len(missing)} [worker: {args.worker}]")
if args.local_only:
print(" (local-only mode — no content sent to external APIs)")
elif args.dry_run:
print(" (dry-run — no API calls)")
else:
print(f" ⚠️ Bookmark content will be sent to LLM API (配置中的 model) for enrichment.")
results = {"done": 0, "skipped": 0, "failed": 0}
for filepath, content, card_id, source_url, category in missing:
if batch.blocked:
break
label = str(filepath.relative_to(BOOKMARKS_DIR))
print(f" → {label}", end=" ", flush=True)
if args.dry_run:
print(f"[dry-run: {card_id}]")
results["done"] += 1
continue
try:
related_ctx = find_related_context(content, [])
classified = classify_content(content, source_type="x-bookmark", current_category=category)
text = _call_llm(api_key, content, card_id, source_url, classified["category"], related_ctx)
if not text:
results["failed"] += 1
print("✗ empty response")
continue
if re.match(r"^SKIPPED", text.strip(), re.IGNORECASE):
results["skipped"] += 1
print("⏭ skipped")
continue
text = apply_category(text, classified["category"])
card_path = CARDS_DIR / f"{card_id}.md"
card_path.write_text(text, encoding="utf-8")
if not batch.publish(card_path, card_id, generated=True):
results["failed"] += 1
continue
results["done"] += 1
print("✓ done")
except Exception as exc:
results["failed"] += 1
print(f"✗ {exc}")
if not args.dry_run and not batch.finish():
return 1
if results["failed"]:
return 1
remaining = len(scan_missing(9999, args.category))
print(f"\n📊 done={results['done']} skipped={results['skipped']} failed={results['failed']} remaining={remaining}")
return 0
import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py
if __name__ == "__main__":
xkb_usage.record(__file__)
raise SystemExit(main())