diff --git a/skills/gitlink-stale/scripts/glapi.py b/skills/gitlink-stale/scripts/glapi.py new file mode 100644 index 0000000..41db09d --- /dev/null +++ b/skills/gitlink-stale/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-stale/scripts/stale.py b/skills/gitlink-stale/scripts/stale.py new file mode 100644 index 0000000..b4eb0ed --- /dev/null +++ b/skills/gitlink-stale/scripts/stale.py @@ -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 `(写操作,需确认)。", ""] + 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()) diff --git a/skills/gitlink-stale/tests/test_stale.py b/skills/gitlink-stale/tests/test_stale.py new file mode 100644 index 0000000..10fa5dd --- /dev/null +++ b/skills/gitlink-stale/tests/test_stale.py @@ -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"]))