"""Pick the landmark releases a task's timeline pins, and fetch their lab logos. A release is a landmark if people bookmarked it: likes are the only engagement signal the listing API returns. Two filters keep the list meaningful rather than merely popular: * repos that repackage someone else's weights (GGUF/ONNX/MLX/quantisations) are not releases, they are formats — dropped by name; * one owner may hold at most PER_OWNER slots, so a single lab shipping five variants of one model does not crowd out the rest of the field. `pins.json`, when a Space has one, is a hand-curated list merged in ahead of the automatic picks — that is how the TTS Space keeps the names it was given. Writes releases.json. Run after fetch_data.py, before build_static.py. """ from __future__ import annotations import datetime as dt import json import re import time import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path HERE = Path(__file__).parent API = "https://huggingface.co/api" UA = {"User-Agent": "audio-spaces/select_releases"} MAX_RELEASES = 52 # ceiling for the big tags AUTO_MIN = 10 # automatic picks always get room, even beside a full pin list # Likes accumulate, so ranking by them alone is a seniority list: a two-year-old # model has had two years to collect what a two-week-old one has had days for. # A release that earns this many likes this quickly is a breakout, and gets a # place regardless of where it sits in the all-time order or whether its owner # has used up their slots. RECENT_DAYS = 120 RECENT_LIKES = 100 MIN_LIKES = 10 # below this it is not a landmark, it is noise PER_OWNER = 4 CANDIDATE_PAGES = 4 # 100 per page, ordered by likes # Conversions and quantisations of another team's weights. Matched on the repo # name only, bounded by separators so "q4" does not catch "Qwen4". PORT = re.compile( r"(?:^|[-_.])(gguf|onnx|mlx|awq|bnb|nvfp4|fp8|fp16|int[48]|[iq][2-8]|q[2-8][-_]?k|" r"coreml|litert|openvino|ct2|tensorrt|trt|quantiz\w*|[48]bit|bf16|unsloth|cpp|w4a16)" r"(?:$|[-_.])", re.I) # The Hub rate-limits, and a commit walk multiplies requests. Space them out, # and back off rather than letting a 429 silently become missing data. THROTTLE = 0.12 RETRIES = 4 def fetch(url: str, timeout: int = 60) -> tuple[object, str]: """(payload, Link header). Raises after RETRIES on anything but 429.""" for attempt in range(RETRIES): time.sleep(THROTTLE) try: req = urllib.request.Request(url, headers=UA) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()), r.headers.get("Link", "") except urllib.error.HTTPError as e: if e.code != 429 or attempt == RETRIES - 1: raise time.sleep(2 ** attempt) # 1s, 2s, 4s raise RuntimeError("unreachable") def get(url: str) -> object: return fetch(url)[0] def candidates(tag: str) -> list[dict]: url = (f"{API}/models?" + urllib.parse.urlencode( {"pipeline_tag": tag, "sort": "likes", "direction": -1, "limit": 100}) + "&expand[]=createdAt&expand[]=likes&expand[]=downloads") rows: list[dict] = [] for _ in range(CANDIDATE_PAGES): page, link = fetch(url, timeout=90) rows += page nxt = re.search(r'<([^>]+)>;\s*rel="next"', link) if not nxt: break url = nxt.group(1) return rows def label_for(repo_id: str) -> str: """A short display name: the repo's own name, minus noise.""" name = repo_id.split("/", 1)[-1] name = re.sub(r"[-_.](v?\d+(\.\d+)*b|base|instruct|pretrained|ft|sft|hf|release)$", "", name, flags=re.I) return name[:26] def avatar_of(owner: str, cache: dict[str, str | None]) -> str | None: if owner in cache: return cache[owner] for kind in ("organizations", "users"): try: cache[owner] = get(f"{API}/{kind}/{owner}/overview").get("avatarUrl") if cache[owner]: return cache[owner] except Exception: continue cache.setdefault(owner, None) return cache[owner] def budget(total_models: int) -> int: """How many landmarks a tag earns. A fixed 52 would pin a fifth of a small tag's entire population, which stops meaning "landmark". Scale with the field, floor at 12 so a young tag still has a timeline, ceiling at MAX_RELEASES so a huge one stays legible. """ return max(12, min(MAX_RELEASES, total_models // 15)) def release_date(repo_id: str, created_at: str) -> str: """When the model actually landed, not when the repo was reserved. `createdAt` is the repo record's birthday, and labs routinely create a repo days or months before pushing weights into it — fishaudio/drama-3 was reserved on 1 Oct and shipped on the 13th. The oldest commit is when content first appeared, so prefer that. Two traps: the commits endpoint paginates, so the oldest entry on page one is not the first commit (Kokoro-82M has 50+); and a super-squashed history reports the squash date, with the original dates gone. Walk to the last page, and fall back to createdAt whenever the walk fails. """ url = f"{API}/models/{urllib.parse.quote(repo_id)}/commits/main" created = created_at[:10] oldest = None for _ in range(12): # bound the walk; 600 commits is plenty try: page, link = fetch(url, timeout=40) except Exception: break if not page: break oldest = page[-1]["date"][:10] # Page one usually settles it: if the oldest commit here already sits at # or before the repo's creation, nothing was pushed late and there is # nothing to find further back. Only an oldest-so-far *after* creation # is ambiguous — either a late push, or more pages to walk. if oldest <= created: return created nxt = re.search(r'<([^>]+)>;\s*rel="next"', link) if not nxt: break url = nxt.group(1) return oldest or created def age_days(created: str) -> int: d = dt.date(int(created[:4]), int(created[5:7]), int(created[8:10])) return (dt.date.today() - d).days def entry(m: dict, cache: dict[str, dict] | None = None) -> dict: prev = (cache or {}).get(m["id"], {}) date = prev["date"] if prev.get("dated") == "commit" else \ release_date(m["id"], m["createdAt"]) return { "date": date, "dated": "commit", "id": m["id"], "label": label_for(m["id"]), "downloads": m.get("downloads", 0), "likes": m.get("likes", 0), } def select(tag: str, pins: list[dict], limit: int, cache: dict[str, dict] | None = None) -> list[dict]: pinned_ids = {p["id"] for p in pins} chosen: list[dict] = [] by_owner: dict[str, int] = {} for p in pins: by_owner[p["id"].split("/")[0]] = by_owner.get(p["id"].split("/")[0], 0) + 1 rows = candidates(tag) # Breakouts first, outside the budget and the per-owner cap: a lab shipping # a hit this month should not be shut out by its own back catalogue. for m in rows: rid, created = m["id"], m.get("createdAt") if not created or rid in pinned_ids: continue if m.get("likes", 0) < RECENT_LIKES or age_days(created) > RECENT_DAYS: continue if PORT.search(rid.split("/", 1)[-1]): continue owner = rid.split("/")[0] by_owner[owner] = by_owner.get(owner, 0) + 1 pinned_ids.add(rid) chosen.append(entry(m, cache)) for m in sorted(rows, key=lambda m: -m.get("likes", 0)): if len(pinned_ids) + len(chosen) >= limit: break rid = m["id"] if rid in pinned_ids or not m.get("createdAt"): continue if m.get("likes", 0) < MIN_LIKES: break # sorted by likes, so the tail is all below if PORT.search(rid.split("/", 1)[-1]): continue owner = rid.split("/")[0] if by_owner.get(owner, 0) >= PER_OWNER: continue by_owner[owner] = by_owner.get(owner, 0) + 1 chosen.append(entry(m, cache)) return chosen def refresh_stats(entries: list[dict], cache: dict[str, dict]) -> None: """Pinned entries keep their hand-written labels but get current figures.""" for e in entries: try: d = get(f"{API}/models/{urllib.parse.quote(e['id'])}") except Exception: continue e["downloads"] = d.get("downloads", e.get("downloads", 0)) e["likes"] = d.get("likes", e.get("likes", 0)) if d.get("createdAt") and e.get("dated") != "commit": e["date"] = release_date(e["id"], d["createdAt"]) e["dated"] = "commit" def cached() -> dict[str, dict]: """What we resolved last time. A release date and a lab avatar do not change, so re-asking for them every day is what triggered the rate limit.""" path = HERE / "releases.json" if not path.exists(): return {} try: return {e["id"]: e for e in json.loads(path.read_text())} except Exception: return {} def main() -> int: tag = sys.argv[1] if len(sys.argv) > 1 else json.loads( (HERE / "task.json").read_text())["tag"] cache = cached() pins_path = HERE / "pins.json" pins = json.loads(pins_path.read_text()) if pins_path.exists() else [] if pins: for p_ in pins: prev = cache.get(p_["id"], {}) if prev.get("dated") == "commit": p_["date"], p_["dated"] = prev["date"], "commit" refresh_stats(pins, cache) total = len(json.loads((HERE / "data.json").read_text())["months"]) and sum( row[1] for row in json.loads((HERE / "data.json").read_text())["months"]) # Pins are kept whatever the budget says, so the budget has to leave room # beyond them — otherwise a pin list at or over the cap starves the picker # and no new release can ever enter the timeline. limit = max(budget(total), len(pins) + AUTO_MIN) entries = pins + select(tag, pins, limit, cache) cache_av: dict[str, str | None] = {} for e in entries: e["owner"] = e["id"].split("/")[0] avatar = avatar_of(e["owner"], cache) if avatar: e["avatar"] = avatar elif "avatar" in e: del e["avatar"] missing = [e["id"] for e in entries if not e.get("avatar")] if missing: print(f" no avatar for {len(missing)}: {', '.join(missing[:4])}", flush=True) entries = [e for e in entries if e.get("avatar")] # the timeline is logos entries.sort(key=lambda e: e["date"]) out = HERE / "releases.json" out.write_text("[\n" + ",\n".join(" " + json.dumps(e) for e in entries) + "\n]\n") print(f"{len(entries)} releases ({len(pins)} pinned, budget {limit} for {total:,} models) -> {out.name}") return 0 if __name__ == "__main__": raise SystemExit(main())