Rotate OpenDota pro-match refresh to stay under rate limits.

Daily now refreshes the 15 oldest watchlist pros, keeps prior cache for the
rest, fails fast on consecutive 429s, and still writes a partial result.

Co-authored-by: Cursor <cursoragent@cursor.com>

web/fetch_pro_matches.py

@@ -7,10 +7,16 @@ Per player: /players/{id}/matches per lobby_type (practice/tournament, and
 ranked with --include-pubs), up to --limit newest each; then /matches/{id}
 for final items + skill builds. League rows do not crowd out ranked pubs.
 
+Daily refresh uses --refresh-limit to rotate the oldest / missing pros so the
+OpenDota anonymous quota is not burned on a full 90-player crawl every night.
+Optional OPENDOTA_API_KEY raises rate limits. Consecutive 429s trip a circuit
+breaker: remaining pros keep prior cache and the script still writes.
+
 Output: web/data/pro_matches.json (Climperor web only; not used by recommend).
 
 Usage:
   python web/fetch_pro_matches.py
+  python web/fetch_pro_matches.py --include-pubs --refresh-limit 15
   python web/fetch_pro_matches.py --include-pubs --limit 8
   python web/fetch_pro_matches.py --players 898754153,Ame
   python web/fetch_pro_matches.py --all-pros --limit-pros 20 --with-team
@@ -36,8 +42,8 @@ from shared.paths import DATA
 from fetch_hero_matches import (
     collect_item_ids,
     extract_player_row,
-    fetch_match,
     load_ability_id_map,
+    opendota_url,
 )
 from fetch_hero_items import load_item_catalog
 from fetch_pro_builds import fetch_pro_index
@@ -47,10 +53,48 @@ OUT = DATA / "pro_matches.json"
 WATCHLIST = DATA / "pro_player_watchlist.json"
 DEFAULT_LIMIT = 8
 DEFAULT_LIMIT_PROS = 40
+DEFAULT_REFRESH_LIMIT = 0  # 0 = refresh all selected pros
+DEFAULT_429_STREAK = 3
 # OpenDota lobby_type: 1=practice, 2=tournament (pro/league biased).
 LOBBY_LEAGUE = (1, 2)
 
 
+class RateLimitTripped(Exception):
+    """OpenDota returned enough consecutive 429s to abort the remaining batch."""
+
+
+class OpenDotaClient:
+    """Fail-fast OpenDota JSON client with consecutive-429 circuit breaker."""
+
+    def __init__(self, *, consecutive_limit: int = DEFAULT_429_STREAK) -> None:
+        self.consecutive_limit = max(1, int(consecutive_limit))
+        self.consecutive_429 = 0
+        self.tripped = False
+
+    def json(self, path: str) -> dict | list:
+        if self.tripped:
+            raise RateLimitTripped("opendota rate-limit circuit open")
+        url = opendota_url(path)
+        try:
+            # No 5/10/20/40s retry chain — daily must finish within budget.
+            data = http_json(url, retries=0)
+        except urllib.error.HTTPError as e:
+            if e.code == 429:
+                self.consecutive_429 += 1
+                _log(
+                    f"HTTP 429 {url} — streak "
+                    f"{self.consecutive_429}/{self.consecutive_limit}"
+                )
+                if self.consecutive_429 >= self.consecutive_limit:
+                    self.tripped = True
+                    raise RateLimitTripped(
+                        f"opendota 429 x{self.consecutive_429}"
+                    ) from e
+            raise
+        self.consecutive_429 = 0
+        return data
+
+
 def _log(msg: str) -> None:
     try:
         print(msg, flush=True)
@@ -113,6 +157,58 @@ def load_watchlist(path: Path) -> list[dict]:
     return out
 
 
