scripts/run_bookmark_worker.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
Bookmark enrichment worker — processes todo items from tiege-queue.json one at a time.
Usage:
python3 scripts/run_bookmark_worker.py --limit 5
python3 scripts/run_bookmark_worker.py --limit 1 --dry-run
python3 scripts/run_bookmark_worker.py --worker myagent --limit 10
python3 scripts/run_bookmark_worker.py --category 01-openclaw-workflows --limit 5
"""
from __future__ import annotations
import argparse
import json
import os
import re
import subprocess
import sys
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import xkb_text
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
import xkb_failures
import xkb_paths
from gbrain_publish import PublicationBatch
from runtime_config import runtime_env
WORKSPACE = xkb_paths.WORKSPACE
BOOKMARKS_DIR = xkb_paths.BOOKMARKS_DIR
QUEUE_PATH = Path(os.getenv("XKB_QUEUE_PATH", str(xkb_paths.XKB_DATA_DIR / "tiege-queue.json")))
CARDS_DIR = xkb_paths.CARDS_DIR
# ── gbrain integration ────────────────────────────────────────────────────────
_GBRAIN_DIR = Path(os.getenv("GBRAIN_DIR", str(Path.home() / "Desktop" / "gbrain")))
_GBRAIN_CLI = str(_GBRAIN_DIR / "src" / "cli.ts")
_GBRAIN_AVAILABLE = (_GBRAIN_DIR / "src" / "cli.ts").exists()
# Load GEMINI_API_KEY for gbrain embed; Hermes injects it via its profile .env.
_GBRAIN_ENV: dict[str, str] = {**os.environ}
# ── Unified LLM helper ────────────────────────────────────────────────────────
_SKILL_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_SKILL_DIR / "scripts"))
from _llm import call as _llm_backend
def _get_api_key() -> str:
return "" # auth handled by _llm.py via openclaw CLI
SYSTEM_PROMPT = """You are a knowledge card generator for a personal learning base. Given the raw content of a single X/Twitter bookmark and optionally a list of related existing cards, output one structured knowledge card in Traditional Chinese.
Strict rules:
- Process ONLY this one bookmark
- Leave sections empty with "無" if uncertain — never hallucinate
- If content is a login page, 404, or homepage noise, output exactly: SKIPPED
- Do NOT use the reader's personal name in any section
Output format (Markdown):
---
id: {id}
type: x-knowledge-card
source_type: x-bookmark
source_url: {source_url}
author: (infer from content, leave blank if unsure)
created_at: (infer from content, leave blank if unsure)
category: {category}
tags: [tag1, tag2, tag3]
sensitivity: public
confidence: medium
---
# <title>
## 1. 核心問題與結論
- **提問**：這篇試圖解答什麼問題？（一句話）
- **結論**：作者給出的答案是什麼？（一句話）
- **可信度說明**：這個結論有沒有數據/實驗/引用支撐，還是只是個人意見？
## 2. Claim 等級
- **等級**：[Attested | Scholarship | Inference]
- Attested：原文直接引用、有具體數據或實驗結果
- Scholarship：作者/領域的分析觀點，有明確來源依據
- Inference：LLM 推論、個人猜測、尚未驗證的假設
- **主要主張**：（一句話說明被標記的核心主張）
- **依據**：（為什麼是這個等級？有什麼支持或限制？）
## 3. 關鍵論點
- 論點一
- 論點二
- 論點三
## 4. False Friends（如有）
這篇涉及哪些看起來像普通詞彙但有特定技術含義的術語？
- term: （術語名稱）
common_misunderstanding: （多數人誤以為是...）
actual_meaning: （在此領域/文章中實際指的是...）
如果沒有：無
## 5. 驚訝點
讀者讀完這篇後，可能感到意外或需要重新思考的是什麼？
（如果沒有明顯驚訝點，填「無」）
## 6. 與現有知識的關係
{related_cards_section}
## 7. 雙語摘要（搜尋索引用）
ZH: <20-40字繁體中文摘要，說明核心發現>
EN: <15-30 word English summary of the core finding>
## 8. 對使用者的價值
- 可追蹤的方向
- 可執行的應用場景
- 適合哪個專案或工作流程
## 9. 原始來源
- Tweet: {source_url}
- Links: (list URLs found in content)
Quality principles: conservative > hallucination, understanding > summary, structured > verbose"""
from _card_prompt import gbrain_put as _gbrain_put
def _find_related_context(bookmark_content: str, top_k: int = 3) -> str:
"""Find related existing cards for context injection.
Uses gbrain hybrid search if available, falls back to keyword search."""
if _GBRAIN_AVAILABLE:
return _find_related_context_gbrain(bookmark_content, top_k)
return _find_related_context_keyword(bookmark_content, top_k)
def _find_related_context_gbrain(bookmark_content: str, top_k: int = 3) -> str:
"""Use gbrain hybrid search (RRF + Gemini) to find related cards."""
try:
sys.path.insert(0, str(Path(__file__).parent))
from xbrain_recall import xbrain_query as gbrain_query
# Use first 300 chars of content as query
query = bookmark_content[:300].replace("\n", " ").strip()
results = gbrain_query(query, limit=top_k, no_expand=True)
if not results:
return "（知識庫中尚無明顯相關的已存 card）"
lines = ["根據知識庫搜尋，以下是最相關的現有 cards，請分析這篇與它們的關係（補充/衝突/重複/延伸）："]
for r in results:
title = r.get("title", r.get("slug", ""))[:60]
chunk = r.get("chunk_text", "")[:120].replace("\n", " ")
lines.append(f"- **{title}**：{chunk}")
lines.append("\n請分析：這篇和以上 cards 的關係是什麼？（補充新面向 / 與某篇結論衝突 / 與某篇重複 / 帶入新概念）")
return "\n".join(lines)
except Exception:
return _find_related_context_keyword(bookmark_content, top_k)
def _find_related_context_keyword(bookmark_content: str, top_k: int = 3) -> str:
"""Fallback: keyword search using search_index.json."""
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"))
items = data.get("items", []) if isinstance(data, dict) else data
except Exception:
return ""
# 斷詞只有一個定義：xkb_text。這裡原本各自帶一份只切 2-gram、
# 而且把整串中文吃成一個 token 的正則。
query_tokens = set(xkb_text.tokenize(bookmark_content[:1000], xkb_text.STOPWORDS))
if not query_tokens:
return ""
scored = []
for item in items:
combined = " ".join([
(item.get("title") or "").lower(),
(item.get("summary") or "").lower(),
" ".join(item.get("tags") or []).lower(),
])
score = sum(1 for t in query_tokens if t in combined)
if score > 1:
scored.append((score, item))
scored.sort(key=lambda x: x[0], reverse=True)
top = scored[:top_k]
if not top:
return "（知識庫中尚無明顯相關的已存 card）"
lines = ["根據知識庫搜尋，以下是最相關的現有 cards，請分析這篇與它們的關係（補充/衝突/重複/延伸）："]
for _, item in top:
title = item.get("title", "")[:60]
summary = (item.get("summary") or "")[:100]
lines.append(f"- **{title}**：{summary}")
lines.append("\n請分析：這篇和以上 cards 的關係是什麼？（補充新面向 / 與某篇結論衝突 / 與某篇重複 / 帶入新概念）")
return "\n".join(lines)
def _call_llm(api_key: str, bookmark_content: str, item: dict, related_context: str = "") -> str:
related_section = related_context if related_context else "（知識庫中尚無明顯相關的已存 card）"
system = SYSTEM_PROMPT.format(
id=item["id"],
source_url=item.get("source_url", ""),
category=item.get("category", ""),
related_cards_section=related_section,
)
user_msg = (
f"Please process this bookmark:\n\nID: {item['id']}\n"
f"Source: {item.get('source_url', '')}\nCategory: {item.get('category', '')}\n\n"
f"--- Raw content ---\n{bookmark_content[:4000]}\n---\n\n"
"Output the knowledge card. If content is low-value (login page/404/noise), output only: SKIPPED"
)
return _llm_backend(system, user_msg)
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
def _load_queue() -> dict:
return json.loads(QUEUE_PATH.read_text(encoding="utf-8"))
def _save_queue(data: dict) -> None:
QUEUE_PATH.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def _read_bookmark(source_path: str) -> str:
"""讀書籤原文；路徑過期時改用 id 找。
佇列存的是當初排隊時的相對路徑，但書籤之後會被歸檔到分類資料夾
（`inbox/123.md` → `99-general/123.md`）。路徑一過期，worker 只會說
「bookmark file not found」然後把項目標成失敗——檔案明明還在。
2026-08-02 有 9 份就是這樣失敗的。id 是穩定的，路徑不是，所以路徑
找不到時就用 id 再找一次。
"""
candidate = Path(source_path)
# 絕對路徑與 .. 都不直接採信，但不能因此連「用 id 再找一次」都放棄——
# BOOKMARKS_DIR 被指到 WORKSPACE 之外時（.env.example 有寫這是支援的），
# sync_tiege_queue 記下來的就是絕對路徑，而那正是這個 fallback 存在的
# 場景。守衛的職責是不信任來源，不是取消復原。
if candidate.is_absolute() or ".." in candidate.parts:
full_path = Path("/nonexistent")
else:
full_path = (WORKSPACE / candidate).resolve()
bookmarks_root = BOOKMARKS_DIR.resolve()
if full_path.is_file() and (full_path == bookmarks_root or bookmarks_root in full_path.parents):
return full_path.read_text(encoding="utf-8", errors="ignore")
moved = next(xkb_paths.BOOKMARKS_DIR.rglob(f"{Path(source_path).stem}.md"), None)
if moved and (moved.resolve() == bookmarks_root or bookmarks_root in moved.resolve().parents):
# relative_to 只是為了訊息好看，但 BOOKMARKS_DIR 在 WORKSPACE 之外時
# 它會拋 ValueError——於是一筆「已經成功找到」的書籤被標成失敗。
# 一行日誌不該有能力讓它在描述的那個動作失敗。
try:
shown = moved.relative_to(WORKSPACE)
except ValueError:
shown = moved
print(f" （書籤已移動：{source_path} → {shown}）", flush=True)
return moved.read_text(encoding="utf-8", errors="ignore")
return ""
def _process_item(item: dict, api_key: str, dry_run: bool, publication=None) -> tuple[str, str]:
"""Returns (status, error). status: done | skipped | failed"""
saved = CARDS_DIR / f"{item['id']}.md"
if saved.exists() and not dry_run:
if publication is not None:
if not publication.publish(saved, item['id']):
return 'failed', 'publication pending'
else:
_gbrain_put(saved, item['id'])
return "done", ""
bookmark_content = _read_bookmark(item["source_path"])
if not bookmark_content:
return "failed", "bookmark file not found"
if dry_run:
preview = bookmark_content[:80].replace("\n", " ")
print(f" preview: {preview}...")
return "done", ""
related_ctx = _find_related_context(bookmark_content)
text = _call_llm(api_key, bookmark_content, item, related_ctx)
if not text:
return "failed", "empty response from API"
if re.match(r"^SKIPPED", text.strip(), re.IGNORECASE):
return "skipped", ""
CARDS_DIR.mkdir(parents=True, exist_ok=True)
card_path = CARDS_DIR / f"{item['id']}.md"
card_path.write_text(text, encoding="utf-8")
if publication is not None:
if not publication.publish(card_path, item["id"], generated=True):
return "failed", "publication pending"
else:
_gbrain_put(card_path, item["id"])
return "done", ""
def _sync_queue() -> bool:
"""Put newly fetched bookmarks on the queue before reading it.
Fetching writes a bookmark into the inbox; a separate step puts it on the
queue this worker reads. Those were two lines of one pipeline until that
pipeline was retired on 2026-08-26 and its stages were split across
scheduler jobs — the job that runs this worker did not inherit the queue
step. The worker then ran on time every night, found nothing to do,
reported success, and 33 bookmarks sat in the inbox accumulating at about
eight a day while every check stayed green.
Doing it here removes the ordering dependency instead of documenting it,
so no future caller can leave the step out. Skipped under --dry-run,
which must not write.
"""
script = Path(__file__).with_name("sync_tiege_queue.py")
if not script.exists():
return False
proc = subprocess.run(
[sys.executable, str(script)],
cwd=str(WORKSPACE), capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=900,
env=xkb_paths.subprocess_env(),
)
if proc.returncode != 0:
error = (proc.stderr or proc.stdout or "").strip()[-300:]
print(f"⚠️ queue sync failed: {error}", file=sys.stderr, flush=True)
return False
output = (proc.stdout or "").strip()
if output:
print(output, flush=True)
return True
def _sync_enriched_index() -> bool:
"""Make cards immediately searchable after a worker batch completes."""
script = Path(__file__).with_name("sync_enriched_index.py")
try:
proc = subprocess.run(
[sys.executable, str(script)],
cwd=str(WORKSPACE),
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=900,
env=xkb_paths.subprocess_env(),
)
except subprocess.TimeoutExpired:
# 逾時會直接把例外丟出去，而呼叫端把這個函式當成回 bool 的。
# 那正好是這支腳本最不該有的結局：卡片寫進磁碟、沒進索引、
# 而且整批以例外收場，看起來像「卡片沒產出」。
msg = "sync_enriched_index 超過 900 秒未完成——卡片已寫入但尚未進索引"
print(msg, file=sys.stderr, flush=True)
xkb_failures.note("sync enriched index", RuntimeError(msg))
return False
output = (proc.stdout or "").strip()
if output:
print("\n" + output, flush=True)
if proc.returncode != 0:
error = (proc.stderr or proc.stdout or "").strip()[-300:]
print(f"⚠️ enriched index sync failed: {error}", file=sys.stderr, flush=True)
return False
return True
def main() -> int:
parser = argparse.ArgumentParser(description="Bookmark enrichment worker")
parser.add_argument("--limit", type=int, default=5, help="Max items to process (default: 5)")
parser.add_argument("--worker", default="worker", help="Worker name recorded in queue")
parser.add_argument("--dry-run", action="store_true", help="Simulate without calling API")
parser.add_argument("--local-only", action="store_true", help="Skip LLM enrichment — index bookmarks without sending content to any API")
parser.add_argument("--category", help="Filter by category slug (e.g. 01-openclaw-workflows)")
args = parser.parse_args()
if args.local_only:
args.dry_run = True # local-only implies dry-run (no API calls)
api_key = runtime_env().get("LLM_API_KEY", "")
if not args.dry_run:
_sync_queue()
data = _load_queue()
items = data["items"]
todo = [i for i in items if i["status"] == "todo"]
if args.category:
todo = [i for i in todo if args.category in i.get("category", "")]
if args.dry_run:
todo = todo[:args.limit]
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(indexer=lambda count: _sync_enriched_index())
return 1
if not todo:
print("✅ No todo items found")
return 0 if args.dry_run or batch.finish(indexer=lambda count: _sync_enriched_index()) else 1
total_todo = len([i for i in items if i["status"] == "todo"])
print(f"📋 Processing {len(todo)}/{total_todo} todo items [worker: {args.worker}]")
if args.local_only:
print(" (local-only mode — no content sent to external APIs; queue will not be mutated)")
elif args.dry_run:
print(" (dry-run mode — no API calls; queue will not be mutated)")
else:
print(f" ⚠️ Bookmark content will be sent to LLM for enrichment (via configured XKB runtime provider).")
if args.dry_run:
for item in todo:
print(f" → {item['id']} [{item.get('category', '')}] ", end="", flush=True)
bookmark_content = _read_bookmark(item["source_path"])
if bookmark_content:
preview = bookmark_content[:80].replace("\n", " ")
print(f"preview: {preview}...")
else:
print("missing bookmark file")
print("\n✅ dry-run complete; queue unchanged")
return 0
generated_attempts = recovered_attempts = 0
id_to_indices: dict[str, list[int]] = defaultdict(list)
for idx, it in enumerate(items):
id_to_indices[it["id"]].append(idx)
results = {"done": 0, "skipped": 0, "failed": 0}
for item in todo:
if batch.blocked:
break
saved = (CARDS_DIR / f"{item['id']}.md").exists()
if saved:
if recovered_attempts >= batch.recovery_limit:
continue
recovered_attempts += 1
else:
if generated_attempts >= args.limit:
continue
generated_attempts += 1
print(f" → {item['id']} [{item.get('category', '')}]", end=" ", flush=True)
indices = id_to_indices[item["id"]]
for idx in indices:
data["items"][idx].update({"status": "processing", "worker": args.worker, "started_at": _now_iso()})
_save_queue(data)
try:
status, error = _process_item(item, api_key, args.dry_run, batch)
title_update = {}
if status == "done" and not args.dry_run:
card_path = CARDS_DIR / f"{item['id']}.md"
if card_path.exists():
m = re.search(r"^# (.+)$", card_path.read_text(encoding="utf-8"), re.MULTILINE)
if m:
title_update = {"title": m.group(1).strip()}
for idx in indices:
data["items"][idx].update({"status": status, "finished_at": _now_iso(), "error": error, **title_update})
results[status] += 1
print(f"✓ {status}")
except Exception as exc:
for idx in indices:
data["items"][idx].update({"status": "failed", "error": str(exc)[:200], "finished_at": _now_iso()})
results["failed"] += 1
print(f"✗ failed: {exc}")
_save_queue(data)
remaining = len([i for i in data["items"] if i["status"] == "todo"])
print(f"\n📊 done={results['done']} skipped={results['skipped']} failed={results['failed']} remaining todo={remaining}")
if not batch.finish(indexer=lambda count: _sync_enriched_index()):
return 1
if results["failed"]:
return 1
return 0
import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py
if __name__ == "__main__":
xkb_usage.record(__file__)
raise SystemExit(main())