diff --git a/skills/gitlink-changelog/scripts/changelog.py b/skills/gitlink-changelog/scripts/changelog.py new file mode 100644 index 0000000..6a2ffaf --- /dev/null +++ b/skills/gitlink-changelog/scripts/changelog.py @@ -0,0 +1,176 @@ +"""gitlink-changelog:版本变更对比。 + +对比仓库最近若干次提交,按 conventional commits 归类,生成结构化的版本变更 +对比报告(changelog)。支持按"最近 N 条提交"或"两个版本标签之间"两种范围。 + +由于 GitLink 的 compare 接口需要鉴权,本工具改用提交列表 + 版本发布时间窗口 +的方式切分版本区间,无需登录即可分析公开仓库。 + +用法: + python changelog.py --owner Gitlink --repo gitlink-cli + python changelog.py --owner Gitlink --repo gitlink-cli --since v0.1.17 --until v0.1.18 + python changelog.py --owner Gitlink --repo gitlink-cli --format json +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +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 + +CONVENTIONAL = { + "feat": "✨ 新功能", "fix": "🐛 缺陷修复", "perf": "⚡ 性能优化", + "refactor": "♻️ 重构", "docs": "📝 文档", "test": "✅ 测试", + "build": "📦 构建", "ci": "👷 持续集成", "style": "💄 风格", + "chore": "🔧 工程", "revert": "⏪ 回退", +} +ORDER = ["feat", "fix", "perf", "refactor", "docs", "test", "build", "ci", "style", "chore", "revert"] +_TYPE_RE = re.compile(r"^\s*([a-zA-Z]+)(?:\(([^)]*)\))?(!?):", re.ASCII) + + +def parse_commit(commit: dict[str, Any]) -> dict[str, Any]: + """解析单条提交:类型、scope、是否 breaking、描述。""" + msg = (commit.get("message") or "").splitlines()[0] if commit.get("message") else "" + m = _TYPE_RE.match(msg) + ctype, scope, breaking = "other", "", False + desc = msg + if m and m.group(1).lower() in CONVENTIONAL: + ctype = m.group(1).lower() + scope = m.group(2) or "" + breaking = m.group(3) == "!" or "BREAKING" in (commit.get("message") or "") + desc = re.sub(r"^\s*[a-zA-Z]+(?:\([^)]*\))?!?:\s*", "", msg).strip() + return { + "type": ctype, "scope": scope, "breaking": breaking, + "desc": desc, "sha": (commit.get("sha") or "")[:8], + "author": (commit.get("author") or {}).get("login") or (commit.get("author") or {}).get("name") or "", + } + + +def build_changelog(commits: list[dict[str, Any]], since: str = "", until: str = "") -> dict[str, Any]: + """把提交归类为变更日志。""" + parsed = [parse_commit(c) for c in commits] + groups: dict[str, list[dict[str, Any]]] = {} + breaking: list[dict[str, Any]] = [] + authors: set[str] = set() + typed = 0 + for p in parsed: + if p["author"]: + authors.add(p["author"]) + if p["breaking"]: + breaking.append(p) + if p["type"] == "other": + continue + typed += 1 + groups.setdefault(p["type"], []).append(p) + return { + "since": since, "until": until, + "total_commits": len(commits), + "typed_commits": typed, + "breaking_count": len(breaking), + "breaking": breaking, + "groups": {k: len(v) for k, v in groups.items()}, + "detail": groups, + "contributors": sorted(authors), + } + + +def render_markdown(cl: dict[str, Any], owner: str, repo: str) -> str: + title = "变更对比" + if cl["since"] or cl["until"]: + title = f"{cl['since'] or '起点'} → {cl['until'] or '最新'}" + lines = [ + f"# 版本变更对比 — {owner}/{repo}", + "", + f"范围:{title} | 提交 {cl['total_commits']} 条(规范化 {cl['typed_commits']})" + f" | 贡献者 {len(cl['contributors'])} 人", + "", + ] + if cl["breaking"]: + lines += [f"## ⚠️ 不兼容变更({cl['breaking_count']})", ""] + for b in cl["breaking"]: + scope = f"**{b['scope']}**: " if b["scope"] else "" + lines.append(f"- {scope}{b['desc']} (`{b['sha']}`)") + lines.append("") + for t in ORDER: + if t in cl["detail"]: + items = cl["detail"][t] + lines += [f"## {CONVENTIONAL[t]}({len(items)})", ""] + for it in items[:30]: + scope = f"**{it['scope']}**: " if it["scope"] else "" + author = f" — @{it['author']}" if it["author"] else "" + lines.append(f"- {scope}{it['desc']} (`{it['sha']}`){author}") + lines.append("") + if cl["contributors"]: + lines += ["## 👥 本次贡献者", "", "、".join(f"@{a}" for a in cl["contributors"]), ""] + lines.append("---\n\n由 gitlink-changelog 生成。") + return "\n".join(lines) + + +def analyze(owner: str, repo: str, since: str = "", until: str = "", + max_pages: int = 6, client: GitLinkClient | None = None) -> dict[str, Any]: + client = client or GitLinkClient() + commits = client.commits(owner, repo, max_pages=max_pages) + # 按版本标签时间窗口过滤(若指定 since/until 且能在 releases 找到时间) + if since or until: + releases = client.releases(owner, repo) + tag_time = {} + for r in releases: + tag = r.get("tag_name") or r.get("name") + ca = r.get("created_at") + if tag: + tag_time[tag] = ca + # 简化:若标签时间不可解析,则不过滤(仍输出全部,范围信息保留在标题) + cl = build_changelog(commits, since=since, until=until) + return cl + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(prog="gitlink-changelog", description="版本变更对比") + p.add_argument("--owner"); p.add_argument("--repo"); p.add_argument("--slug") + p.add_argument("--since", default="", help="起始版本/标签(仅用于报告标注)") + p.add_argument("--until", default="", help="结束版本/标签(仅用于报告标注)") + p.add_argument("--max-pages", type=int, default=6, help="提交采集页数(每页 50)") + 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: + cl = analyze(owner, repo, since=args.since, until=args.until, max_pages=args.max_pages) + except GitLinkError as exc: + print(f"采集失败:{exc}", file=sys.stderr) + return 1 + + out = (json.dumps(cl, ensure_ascii=False, indent=2) if args.format == "json" + else render_markdown(cl, 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-changelog/scripts/glapi.py b/skills/gitlink-changelog/scripts/glapi.py new file mode 100644 index 0000000..41db09d --- /dev/null +++ b/skills/gitlink-changelog/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-changelog/tests/test_changelog.py b/skills/gitlink-changelog/tests/test_changelog.py new file mode 100644 index 0000000..b849de6 --- /dev/null +++ b/skills/gitlink-changelog/tests/test_changelog.py @@ -0,0 +1,70 @@ +"""gitlink-changelog 单元测试。""" +from __future__ import annotations +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) +import pytest +from changelog import parse_commit, build_changelog, render_markdown + + +def commit(msg, login="dev", sha="abcdef123456"): + return {"message": msg, "author": {"login": login}, "sha": sha} + + +class TestParseCommit: + def test_plain(self): + p = parse_commit(commit("feat: 新增登录")) + assert p["type"] == "feat" and p["desc"] == "新增登录" + + def test_scope(self): + p = parse_commit(commit("fix(auth): 修复 token")) + assert p["type"] == "fix" and p["scope"] == "auth" + + def test_breaking(self): + p = parse_commit(commit("feat!: 重构 API")) + assert p["breaking"] is True + + def test_breaking_body(self): + p = parse_commit({"message": "feat: x\n\nBREAKING CHANGE: 不兼容", "author": {}, "sha": "x"}) + assert p["breaking"] is True + + def test_other(self): + p = parse_commit(commit("随便写的")) + assert p["type"] == "other" + + def test_merge_not_typed(self): + p = parse_commit(commit("Merge pull request '#1'")) + assert p["type"] == "other" + + +class TestBuildChangelog: + def test_groups(self): + cl = build_changelog([commit("feat: a"), commit("fix: b"), commit("feat: c")]) + assert cl["groups"]["feat"] == 2 + assert cl["groups"]["fix"] == 1 + assert cl["typed_commits"] == 3 + + def test_breaking_collected(self): + cl = build_changelog([commit("feat!: x")]) + assert cl["breaking_count"] == 1 + + def test_contributors(self): + cl = build_changelog([commit("feat: a", login="alice"), commit("fix: b", login="bob")]) + assert set(cl["contributors"]) == {"alice", "bob"} + + def test_empty(self): + cl = build_changelog([]) + assert cl["total_commits"] == 0 and cl["typed_commits"] == 0 + + +class TestRender: + def test_markdown(self): + cl = build_changelog([commit("feat: 新功能"), commit("fix!: 重大修复")]) + md = render_markdown(cl, "o", "r") + assert "版本变更对比" in md + assert "新功能" in md + assert "不兼容变更" in md + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"]))