+def load_existing(path: Path) -> dict:
+    """Load prior pro_matches.json; empty dict when missing/unreadable."""
+    if not path.is_file():
+        return {}
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return raw if isinstance(raw, dict) else {}
+
+
+def parse_ts(value: object) -> datetime | None:
+    if not isinstance(value, str) or not value.strip():
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def cell_fetched_at(cell: dict | None) -> datetime | None:
+    if not isinstance(cell, dict):
+        return None
+    return parse_ts(cell.get("fetched_at"))
+
+
+def select_refresh_batch(
+    picked: list[tuple[int, dict]],
+    existing_by_pro: dict[str, dict],
+    refresh_limit: int,
+) -> tuple[list[tuple[int, dict]], list[tuple[int, dict]]]:
+    """Split watchlist into (to_refresh, to_retain) by staleness.
+
+    Missing / unscored cells sort oldest. ``refresh_limit <= 0`` refreshes all.
+    """
+    if refresh_limit <= 0 or refresh_limit >= len(picked):
+        return list(picked), []
+
+    ranked: list[tuple[float, int, tuple[int, dict]]] = []
+    for idx, item in enumerate(picked):
+        aid, _prof = item
+        ts = cell_fetched_at(existing_by_pro.get(str(aid)))
+        # Missing timestamp => oldest (refresh first).
+        score = ts.timestamp() if ts is not None else float("-inf")
+        ranked.append((score, idx, item))
+    ranked.sort(key=lambda row: (row[0], row[1]))
+    to_refresh = [item for _score, _idx, item in ranked[:refresh_limit]]
+    refresh_ids = {aid for aid, _ in to_refresh}
+    to_retain = [item for item in picked if item[0] not in refresh_ids]
+    return to_refresh, to_retain
+
+
 def parse_pro_filter(raw: str, pro_index: dict[int, dict]) -> list[int]:
     """Comma-separated account ids or registered pro names."""
     if not raw.strip():
@@ -219,6 +315,7 @@ def player_match_metas(
     account_id: int,
     limit: int,
     *,
+    client: OpenDotaClient,
     lobby_types: tuple[int, ...] = LOBBY_LEAGUE,
 ) -> list[dict]:
     """Recent match list rows for a pro (deduped, newest first).
@@ -229,9 +326,11 @@ def player_match_metas(
     """
     by_id: dict[int, dict] = {}
     for lt in lobby_types:
-        url = f"{OPENDOTA}/players/{account_id}/matches?limit={limit}&lobby_type={lt}"
+        path = f"/players/{account_id}/matches?limit={limit}&lobby_type={lt}"
         try:
