scripts/fetch_github_repos.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
fetch_github_repos.py
從 GitHub 抓取使用者的 fork/star repo，生成知識卡片（repo-level only，不做整包 repo ingest），
加入 search_index.json。
Usage:
python3 scripts/fetch_github_repos.py --forks
python3 scripts/fetch_github_repos.py --stars
python3 scripts/fetch_github_repos.py --forks --stars
python3 scripts/fetch_github_repos.py --forks --limit 20
python3 scripts/fetch_github_repos.py --forks --dry-run
Exit codes:
0 — success, 0 new cards
1 — error
2 — success, >= 1 new cards added
"""
import argparse
import base64
import json
import re
import subprocess
import sys
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
from runtime_config import runtime_env
# ── Shared card prompt module ─────────────────────────────────────────────────
sys.path.insert(0, str(Path(__file__).parent))
from gbrain_publish import PublicationBatch
from xkb_frontmatter import parse as extract_frontmatter
from _card_prompt import (
build_prompt, extract_summary, find_related_context,
llm_call as _llm_call, SOURCE_LABELS, gbrain_put as _gbrain_put,
)
from category_classifier import classify_content
from category_classifier import apply_category
SOURCE_LABELS["github_fork"] = "GitHub 倉庫（Fork）"
SOURCE_LABELS["github_star"] = "GitHub 倉庫（Star）"
# ── Paths ──────────────────────────────────────────────────────────────────────
import xkb_paths
WORKSPACE_DIR = xkb_paths.WORKSPACE
BOOKMARKS_DIR = xkb_paths.BOOKMARKS_DIR
CARDS_DIR = xkb_paths.CARDS_DIR
INDEX_FILE = xkb_paths.INDEX_FILE
GITHUB_RAW_DIR = BOOKMARKS_DIR / "github" # raw metadata per repo (source layer)
CARD_CATEGORIES = [
"ai-tools", "developer-tools", "workflows", "data",
"startup", "design", "tech", "learning", "other"
]
def load_env_key() -> str:
return runtime_env().get("LLM_API_KEY", "")
# llm_call imported from _card_prompt as _llm_call
def gh_api(endpoint: str) -> list:
"""Call gh CLI with pagination and return merged list."""
result = subprocess.run(
["gh", "api", "--paginate", endpoint],
capture_output=True, text=True
)
if result.returncode != 0:
print(f" [ERROR] gh api {endpoint}: {result.stderr.strip()}", file=sys.stderr)
return []
text = result.stdout.strip()
# gh --paginate may return multiple JSON arrays on separate lines
items = []
for line in text.splitlines():
line = line.strip()
if not line:
continue
try:
parsed = json.loads(line)
if isinstance(parsed, list):
items.extend(parsed)
elif isinstance(parsed, dict):
items.append(parsed)
except Exception:
pass
return items or json.loads(text) if text else []
def get_readme_preview(full_name: str, max_chars: int = 500) -> str:
"""Fetch README first N chars. Returns empty string on any error — not fatal."""
result = subprocess.run(
["gh", "api", f"repos/{full_name}/readme", "--jq", ".content"],
capture_output=True, text=True
)
if result.returncode != 0 or not result.stdout.strip():
return ""
try:
decoded = base64.b64decode(result.stdout.strip()).decode("utf-8", errors="ignore")
decoded = re.sub(r'#+\s*', '', decoded)
decoded = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', decoded)
decoded = re.sub(r'!\[.*?\]\(.*?\)', '', decoded)
return decoded[:max_chars].strip()
except Exception:
return ""
def load_index() -> dict:
if INDEX_FILE.exists():
return json.loads(INDEX_FILE.read_text(encoding="utf-8"))
return {"version": "1.1", "items": []}
# extract_summary imported from _card_prompt
def filter_stars(repos: list) -> list:
return [r for r in repos
if not r.get("archived")
and (r.get("description") or r.get("topics"))]
def make_dedup_key(url: str, action_type: str) -> str:
"""Dedup key is (source_url, source_type) — fork and star of same repo are different entries."""
return f"{url}|{action_type}"
def process_repos(repos: list, action_type: str, dry_run: bool, api_key: str,
existing_keys: set, limit: int,
existing_items: list[dict] | None = None, publication=None) -> tuple[int, list[dict]]:
"""
Returns (count_processed, new_index_items).
Does NOT write to index — caller batches and saves once.
"""
subdir = GITHUB_RAW_DIR / (action_type.replace("github_", "") + "s")
subdir.mkdir(parents=True, exist_ok=True)
CARDS_DIR.mkdir(parents=True, exist_ok=True)
new_items = []
processed = 0
attempted = 0
batch = publication or PublicationBatch(_gbrain_put)
for repo in repos:
if batch.blocked:
break
full_name = repo.get("full_name", "")
url = repo.get("html_url", f"https://github.com/{full_name}")
dedup_key = make_dedup_key(url, action_type)
card_id = f"{action_type}-{full_name.replace('/', '-')}"
saved = CARDS_DIR / f"{card_id}.md"
if not dry_run and batch.existing(saved, card_id):
continue
if dedup_key in existing_keys:
print(f" [SKIP] 已存在：{full_name} ({action_type})")
continue
if limit and attempted >= limit:
continue
attempted += 1
description = repo.get("description") or "(no description)"
language = repo.get("language") or "unknown"
topics = ", ".join(repo.get("topics", [])) or "(none)"
stars = repo.get("stargazers_count", 0)
print(f" 📦 {full_name} [{language}] {'★'+str(stars) if stars else ''}")
if dry_run:
print(f" → {description[:80]}")
processed += 1
continue
# Save raw metadata (source layer)
safe_name = full_name.replace("/", "__")
raw_path = subdir / f"{safe_name}.json"
raw_path.write_text(json.dumps({
"full_name": full_name, "url": url, "description": description,
"language": language, "topics": repo.get("topics", []),
"stars": stars, "archived": repo.get("archived", False),
"pushed_at": repo.get("pushed_at", ""), "action_type": action_type,
"fetched_at": datetime.now(timezone.utc).isoformat(),
}, ensure_ascii=False, indent=2), encoding="utf-8")
# Fetch README preview (non-fatal)
readme = get_readme_preview(full_name)
# Card ID (also used in frontmatter and filename)
card_id = f"{action_type}-{full_name.replace('/', '-')}"
# Generate 9-section card via shared _card_prompt
content = (
f"Repo: {full_name}\n"
f"Description: {description}\n"
f"Language: {language}\n"
f"Topics: {topics}\n"
f"Stars: {stars}\n"
f"Action: {action_type} "
f"(github_fork = 主動複製並可能修改；github_star = 標記感興趣)\n\n"
f"README:\n{readme or '(not available)'}"
)
related = find_related_context(content, existing_items or [])
classified = classify_content(content, source_type=action_type, current_category="tech")
prompt = build_prompt(
content=content,
card_id=card_id,
source_type=action_type,
source_url=url,
category=classified["category"],
related_context=related,
)
try:
card_content = _llm_call(prompt, api_key)
except Exception as e:
print(f" ❌ LLM 失敗：{e}")
continue
# Inject id into frontmatter if missing
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, classified["category"])
# Save LLM card to memory/cards/
card_path = CARDS_DIR / f"{card_id}.md"
card_path.write_text(card_content, encoding="utf-8")
batch.publish(card_path, card_id, generated=True)
print(f" 💾 memory/cards/{card_id}.md")
# Build index item
fm = extract_frontmatter(card_content)
summary = extract_summary(card_content)
tags_raw = fm.get("tags", "")
tags = list({t.strip() for t in tags_raw.split(",") if t.strip()})
if language and language != "unknown":
tags.append(language.lower())
new_items.append({
"path": str(card_path),
"relative_path": f"cards/{card_id}.md",
"title": fm.get("title", full_name),
"category": fm.get("category", classified["category"]),
"tags": list(set(tags)),
"summary": summary,
"source_url": url,
"source_type": action_type,
"searchable": f"{full_name} {description} {topics} {summary} {' '.join(tags)}",
"mtime": datetime.now(timezone.utc).isoformat(),
"size": card_path.stat().st_size,
"enriched": True,
"language": language,
})
processed += 1
return processed, new_items
def main():
parser = argparse.ArgumentParser(
description="Fetch GitHub forks/stars into XKB (repo-level cards only)")
parser.add_argument("--forks", action="store_true")
parser.add_argument("--stars", action="store_true")
parser.add_argument("--limit", type=int, default=0, help="Max repos per type (0=all)")
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--no-filter", action="store_true", help="Disable star quality filter")
args = parser.parse_args()
if not args.forks and not args.stars:
parser.error("Specify --forks and/or --stars")
api_key = "" if args.dry_run else load_env_key()
if not api_key and not args.dry_run:
print("[ERROR] LLM_API_KEY not found")
return 1
# Build dedup key set from existing index
index_data = load_index()
existing_items_list = index_data.get("items", [])
existing_keys = {
make_dedup_key(item.get("source_url", ""), item.get("source_type", ""))
for item in existing_items_list
}
batch = PublicationBatch(_gbrain_put)
if not args.dry_run and not batch.ready():
return 1
all_new_items: list[dict] = []
total = 0
if args.forks:
print("\n🔀 Fetching forks...")
repos = gh_api("user/repos?type=fork&per_page=100")
print(f" Found {len(repos)} forks")
n, items = process_repos(repos, "github_fork", args.dry_run, api_key,
existing_keys, args.limit, existing_items_list, batch)
print(f" ✅ Processed {n} fork cards")
total += n
all_new_items.extend(items)
if args.stars and not batch.blocked:
print("\n⭐ Fetching stars...")
repos = gh_api("user/starred?per_page=100")
if not args.no_filter:
before = len(repos)
repos = filter_stars(repos)
print(f" Found {before} stars → {len(repos)} after filtering")
else:
print(f" Found {len(repos)} stars (no filter)")
n, items = process_repos(repos, "github_star", args.dry_run, api_key,
existing_keys, args.limit, existing_items_list, batch)
print(f" ✅ Processed {n} star cards")
total += n
all_new_items.extend(items)
# Rebuild once from the cards on disk.
if not args.dry_run and not batch.finish():
return 1
if total and not args.dry_run:
print(f"\n✅ 完成：共新增 {total} 張 GitHub 知識卡片，已寫入索引")
print("💡 執行以下指令更新語意索引：")
print(" python3 scripts/build_vector_index.py --incremental")
else:
print(f"\n✅ 完成：共新增 {total} 張 GitHub 知識卡片")
# Exit code 2 = new cards added (for run_github_sync.sh to detect)
return 2 if total > 0 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())