#!/usr/bin/env python3 """ Fetch latest GitHub releases for each plugin in plugin-catalog.json, download the primary .zip asset, compute SHA256, and produce a Jellyfin plugin-repository manifest. Usage: python3 fetch_releases.py GITHUB_TOKEN=ghp_xxx python3 fetch_releases.py """ import base64 import hashlib import json import os import re import sys import time import uuid import zipfile from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse from urllib.request import Request, urlopen from urllib.error import HTTPError CATALOG = Path(__file__).with_name("plugin-catalog.json") MANIFEST = Path(__file__).with_name("jellyfin-plugin-repo.json") CACHE_DIR = Path(__file__).with_name(".release-cache") DOWNLOAD_DIR = Path(__file__).with_name("plugin-zips") DEFAULT_TARGET_ABI = os.environ.get("DEFAULT_TARGET_ABI", "10.8.0.0") GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") # UUID namespace so the same owner/repo always yields the same GUID. GUID_NS = uuid.UUID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") def github_api_request(url: str) -> tuple[int, dict | None]: req = Request(url, headers={"Accept": "application/vnd.github+json"}) if GITHUB_TOKEN: req.add_header("Authorization", f"Bearer {GITHUB_TOKEN}") try: with urlopen(req, timeout=30) as resp: return resp.status, json.loads(resp.read().decode("utf-8")) except HTTPError as e: body = e.read().decode("utf-8", errors="ignore") return e.code, {"error": body} def fetch_release(owner: str, repo: str) -> dict | None: cache_file = CACHE_DIR / f"{owner}_{repo}.json" if cache_file.exists(): cached = json.loads(cache_file.read_text(encoding="utf-8")) # If cached result is not an error, reuse it. if "error" not in cached: return cached url = f"https://api.github.com/repos/{owner}/{repo}/releases/latest" status, data = github_api_request(url) cache_file.write_text(json.dumps(data, indent=2), encoding="utf-8") if status != 200: print(f" ! {owner}/{repo}: GitHub API HTTP {status}") return None return data def download_asset(asset_url: str, dest: Path) -> bool: dest.parent.mkdir(parents=True, exist_ok=True) req = Request(asset_url, headers={"Accept": "application/octet-stream"}) if GITHUB_TOKEN: req.add_header("Authorization", f"Bearer {GITHUB_TOKEN}") try: with urlopen(req, timeout=60) as resp, dest.open("wb") as f: downloaded = 0 while True: chunk = resp.read(64 * 1024) if not chunk: break f.write(chunk) downloaded += len(chunk) if downloaded % (512 * 1024) < 64 * 1024: print(f" ... {downloaded // 1024} KB", end="\r") print(f" ✓ {downloaded // 1024} KB") return True except HTTPError as e: print(f" ! download failed HTTP {e.code}: {asset_url}") return False def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(64 * 1024), b""): h.update(chunk) return h.hexdigest() def looks_like_plugin_zip(name: str) -> bool: lower = name.lower() return lower.endswith(".zip") and ("plugin" in lower or "jellyfin" in lower) def pick_asset(assets: list[dict]) -> dict | None: if not assets: return None # Prefer assets that look like a plugin zip. for asset in assets: if looks_like_plugin_zip(asset["name"]): return asset # Fallback to first .zip. for asset in assets: if asset["name"].lower().endswith(".zip"): return asset # Last resort: first asset. return assets[0] def make_guid(owner: str, repo: str) -> str: return str(uuid.uuid5(GUID_NS, f"{owner}/{repo}")) def main() -> int: if not CATALOG.exists(): print(f"Missing catalog: {CATALOG}. Run parse_plugins.py first.") return 1 plugins = json.loads(CATALOG.read_text(encoding="utf-8")) CACHE_DIR.mkdir(exist_ok=True) DOWNLOAD_DIR.mkdir(exist_ok=True) manifest: list[dict] = [] skipped: list[tuple[str, str]] = [] downloaded: list[tuple[str, str, str]] = [] no_release: list[str] = [] for idx, plugin in enumerate(plugins, 1): gh = plugin.get("github") if not gh: skipped.append((plugin["name"], "no GitHub URL")) continue owner, repo = gh["owner"], gh["repo"] print(f"[{idx}/{len(plugins)}] {owner}/{repo}") release = fetch_release(owner, repo) if release is None: # Distinguish rate-limit from missing release using cache file. cache_file = CACHE_DIR / f"{owner}_{repo}.json" if cache_file.exists(): cached = json.loads(cache_file.read_text(encoding="utf-8")) if isinstance(cached, dict) and "error" in cached: msg = cached.get("error", "") if "rate limit" in msg.lower() or "API rate limit exceeded" in msg: print(" ! GitHub API rate limit hit. Stop and retry later or set GITHUB_TOKEN.") break no_release.append(f"{owner}/{repo}") continue assets = release.get("assets", []) asset = pick_asset(assets) if asset is None: no_release.append(f"{owner}/{repo}") continue asset_url = asset["browser_download_url"] asset_name = asset["name"] safe_name = re.sub(r"[^\w\-.]", "_", asset_name) zip_path = DOWNLOAD_DIR / f"{owner}_{repo}_{safe_name}" if not zip_path.exists() or zip_path.stat().st_size == 0: print(f" ↓ {asset_name}") if not download_asset(asset_url, zip_path): skipped.append((plugin["name"], "download failed")) continue else: print(f" ✓ {asset_name} already cached") checksum = sha256_file(zip_path) version = release["tag_name"].lstrip("v") published = release.get("published_at", "") # Try to extract targetAbi from the zip if it contains a .dll with # a Jellyfin plugin manifest (meta.json). Otherwise use default. target_abi = DEFAULT_TARGET_ABI try: with zipfile.ZipFile(zip_path, "r") as zf: for name in zf.namelist(): if name.lower().endswith("meta.json"): meta = json.loads(zf.read(name)) target_abi = meta.get("targetAbi", target_abi) break except Exception: pass manifest.append( { "category": plugin["category"], "guid": make_guid(owner, repo), "name": plugin["name"], "description": plugin["description"], "overview": plugin["description"], "owner": owner, "imageUrl": "", "versions": [ { "version": version, "changelog": release.get("body", "") or "", "targetAbi": target_abi, "sourceUrl": f"file://{zip_path.resolve()}", "checksum": checksum, "timestamp": published, } ], } ) downloaded.append((plugin["name"], version, str(zip_path))) time.sleep(0.25) MANIFEST.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") print("\n" + "=" * 60) print(f"Manifest written to: {MANIFEST}") print(f"Plugins in manifest: {len(manifest)}") print(f"Downloaded zips: {len(downloaded)}") print(f"No release / no asset: {len(no_release)}") print(f"Skipped: {len(skipped)}") if no_release: print("\nRepos without a usable release/asset:") for r in no_release: print(f" - {r}") if skipped: print("\nSkipped:") for name, reason in skipped: print(f" - {name}: {reason}") return 0 if __name__ == "__main__": sys.exit(main())