105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
||
"""CNB CodeBuddy NPC 通用监控脚本 — 自动盯所有 open Issue + PR。
|
||
|
||
设计：
|
||
- 自动列出仓库所有 open 的 Issue（含 NPC 评论摘要）
|
||
- 列出所有 open PR（NPC 推的 MR）
|
||
- 输出稳定（无时间戳），任何新 Issue / 新评论 / 新 PR 都会让输出变化
|
||
- 配合 cronjob monitor_script 使用：有变化才触发 agent，静默零消耗
|
||
"""
|
||
import json
|
||
import os
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
API = "https://api.cnb.cool"
|
||
REPO = os.environ.get("CNB_REPO", "muchen-org/download-webhook")
|
||
|
||
|
||
def load_token():
|
||
"""从 ~/.hermes/.env 读取 CNB_TOKEN"""
|
||
env_path = os.path.expanduser("~/.hermes/.env")
|
||
try:
|
||
with open(env_path) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line.startswith("CNB_TOKEN="):
|
||
return line.split("=", 1)[1].strip()
|
||
except FileNotFoundError:
|
||
pass
|
||
return os.environ.get("CNB_TOKEN", "")
|
||
|
||
|
||
def api_get(path):
|
||
req = urllib.request.Request(
|
||
f"{API}{path}",
|
||
headers={
|
||
"Authorization": f"Bearer {load_token()}",
|
||
"accept": "application/json",
|
||
},
|
||
method="GET",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=20) as resp:
|
||
return json.loads(resp.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
return {"error": e.code}
|
||
except Exception as e:
|
||
return {"error": str(e)}
|
||
|
||
|
||
def get_list(payload, key="items"):
|
||
"""兼容 list / {items: [...]} 两种返回"""
|
||
if isinstance(payload, list):
|
||
return payload
|
||
if isinstance(payload, dict) and key in payload:
|
||
return payload[key]
|
||
if isinstance(payload, dict) and "data" in payload:
|
||
return payload["data"]
|
||
return []
|
||
|
||
|
||
def main():
|
||
lines = []
|
||
|
||
# 1. 所有 open Issue
|
||
issues = api_get(f"/{REPO}/-/issues?state=open")
|
||
issue_list = get_list(issues)
|
||
open_issues = [i for i in issue_list if i.get("state") == "open"]
|
||
lines.append(f"Open issues: {len(open_issues)}")
|
||
for issue in sorted(open_issues, key=lambda x: int(x.get("number", 0))):
|
||
num = issue.get("number", "?")
|
||
comments = api_get(f"/{REPO}/-/issues/{num}/comments")
|
||
comment_list = get_list(comments)
|
||
# 只显示 NPC 评论（作者含 npc 或模型名）
|
||
npc_comments = [
|
||
c for c in comment_list
|
||
if "npc" in (c.get("author", {}).get("username", "")).lower()
|
||
or "npc" in (c.get("author", {}).get("nickname", "")).lower()
|
||
or (c.get("author", {}).get("username", "") or "").startswith("npc/")
|
||
or (c.get("author", {}).get("nickname", "") or "").startswith("npc/")
|
||
or c.get("author", {}).get("username", "") == "deepseek-v4-flash"
|
||
]
|
||
title = (issue.get("title") or "").replace("\n", " ")
|
||
lines.append(f" #{num} open: {title[:70]}")
|
||
for c in npc_comments:
|
||
body = (c.get("body") or "").replace("\n", " ")[:80]
|
||
lines.append(f" NPC: {body}")
|
||
if not npc_comments:
|
||
lines.append(f" (no NPC activity)")
|
||
|
||
# 2. 所有 open PR
|
||
pulls = api_get(f"/{REPO}/-/pulls?state=open")
|
||
pull_list = get_list(pulls)
|
||
open_pulls = [p for p in pull_list if p.get("state") == "open"]
|
||
lines.append(f"Open PRs: {len(open_pulls)}")
|
||
for p in sorted(open_pulls, key=lambda x: int(x.get("number", 0))):
|
||
author = p.get("author", {}).get("nickname", p.get("author", {}).get("username", "?"))
|
||
lines.append(f" PR#{p.get('number', '?')} [{author}]: {(p.get('title') or '')[:60]}")
|
||
|
||
print("\n".join(lines))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|