diff --git a/skills/gitlink-metrics/scripts/glapi.py b/skills/gitlink-metrics/scripts/glapi.py new file mode 100644 index 0000000..41db09d --- /dev/null +++ b/skills/gitlink-metrics/scripts/glapi.py @@ -0,0 +1,241 @@ +"""GitLink 公开 API 共享客户端。 + +供 gitlink-skills-pack 下各 Skill 的脚本复用。仅依赖 Python 标准库, +无需第三方包,便于在受限环境或 Agent 沙箱中运行。 + +数据全部来自 GitLink 平台公开接口(https://www.gitlink.org.cn/api), +默认无需 token;如需访问私有仓库,可传入 token。 + +所有方法均为只读,不修改任何远程数据。 +""" + +from __future__ import annotations + +import base64 +import json +import time +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path +from typing import Any + +API_BASE = "https://www.gitlink.org.cn/api" +USER_AGENT = "gitlink-skills-pack/1.0 (+https://www.gitlink.org.cn)" +DEFAULT_TIMEOUT = 30 +COMMIT_PAGE_SIZE = 50 # GitLink commits 接口每页硬上限 + + +class GitLinkError(RuntimeError): + """API 调用中不可恢复的错误。""" + + +class GitLinkClient: + """GitLink 公开数据接口客户端。 + + 带可选文件缓存:同一资源重复读取不重复打网,对平台友好。 + """ + + def __init__(self, base: str = API_BASE, token: str | None = None, + timeout: int = DEFAULT_TIMEOUT, cache_dir: Path | None = None) -> None: + self.base = base.rstrip("/") + self.token = token + self.timeout = timeout + self.cache_dir = cache_dir + if self.cache_dir: + self.cache_dir.mkdir(parents=True, exist_ok=True) + + # ------------------------------------------------------------------ + # 底层请求 + # ------------------------------------------------------------------ + def _cache_path(self, url: str) -> Path | None: + if not self.cache_dir: + return None + safe = urllib.parse.quote(url, safe="") + return self.cache_dir / f"{safe}.json" + + def get(self, path: str, query: dict[str, Any] | None = None) -> Any: + """GET 请求,返回解析后的 JSON(dict/list)或 None。""" + url = f"{self.base}/{path.lstrip('/')}" + if query: + url = f"{url}?{urllib.parse.urlencode(query)}" + + cache_path = self._cache_path(url) + if cache_path and cache_path.exists(): + return json.loads(cache_path.read_text(encoding="utf-8")) + + headers = {"Accept": "application/json", "User-Agent": USER_AGENT} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + + req = urllib.request.Request(url, headers=headers) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + raw = resp.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + raise GitLinkError(f"HTTP {exc.code}: {url}") from exc + except urllib.error.URLError as exc: + raise GitLinkError(f"网络错误: {url} -> {exc.reason}") from exc + + text = raw.strip() + if not text or text in ("null", "{}", "[]"): + data: Any = None + elif text[0] in "{[": + try: + data = json.loads(text) + except json.JSONDecodeError as exc: + raise GitLinkError(f"响应非 JSON: {url}") from exc + else: + raise GitLinkError(f"响应非 JSON(可能是 HTML): {url}") + + if cache_path is not None: + cache_path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + return data + + # ------------------------------------------------------------------ + # 资源访问(高层封装) + # ------------------------------------------------------------------ + def repo_info(self, owner: str, repo: str) -> dict[str, Any]: + """仓库元信息。""" + data = self.get(f"{owner}/{repo}.json") + return data if isinstance(data, dict) else {} + + def issues(self, owner: str, repo: str, limit: int = 50, + page: int = 1) -> list[dict[str, Any]]: + """Issue 列表。""" + data = self.get(f"{owner}/{repo}/issues.json", {"page": page, "limit": limit}) + return _extract_list(data, ("issues",)) + + def issue_detail(self, owner: str, repo: str, number: int) -> dict[str, Any]: + """单个 Issue 详情(含完整字段)。""" + data = self.get(f"{owner}/{repo}/issues/{number}.json") + return data if isinstance(data, dict) else {} + + def pulls(self, owner: str, repo: str, limit: int = 50, + page: int = 1) -> list[dict[str, Any]]: + """PR 列表。""" + data = self.get(f"{owner}/{repo}/pulls.json", {"page": page, "limit": limit}) + return _extract_list(data, ("issues", "pulls")) + + def contributors(self, owner: str, repo: str) -> list[dict[str, Any]]: + """贡献者列表。""" + data = self.get(f"{owner}/{repo}/contributors.json") + return _extract_list(data, ("list",)) + + def commits(self, owner: str, repo: str, max_pages: int = 4) -> list[dict[str, Any]]: + """提交列表(按需翻页,每页 50 条,以 total_count 为终止依据)。""" + out: list[dict[str, Any]] = [] + total: int | None = None + for page in range(1, max(1, max_pages) + 1): + data = self.get(f"{owner}/{repo}/commits.json", + {"page": page, "limit": COMMIT_PAGE_SIZE}) + if total is None and isinstance(data, dict): + total = _safe_int(data.get("total_count")) or None + page_items = _extract_list(data, ("commits",)) + if not page_items: + break + out.extend(page_items) + if total is not None and len(out) >= total: + break + return out + + def list_dir(self, owner: str, repo: str, path: str = "", + ref: str = "master") -> list[dict[str, Any]]: + """列出目录下的条目(文件与子目录)。 + + 返回的每个 entry 含 name / path / type(file|dir) / sha / size, + 文件类型的 entry 还可能直接带明文 content。 + """ + data = self.get(f"{owner}/{repo}/sub_entries.json", + {"filepath": path, "ref": ref}) + # 查询目录时 entries 为 list;查询单文件时 entries 为单个 dict。 + # 统一归一化为 list,便于下游处理。 + if isinstance(data, dict): + entries = data.get("entries") + if isinstance(entries, dict): + return [entries] + if isinstance(entries, list): + return entries + return _extract_list(data, ("entries",)) + + def file_content(self, owner: str, repo: str, filepath: str, + ref: str = "master") -> str | None: + """读取单个文件的文本内容。 + + GitLink 的 sub_entries 接口对单文件查询会在 entries 中返回明文 content, + 据此取出。文件不存在或无内容时返回 None。 + """ + entries = self.list_dir(owner, repo, filepath, ref) + target = filepath.rsplit("/", 1)[-1] + for entry in entries: + if entry.get("type") == "file" and entry.get("name") == target: + content = entry.get("content") + if isinstance(content, str): + return content + # 回退:部分情况下单文件查询 entries 仅一项 + if len(entries) == 1 and entries[0].get("type") == "file": + content = entries[0].get("content") + if isinstance(content, str): + return content + return None + + def readme(self, owner: str, repo: str, ref: str = "master") -> str | None: + """读取仓库 README(自动 base64 解码)。""" + data = self.get(f"{owner}/{repo}/readme.json", {"ref": ref}) + if not isinstance(data, dict): + return None + content = data.get("content") + if not isinstance(content, str): + return None + # 注意:GitLink 的 readme.json 虽然 encoding 标为 base64, + # 实测 content 多为明文 Markdown。先探测明文特征,命中则直接返回; + # 否则再尝试 base64 解码。 + stripped = content.lstrip() + if stripped.startswith(("#", "<", "[", "-", "*", "本", "这", "项")) or "\n" in content[:200]: + return content + try: + raw = base64.b64decode(content.encode("ascii", "ignore")) + decoded = raw.decode("utf-8", errors="replace") + # 解码结果若不像文本(大量替换符),回退为原文 + if decoded.count("\ufffd") > len(decoded) * 0.1: + return content + return decoded + except (ValueError, TypeError): + return content + + +# ---------------------------------------------------------------------------- +# 辅助 +# ---------------------------------------------------------------------------- + +def _extract_list(payload: Any, keys: tuple[str, ...]) -> list[Any]: + """从可能嵌套的响应中提取第一个匹配键的列表。""" + if isinstance(payload, list): + return payload + if isinstance(payload, dict): + for key in keys: + value = payload.get(key) + if isinstance(value, list): + return value + return [] + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def split_owner_repo(slug: str) -> tuple[str, str]: + """把 'owner/repo' 或完整 URL 解析为 (owner, repo)。""" + s = slug.strip() + if s.startswith("http"): + parts = urllib.parse.urlparse(s).path.strip("/").split("/") + if len(parts) >= 2: + return parts[0], parts[1].replace(".git", "") + raise GitLinkError(f"无法从 URL 解析 owner/repo: {slug}") + if "/" in s: + owner, repo = s.split("/", 1) + return owner, repo.replace(".git", "") + raise GitLinkError(f"格式应为 owner/repo: {slug}") diff --git a/skills/gitlink-metrics/scripts/metrics.py b/skills/gitlink-metrics/scripts/metrics.py new file mode 100644 index 0000000..252a64f --- /dev/null +++ b/skills/gitlink-metrics/scripts/metrics.py @@ -0,0 +1,216 @@ +"""gitlink-metrics:仓库量化指标看板。 + +把仓库的协作数据转化为一组量化指标,生成数据驱动的运营看板: +- 规模指标:Issue/PR/贡献者/版本数 +- PR 合并率与开放占比 +- 提交活跃趋势(按月) +- 贡献集中度:基尼系数、CR3/CR5、巴士因子 +- 多维评分卡(0-100):活跃度 / 协作 / 社区 / 可维护性 + +相比 gitlink-insight(偏定性的健康度与周报),本技能聚焦**量化指标与评分卡**, +用可计算、可对比的数字刻画仓库状态,功能更偏数据分析。 + +数据来自 GitLink 公开 API(只读),无需登录。 + +用法: + python metrics.py --owner Gitlink --repo gitlink-cli + python metrics.py --owner Gitlink --repo gitlink-cli --format json +""" + +from __future__ import annotations + +import argparse +import json +import sys +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from glapi import GitLinkClient, GitLinkError, split_owner_repo + +if hasattr(sys.stdout, "reconfigure"): + try: + sys.stdout.reconfigure(encoding="utf-8") + except Exception: + pass + + +def gini(values: list[float]) -> float: + xs = sorted(v for v in values if v >= 0) + n = len(xs) + total = sum(xs) + if n == 0 or total == 0: + return 0.0 + cum = sum(i * x for i, x in enumerate(xs, start=1)) + return round((2 * cum) / (n * total) - (n + 1) / n, 3) + + +def bus_factor(values: list[int]) -> int: + total = sum(values) + if total <= 0: + return 0 + cum = 0 + for i, v in enumerate(sorted(values, reverse=True), start=1): + cum += v + if cum / total >= 0.5: + return i + return len(values) + + +def monthly_commits(commits: list[dict[str, Any]]) -> dict[str, int]: + months: Counter[str] = Counter() + for c in commits: + ts = c.get("timestamp") + try: + dt = datetime.fromtimestamp(int(ts), tz=timezone.utc) + months[f"{dt.year:04d}-{dt.month:02d}"] += 1 + except (TypeError, ValueError, OSError): + continue + return dict(sorted(months.items())) + + +def pr_metrics(pulls: list[dict[str, Any]]) -> dict[str, Any]: + merged = sum(1 for p in pulls if p.get("pull_request_status") == 1) + closed = sum(1 for p in pulls if p.get("pull_request_status") == 2) + opened = sum(1 for p in pulls if p.get("pull_request_status") in (0, None)) + total = len(pulls) + return { + "total": total, "merged": merged, "closed": closed, "open": opened, + "merge_rate": round(merged / total * 100, 1) if total else 0.0, + } + + +def score_card(info, prm, contribs_vals, months) -> dict[str, int]: + """多维评分卡(各 0-100)。""" + # 活跃度:近月提交量 + recent = list(months.values())[-3:] if months else [] + avg_recent = sum(recent) / len(recent) if recent else 0 + activity = min(100, int(avg_recent * 2)) + # 协作:PR 合并率 + collab = int(prm["merge_rate"]) + # 社区:贡献者数 + star/fork + community = min(100, len(contribs_vals) * 4 + int(info.get("praises_count") or 0) + + int(info.get("forked_count") or 0)) + # 可维护性:巴士因子越高越好(封顶) + bf = bus_factor(contribs_vals) + maintain = min(100, bf * 20) + overall = round((activity + collab + community + maintain) / 4) + return {"活跃度": activity, "协作": collab, "社区": community, + "可维护性": maintain, "综合": overall} + + +def analyze(owner: str, repo: str, client: GitLinkClient | None = None, + commit_pages: int = 6) -> dict[str, Any]: + client = client or GitLinkClient() + info = client.repo_info(owner, repo) + pulls = client.pulls(owner, repo, limit=50) + commits = client.commits(owner, repo, max_pages=commit_pages) + contributors = client.contributors(owner, repo) + contribs_vals = [int(c.get("contributions") or 0) for c in contributors] + months = monthly_commits(commits) + prm = pr_metrics(pulls) + return { + "owner": owner, "repo": repo, + "scale": { + "issues": info.get("issues_count"), + "pulls": info.get("pull_requests_count"), + "contributors": len(contributors), + "releases": info.get("version_releases_count"), + "stars": info.get("praises_count"), + "forks": info.get("forked_count"), + }, + "pr_metrics": prm, + "monthly_commits": months, + "concentration": { + "gini": gini([float(v) for v in contribs_vals]), + "cr3": round(sum(sorted(contribs_vals, reverse=True)[:3]) / (sum(contribs_vals) or 1) * 100, 1), + "cr5": round(sum(sorted(contribs_vals, reverse=True)[:5]) / (sum(contribs_vals) or 1) * 100, 1), + "bus_factor": bus_factor(contribs_vals), + }, + "score_card": score_card(info, prm, contribs_vals, months), + } + + +def _bar(n: int, peak: int, width: int = 18) -> str: + filled = int(round(n / peak * width)) if peak else 0 + return "█" * filled + "·" * (width - filled) + + +def render_dashboard(m: dict[str, Any], owner: str, repo: str) -> str: + sc = m["scale"]; prm = m["pr_metrics"]; con = m["concentration"]; card = m["score_card"] + lines = [ + f"# 仓库指标看板 — {owner}/{repo}", + "", + f"综合评分:**{card['综合']}/100**", + "", + "## 一、规模指标", + "", + f"| Issue | PR | 贡献者 | 版本 | Star | Fork |", + f"|:--:|:--:|:--:|:--:|:--:|:--:|", + f"| {sc['issues']} | {sc['pulls']} | {sc['contributors']} | {sc['releases']} | {sc['stars']} | {sc['forks']} |", + "", + "## 二、PR 指标", + "", + f"- 合并率:**{prm['merge_rate']}%**(合并 {prm['merged']} / 开放 {prm['open']} / 关闭 {prm['closed']},共 {prm['total']})", + "", + "## 三、提交活跃趋势(按月)", + "", + ] + months = m["monthly_commits"] + if months: + peak = max(months.values()) + for mo, n in months.items(): + lines.append(f"- `{mo}` {_bar(n, peak)} {n}") + else: + lines.append("- (无可定位时间的提交)") + lines += [ + "", "## 四、贡献集中度", "", + f"- 基尼系数:{con['gini']}(0=均衡,1=集中)", + f"- CR3 / CR5:{con['cr3']}% / {con['cr5']}%", + f"- 巴士因子:{con['bus_factor']}(约需多少人离开会影响过半知识延续)", + "", "## 五、多维评分卡", "", + "| 维度 | 评分 |", "|------|:----:|", + ] + for k in ("活跃度", "协作", "社区", "可维护性", "综合"): + lines.append(f"| {k} | {card[k]}/100 |") + lines += ["", "---", "", "由 gitlink-metrics 生成。所有指标基于 GitLink 公开数据,分析只读。"] + return "\n".join(lines) + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(prog="gitlink-metrics", description="仓库量化指标看板") + p.add_argument("--owner"); p.add_argument("--repo"); p.add_argument("--slug") + p.add_argument("--commit-pages", type=int, default=6) + p.add_argument("--format", choices=["markdown", "json"], default="markdown") + p.add_argument("--output", type=Path) + args = p.parse_args(argv) + + if args.slug: + owner, repo = split_owner_repo(args.slug) + elif args.owner and args.repo: + owner, repo = args.owner, args.repo + else: + print("错误:请用 --owner/--repo 或 --slug 指定仓库。", file=sys.stderr) + return 2 + + try: + m = analyze(owner, repo, commit_pages=args.commit_pages) + except GitLinkError as exc: + print(f"采集失败:{exc}", file=sys.stderr) + return 1 + + out = (json.dumps(m, ensure_ascii=False, indent=2) if args.format == "json" + else render_dashboard(m, owner, repo)) + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(out, encoding="utf-8") + print(f"已写入 {args.output}") + else: + print(out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/gitlink-metrics/tests/test_metrics.py b/skills/gitlink-metrics/tests/test_metrics.py new file mode 100644 index 0000000..d1f4298 --- /dev/null +++ b/skills/gitlink-metrics/tests/test_metrics.py @@ -0,0 +1,96 @@ +"""gitlink-metrics 单元测试。""" +from __future__ import annotations +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) +import pytest +from metrics import gini, bus_factor, monthly_commits, pr_metrics, score_card, analyze, render_dashboard + + +class TestGini: + def test_equal(self): + assert gini([10, 10, 10]) == pytest.approx(0.0, abs=0.01) + + def test_concentrated(self): + assert gini([100, 1, 1]) > 0.5 + + def test_empty(self): + assert gini([]) == 0.0 + + +class TestBusFactor: + def test_single_dominant(self): + assert bus_factor([60, 20, 20]) == 1 + + def test_even(self): + assert bus_factor([25, 25, 25, 25]) == 2 + + def test_empty(self): + assert bus_factor([]) == 0 + + +class TestMonthlyCommits: + def test_groups_by_month(self): + import datetime + ts1 = int(datetime.datetime(2026, 1, 15, tzinfo=datetime.timezone.utc).timestamp()) + ts2 = int(datetime.datetime(2026, 1, 20, tzinfo=datetime.timezone.utc).timestamp()) + ts3 = int(datetime.datetime(2026, 2, 1, tzinfo=datetime.timezone.utc).timestamp()) + m = monthly_commits([{"timestamp": ts1}, {"timestamp": ts2}, {"timestamp": ts3}]) + assert m["2026-01"] == 2 and m["2026-02"] == 1 + + def test_invalid_ts(self): + assert monthly_commits([{"timestamp": None}]) == {} + + +class TestPRMetrics: + def test_rates(self): + pulls = [{"pull_request_status": 1}, {"pull_request_status": 1}, + {"pull_request_status": 0}, {"pull_request_status": 2}] + r = pr_metrics(pulls) + assert r["merged"] == 2 and r["open"] == 1 and r["closed"] == 1 + assert r["merge_rate"] == 50.0 + + def test_empty(self): + assert pr_metrics([])["merge_rate"] == 0.0 + + +class TestScoreCard: + def test_all_dims(self): + info = {"praises_count": 5, "forked_count": 10} + prm = {"merge_rate": 80} + card = score_card(info, prm, [50, 30, 20], {"2026-05": 40, "2026-06": 30}) + for k in ("活跃度", "协作", "社区", "可维护性", "综合"): + assert 0 <= card[k] <= 100 + assert card["协作"] == 80 + + +class TestAnalyze: + class FakeClient: + def repo_info(self, o, r): + return {"issues_count": 10, "pull_requests_count": 5, "version_releases_count": 2, + "praises_count": 3, "forked_count": 4} + def pulls(self, o, r, limit=50): + return [{"pull_request_status": 1}, {"pull_request_status": 0}] + def commits(self, o, r, max_pages=6): + import datetime + ts = int(datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc).timestamp()) + return [{"timestamp": ts}] + def contributors(self, o, r): + return [{"login": "a", "contributions": 80}, {"login": "b", "contributions": 20}] + + def test_full(self): + m = analyze("o", "r", client=self.FakeClient()) + assert m["scale"]["contributors"] == 2 + assert m["pr_metrics"]["merge_rate"] == 50.0 + assert "综合" in m["score_card"] + + def test_render(self): + m = analyze("o", "r", client=self.FakeClient()) + md = render_dashboard(m, "o", "r") + assert "仓库指标看板" in md + assert "多维评分卡" in md + assert "巴士因子" in md + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"]))