#!/usr/bin/env python3 """Refresh GitHub star counts for plugin-catalog repos, at most once a day. Writes ``website/static/api/plugin-stars.json``:: {"fetched_at": "", "stars": {"owner/repo": 123, ...}} ``extract-plugins.py`` merges these into ``plugins.json`` so the catalog page can rank entries by stars. Rate-limit discipline follows the skills-index rule: GitHub is only ever consulted from the scheduled ``skills-index.yml`` run (twice daily, ``--probe``), which uploads the result as an artifact. Docs deploys run with ``--reuse-only`` and NEVER touch the API: they take the scheduled artifact when given one, else the live site's copy (one CDN GET), else whatever is on disk, else an empty map (the page then ranks alphabetically). The probe itself is ONE GraphQL request for every catalog repo (aliased ``repository`` fields), not one REST call per repo. Any failure (rate limit, network, bad token) keeps the previous counts instead of regressing them to zero. """ from __future__ import annotations import argparse import json import os import re import sys import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path import yaml REPO_ROOT = Path(__file__).resolve().parents[2] DEFAULT_CATALOG_DIR = REPO_ROOT / "plugin-catalog" DEFAULT_OUTPUT = REPO_ROOT / "website" / "static" / "api" / "plugin-stars.json" LIVE_URL = "https://hermes-agent.nousresearch.com/docs/api/plugin-stars.json" _GITHUB_REPO_RE = re.compile(r"^https://github\.com/([^/\s]+)/([^/\s#?]+?)(?:\.git)?/?$") def _log(msg: str) -> None: print(f"[fetch-plugin-stars] {msg}", file=sys.stderr) def github_slug(repo_url: str) -> str | None: """``owner/repo`` for a github.com URL, else None (non-GitHub hosts are never probed).""" m = _GITHUB_REPO_RE.match(repo_url.strip()) return f"{m.group(1)}/{m.group(2)}" if m else None def catalog_slugs(catalog_dir: Path) -> list[str]: slugs: set[str] = set() for path in sorted(catalog_dir.glob("*.yaml")): if path.name == "removed.yaml": continue try: raw = yaml.safe_load(path.read_text(encoding="utf-8")) except (yaml.YAMLError, OSError): continue slug = github_slug(str((raw or {}).get("repo") or "")) if isinstance(raw, dict) else None if slug: slugs.add(slug) return sorted(slugs) def _http_json(url: str, headers: dict[str, str], timeout: float = 15.0): req = urllib.request.Request(url, headers={"User-Agent": "hermes-agent-docs", **headers}) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8")) def load_previous(output: Path, live_url: str | None) -> dict: """Newest of {live site copy, on-disk copy}; ``{}`` when neither exists.""" candidates: list[dict] = [] if live_url: try: data = _http_json(live_url, {}) if isinstance(data, dict) and isinstance(data.get("stars"), dict): candidates.append(data) except (urllib.error.URLError, OSError, ValueError) as e: _log(f"live cache unavailable ({e}); continuing without it") if output.is_file(): try: data = json.loads(output.read_text(encoding="utf-8")) if isinstance(data, dict) and isinstance(data.get("stars"), dict): candidates.append(data) except (OSError, ValueError): pass return max(candidates, key=lambda d: str(d.get("fetched_at") or ""), default={}) _GRAPHQL_URL = "https://api.github.com/graphql" def _graphql(query: str, token: str) -> dict: req = urllib.request.Request( _GRAPHQL_URL, data=json.dumps({"query": query}).encode("utf-8"), method="POST", headers={"User-Agent": "hermes-agent-docs", "Authorization": f"Bearer {token}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=30.0) as resp: return json.loads(resp.read().decode("utf-8")) def stars_query(slugs: list[str]) -> str: fields = "\n".join( f'r{i}: repository(owner: {json.dumps(slug.split("/", 1)[0])}, name: {json.dumps(slug.split("/", 1)[1])})' " { stargazerCount }" for i, slug in enumerate(slugs)) return "query {\n" + fields + "\n}" def probe_stars(slugs: list[str], previous: dict[str, int], token: str | None) -> dict[str, int]: """One GraphQL request for all repos; on failure keep every previous count (never regress to 0).""" if not slugs: return {} if not token: _log("no GITHUB_TOKEN; keeping previous counts without probing") return {s: previous[s] for s in slugs if s in previous} try: payload = _graphql(stars_query(slugs), token) except (urllib.error.URLError, OSError, ValueError) as e: _log(f"GraphQL probe failed ({e}); keeping previous counts") return {s: previous[s] for s in slugs if s in previous} data = payload.get("data") or {} for err in payload.get("errors") or []: _log(f"GraphQL: {err.get('message')}") # e.g. a renamed/deleted repo; its previous count is kept stars: dict[str, int] = {} for i, slug in enumerate(slugs): node = data.get(f"r{i}") if isinstance(node, dict) and isinstance(node.get("stargazerCount"), int): stars[slug] = node["stargazerCount"] elif slug in previous: stars[slug] = previous[slug] return stars def main(catalog_dir: Path = DEFAULT_CATALOG_DIR, output: Path = DEFAULT_OUTPUT, probe: bool = False, live_url: str | None = LIVE_URL, token: str | None = None) -> int: previous = load_previous(output, live_url) output.parent.mkdir(parents=True, exist_ok=True) if not probe: output.write_text(json.dumps(previous or {"fetched_at": None, "stars": {}}, separators=(",", ":")), encoding="utf-8") print(f"Reused plugin stars from {previous.get('fetched_at')} " f"({len(previous.get('stars', {}))} repos, no GitHub calls)") return 0 slugs = catalog_slugs(catalog_dir) prev_stars = {k: int(v) for k, v in (previous.get("stars") or {}).items() if isinstance(v, (int, float))} stars = probe_stars(slugs, prev_stars, token or os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")) probed = stars != prev_stars or not previous fetched_at = datetime.now(timezone.utc).isoformat() if probed or stars else str(previous.get("fetched_at") or "") output.write_text(json.dumps({"fetched_at": fetched_at, "stars": stars}, separators=(",", ":")), encoding="utf-8") print(f"Probed {len(slugs)} repos in one GraphQL request, wrote {len(stars)} star counts to {output}") return 0 if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--catalog-dir", type=Path, default=DEFAULT_CATALOG_DIR) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) parser.add_argument("--probe", action="store_true", help="call GitHub (one GraphQL request); only the scheduled skills-index run does this") parser.add_argument("--no-live", action="store_true", help="do not consult the live site's cache") args = parser.parse_args() sys.exit(main(catalog_dir=args.catalog_dir, output=args.output, probe=args.probe, live_url=None if args.no_live else LIVE_URL))