feat(skills): 补充 gitlink-stale 配套脚本与测试

将 SKILL.md 方式A引用的脚本(scripts/)与单元测试(tests/)纳入 PR,使其自包含可运行。
This commit is contained in:
Ct201314 2026-06-12 18:01:47 +08:00
parent 1ae613717b
commit 8c6a00f407
3 changed files with 521 additions and 0 deletions

View File

@ -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 请求,返回解析后的 JSONdict/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}")

View File

@ -0,0 +1,194 @@
"""gitlink-stale陈旧 Issue/PR 清理。
检测仓库中久未更新的开放 Issue PR按陈旧程度分级活跃/留意/陈旧/僵尸
生成清理建议清单帮助维护者控制积压
数据来自 GitLink 公开 API只读无需登录生成的只是建议清单
是否关闭由维护者通过 gitlink-cli 自行决定
用法
python stale.py --owner Gitlink --repo gitlink-cli
python stale.py --owner Gitlink --repo gitlink-cli --format json
"""
from __future__ import annotations
import argparse
import json
import re
import sys
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 parse_relative_days(text: str) -> int | None:
"""把相对时间(如 "3天前"/"2个月前"/"1年前"/"刚刚")粗略换算成天数。"""
if not text:
return None
t = str(text).strip()
if "" in t or "分钟" in t or "" in t or "小时" in t:
return 0
m = re.search(r"(\d+)\s*天", t)
if m:
return int(m.group(1))
m = re.search(r"(\d+)\s*周", t)
if m:
return int(m.group(1)) * 7
m = re.search(r"(\d+)\s*个?月", t)
if m:
return int(m.group(1)) * 30
m = re.search(r"(\d+)\s*年", t)
if m:
return int(m.group(1)) * 365
return None
# 陈旧度分级阈值(天)
def grade(days: int | None) -> str:
if days is None:
return "未知"
if days <= 14:
return "活跃"
if days <= 45:
return "留意"
if days <= 120:
return "陈旧"
return "僵尸"
GRADE_EMOJI = {"活跃": "🟢", "留意": "🟡", "陈旧": "🟠", "僵尸": "🔴", "未知": ""}
def _age_days(item: dict[str, Any]) -> int | None:
"""优先用 updated_at/created_at 的相对时间估算停滞天数。"""
for key in ("updated_at", "created_at", "time_ago", "pr_time"):
v = item.get(key)
d = parse_relative_days(v) if v else None
if d is not None:
return d
return None
def analyze_stale(issues: list[dict[str, Any]], pulls: list[dict[str, Any]]) -> dict[str, Any]:
"""分析开放 Issue/PR 的陈旧度。"""
def _scan(items, kind):
out = []
for it in items:
# 只看开放项
if kind == "pr" and it.get("pull_request_status") not in (0, None):
continue
if kind == "issue":
status = str(it.get("issue_status") or "")
if "" in status or "closed" in status.lower():
continue
days = _age_days(it)
g = grade(days)
out.append({
"kind": kind,
"id": str(it.get("pull_request_number") or it.get("id") or ""),
"title": str(it.get("name") or it.get("subject") or it.get("title") or "")[:60],
"age_days": days,
"grade": g,
"author": it.get("author_login") or it.get("author_name") or "",
})
return out
records = _scan(issues, "issue") + _scan(pulls, "pr")
by_grade: dict[str, int] = {}
for r in records:
by_grade[r["grade"]] = by_grade.get(r["grade"], 0) + 1
# 需要清理的:陈旧 + 僵尸,按停滞天数降序
cleanup = sorted(
[r for r in records if r["grade"] in ("陈旧", "僵尸")],
key=lambda x: x["age_days"] or 0, reverse=True,
)
return {
"total_open": len(records),
"by_grade": by_grade,
"cleanup_count": len(cleanup),
"cleanup": cleanup,
"records": records,
}
def render_report(result: dict[str, Any], owner: str, repo: str) -> str:
lines = [
f"# 陈旧 Issue/PR 清理报告 — {owner}/{repo}",
"",
f"开放项共 {result['total_open']} 个。陈旧度分布:",
"",
]
for g in ("活跃", "留意", "陈旧", "僵尸", "未知"):
n = result["by_grade"].get(g, 0)
if n:
lines.append(f"- {GRADE_EMOJI[g]} {g}{n}")
lines.append("")
lines.append("> 分级标准:活跃 ≤14 天 / 留意 ≤45 天 / 陈旧 ≤120 天 / 僵尸 >120 天(按最近更新估算)。")
lines.append("")
if result["cleanup"]:
lines += [f"## 建议处理({result['cleanup_count']} 个陈旧/僵尸项)", "",
"| 类型 | 编号 | 标题 | 停滞 | 等级 |", "|:----:|:----:|------|:----:|:----:|"]
for c in result["cleanup"][:30]:
kind = "Issue" if c["kind"] == "issue" else "PR"
age = f"{c['age_days']}" if c["age_days"] is not None else "未知"
lines.append(f"| {kind} | {c['id']} | {c['title']} | {age} | {GRADE_EMOJI[c['grade']]}{c['grade']} |")
lines += ["", "## 建议行动", "",
"1. 对「僵尸」项:评估是否仍有意义,无意义可关闭或加 `wontfix` 标签。",
"2. 对「陈旧」项:@相关人确认进展,或补充信息后重新激活。",
"3. 关闭操作:`gitlink-cli issue +close --number <web序号>`(写操作,需确认)。", ""]
else:
lines += ["🎉 没有发现明显陈旧的开放项,积压控制良好。", ""]
lines.append("---\n\n由 gitlink-stale 生成。分析只读,关闭操作由维护者确认执行。")
return "\n".join(lines)
def analyze(owner: str, repo: str, client: GitLinkClient | None = None) -> dict[str, Any]:
client = client or GitLinkClient()
issues = client.issues(owner, repo, limit=50)
pulls = client.pulls(owner, repo, limit=50)
return analyze_stale(issues, pulls)
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(prog="gitlink-stale", description="陈旧 Issue/PR 清理")
p.add_argument("--owner"); p.add_argument("--repo"); p.add_argument("--slug")
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:
result = analyze(owner, repo)
except GitLinkError as exc:
print(f"采集失败:{exc}", file=sys.stderr)
return 1
out = (json.dumps(result, ensure_ascii=False, indent=2) if args.format == "json"
else render_report(result, 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())

View File

@ -0,0 +1,86 @@
"""gitlink-stale 单元测试。"""
from __future__ import annotations
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
import pytest
from stale import parse_relative_days, grade, analyze_stale, render_report
class TestParseRelativeDays:
def test_days(self):
assert parse_relative_days("3天前") == 3
def test_weeks(self):
assert parse_relative_days("2周前") == 14
def test_months(self):
assert parse_relative_days("2个月前") == 60
def test_years(self):
assert parse_relative_days("1年前") == 365
def test_recent(self):
assert parse_relative_days("刚刚") == 0
assert parse_relative_days("3小时前") == 0
def test_unknown(self):
assert parse_relative_days("") is None
assert parse_relative_days("奇怪的值") is None
class TestGrade:
def test_levels(self):
assert grade(5) == "活跃"
assert grade(30) == "留意"
assert grade(90) == "陈旧"
assert grade(200) == "僵尸"
assert grade(None) == "未知"
class TestAnalyzeStale:
def test_filters_closed(self):
issues = [
{"id": "1", "name": "open old", "updated_at": "200天前", "issue_status": "正在解决"},
{"id": "2", "name": "closed", "updated_at": "300天前", "issue_status": "关闭"},
]
r = analyze_stale(issues, [])
# 关闭的不计入
assert r["total_open"] == 1
assert r["by_grade"]["僵尸"] == 1
def test_pr_open_only(self):
pulls = [
{"pull_request_number": "1", "name": "open pr", "pr_time": "100天前", "pull_request_status": 0},
{"pull_request_number": "2", "name": "merged", "pr_time": "100天前", "pull_request_status": 1},
]
r = analyze_stale([], pulls)
assert r["total_open"] == 1
def test_cleanup_list(self):
issues = [{"id": "1", "name": "zombie", "updated_at": "200天前", "issue_status": "open"}]
r = analyze_stale(issues, [])
assert r["cleanup_count"] == 1
assert r["cleanup"][0]["grade"] == "僵尸"
def test_empty(self):
r = analyze_stale([], [])
assert r["total_open"] == 0 and r["cleanup_count"] == 0
class TestRender:
def test_report(self):
issues = [{"id": "1", "name": "old issue", "updated_at": "150天前", "issue_status": "open"}]
r = analyze_stale(issues, [])
md = render_report(r, "o", "r")
assert "陈旧 Issue/PR 清理报告" in md
assert "建议处理" in md
def test_clean_repo(self):
r = analyze_stale([{"id": "1", "name": "x", "updated_at": "1天前", "issue_status": "open"}], [])
md = render_report(r, "o", "r")
assert "积压控制良好" in md
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))