#!/usr/bin/env python3 """Collect release context between a target tag and its comparison baseline.""" from __future__ import annotations import argparse import json import re import subprocess import sys from pathlib import Path SEMVER_TAG_RE = re.compile(r"^v(\d+)\.(\d+)\.(\d+)$") def run_git( repo: Path, args: list[str], allow_failure: bool = False, timeout_seconds: int = 15, ) -> str: try: result = subprocess.run( ["git", *args], cwd=repo, capture_output=True, text=True, check=False, timeout=timeout_seconds, ) except subprocess.TimeoutExpired: if allow_failure: return "" raise RuntimeError(f"git command timed out: git {' '.join(args)}") if result.returncode != 0 and not allow_failure: raise RuntimeError(result.stderr.strip() or "git command failed") return result.stdout.strip() def parse_semver(tag: str) -> tuple[int, int, int] | None: match = SEMVER_TAG_RE.match(tag) if not match: return None return tuple(int(part) for part in match.groups()) def list_reachable_tags(repo: Path) -> list[str]: output = run_git(repo, ["tag", "--merged", "HEAD"]) return [line.strip() for line in output.splitlines() if line.strip()] def choose_previous_tag(repo: Path, tags: list[str], target_tag: str | None) -> tuple[str | None, str | None]: semver_tags: list[tuple[tuple[int, int, int], str]] = [] other_tags: list[str] = [] target_semver = parse_semver(target_tag) if target_tag else None for tag in tags: if target_tag and tag == target_tag: continue parsed = parse_semver(tag) if parsed is None: other_tags.append(tag) continue if target_semver is not None and parsed >= target_semver: continue semver_tags.append((parsed, tag)) if semver_tags: semver_tags.sort(key=lambda item: item[0], reverse=True) return semver_tags[0][1], "semver" if not other_tags: return None, None candidates: list[tuple[int, str]] = [] for tag in other_tags: ts = run_git(repo, ["log", "-1", "--format=%ct", tag], allow_failure=True) try: candidates.append((int(ts), tag)) except ValueError: continue if not candidates: return other_tags[-1], "fallback" candidates.sort(reverse=True) return candidates[0][1], "fallback" def ensure_tag_absent(repo: Path, tag: str) -> None: local = run_git(repo, ["tag", "--list", tag]) if local: raise ValueError(f"target tag already exists locally: {tag}") remotes_output = run_git(repo, ["remote"]) for remote in [line.strip() for line in remotes_output.splitlines() if line.strip()]: remote_hit = run_git( repo, ["ls-remote", "--tags", remote, tag], allow_failure=True, timeout_seconds=8, ) if remote_hit: raise ValueError(f"target tag already exists on remote {remote}: {tag}") def ensure_tag_exists(repo: Path, tag: str) -> None: hit = run_git(repo, ["rev-parse", "--verify", "--quiet", tag], allow_failure=True) if not hit: raise ValueError(f"previous tag does not exist locally: {tag}") def get_commit_list(repo: Path, rev_range: str) -> list[dict[str, str]]: output = run_git(repo, ["log", "--reverse", "--date=short", "--pretty=format:%H%x09%ad%x09%s", rev_range]) commits: list[dict[str, str]] = [] for line in output.splitlines(): commit_hash, date_str, subject = line.split("\t", 2) commits.append( { "hash": commit_hash, "short_hash": commit_hash[:7], "date": date_str, "subject": subject, } ) return commits def get_changed_files(repo: Path, rev_range: str) -> list[str]: output = run_git(repo, ["diff", "--name-only", rev_range]) return [line.strip() for line in output.splitlines() if line.strip()] def get_numstat(repo: Path, rev_range: str) -> dict[str, int]: output = run_git(repo, ["diff", "--numstat", rev_range]) changed_files = 0 insertions = 0 deletions = 0 for line in output.splitlines(): parts = line.split("\t") if len(parts) < 3: continue added, removed = parts[0], parts[1] changed_files += 1 if added.isdigit(): insertions += int(added) if removed.isdigit(): deletions += int(removed) return { "changed_files": changed_files, "insertions": insertions, "deletions": deletions, } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo", default=".", help="Repository root. Defaults to current directory.") parser.add_argument("--tag", help="Target release tag to validate, for example v1.2.3.") parser.add_argument("--previous-tag", help="Explicit comparison baseline.") return parser def main() -> int: args = build_parser().parse_args() repo_path = Path(args.repo).expanduser().resolve() if not (repo_path / ".git").exists(): print(json.dumps({"error": f"not a git repository: {repo_path}"})) return 1 if args.tag and parse_semver(args.tag) is None: print(json.dumps({"error": f"invalid tag format: {args.tag}", "expected": "vx.y.z"})) return 1 try: if args.tag: ensure_tag_absent(repo_path, args.tag) if args.previous_tag: ensure_tag_exists(repo_path, args.previous_tag) except ValueError as exc: print(json.dumps({"error": str(exc)})) return 1 tags = list_reachable_tags(repo_path) previous_tag = args.previous_tag previous_tag_source = "explicit" if previous_tag else None if previous_tag is None: previous_tag, previous_tag_source = choose_previous_tag(repo_path, tags, args.tag) rev_range = "HEAD" if previous_tag: rev_range = f"{previous_tag}..HEAD" try: commits = get_commit_list(repo_path, rev_range) changed_files = get_changed_files(repo_path, rev_range) stats = get_numstat(repo_path, rev_range) head_commit = run_git(repo_path, ["rev-parse", "HEAD"]) except RuntimeError as exc: print(json.dumps({"error": str(exc)})) return 1 payload: dict[str, object] = { "repo": str(repo_path), "target_tag": args.tag, "previous_tag": previous_tag, "previous_tag_source": previous_tag_source, "rev_range": rev_range, "head_commit": head_commit, "commit_count": len(commits), "commits": commits, "changed_files": changed_files, } payload.update(stats) print(json.dumps(payload, indent=2, ensure_ascii=False)) return 0 if __name__ == "__main__": sys.exit(main())