feat(skills): 补充 gitlink-metrics 配套脚本与测试
将 SKILL.md 方式A引用的脚本（scripts/）与单元测试（tests/）纳入 PR，使其自包含可运行。
This commit is contained in:
parent
66793a66ae
commit
64283c7939
|
|
@ -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}")
|
||||
|
|
@ -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())
|
||||
|
|
@ -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"]))
|
||||
Loading…
Reference in New Issue