🐛 fix(release): 修复自动更新日志缺少贡献者署名

- 基于关联 PR 与提交元数据解析外部贡献者账号
- 生成 GitHub 原生用户提及以展示贡献者头像名单
- 增加 API 熔断、版本历史告警与 CI 回归测试
This commit is contained in:
Syngnat
2026-07-19 17:39:36 +08:00
parent c4f24a6245
commit ea9ccb3648
5 changed files with 768 additions and 55 deletions

View File

@@ -0,0 +1,469 @@
#!/usr/bin/env python3
"""Generate categorized GitHub release notes with contributor attribution.
For every non-merge commit in the release range, the generator first resolves
the author of an associated merged pull request. It falls back to GitHub's
commit metadata and then to a GitHub noreply email address. External
contributors are mentioned in the resulting Markdown so GitHub can render its
native contributor names and avatar list on the release page.
"""
from __future__ import annotations
import argparse
import http.client
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence
GITHUB_LOGIN_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?$")
GITHUB_NOREPLY_RE = re.compile(
r"^(?:\d+\+)?(?P<login>[A-Za-z0-9-]+(?:\[bot\])?)@users\.noreply\.github\.com$",
re.IGNORECASE,
)
CONTRIBUTED_BY_RE = re.compile(
r"\(\s*contributed\s+by\s+(?:\*\*)?@[A-Za-z0-9-]+(?:\*\*)?\s*\)",
re.IGNORECASE,
)
@dataclass(frozen=True)
class Commit:
sha: str
subject: str
author_name: str
author_email: str
class GitHubUnavailableError(RuntimeError):
"""Raised after GitHub API retries are exhausted."""
class GitHubClient:
"""Small read-only GitHub REST client used by the release workflow."""
def __init__(self, token: str = "", api_url: str = "https://api.github.com") -> None:
self._token = token.strip()
self._api_url = api_url.rstrip("/")
self._available = True
@property
def available(self) -> bool:
return self._available
def get_json(self, path: str) -> Any:
if not self._available:
raise GitHubUnavailableError("GitHub API disabled after an earlier request failure")
url = f"{self._api_url}/{path.lstrip('/')}"
headers = {
"Accept": "application/vnd.github+json",
"User-Agent": "GoNavi-release-notes",
"X-GitHub-Api-Version": "2022-11-28",
}
if self._token:
headers["Authorization"] = f"Bearer {self._token}"
for attempt in range(3):
request = urllib.request.Request(url, headers=headers)
try:
with urllib.request.urlopen(request, timeout=15) as response:
return json.load(response)
except urllib.error.HTTPError as exc:
retryable = exc.code in {429, 500, 502, 503, 504}
if retryable and attempt < 2:
time.sleep(2**attempt)
continue
if exc.code in {401, 403, 429} or exc.code >= 500:
self._available = False
raise GitHubUnavailableError(
f"GitHub API returned HTTP {exc.code} for {path}"
) from exc
raise
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
self._available = False
raise GitHubUnavailableError(
f"GitHub API returned invalid JSON for {path}"
) from exc
except (OSError, http.client.HTTPException) as exc:
if attempt < 2:
time.sleep(2**attempt)
continue
self._available = False
raise GitHubUnavailableError(
f"GitHub API could not be reached for {path}"
) from exc
self._available = False
raise GitHubUnavailableError(f"GitHub request failed: {path}")
def run_git(*args: str) -> str:
completed = subprocess.run(
["git", *args],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
return completed.stdout
def resolve_previous_tag(tag: str) -> str:
"""Return the preceding release tag by creation date.
Release branches are not guaranteed to have linear ancestry, so the
nearest reachable tag can be much older than the actual previous release.
This keeps the established release ordering while explicitly excluding the
tag currently being published.
"""
output = run_git("tag", "--list", "v*", "--sort=-creatordate")
tags = [candidate for candidate in output.splitlines() if candidate]
current = tag.casefold()
for index, candidate in enumerate(tags):
if candidate.casefold() == current:
return tags[index + 1] if index + 1 < len(tags) else ""
return tags[0] if tags else ""
def warn_if_release_history_diverged(previous_tag: str, tag: str) -> None:
if not previous_tag:
return
try:
run_git("merge-base", "--is-ancestor", previous_tag, tag)
except subprocess.CalledProcessError as exc:
if exc.returncode != 1:
raise
print(
f"warning: {previous_tag} is not an ancestor of {tag}; "
"release notes will include commits reachable only from the new tag",
file=sys.stderr,
)
def read_commits(tag: str, previous_tag: str) -> list[Commit]:
revision_range = f"{previous_tag}..{tag}" if previous_tag else tag
output = run_git(
"log",
revision_range,
"--no-merges",
"-z",
"--pretty=format:%H%x00%s%x00%an%x00%ae",
)
if not output:
return []
fields = output.split("\x00")
if len(fields) % 4 != 0:
raise ValueError("unexpected git log record while generating release notes")
return [Commit(*fields[index : index + 4]) for index in range(0, len(fields), 4)]
def login_from_noreply_email(author_email: str) -> str:
match = GITHUB_NOREPLY_RE.fullmatch(author_email.strip())
return match.group("login") if match else ""
def _login_from_payload(payload: Any) -> str:
if not isinstance(payload, Mapping):
return ""
login = payload.get("login")
return login.strip() if isinstance(login, str) else ""
def resolve_contributor_login(
*,
pulls: Any,
commit_payload: Any,
author_email: str,
repository: str,
) -> str:
"""Resolve attribution, preferring the author of the merged pull request."""
candidates: list[Mapping[str, Any]] = []
if isinstance(pulls, list):
for pull in pulls:
if not isinstance(pull, Mapping) or not pull.get("merged_at"):
continue
base = pull.get("base")
base_repo = base.get("repo") if isinstance(base, Mapping) else None
full_name = base_repo.get("full_name") if isinstance(base_repo, Mapping) else None
if not isinstance(full_name, str) or full_name.casefold() != repository.casefold():
continue
if _login_from_payload(pull.get("user")):
candidates.append(pull)
if candidates:
# Prefer the PR that originally introduced the commit. A newer
# backport or release-branch PR may otherwise steal attribution from
# the original contributor.
selected = min(candidates, key=lambda pull: str(pull.get("merged_at", "")))
return _login_from_payload(selected.get("user"))
if isinstance(commit_payload, Mapping):
login = _login_from_payload(commit_payload.get("author"))
if login:
return login
return login_from_noreply_email(author_email)
def _api_path(repository: str, suffix: str) -> str:
owner, name = repository.split("/", 1)
return f"repos/{urllib.parse.quote(owner, safe='')}/{urllib.parse.quote(name, safe='')}/{suffix}"
def infer_login_from_commit(commit: Commit, repository: str) -> str:
login = login_from_noreply_email(commit.author_email)
if login:
return login
owner = repository.split("/", 1)[0]
if commit.author_name.strip().casefold() == owner.casefold():
return owner
return ""
def fetch_commit_attribution(
*,
client: GitHubClient,
repository: str,
commit: Commit,
) -> str:
encoded_sha = urllib.parse.quote(commit.sha, safe="")
pulls: Any = []
commit_payload: Any = {}
try:
pulls = client.get_json(_api_path(repository, f"commits/{encoded_sha}/pulls"))
except (OSError, ValueError, GitHubUnavailableError) as exc:
print(
f"warning: unable to query associated pull requests for {commit.sha[:12]}: {exc}",
file=sys.stderr,
)
if not client.available:
return infer_login_from_commit(commit, repository)
# A merged PR author is the most accurate attribution. Avoid the second API
# request when it is already available.
login = resolve_contributor_login(
pulls=pulls,
commit_payload={},
author_email="",
repository=repository,
)
if login:
return login
try:
commit_payload = client.get_json(_api_path(repository, f"commits/{encoded_sha}"))
except (OSError, ValueError, GitHubUnavailableError) as exc:
print(
f"warning: unable to query commit author for {commit.sha[:12]}: {exc}",
file=sys.stderr,
)
return resolve_contributor_login(
pulls=pulls,
commit_payload=commit_payload,
author_email=commit.author_email,
repository=repository,
)
def load_attributions(path: Path) -> dict[str, str]:
payload = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise ValueError("attributions file must contain a JSON object")
attributions: dict[str, str] = {}
for sha, login in payload.items():
if isinstance(sha, str) and isinstance(login, str):
attributions[sha] = login.strip()
return attributions
def collect_attributions(
*,
commits: Sequence[Commit],
repository: str,
fixture_path: Path | None,
) -> dict[str, str]:
if fixture_path is not None:
fixture = load_attributions(fixture_path)
return {
commit.sha: fixture.get(commit.sha, "") or login_from_noreply_email(commit.author_email)
for commit in commits
}
token = os.environ.get("GITHUB_TOKEN", "")
if not token:
print(
"warning: GITHUB_TOKEN is not set; GitHub API rate limits may prevent complete attribution",
file=sys.stderr,
)
client = GitHubClient(
token=token,
api_url=os.environ.get("GITHUB_API_URL", "https://api.github.com"),
)
attributions: dict[str, str] = {}
for commit in commits:
if client.available:
attributions[commit.sha] = fetch_commit_attribution(
client=client,
repository=repository,
commit=commit,
)
else:
attributions[commit.sha] = infer_login_from_commit(commit, repository)
return attributions
def is_external_contributor(login: str, repository: str) -> bool:
login = login.strip()
owner = repository.split("/", 1)[0]
if not login or login.casefold() == owner.casefold():
return False
lowered = login.casefold()
if lowered.endswith("[bot]") or lowered in {"github-actions", "github-actions[bot]"}:
return False
return GITHUB_LOGIN_RE.fullmatch(login) is not None
def category_for_subject(subject: str) -> str:
lowered = subject.casefold()
if subject.startswith("") or "feat" in lowered:
return "feature"
if subject.startswith("🐛") or "fix" in lowered:
return "fix"
if subject.startswith("") or "perf" in lowered:
return "performance"
if subject.startswith("♻️") or "refactor" in lowered:
return "refactor"
if subject.startswith("🌐"):
return "i18n"
return "other"
def render_release_notes(
*,
commits: Sequence[Commit],
attributions: Mapping[str, str],
repository: str,
tag: str,
previous_tag: str,
repository_url: str,
) -> str:
sections = [
("feature", "## ✨ 新功能"),
("fix", "## 🐛 问题修复"),
("performance", "## ⚡ 性能优化"),
("refactor", "## ♻️ 重构"),
("i18n", "## 🌐 国际化"),
("other", "## 🔧 其他变更"),
]
categorized: dict[str, list[str]] = {key: [] for key, _ in sections}
for commit in commits:
subject = commit.subject.strip()
if not subject:
continue
login = attributions.get(commit.sha, "").strip()
attribution = ""
if is_external_contributor(login, repository) and not CONTRIBUTED_BY_RE.search(subject):
# A real GitHub @mention makes the release page render its native,
# deduplicated Contributors avatar row; do not duplicate it in HTML.
attribution = f" (contributed by **@{login}**)"
categorized[category_for_subject(subject)].append(f"- {subject}{attribution}")
blocks: list[str] = []
for key, heading in sections:
if categorized[key]:
blocks.append(f"{heading}\n\n" + "\n".join(categorized[key]))
if not blocks:
blocks.append("暂无提交记录。")
if previous_tag:
base_url = repository_url.rstrip("/")
previous_url = urllib.parse.quote(previous_tag, safe="")
tag_url = urllib.parse.quote(tag, safe="")
blocks.append(
"---\n"
f"**完整变更**: [{previous_tag}...{tag}]"
f"({base_url}/compare/{previous_url}...{tag_url})"
)
return "\n\n".join(blocks) + "\n"
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Generate categorized release notes with GitHub contributor mentions"
)
parser.add_argument("--repo", required=True, help="GitHub repository in owner/name form")
parser.add_argument("--tag", required=True, help="Release tag")
parser.add_argument(
"--previous-tag",
default="",
help="Previous tag (default: preceding v* tag by creation date)",
)
parser.add_argument("--repository-url", required=True, help="Repository web URL")
parser.add_argument(
"--attributions-file",
type=Path,
help="Offline JSON mapping of commit SHA to GitHub login (tests/local use)",
)
parser.add_argument("--output", type=Path, required=True, help="Output Markdown file")
args = parser.parse_args(argv)
if not re.fullmatch(r"[^/\s]+/[^/\s]+", args.repo):
parser.error("--repo must use owner/name format")
return args
def main(argv: Sequence[str] | None = None) -> int:
args = parse_args(argv)
previous_tag = args.previous_tag.strip() or resolve_previous_tag(args.tag)
warn_if_release_history_diverged(previous_tag, args.tag)
revision_range = f"{previous_tag}..{args.tag}" if previous_tag else args.tag
commits = read_commits(args.tag, previous_tag)
attributions = collect_attributions(
commits=commits,
repository=args.repo,
fixture_path=args.attributions_file,
)
body = render_release_notes(
commits=commits,
attributions=attributions,
repository=args.repo,
tag=args.tag,
previous_tag=previous_tag,
repository_url=args.repository_url,
)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(body, encoding="utf-8")
contributors = {
login
for login in attributions.values()
if is_external_contributor(login, args.repo)
}
print(
f"wrote {args.output} ({len(commits)} commits, {len(contributors)} external contributors, "
f"range={revision_range})"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -0,0 +1,285 @@
#!/usr/bin/env python3
import importlib.util
import io
import json
import os
import pathlib
import subprocess
import sys
import tempfile
import unittest
from contextlib import redirect_stderr
from unittest import mock
ROOT = pathlib.Path(__file__).resolve().parents[1]
MODULE_PATH = pathlib.Path(__file__).with_name("generate-release-notes.py")
SPEC = importlib.util.spec_from_file_location("generate_release_notes", MODULE_PATH)
MODULE = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
sys.modules[SPEC.name] = MODULE
SPEC.loader.exec_module(MODULE)
class GenerateReleaseNotesTests(unittest.TestCase):
def test_resolves_previous_release_tag_even_when_history_is_not_linear(self) -> None:
with mock.patch.object(
MODULE,
"run_git",
return_value="v1.2.0\nv1.1.0\nv1.0.0\nv0.9.0\n",
):
self.assertEqual(MODULE.resolve_previous_tag("v1.1.0"), "v1.0.0")
def test_warns_when_release_tags_have_diverged(self) -> None:
error = subprocess.CalledProcessError(1, ["git", "merge-base"])
stderr = io.StringIO()
with mock.patch.object(MODULE, "run_git", side_effect=error), redirect_stderr(stderr):
MODULE.warn_if_release_history_diverged("v1.0.0", "v1.1.0")
self.assertIn("v1.0.0 is not an ancestor of v1.1.0", stderr.getvalue())
def test_renders_external_contributor_mentions_without_tagging_owner_or_bots(self) -> None:
external_sha = "a" * 40
owner_sha = "b" * 40
bot_sha = "c" * 40
body = MODULE.render_release_notes(
commits=[
MODULE.Commit(external_sha, "✨ feat(export): add contributor notes", "Outside", "outside@example.test"),
MODULE.Commit(owner_sha, "🐛 fix(release): keep owner entry", "Syngnat", "owner@example.test"),
MODULE.Commit(bot_sha, "🔧 chore(deps): update action", "Bot", "bot@example.test"),
],
attributions={
external_sha: "OutsideUser",
owner_sha: "Syngnat",
bot_sha: "dependabot[bot]",
},
repository="Syngnat/GoNavi",
tag="v1.1.0",
previous_tag="v1.0.0",
repository_url="https://github.com/Syngnat/GoNavi",
)
self.assertIn(
"- ✨ feat(export): add contributor notes (contributed by **@OutsideUser**)",
body,
)
self.assertIn("- 🐛 fix(release): keep owner entry", body)
self.assertNotIn("keep owner entry (contributed by", body)
self.assertNotIn("dependabot[bot]", body)
self.assertEqual(body.count("**@OutsideUser**"), 1)
def test_prefers_merged_pull_request_author_for_commit_attribution(self) -> None:
pulls = [
{
"number": 600,
"merged_at": "2026-07-17T13:49:59Z",
"user": {"login": "OutsideUser"},
"base": {"repo": {"full_name": "Syngnat/GoNavi"}},
},
{
"number": 666,
"merged_at": "2026-07-18T13:49:59Z",
"user": {"login": "Syngnat"},
"base": {"repo": {"full_name": "Syngnat/GoNavi"}},
}
]
commit_payload = {"author": {"login": "Syngnat"}}
login = MODULE.resolve_contributor_login(
pulls=pulls,
commit_payload=commit_payload,
author_email="owner@example.test",
repository="Syngnat/GoNavi",
)
self.assertEqual(login, "OutsideUser")
def test_api_outage_uses_noreply_fallback_and_opens_the_circuit(self) -> None:
class UnavailableClient:
available = True
calls = 0
def get_json(self, _path: str):
self.calls += 1
self.available = False
raise MODULE.GitHubUnavailableError("offline")
client = UnavailableClient()
commit = MODULE.Commit(
"d" * 40,
"✨ feat(release): external contribution (#666)",
"Outside User",
"123+OutsideUser@users.noreply.github.com",
)
with redirect_stderr(io.StringIO()):
login = MODULE.fetch_commit_attribution(
client=client,
repository="Syngnat/GoNavi",
commit=commit,
)
self.assertEqual(login, "OutsideUser")
self.assertEqual(client.calls, 1)
def test_commit_api_login_wins_over_historical_noreply_login(self) -> None:
class RecordingClient:
available = True
paths: list[str] = []
def get_json(self, path: str):
self.paths.append(path)
if path.endswith("/pulls"):
return []
return {"author": {"login": "CurrentLogin"}}
client = RecordingClient()
commit = MODULE.Commit(
"f" * 40,
"🐛 fix(release): preserve current login",
"Old Login",
"123+OldLogin@users.noreply.github.com",
)
login = MODULE.fetch_commit_attribution(
client=client,
repository="Syngnat/GoNavi",
commit=commit,
)
self.assertEqual(login, "CurrentLogin")
self.assertEqual(len(client.paths), 2)
def test_real_client_opens_circuit_for_incomplete_http_response(self) -> None:
client = MODULE.GitHubClient(token="test-token")
incomplete = MODULE.http.client.IncompleteRead(b"", 1)
with mock.patch.object(
MODULE.urllib.request,
"urlopen",
side_effect=incomplete,
) as urlopen, mock.patch.object(MODULE.time, "sleep"):
with self.assertRaises(MODULE.GitHubUnavailableError):
client.get_json("repos/Syngnat/GoNavi/commits/deadbeef")
with self.assertRaises(MODULE.GitHubUnavailableError):
client.get_json("repos/Syngnat/GoNavi/commits/deadbeef")
self.assertFalse(client.available)
self.assertEqual(urlopen.call_count, 3)
def test_git_log_parser_allows_control_separators_in_subject(self) -> None:
sha = "e" * 40
output = f"{sha}\0✨ feat: keep \x1e and \x1f\0Outside User\0outside@example.test"
with mock.patch.object(MODULE, "run_git", return_value=output):
commits = MODULE.read_commits("v1.1.0", "v1.0.0")
self.assertEqual(len(commits), 1)
self.assertEqual(commits[0].subject, "✨ feat: keep \x1e and \x1f")
def test_cli_generates_notes_from_git_range_and_offline_attribution_fixture(self) -> None:
with tempfile.TemporaryDirectory(prefix="gonavi-release-notes-") as tmp:
repo = pathlib.Path(tmp)
self.run_git(repo, "init")
self.run_git(repo, "config", "user.name", "Syngnat")
self.run_git(repo, "config", "user.email", "owner@example.test")
(repo / "fixture.txt").write_text("baseline\n", encoding="utf-8")
self.run_git(repo, "add", "fixture.txt")
self.run_git(repo, "commit", "-m", "🔧 chore: baseline")
self.run_git(repo, "tag", "v1.0.0")
(repo / "fixture.txt").write_text("baseline\nowner\n", encoding="utf-8")
self.run_git(repo, "add", "fixture.txt")
self.run_git(repo, "commit", "-m", "🐛 fix(release): owner fix")
owner_sha = self.run_git(repo, "rev-parse", "HEAD").strip()
(repo / "fixture.txt").write_text("baseline\nowner\nexternal\n", encoding="utf-8")
self.run_git(repo, "add", "fixture.txt")
contributor_env = {
**os.environ,
"GIT_AUTHOR_NAME": "Outside User",
"GIT_AUTHOR_EMAIL": "123+OutsideUser@users.noreply.github.com",
"GIT_COMMITTER_NAME": "Outside User",
"GIT_COMMITTER_EMAIL": "123+OutsideUser@users.noreply.github.com",
}
self.run_git(repo, "commit", "-m", "✨ feat(release): external feature (#666)", env=contributor_env)
external_sha = self.run_git(repo, "rev-parse", "HEAD").strip()
self.run_git(repo, "tag", "v1.1.0")
attributions = repo / "attributions.json"
attributions.write_text(
json.dumps({owner_sha: "Syngnat", external_sha: "OutsideUser"}),
encoding="utf-8",
)
output = repo / "release-notes.md"
subprocess.check_call(
[
sys.executable,
str(MODULE_PATH),
"--repo",
"Syngnat/GoNavi",
"--tag",
"v1.1.0",
"--previous-tag",
"v1.0.0",
"--repository-url",
"https://github.com/Syngnat/GoNavi",
"--attributions-file",
str(attributions),
"--output",
str(output),
],
cwd=repo,
env={key: value for key, value in os.environ.items() if key != "GITHUB_TOKEN"},
)
body = output.read_text(encoding="utf-8")
self.assertIn("## ✨ 新功能", body)
self.assertIn("external feature (#666) (contributed by **@OutsideUser**)", body)
self.assertIn("## 🐛 问题修复", body)
self.assertNotIn("owner fix (contributed by", body)
self.assertIn(
"[v1.0.0...v1.1.0](https://github.com/Syngnat/GoNavi/compare/v1.0.0...v1.1.0)",
body,
)
def test_release_workflow_uses_tested_generator_and_pull_request_metadata(self) -> None:
release = (ROOT / ".github" / "workflows" / "release.yml").read_text(encoding="utf-8")
dev_build = (ROOT / ".github" / "workflows" / "dev-build.yml").read_text(encoding="utf-8")
self.assertIn("pull-requests: read", release)
self.assertIn("python3 tools/generate-release-notes.py", release)
self.assertNotIn("git log \"$RANGE\" --no-merges --pretty=format:'%s'", release)
changelog_step = release.split("- name: Generate Changelog", 1)[1].split(
"- name: Create Release", 1
)[0]
for expected in (
"GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}",
'CHANGELOG_FILE="$RUNNER_TEMP/changelog.md"',
'--repo "${{ github.repository }}"',
'--tag "${{ github.ref_name }}"',
'--repository-url "${{ github.server_url }}/${{ github.repository }}"',
'--output "$CHANGELOG_FILE"',
'echo "changelog_file=$CHANGELOG_FILE" >> "$GITHUB_OUTPUT"',
):
self.assertIn(expected, changelog_step)
create_release_step = release.split("- name: Create Release", 1)[1]
self.assertIn(
"body_path: ${{ steps.changelog.outputs.changelog_file }}",
create_release_step,
)
for workflow in (release, dev_build):
self.assertIn("python3 tools/generate-release-notes.test.py", workflow)
@staticmethod
def run_git(repo: pathlib.Path, *args: str, env=None) -> str:
return subprocess.check_output(
["git", *args],
cwd=repo,
env=env,
text=True,
encoding="utf-8",
stderr=subprocess.STDOUT,
)
if __name__ == "__main__":
unittest.main()