Publish agent-radar via gitea-publish skill
This commit is contained in:
gitea-publish
Executable
+215
@@ -0,0 +1,215 @@
|
||||
"""agent-radar 共享基础库：GitHub API 封装 + YAML 精确回写。
|
||||
|
||||
设计要点
|
||||
--------
|
||||
1. GitHub API 优先使用 GITHUB_TOKEN，缺失时匿名降级（限额 60 次/小时）。
|
||||
2. YAML 回写采用**行级精确替换**而不是 yaml.dump —— dump 会冲掉档案里的
|
||||
所有注释和字段顺序，而档案是靠注释承载"为什么这么填"的。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
DATA = ROOT / "data"
|
||||
PROJECTS = DATA / "projects"
|
||||
SNAPSHOTS = DATA / "snapshots"
|
||||
ARTICLES = ROOT / "articles"
|
||||
SCHEMA = ROOT / "schema"
|
||||
|
||||
GH_API = "https://api.github.com"
|
||||
TOKEN = os.environ.get("GITHUB_TOKEN", "").strip()
|
||||
|
||||
_last_call = 0.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- GitHub API
|
||||
def gh_get(path: str, timeout: int = 25, retries: int = 2) -> dict | list | None:
|
||||
"""GET GitHub API。失败返回 None（调用方负责降级）。
|
||||
|
||||
内置限速：匿名 1.5s 间隔、带 token 0.3s 间隔，避免触发 403 限流。
|
||||
|
||||
retries 是必要的：releases 这类接口响应体可达数百 KB，
|
||||
偶发 IncompleteRead（连接提前关闭）。不加重试会静默丢掉版本节奏数据，
|
||||
而"样本不足"和"真的没发版"在结果上无法区分。
|
||||
"""
|
||||
global _last_call
|
||||
url = f"{GH_API}/{path.lstrip('/')}"
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
headers={
|
||||
"Accept": "application/vnd.github+json",
|
||||
"User-Agent": "agent-radar",
|
||||
**({"Authorization": f"Bearer {TOKEN}"} if TOKEN else {}),
|
||||
},
|
||||
)
|
||||
|
||||
last_err = None
|
||||
for attempt in range(retries + 1):
|
||||
wait = 0.3 if TOKEN else 1.5
|
||||
gap = time.time() - _last_call
|
||||
if gap < wait:
|
||||
time.sleep(wait - gap)
|
||||
_last_call = time.time()
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
return json.loads(r.read().decode("utf-8"))
|
||||
except urllib.error.HTTPError as e:
|
||||
# 4xx 是确定性失败，重试无意义
|
||||
if e.code == 403:
|
||||
print(f" [warn] GitHub API 限流或禁止访问 (403): {path}")
|
||||
elif e.code == 404:
|
||||
print(f" [warn] 资源不存在 (404): {path}")
|
||||
else:
|
||||
print(f" [warn] HTTP {e.code}: {path}")
|
||||
return None
|
||||
except Exception as e: # IncompleteRead / 超时 / 连接重置
|
||||
last_err = e
|
||||
if attempt < retries:
|
||||
time.sleep(1.5 * (attempt + 1))
|
||||
continue
|
||||
print(f" [warn] 请求失败（已重试 {retries} 次）: {path} -> {last_err}")
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- YAML 读写
|
||||
def load_yaml(path) -> dict:
|
||||
"""接受 str 或 Path —— 调用方传字符串时不应该炸。"""
|
||||
p = Path(path)
|
||||
return yaml.safe_load(p.read_text(encoding="utf-8")) or {}
|
||||
|
||||
|
||||
def today() -> str:
|
||||
return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
def _fmt(v) -> str:
|
||||
"""把 Python 值渲染成 YAML 标量文本。"""
|
||||
if isinstance(v, bool):
|
||||
return "true" if v else "false"
|
||||
if v is None:
|
||||
return '""'
|
||||
if isinstance(v, (int, float)):
|
||||
return str(v)
|
||||
if isinstance(v, dict):
|
||||
inner = ", ".join(f"{k}: {_fmt(x)}" for k, x in v.items())
|
||||
return "{" + inner + "}"
|
||||
if isinstance(v, list):
|
||||
inner = ", ".join(_fmt(x) for x in v)
|
||||
return "[" + inner + "]"
|
||||
s = str(v)
|
||||
# 含冒号空格、井号、引号或为空的字符串需要加引号
|
||||
if s == "" or re.search(r'[:#\n"]|^\s|\s$', s) or re.match(r"^[-?*&!|>%@`]", s):
|
||||
return json.dumps(s, ensure_ascii=False)
|
||||
return s
|
||||
|
||||
|
||||
def update_block(text: str, block_key: str, values: dict, indent: int | None = None) -> str:
|
||||
"""在 YAML 文本里定位 `<block_key>:` 块，替换其中的键值。
|
||||
|
||||
只改块内已有键的值；块内没有而 values 里有的，追加到块末尾。
|
||||
块外内容（含所有注释）原样保留。
|
||||
|
||||
indent 传 None 时**自动探测**第一个同名块 —— 推荐用这个。
|
||||
硬编码缩进是很脆的：schema 一改层级，块的实际缩进就变了，
|
||||
而失败表现只是"静默没写进去"，很容易被忽略。
|
||||
"""
|
||||
lines = text.split("\n")
|
||||
# `(\s*)` 而不是 `(\s+)`：顶层块的缩进是 0，用 `+` 会匹配不到
|
||||
# （article: 就是顶层块，这个坑踩过一次）
|
||||
pat = re.compile(rf"^(\s*){re.escape(block_key)}:\s*$")
|
||||
|
||||
start = None
|
||||
base = None
|
||||
for i, ln in enumerate(lines):
|
||||
m = pat.match(ln)
|
||||
if m and (indent is None or len(m.group(1)) == indent):
|
||||
start, base = i, len(m.group(1))
|
||||
break
|
||||
if start is None:
|
||||
raise ValueError(f"未找到 `{block_key}:` 块（indent={indent}）")
|
||||
|
||||
# 块的结束：下一个缩进 <= base 的非空、非注释行
|
||||
end = len(lines)
|
||||
for j in range(start + 1, len(lines)):
|
||||
ln = lines[j]
|
||||
if not ln.strip() or ln.lstrip().startswith("#"):
|
||||
continue
|
||||
if len(ln) - len(ln.lstrip()) <= base:
|
||||
end = j
|
||||
break
|
||||
|
||||
child = base + 2
|
||||
seen: set[str] = set()
|
||||
for k in range(start + 1, end):
|
||||
m = re.match(rf"^(\s{{{child},}})([A-Za-z_][\w-]*):\s*(.*)$", lines[k])
|
||||
if not m:
|
||||
continue
|
||||
key = m.group(2)
|
||||
if key in values:
|
||||
lines[k] = f"{' ' * child}{key}: {_fmt(values[key])}"
|
||||
seen.add(key)
|
||||
|
||||
missing = [f"{' ' * child}{k}: {_fmt(v)}" for k, v in values.items() if k not in seen]
|
||||
if missing:
|
||||
lines[end:end] = missing
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 快照
|
||||
def snapshot_path(pid: str) -> Path:
|
||||
return SNAPSHOTS / f"{pid}.jsonl"
|
||||
|
||||
|
||||
def append_snapshot(pid: str, metrics: dict) -> None:
|
||||
"""追加一行快照。同一天重复运行会覆盖当天那行（保持"一天一个点"）。"""
|
||||
SNAPSHOTS.mkdir(parents=True, exist_ok=True)
|
||||
p = snapshot_path(pid)
|
||||
day = metrics.get("data_date") or today()
|
||||
|
||||
rows = []
|
||||
if p.exists():
|
||||
rows = [r for r in p.read_text(encoding="utf-8").splitlines() if r.strip()]
|
||||
rows = [r for r in rows if json.loads(r).get("data_date") != day]
|
||||
rows.append(json.dumps(metrics, ensure_ascii=False))
|
||||
p.write_text("\n".join(rows) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def last_snapshot(pid: str) -> dict | None:
|
||||
p = snapshot_path(pid)
|
||||
if not p.exists():
|
||||
return None
|
||||
rows = [r for r in p.read_text(encoding="utf-8").splitlines() if r.strip()]
|
||||
return json.loads(rows[-1]) if rows else None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 许可证分级
|
||||
def classify_license(spdx: str, license_name: str = "") -> dict:
|
||||
"""按 schema/license-tiers.yaml 判定成色分级与镜像可见性。
|
||||
|
||||
返回 {level, redistributable, visibility}。
|
||||
无法识别时返回 L3（保守策略：宁可私有，不可误公开）。
|
||||
"""
|
||||
tiers = load_yaml(SCHEMA / "license-tiers.yaml").get("tiers", [])
|
||||
key = (spdx or "").strip()
|
||||
for t in tiers:
|
||||
match = [str(x) for x in t.get("match_spdx", []) if x is not None]
|
||||
if key in match:
|
||||
return {
|
||||
"level": t["level"],
|
||||
"redistributable": bool(t.get("redistributable")),
|
||||
"visibility": t.get("visibility", "private"),
|
||||
}
|
||||
# 未命中：按 L3 处理（源码可得但许可不明 → 不公开）
|
||||
return {"level": "L3", "redistributable": False, "visibility": "private"}
|
||||
Reference in New Issue
Block a user