-            raw = http_json(url)
+            raw = client.json(path)
+        except RateLimitTripped:
+            raise
         except (
             urllib.error.HTTPError,
             urllib.error.URLError,
@@ -269,9 +368,26 @@ def player_match_metas(
     )
 
 
+def fetch_match_detail(client: OpenDotaClient, match_id: int) -> dict | None:
+    try:
+        raw = client.json(f"/matches/{match_id}")
+    except RateLimitTripped:
+        raise
+    except (
+        urllib.error.HTTPError,
+        urllib.error.URLError,
+        TimeoutError,
+        json.JSONDecodeError,
+        OSError,
+    ):
+        return None
+    return raw if isinstance(raw, dict) else None
+
+
 def fetch_player_matches(
     account_id: int,
     *,
+    client: OpenDotaClient,
     limit: int,
     id_map: dict[int, str],
     catalog: dict[int, dict],
@@ -279,7 +395,9 @@ def fetch_player_matches(
     delay: float,
     lobby_types: tuple[int, ...],
 ) -> list[dict]:
-    metas = player_match_metas(account_id, limit, lobby_types=lobby_types)
+    metas = player_match_metas(
+        account_id, limit, client=client, lobby_types=lobby_types
+    )
     rows: list[dict] = []
     for meta in metas:
         try:
@@ -295,7 +413,7 @@ def fetch_player_matches(
         except (TypeError, ValueError):
             lt_i = None
         row_origin = "public" if lt_i == 7 else "pro"
-        detail = fetch_match(mid)
+        detail = fetch_match_detail(client, mid)
         if delay > 0:
             time.sleep(delay)
         if not detail:
@@ -363,6 +481,32 @@ def build_indexes(
     return out
 
 
+def retain_cell(
+    aid: int,
+    prof: dict,
+    existing: dict | None,
+) -> dict:
+    """Keep prior matches for a pro not refreshed this round."""
+    if isinstance(existing, dict) and isinstance(existing.get("matches"), list):
+        cell = dict(existing)
+        cell["account_id"] = aid
+        for key in ("name", "team_tag", "team_name", "country_code"):
+            if prof.get(key) and not cell.get(key):
+                cell[key] = prof.get(key)
+        cell["match_count"] = len(cell.get("matches") or [])
+        return cell
+    return {
+        "account_id": aid,
+        "name": prof.get("name"),
+        "team_tag": prof.get("team_tag"),
+        "team_name": prof.get("team_name"),
+        "country_code": prof.get("country_code"),
+        "match_count": 0,
+        "matches": [],
+        "fetched_at": None,
+    }
+
+
 def write_out(
     path: Path,
     *,
@@ -374,6 +518,10 @@ def write_out(
     limit_pros: int,
     lobby_types: tuple[int, ...],
     player_source: str,
+    refreshed_count: int,
+    retained_count: int,
+    refresh_limit: int,
+    rate_limited: bool,
 ) -> None:
     used = collect_item_ids(by_hero)
     items_out = {
@@ -402,10 +550,15 @@ def write_out(
             "pro_count": len(by_pro),
             "match_count": match_count,
             "hero_count": len(by_hero),
+            "refresh_limit": refresh_limit,
+            "refreshed_count": refreshed_count,
+            "retained_count": retained_count,
+            "rate_limited": bool(rate_limited),
             "note_zh": (
                 "OpenDota 明星选手近期联赛/锦标赛对局（可选含天梯 lobby_type=7）；"
                 "默认名单见 web/data/pro_player_watchlist.json；"
                 "每种 lobby 各保留最近 limit 场，避免联赛挤掉天梯；"
+                "daily 按 fetched_at 轮换最陈旧选手（--refresh-limit）；"
                 "lobby_type 1=训练/practice、2=tournament、7=ranked；"
                 "含终局出装、加点与联赛名（若有）。"
             ),
@@ -431,7 +584,7 @@ def main() -> None:
         "--limit",
         type=int,
         default=DEFAULT_LIMIT,
-        help=f"Matches per pro (default: {DEFAULT_LIMIT})",
+        help=f"Matches per lobby per pro (default: {DEFAULT_LIMIT})",
     )
     ap.add_argument(
         "--limit-pros",
@@ -439,6 +592,16 @@ def main() -> None:
         default=DEFAULT_LIMIT_PROS,
         help=f"Max pros when --all-pros (default: {DEFAULT_LIMIT_PROS})",
     )
+    ap.add_argument(
+        "--refresh-limit",
+        type=int,
+        default=DEFAULT_REFRESH_LIMIT,
+        help=(
+            "Max pros to refresh this run by oldest fetched_at "
+            f"(0=all; default: {DEFAULT_REFRESH_LIMIT}). "
+            "Ignored when --players is set."
+        ),
+    )
     ap.add_argument(
         "--players",
         default="",
@@ -466,15 +629,25 @@ def main() -> None:
         action="store_true",
         help="Also include ranked pub lobby_type=7 (high-MMR scrims)",
     )
+    ap.add_argument(
+        "--429-streak",
+        dest="streak_429",
+        type=int,
+        default=DEFAULT_429_STREAK,
+        help=f"Consecutive 429s before aborting remaining pros (default: {DEFAULT_429_STREAK})",
+    )
     args = ap.parse_args()
 
     limit = max(1, int(args.limit))
     limit_pros = max(1, int(args.limit_pros))
+    refresh_limit = max(0, int(args.refresh_limit))
     active_days = int(args.active_days) if args.active_days > 0 else None
     lobby_types: tuple[int, ...] = LOBBY_LEAGUE
     if args.include_pubs:
         lobby_types = LOBBY_LEAGUE + (7,)
 
+    client = OpenDotaClient(consecutive_limit=int(args.streak_429))
+
     _log("loading pro players ...")
     pro_index = fetch_pro_index()
     _log(f"  {len(pro_index)} registered pros")
@@ -508,6 +681,25 @@ def main() -> None:
     if not picked:
         raise SystemExit("No pros matched filters")
 
+    existing = load_existing(args.out)
+    existing_by_pro = existing.get("by_pro") if isinstance(existing.get("by_pro"), dict) else {}
+    assert isinstance(existing_by_pro, dict)
+
+    # Explicit --players means refresh those fully (no rotation).
+    if player_ids:
+        to_refresh, to_retain = list(picked), []
+        effective_refresh_limit = 0
+    else:
+        to_refresh, to_retain = select_refresh_batch(
+            picked, existing_by_pro, refresh_limit
+        )
+        effective_refresh_limit = refresh_limit
+
+    _log(
+        f"  refresh batch={len(to_refresh)} retain={len(to_retain)} "
+        f"refresh_limit={effective_refresh_limit or 'all'}"
+    )
+
     heroes = hero_table()
     id_to_key = {int(h["id"]): h["key"] for h in heroes}
     catalog, _ = load_item_catalog()
@@ -516,20 +708,57 @@ def main() -> None:
     by_pro: dict[str, dict] = {}
     pros_meta: dict[str, dict] = {}
     total_matches = 0
+    refreshed_count = 0
+    rate_limited = False
+    now_iso = datetime.now(timezone.utc).isoformat()
 
-    for i, (aid, prof) in enumerate(picked, 1):
-        label = prof.get("name") or prof.get("team_tag") or str(aid)
-        _log(f"[{i}/{len(picked)}] {label} ({aid}) ...")
-        matches = fetch_player_matches(
-            aid,
-            limit=limit,
-            id_map=id_map,
-            catalog=catalog,
-            id_to_key=id_to_key,
-            delay=args.delay,
-            lobby_types=lobby_types,
-        )
+    for aid, prof in to_retain:
         sid = str(aid)
+        cell = retain_cell(aid, prof, existing_by_pro.get(sid))
+        by_pro[sid] = cell
+        pros_meta[sid] = {
+            "account_id": aid,
+            "name": cell.get("name") or prof.get("name"),
+            "team_tag": cell.get("team_tag") or prof.get("team_tag"),
+            "team_name": cell.get("team_name") or prof.get("team_name"),
+            "country_code": cell.get("country_code") or prof.get("country_code"),
+        }
+        total_matches += len(cell.get("matches") or [])
+
+    for i, (aid, prof) in enumerate(to_refresh, 1):
+        label = prof.get("name") or prof.get("team_tag") or str(aid)
+        sid = str(aid)
+        _log(f"[{i}/{len(to_refresh)}] {label} ({aid}) ...")
+        try:
+            matches = fetch_player_matches(
+                aid,
+                client=client,
+                limit=limit,
+                id_map=id_map,
+                catalog=catalog,
+                id_to_key=id_to_key,
+                delay=args.delay,
+                lobby_types=lobby_types,
+            )
+        except RateLimitTripped as exc:
+            rate_limited = True
+            _log(f"  rate-limited; keeping prior cache for remaining ({exc})")
+            # Keep old (or empty) for this pro and every leftover refresh target.
+            remaining = to_refresh[i - 1 :]
+            for raid, rprof in remaining:
+                rsid = str(raid)
+                cell = retain_cell(raid, rprof, existing_by_pro.get(rsid))
+                by_pro[rsid] = cell
+                pros_meta[rsid] = {
+                    "account_id": raid,
+                    "name": cell.get("name") or rprof.get("name"),
+                    "team_tag": cell.get("team_tag") or rprof.get("team_tag"),
+                    "team_name": cell.get("team_name") or rprof.get("team_name"),
+                    "country_code": cell.get("country_code") or rprof.get("country_code"),
+                }
+                total_matches += len(cell.get("matches") or [])
+            break
+
         by_pro[sid] = {
             "account_id": aid,
             "name": prof.get("name"),
@@ -538,6 +767,7 @@ def main() -> None:
             "country_code": prof.get("country_code"),
             "match_count": len(matches),
             "matches": matches,
+            "fetched_at": now_iso,
         }
         pros_meta[sid] = {
             "account_id": aid,
@@ -546,9 +776,11 @@ def main() -> None:
             "team_name": prof.get("team_name"),
             "country_code": prof.get("country_code"),
         }
+        refreshed_count += 1
         total_matches += len(matches)
         _log(f"  {len(matches)} matches")
 
+    retained_count = len(by_pro) - refreshed_count
     by_hero = build_indexes(by_pro, id_to_key)
     write_out(
         args.out,
@@ -560,9 +792,15 @@ def main() -> None:
         limit_pros=len(picked),
         lobby_types=lobby_types,
         player_source=player_source,
+        refreshed_count=refreshed_count,
+        retained_count=max(0, retained_count),
+        refresh_limit=effective_refresh_limit,
+        rate_limited=rate_limited,
     )
     _log(
-        f"done pros={len(by_pro)} matches={total_matches} heroes={len(by_hero)} → {args.out}"
+        f"done pros={len(by_pro)} refreshed={refreshed_count} "
+        f"retained={max(0, retained_count)} matches={total_matches} "
+        f"heroes={len(by_hero)} rate_limited={rate_limited} → {args.out}"
     )