#!/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())