|
#!/usr/bin/env python3 |
|
"""Report new major-version tags for GitHub Actions user chooses. |
|
|
|
A "major" is the floating tag people pin in workflows (actions/checkout@v7). |
|
The first run records the highest major. |
|
Subsequent runs flag only a real bump (v6 -> v7). |
|
|
|
Requires `gh` already logged in (`gh auth status`). |
|
|
|
Examples: |
|
./check_action_majors.py |
|
./check_action_majors.py actions/upload-artifact actions/cache |
|
./check_action_majors.py --file actions.txt |
|
./check_action_majors.py --scan ~/src/mumps --scan ~/src/ffilesystem |
|
./check_action_majors.py --fail-on-new |
|
""" |
|
|
|
import argparse |
|
import asyncio |
|
import json |
|
import re |
|
import sys |
|
from dataclasses import dataclass, field |
|
from datetime import datetime, timezone |
|
from pathlib import Path |
|
|
|
|
|
STATE_NAME = "action-majors-state.json" |
|
TAG_RE = re.compile(r"^v(\d+)(?:\.(\d+))?(?:\.(\d+))?([-+].+)?$") |
|
USES_RE = re.compile( |
|
r"""(?im)^\s*-?\s*uses:\s*['\"]?""" |
|
r"""([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)""" |
|
r"""@v(\d+)(?:\.\d+)*""" |
|
r"""['\"]?\s*(?:#.*)?$""" |
|
) |
|
REPO_RE = re.compile( |
|
r"^(?:https?://github\.com/)?" |
|
r"([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)" |
|
r"(?:@v(\d+))?" |
|
r"/?$" |
|
) |
|
|
|
|
|
@dataclass |
|
class Version: |
|
major: int |
|
minor: int |
|
patch: int |
|
raw: str |
|
|
|
def as_tuple(self) -> tuple[int, int, int]: |
|
return (self.major, self.minor, self.patch) |
|
|
|
|
|
@dataclass |
|
class ActionReport: |
|
repo: str |
|
latest_major: int | None = None |
|
latest_tag: str | None = None |
|
floating_tag: str | None = None |
|
published: str | None = None |
|
pin: int | None = None |
|
baseline: int | None = None |
|
error: str | None = None |
|
|
|
|
|
@dataclass |
|
class FetchResult: |
|
tags: list[str] = field(default_factory=list) |
|
releases: dict[str, str] = field(default_factory=dict) # tag -> YYYY-MM-DD |
|
|
|
|
|
def parse_tag(name: str, include_prerelease: bool = False) -> Version | None: |
|
match = TAG_RE.fullmatch(name.strip()) |
|
if not match: |
|
return None |
|
if match.group(4) and not include_prerelease: |
|
return None |
|
return Version( |
|
major=int(match.group(1)), |
|
minor=int(match.group(2) or 0), |
|
patch=int(match.group(3) or 0), |
|
raw=name.strip(), |
|
) |
|
|
|
|
|
def normalize_repo(text: str) -> tuple[str, int | None]: |
|
match = REPO_RE.fullmatch(text.strip()) |
|
if not match: |
|
raise ValueError(f"not an owner/repo: {text}") |
|
pin = int(match.group(2)) if match.group(2) else None |
|
return match.group(1), pin |
|
|
|
|
|
async def gh(*args: str) -> str: |
|
proc = await asyncio.create_subprocess_exec( |
|
"gh", |
|
*args, |
|
stdout=asyncio.subprocess.PIPE, |
|
stderr=asyncio.subprocess.PIPE, |
|
) |
|
stdout, stderr = await proc.communicate() |
|
if proc.returncode != 0: |
|
raise RuntimeError(stderr.decode().strip() or f"gh exited {proc.returncode}") |
|
return stdout.decode() |
|
|
|
|
|
async def fetch(repo: str) -> FetchResult: |
|
tags, releases = await asyncio.gather( |
|
gh("api", "--paginate", f"repos/{repo}/tags", "--jq", ".[].name"), |
|
gh( |
|
"api", |
|
"--paginate", |
|
f"repos/{repo}/releases", |
|
"--jq", |
|
r'.[] | [.tag_name, (.published_at // "")] | @tsv', |
|
), |
|
return_exceptions=True, |
|
) |
|
if isinstance(tags, BaseException): |
|
raise tags |
|
if isinstance(releases, BaseException): |
|
raise releases |
|
result = FetchResult(tags=[line.strip() for line in tags.splitlines() if line.strip()]) |
|
for line in releases.splitlines(): |
|
if not line.strip(): |
|
continue |
|
tag, _, published = line.partition("\t") |
|
result.releases[tag.strip()] = published.strip()[:10] |
|
return result |
|
|
|
|
|
def summarize( |
|
repo: str, |
|
fetched: FetchResult, |
|
pin: int | None, |
|
baseline: int | None, |
|
include_prerelease: bool, |
|
) -> ActionReport: |
|
versions = [ |
|
parsed |
|
for name in fetched.tags |
|
if (parsed := parse_tag(name, include_prerelease)) is not None |
|
] |
|
report = ActionReport(repo=repo, pin=pin, baseline=baseline) |
|
if not versions: |
|
report.error = "no vN tags found" |
|
return report |
|
latest_major = max(v.major for v in versions) |
|
in_major = [v for v in versions if v.major == latest_major] |
|
# Prefer a real semver tag (v7.0.1) over the floating major (v7 -> 7.0.0). |
|
concrete = [v for v in in_major if v.raw != f"v{v.major}"] |
|
best = max(concrete or in_major, key=lambda v: v.as_tuple()) |
|
floating = f"v{latest_major}" |
|
report.latest_major = latest_major |
|
report.latest_tag = best.raw |
|
report.floating_tag = floating if any(v.raw == floating for v in versions) else None |
|
report.published = fetched.releases.get(best.raw) or fetched.releases.get(floating) |
|
return report |
|
|
|
|
|
def status_of(report: ActionReport, had_baseline: bool) -> str: |
|
if report.error or report.latest_major is None: |
|
return "error" |
|
if not had_baseline: |
|
return "baseline" |
|
if report.baseline is None: |
|
return "new action" |
|
if report.latest_major > report.baseline: |
|
return "NEW MAJOR" |
|
if report.latest_major < report.baseline: |
|
return "major decreased" |
|
return "unchanged" |
|
|
|
|
|
def load_state(path: Path) -> dict: |
|
if not path.exists(): |
|
return {} |
|
data = json.loads(path.read_text()) |
|
return data if isinstance(data, dict) else {} |
|
|
|
|
|
def save_state(path: Path, reports: list[ActionReport]) -> None: |
|
previous = load_state(path) |
|
now = datetime.now(timezone.utc).strftime("%Y-%m-%d") |
|
for report in reports: |
|
if report.latest_major is None: |
|
continue |
|
previous[report.repo] = { |
|
"major": report.latest_major, |
|
"tag": report.latest_tag, |
|
"floating": report.floating_tag, |
|
"published": report.published, |
|
"seen": now, |
|
} |
|
path.parent.mkdir(parents=True, exist_ok=True) |
|
path.write_text(json.dumps(previous, indent=2) + "\n") |
|
|
|
|
|
def collect_actions(repos: list[str], state: dict) -> list[tuple[str, int | None]]: |
|
found: dict[str, int | None] = {} |
|
|
|
def add(repo: str, pin: int | None) -> None: |
|
current = found.get(repo) |
|
if current is None or (pin is not None and (current is None or pin > current)): |
|
found[repo] = pin if pin is not None else current |
|
|
|
if not repos: |
|
for repo in state: |
|
add(repo, None) |
|
for text in repos: |
|
add(*normalize_repo(text)) |
|
return list(found.items()) |
|
|
|
|
|
def render(reports: list[ActionReport], had_baseline: bool) -> int: |
|
rows = [] |
|
new_count = 0 |
|
for report in reports: |
|
status = status_of(report, had_baseline) |
|
if status == "NEW MAJOR": |
|
new_count += 1 |
|
if report.error: |
|
cells = (report.repo, "-", "-", "-", "-", "-", f"error: {report.error}") |
|
else: |
|
if report.pin is not None and report.latest_major and report.latest_major > report.pin: |
|
if status == "unchanged": |
|
status = f"behind pin (v{report.pin})" |
|
elif status == "NEW MAJOR": |
|
status = f"NEW MAJOR; pin is v{report.pin}" |
|
cells = ( |
|
report.repo, |
|
report.floating_tag or f"v{report.latest_major}", |
|
report.latest_tag or "-", |
|
report.published or "-", |
|
f"v{report.baseline}" if report.baseline is not None else "-", |
|
f"v{report.pin}" if report.pin is not None else "-", |
|
status, |
|
) |
|
rows.append(cells) |
|
|
|
headers = ("action", "latest", "release", "published", "baseline", "pin", "status") |
|
width = [len(h) for h in headers] |
|
for cells in rows: |
|
for i, cell in enumerate(cells): |
|
width[i] = max(width[i], len(cell)) |
|
|
|
def line(cells: tuple[str, ...]) -> str: |
|
return " ".join(cell.ljust(width[i]) for i, cell in enumerate(cells)) |
|
|
|
print(line(headers)) |
|
print(line(tuple("-" * n for n in width))) |
|
for cells in rows: |
|
print(line(cells)) |
|
print() |
|
if not had_baseline: |
|
print("No previous baseline. Recorded current majors; re-run later to detect bumps.") |
|
elif new_count: |
|
print(f"{new_count} action(s) published a new major since the last run.") |
|
else: |
|
print("No new majors since the last run.") |
|
return new_count |
|
|
|
|
|
async def check_action( |
|
repo: str, |
|
pin: int | None, |
|
baseline: int | None, |
|
include_prerelease: bool, |
|
) -> ActionReport: |
|
try: |
|
return summarize(repo, await fetch(repo), pin, baseline, include_prerelease) |
|
except RuntimeError as exc: |
|
return ActionReport(repo=repo, pin=pin, baseline=baseline, error=str(exc)) |
|
|
|
|
|
async def async_main(argv: list[str] | None = None) -> int: |
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
|
parser.add_argument("repos", nargs="*", help="owner/repo or owner/repo@vN (pin is compared, not fetched)") |
|
parser.add_argument( |
|
"--state", |
|
type=Path, |
|
default=Path(__file__).resolve().with_name(STATE_NAME), |
|
help=f"baseline file (default: {STATE_NAME} next to this script)", |
|
) |
|
parser.add_argument("--no-save", action="store_true", help="do not update the baseline file") |
|
parser.add_argument("--include-prerelease", action="store_true", help="count vN.N.N-beta tags as majors") |
|
parser.add_argument("--fail-on-new", action="store_true", help="exit 1 when a new major is found") |
|
parser.add_argument("--json", action="store_true", help="print JSON instead of a table") |
|
args = parser.parse_args(argv) |
|
|
|
state = load_state(args.state) |
|
actions = collect_actions(args.repos, state) |
|
if not actions: |
|
parser.error("no actions to check") |
|
|
|
had_baseline = bool(state) |
|
|
|
def baseline_for(repo: str) -> int | None: |
|
entry = state.get(repo) |
|
if isinstance(entry, dict) and isinstance(entry.get("major"), int): |
|
return entry["major"] |
|
return None |
|
|
|
reports = await asyncio.gather( |
|
*( |
|
check_action( |
|
repo, |
|
pin, |
|
baseline_for(repo), |
|
args.include_prerelease, |
|
) |
|
for repo, pin in actions |
|
) |
|
) |
|
|
|
if args.json: |
|
payload = [ |
|
{ |
|
"repo": report.repo, |
|
"latest_major": report.latest_major, |
|
"latest_tag": report.latest_tag, |
|
"floating_tag": report.floating_tag, |
|
"published": report.published, |
|
"pin": report.pin, |
|
"baseline": report.baseline, |
|
"status": status_of(report, had_baseline), |
|
"error": report.error, |
|
"releases": f"https://github.com/{report.repo}/releases", |
|
} |
|
for report in reports |
|
] |
|
print(json.dumps(payload, indent=2)) |
|
new_count = sum(1 for item in payload if item["status"] == "NEW MAJOR") |
|
else: |
|
new_count = render(reports, had_baseline) |
|
print() |
|
for report in reports: |
|
if status_of(report, had_baseline) == "NEW MAJOR": |
|
print(f"https://github.com/{report.repo}/releases") |
|
|
|
if not args.no_save: |
|
save_state(args.state, reports) |
|
print(f"\nbaseline: {args.state}", file=sys.stderr) |
|
|
|
if args.fail_on_new and new_count: |
|
return 1 |
|
if any(report.error for report in reports): |
|
return 2 |
|
return 0 |
|
|
|
|
|
def main(argv: list[str] | None = None) -> int: |
|
return asyncio.run(async_main(argv)) |
|
|
|
|
|
if __name__ == "__main__": |
|
raise SystemExit(main()) |