feat(ce-release-notes): add skill for browsing plugin release history (#589)
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Trevin Chow
@@ -0,0 +1,279 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
list-plugin-releases.py — Fetch compound-engineering plugin releases from GitHub.
|
||||
|
||||
Output: a single JSON object on stdout. Always exits 0; failures are encoded
|
||||
in the contract, never raised.
|
||||
|
||||
Usage:
|
||||
python3 list-plugin-releases.py [--limit N] [--api-base URL]
|
||||
|
||||
Environment:
|
||||
CE_RELEASE_NOTES_GH_BIN Override the gh binary path (default: "gh"). Used
|
||||
by the test harness; leave unset in production.
|
||||
|
||||
Contract:
|
||||
Success:
|
||||
{"ok": true, "source": "gh"|"anon", "fetched_at": "ISO8601",
|
||||
"releases": [{tag, version, name, published_at, url, body, linked_prs}]}
|
||||
Failure:
|
||||
{"ok": false, "error": {"code": "rate_limit"|"network_outage",
|
||||
"message": "...", "user_hint": "..."}}
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
|
||||
OWNER = "EveryInc"
|
||||
REPO = "compound-engineering-plugin"
|
||||
TAG_PREFIX = "compound-engineering-v"
|
||||
DEFAULT_API_BASE = "https://api.github.com"
|
||||
GH_TIMEOUT_SECS = 10
|
||||
ANON_TIMEOUT_SECS = 10
|
||||
RELEASES_URL = "https://github.com/" + OWNER + "/" + REPO + "/releases"
|
||||
PR_REGEX = re.compile(r"\[#(\d+)\]")
|
||||
|
||||
|
||||
def _now_iso():
|
||||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
|
||||
def _extract_linked_prs(body):
|
||||
if not body:
|
||||
return []
|
||||
seen = set()
|
||||
out = []
|
||||
for m in PR_REGEX.finditer(body):
|
||||
n = int(m.group(1))
|
||||
if n not in seen:
|
||||
seen.add(n)
|
||||
out.append(n)
|
||||
return out
|
||||
|
||||
|
||||
def _version_from_tag(tag):
|
||||
if tag.startswith(TAG_PREFIX):
|
||||
return tag[len(TAG_PREFIX):]
|
||||
return tag
|
||||
|
||||
|
||||
def _normalize_release(raw):
|
||||
"""Coerce a raw release dict (gh shape OR API shape) into the contract shape."""
|
||||
tag = raw.get("tagName") or raw.get("tag_name") or ""
|
||||
if not tag:
|
||||
return None
|
||||
body = raw.get("body") or ""
|
||||
return {
|
||||
"tag": tag,
|
||||
"version": _version_from_tag(tag),
|
||||
"name": raw.get("name") or "",
|
||||
"published_at": raw.get("publishedAt") or raw.get("published_at") or "",
|
||||
"url": raw.get("html_url") or raw.get("url") or "",
|
||||
"body": body,
|
||||
"linked_prs": _extract_linked_prs(body),
|
||||
}
|
||||
|
||||
|
||||
def _filter_and_sort(raw_list):
|
||||
out = []
|
||||
for raw in raw_list:
|
||||
if not isinstance(raw, dict):
|
||||
continue
|
||||
norm = _normalize_release(raw)
|
||||
if norm is None:
|
||||
continue
|
||||
if not norm["tag"].startswith(TAG_PREFIX):
|
||||
continue
|
||||
out.append(norm)
|
||||
out.sort(key=lambda r: r["published_at"], reverse=True)
|
||||
return out
|
||||
|
||||
|
||||
def attempt_gh(limit):
|
||||
"""
|
||||
Try to fetch via gh. Returns (success, releases).
|
||||
success=True → caller emits the result with source="gh"
|
||||
success=False → caller falls back to attempt_anon
|
||||
Falls back when: gh missing, gh exits non-zero, gh times out, gh stdout is
|
||||
not parseable JSON, or gh returns zero plugin tags (covers the GitHub
|
||||
Enterprise silent-empty case).
|
||||
"""
|
||||
gh_bin = os.environ.get("CE_RELEASE_NOTES_GH_BIN", "gh")
|
||||
# `gh release list --json` does NOT expose `body` or `url` (only metadata
|
||||
# fields). `gh api` returns the full GitHub Releases API response shape
|
||||
# (tag_name, html_url, body, published_at, ...) and uses gh's auth so
|
||||
# there is no rate limit. The normalizer already handles this shape.
|
||||
cmd = [
|
||||
gh_bin,
|
||||
"api",
|
||||
"/repos/" + OWNER + "/" + REPO + "/releases?per_page=" + str(limit),
|
||||
]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
timeout=GH_TIMEOUT_SECS,
|
||||
check=False,
|
||||
)
|
||||
except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
|
||||
return False, None
|
||||
if result.returncode != 0:
|
||||
return False, None
|
||||
try:
|
||||
raw_list = json.loads(result.stdout)
|
||||
except json.JSONDecodeError:
|
||||
return False, None
|
||||
if not isinstance(raw_list, list):
|
||||
return False, None
|
||||
releases = _filter_and_sort(raw_list)
|
||||
if not releases:
|
||||
return False, None
|
||||
return True, releases
|
||||
|
||||
|
||||
def _format_reset_hint(reset_unix):
|
||||
secs_until = max(0, reset_unix - int(time.time()))
|
||||
minutes = (secs_until + 59) // 60
|
||||
if minutes <= 1:
|
||||
return "less than a minute"
|
||||
return str(minutes) + " minutes"
|
||||
|
||||
|
||||
def attempt_anon(limit, api_base):
|
||||
"""
|
||||
Fetch via the anonymous GitHub API.
|
||||
Returns (status, payload):
|
||||
"ok" → payload = {"releases": [...]}
|
||||
"rate_limit" → payload = {"reset_hint": "N minutes"}
|
||||
"network_outage" → payload = {"detail": "..."}
|
||||
"""
|
||||
url = api_base + "/repos/" + OWNER + "/" + REPO + "/releases?per_page=" + str(limit)
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
headers={
|
||||
"Accept": "application/vnd.github+json",
|
||||
"User-Agent": "ce-release-notes-skill",
|
||||
},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=ANON_TIMEOUT_SECS) as resp:
|
||||
body = resp.read()
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 403:
|
||||
remaining = e.headers.get("X-RateLimit-Remaining")
|
||||
if remaining == "0":
|
||||
try:
|
||||
reset_unix = int(e.headers.get("X-RateLimit-Reset") or "0")
|
||||
except ValueError:
|
||||
reset_unix = 0
|
||||
return "rate_limit", {"reset_hint": _format_reset_hint(reset_unix)}
|
||||
return "network_outage", {"detail": "HTTP " + str(e.code)}
|
||||
except urllib.error.URLError as e:
|
||||
return "network_outage", {"detail": "network error: " + str(e.reason)}
|
||||
except Exception as e:
|
||||
return "network_outage", {"detail": "unexpected: " + type(e).__name__}
|
||||
|
||||
try:
|
||||
raw_list = json.loads(body)
|
||||
except json.JSONDecodeError:
|
||||
return "network_outage", {"detail": "malformed JSON from API"}
|
||||
if not isinstance(raw_list, list):
|
||||
return "network_outage", {"detail": "unexpected API response shape"}
|
||||
return "ok", {"releases": _filter_and_sort(raw_list)}
|
||||
|
||||
|
||||
def emit(obj):
|
||||
sys.stdout.write(json.dumps(obj))
|
||||
sys.stdout.write("\n")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Fetch compound-engineering plugin releases from GitHub."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=40,
|
||||
help="Number of raw releases to fetch (default: 40).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--api-base",
|
||||
default=DEFAULT_API_BASE,
|
||||
help="Override the GitHub API base URL (test harness use).",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
success, releases = attempt_gh(args.limit)
|
||||
if success:
|
||||
emit(
|
||||
{
|
||||
"ok": True,
|
||||
"source": "gh",
|
||||
"fetched_at": _now_iso(),
|
||||
"releases": releases,
|
||||
}
|
||||
)
|
||||
return
|
||||
|
||||
status, payload = attempt_anon(args.limit, args.api_base)
|
||||
if status == "ok":
|
||||
emit(
|
||||
{
|
||||
"ok": True,
|
||||
"source": "anon",
|
||||
"fetched_at": _now_iso(),
|
||||
"releases": payload["releases"],
|
||||
}
|
||||
)
|
||||
return
|
||||
|
||||
if status == "rate_limit":
|
||||
message = (
|
||||
"GitHub anonymous API rate limit hit (resets in "
|
||||
+ payload["reset_hint"]
|
||||
+ ")."
|
||||
)
|
||||
user_hint = (
|
||||
"Install and authenticate `gh` to remove this limit, or open "
|
||||
+ RELEASES_URL
|
||||
+ " directly."
|
||||
)
|
||||
emit(
|
||||
{
|
||||
"ok": False,
|
||||
"error": {
|
||||
"code": "rate_limit",
|
||||
"message": message,
|
||||
"user_hint": user_hint,
|
||||
},
|
||||
}
|
||||
)
|
||||
return
|
||||
|
||||
message = "Could not reach the GitHub Releases API."
|
||||
user_hint = (
|
||||
"Check your network connection, or open " + RELEASES_URL + " directly."
|
||||
)
|
||||
emit(
|
||||
{
|
||||
"ok": False,
|
||||
"error": {
|
||||
"code": "network_outage",
|
||||
"message": message,
|
||||
"user_hint": user_hint,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user