264 lines
9.5 KiB
Python
264 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
github-weekly-digest.py — GitHub 精选周报（牧尘定制版）
|
||
|
||
流程：
|
||
1. 抓取 RSSHub /github/topics/{topic} 多主题（ai / machine-learning / frontend / developer-tools / agents）
|
||
2. 收集候选仓库（title / link / desc）
|
||
3. LLM 筛选：按牧尘兴趣（AI 基建、开源工具、独立开发、前端设计）精选 3-5 个
|
||
4. 输出周报（有新增才推送）
|
||
|
||
用法：
|
||
python3 ~/.hermes/scripts/github-weekly-digest.py [--dry-run] [--force]
|
||
|
||
配置（环境变量可覆盖）：
|
||
RSSHUB_BASE - RSSHub 地址，默认 http://127.0.0.1:1200
|
||
ZHIYI_BASE - 织忆地址，默认 http://localhost:7821
|
||
ZHIYI_KEY - 织忆 API key
|
||
NEWAPI_BASE - NewAPI 地址，默认 http://127.0.0.1:3000/v1
|
||
NEWAPI_KEY - NewAPI key（筛选用免费模型）
|
||
"""
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
import xml.etree.ElementTree as ET
|
||
from datetime import datetime
|
||
|
||
RSSHUB_BASE = os.environ.get("RSSHUB_BASE", "http://127.0.0.1:1200")
|
||
ZHIYI_BASE = os.environ.get("ZHIYI_BASE", "http://localhost:7821")
|
||
ZHIYI_KEY = os.environ.get("ZHIYI_KEY", "zhiyi-dev-key-2026")
|
||
NEWAPI_BASE = os.environ.get("NEWAPI_BASE", "http://127.0.0.1:3000/v1")
|
||
NEWAPI_KEY = os.environ.get("NEWAPI_KEY", "")
|
||
AGNES_KEY = ""
|
||
try:
|
||
for _l in open(os.path.expanduser("~/.hermes/.env"), encoding="utf-8"):
|
||
if _l.startswith("AGNES_API_KEY=") and not _l.startswith("#"):
|
||
AGNES_KEY = _l.strip().split("=", 1)[1]
|
||
break
|
||
except Exception:
|
||
pass
|
||
AGNES_BASE = "https://apihub.agnes-ai.com/v1"
|
||
LOCAL_BASE = "http://127.0.0.1:8080/v1" # 本地 llama-server（3B/7B 常驻）
|
||
USE_LOCAL_LLM = os.environ.get("GH_DIGEST_LOCAL", "1") == "1" # 默认走本地
|
||
|
||
def _load_newapi_key():
|
||
"""从 ~/.hermes/config.yaml 读取 newapi-local provider 的 api_key（config 是权威源）"""
|
||
if AGNES_KEY:
|
||
return AGNES_KEY
|
||
if NEWAPI_KEY:
|
||
return NEWAPI_KEY
|
||
try:
|
||
import yaml
|
||
cfg_path = os.path.expanduser("~/.hermes/config.yaml")
|
||
with open(cfg_path) as f:
|
||
cfg = yaml.safe_load(f)
|
||
prov = (cfg.get("providers") or {}).get("newapi-local") or {}
|
||
return prov.get("api_key", "")
|
||
except Exception:
|
||
return ""
|
||
|
||
STATE_FILE = os.path.expanduser("~/.hermes/data/github_digest_seen.json")
|
||
MAX_SEEN = 300
|
||
|
||
# 订阅主题（RSSHub /github/topics/{topic} 已验证可用）
|
||
TOPICS = [
|
||
("ai", "AI"),
|
||
("machine-learning", "机器学习"),
|
||
("frontend", "前端"),
|
||
("developer-tools", "开发者工具"),
|
||
("agents", "AI Agent"),
|
||
("llm", "LLM"),
|
||
]
|
||
|
||
# 牧尘兴趣画像（筛选 prompt 用，简短版避免长 prompt 导致小模型空响应）
|
||
INTEREST_PROFILE = """用户兴趣：独立开发者、AI基建(LLM网关/Agent/记忆/RAG)、前端设计(React/Tailwind/shadcn)、财务自动化(金蝶K3/OCR)。偏好MIT开源、自托管、零配置工具。不感兴趣：游戏/社交/学习资源合集。"""
|
||
|
||
|
||
def fetch_rss(path, retries=3):
|
||
url = RSSHUB_BASE + path
|
||
last_exc = None
|
||
for attempt in range(retries):
|
||
try:
|
||
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
|
||
with urllib.request.urlopen(req, timeout=45) as resp:
|
||
return resp.read().decode("utf-8", "ignore")
|
||
except Exception as e:
|
||
last_exc = e
|
||
if attempt < retries - 1:
|
||
time.sleep(3 * (attempt + 1))
|
||
raise last_exc
|
||
|
||
|
||
def parse_items(xml_text):
|
||
items = []
|
||
try:
|
||
root = ET.fromstring(xml_text)
|
||
except ET.ParseError:
|
||
return items
|
||
for it in root.iter("item"):
|
||
title = (it.findtext("title") or "").strip()
|
||
link = (it.findtext("link") or "").strip()
|
||
desc = re.sub(r"<[^>]+>", "", it.findtext("description") or "").strip()
|
||
items.append({"title": title, "link": link, "desc": desc[:200]})
|
||
for it in root.iter("{http://www.w3.org/2005/Atom}entry"):
|
||
title = (it.findtext("{http://www.w3.org/2005/Atom}title") or "").strip()
|
||
link_el = it.find("{http://www.w3.org/2005/Atom}link")
|
||
link = (link_el.get("href") if link_el is not None else "").strip()
|
||
desc = re.sub(r"<[^>]+>", "", it.findtext("{http://www.w3.org/2005/Atom}summary") or "").strip()
|
||
items.append({"title": title, "link": link, "desc": desc[:200]})
|
||
return items
|
||
|
||
|
||
def load_seen():
|
||
if os.path.exists(STATE_FILE):
|
||
try:
|
||
with open(STATE_FILE) as f:
|
||
return set(json.load(f))
|
||
except Exception:
|
||
return set()
|
||
return set()
|
||
|
||
|
||
def save_seen(seen):
|
||
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
||
with open(STATE_FILE, "w") as f:
|
||
json.dump(sorted(seen)[-MAX_SEEN:], f, ensure_ascii=False)
|
||
|
||
|
||
def llm_select(candidates):
|
||
"""用免费模型筛选 3-5 个最值得推荐的仓库。返回选中项列表。"""
|
||
key = _load_newapi_key()
|
||
if not key or not candidates:
|
||
return candidates[:5]
|
||
|
||
cand_text = "\n".join(
|
||
f"{i+1}. {c['title']} | {c['link']} | {c['desc'][:80]}"
|
||
for i, c in enumerate(candidates)
|
||
)
|
||
prompt = f"""{INTEREST_PROFILE}
|
||
|
||
候选仓库：
|
||
{cand_text}
|
||
|
||
选3个最值得关注的，只输出编号，逗号分隔（如: 4,5,8）。不要其他内容。
|
||
"""
|
||
payload = json.dumps({
|
||
"model": "qwen3b",
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"temperature": 0.3,
|
||
"max_tokens": 100,
|
||
}).encode("utf-8")
|
||
base = LOCAL_BASE if USE_LOCAL_LLM else (AGNES_BASE if AGNES_KEY else NEWAPI_BASE)
|
||
req = urllib.request.Request(
|
||
base + "/chat/completions",
|
||
data=payload,
|
||
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||
data = json.loads(resp.read().decode("utf-8"))
|
||
text = data["choices"][0]["message"].get("content") or ""
|
||
# 解析逗号/空格分隔编号（如 "4,5,8" 或 "4 5 8"）
|
||
nums = [int(x) for x in re.findall(r"\d+", text)]
|
||
result = []
|
||
for n in nums:
|
||
idx = n - 1
|
||
if 0 <= idx < len(candidates) and len(result) < 5:
|
||
c = dict(candidates[idx])
|
||
c["reason"] = f"来自 GitHub Topics「{c.get('topic','')}」热榜，与你关注方向相关"
|
||
result.append(c)
|
||
if result:
|
||
return result
|
||
print("[warn] LLM 未返回有效编号，退回候选前5条", file=sys.stderr)
|
||
return candidates[:5]
|
||
except Exception as e:
|
||
print(f"[warn] LLM 筛选失败，退回前5条: {e}", file=sys.stderr)
|
||
return candidates[:5]
|
||
|
||
|
||
def main():
|
||
dry = "--dry-run" in sys.argv
|
||
force = "--force" in sys.argv
|
||
seen = load_seen()
|
||
candidates = []
|
||
errors = []
|
||
|
||
for topic, label in TOPICS:
|
||
try:
|
||
xml_text = fetch_rss(f"/github/topics/{topic}")
|
||
items = parse_items(xml_text)
|
||
except Exception as e:
|
||
errors.append(f"{label}: {type(e).__name__}: {e}")
|
||
continue
|
||
for it in items[:6]: # 每主题最多 6 个候选
|
||
if not it["link"]:
|
||
continue
|
||
# 过滤最近推荐过的（避免连续重复），force 时忽略
|
||
if it["link"] in seen and not force:
|
||
continue
|
||
candidates.append({**it, "topic": label})
|
||
|
||
# 去重（跨主题可能重复）
|
||
seen_links = set()
|
||
uniq = []
|
||
for c in candidates:
|
||
if c["link"] not in seen_links:
|
||
seen_links.add(c["link"])
|
||
uniq.append(c)
|
||
candidates = uniq
|
||
|
||
if not candidates:
|
||
# 全部候选都是最近推荐过的 → 本周推"回顾"而不是静默（周报是周期性交付，不能哑火）
|
||
print(f"📚 GitHub 精选周报（{datetime.now().strftime('%m-%d')}）")
|
||
print("本周无新候选——最近推荐的都已覆盖。下周继续观察新项目。")
|
||
return
|
||
|
||
# 启发式预筛：候选过多时先按兴趣关键词过滤，降 LLM prompt 长度
|
||
if len(candidates) > 12:
|
||
INTEREST_KW = [
|
||
"agent", "llm", "gpt", "rag", "memory", "mcp", "ai", "model",
|
||
"react", "tailwind", "ui", "design", "css", "frontend",
|
||
"tool", "cli", "automation", "workflow", "self-host", "docker",
|
||
"ocr", "excel", "account", "finance", "pdf",
|
||
]
|
||
scored = []
|
||
for c in candidates:
|
||
hay = (c["title"] + " " + c["desc"]).lower()
|
||
score = sum(1 for kw in INTEREST_KW if kw in hay)
|
||
scored.append((score, c))
|
||
scored.sort(key=lambda x: -x[0])
|
||
candidates = [c for s, c in scored[:12]]
|
||
|
||
# LLM 筛选
|
||
picks = llm_select(candidates)
|
||
|
||
# 更新 seen：只保留"最近推荐的"（最多 20 个），供下周过滤避免重复
|
||
# 注意：dry-run 不写状态（模拟无副作用）
|
||
if not dry:
|
||
for p in picks:
|
||
seen.add(p["link"])
|
||
save_seen(seen)
|
||
|
||
# 输出周报
|
||
lines = []
|
||
lines.append(f"📚 GitHub 精选周报（{datetime.now().strftime('%m-%d')}）")
|
||
lines.append(f"本周候选 {len(candidates)} 个，为你精选 {len(picks)} 个：")
|
||
lines.append("")
|
||
for i, p in enumerate(picks, 1):
|
||
lines.append(f"**{i}. {p['title']}**")
|
||
lines.append(f"`{p['link']}`")
|
||
if p.get("reason"):
|
||
lines.append(f"💡 {p['reason']}")
|
||
lines.append("")
|
||
if errors:
|
||
lines.append(f"⚠️ 部分主题抓取失败: {'; '.join(errors[:3])}")
|
||
print("\n".join(lines))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|