agent-radar/scripts/_common.py

"""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 upsert_block(
    text: str, block_key: str, values: dict, parent: str, indent: int | None = None
) -> str:
    """update_block 的容错版：块不存在时，在 parent 块末尾**新建**它。

    为什么需要这个：手工补完档案时习惯整块替换（ecosystem、business 等），
    一不留神就会把块内的子块（如 ecosystem.metrics）一起覆盖掉。
    update_block 遇到块缺失只会抛 ValueError，被上层 catch 成一条警告，
    表现是「指标栏静默变空」——很难察觉，等发现时原始数据已经没了。
    有了 upsert，这类误删能自动长回来。
    """
    try:
        return update_block(text, block_key, values, indent)
    except ValueError:
        pass

    lines = text.split("\n")
    pat = re.compile(rf"^(\s*){re.escape(parent)}:\s*$")
    pstart = None
    pbase = 0
    for i, ln in enumerate(lines):
        m = pat.match(ln)
        if m:
            pstart, pbase = i, len(m.group(1))
            break
    if pstart is None:
        raise ValueError(f"未找到父块 `{parent}:`，无法新建 `{block_key}:`")

    pend = len(lines)
    for j in range(pstart + 1, len(lines)):
        ln = lines[j]
        if not ln.strip() or ln.lstrip().startswith("#"):
            continue
        if len(ln) - len(ln.lstrip()) <= pbase:
            pend = j
            break

    child = pbase + 2
    grand = child + 2
    new = [f"{' ' * child}{block_key}:"] + [
        f"{' ' * grand}{k}: {_fmt(v)}" for k, v in values.items()
    ]
    lines[pend:pend] = new
    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"}