"""Fetch STRATZ hero meta (weekly WR/pick by bracket + position) and matchup tops. Requires STRATZ_API_TOKEN (or KEYZOO_ASSET_API_KEY) in the environment / .env. Output is for the Climperor web site only — never merge into relations.json or recommend. Data sources (heroStats GraphQL): - winWeek(take=N, bracketIds): per-medal weekly pick/win → `weeks` / `latest` - winWeek(take=1, bracketIds, positionIds): per-medal per-position latest week → `positions` (same time window as headline cards; exact medal, not basic merge) - matchUp: counter / countered / synergy tops → stratz_matchup_tops.json (global aggregate — no bracket / position / week filter) Usage: python web/fetch_stratz_meta.py python web/fetch_stratz_meta.py --weeks 8 --delay 0.25 python web/fetch_stratz_meta.py --skip-matchups python web/fetch_stratz_meta.py --matchups-only python web/fetch_stratz_meta.py --resume-matchups # interrupt resume only """ from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import argparse import json import os import tempfile import time import urllib.error import urllib.request from datetime import datetime, timezone from shared.grid import hero_table from shared.paths import DATA, ROOT API = "https://api.stratz.com/graphql" OUT_META = DATA / "stratz_hero_meta.json" OUT_MATCHUPS = DATA / "stratz_matchup_tops.json" # UI medal keys ↔ STRATZ RankBracket enum BRACKET_ORDER = ( "herald", "guardian", "crusader", "archon", "legend", "ancient", "divine", "immortal", ) BRACKET_ENUM = { "herald": "HERALD", "guardian": "GUARDIAN", "crusader": "CRUSADER", "archon": "ARCHON", "legend": "LEGEND", "ancient": "ANCIENT", "divine": "DIVINE", "immortal": "IMMORTAL", } # Medal → RankBracketBasicEnum (legacy; position stats now use exact medal via winWeek) BRACKET_TO_BASIC = { "herald": "HERALD_GUARDIAN", "guardian": "HERALD_GUARDIAN", "crusader": "CRUSADER_ARCHON", "archon": "CRUSADER_ARCHON", "legend": "LEGEND_ANCIENT", "ancient": "LEGEND_ANCIENT", "divine": "DIVINE_IMMORTAL", "immortal": "DIVINE_IMMORTAL", } POSITION_ORDER = ( "POSITION_1", "POSITION_2", "POSITION_3", "POSITION_4", "POSITION_5", ) ATTRIBUTION = "https://stratz.com" MATCHUP_SCOPE = { "kind": "global_aggregate", "bracket": None, "position": None, "week": None, "label_zh": "全局聚合（未按段位 / 分路 / 周过滤）", "note": ( "STRATZ heroStats.matchUp without bracketIds/positionIds/week. " "advantage is upstream synergy (relative), not raw win-rate pp. " "Web-only; do not merge into relations.json." ), } MATCHUP_NOTE = ( "counters = positive vs advantage (STRATZ synergy); " "countered = heroes this hero loses to (negated advantage, mirrored wr); " "synergies = with synergy. " "advantage ≠ win-rate percentage points. " "Web-only; do not merge into relations.json." ) def load_token() -> str: for key in ( "STRATZ_API_TOKEN", "KEYZOO_ASSET_API_KEY", "KEYZOO_ASSET_SECRET_API_KEY", "KEYZOO_ASSET_TOKEN", ): env = os.environ.get(key, "").strip() if env: return env path = ROOT / ".env" if path.is_file(): for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) if k.strip() == "STRATZ_API_TOKEN": return v.strip().strip('"').strip("'") raise SystemExit( "STRATZ_API_TOKEN missing. Set env / keyzoo inject or add to .env " "(see .env.example)." ) def gql(token: str, query: str, variables: dict | None = None) -> dict: body: dict = {"query": query} if variables: body["variables"] = variables req = urllib.request.Request( API, data=json.dumps(body).encode(), headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "User-Agent": "climperor", }, method="POST", ) with urllib.request.urlopen(req, timeout=120) as resp: payload = json.loads(resp.read().decode()) if payload.get("errors"): raise RuntimeError(str(payload["errors"][:2])) return payload["data"] def _pw(pick: int, win: int) -> dict: pick = max(0, int(pick or 0)) win = max(0, int(win or 0)) wr = (win / pick) if pick > 0 else None return {"pick": pick, "win": win, "wr": wr} def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def write_json_atomic(path: Path, payload: dict) -> None: """Write JSON via temp file then replace, so partial writes never corrupt cache.""" path.parent.mkdir(parents=True, exist_ok=True) text = json.dumps(payload, ensure_ascii=False, indent=2) + "\n" fd, tmp_name = tempfile.mkstemp( prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent), ) tmp_path = Path(tmp_name) try: with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(text) os.replace(tmp_path, path) except Exception: try: tmp_path.unlink(missing_ok=True) except OSError: pass raise def fetch_weeks_for_bracket( token: str, hero_ids: list[int], bracket: str, take: int ) -> list[dict]: enum = BRACKET_ENUM[bracket] # GraphQL list variable as inline enums (Short! list for heroIds). ids_lit = ", ".join(str(i) for i in hero_ids) query = f""" query {{ heroStats {{ winWeek(heroIds: [{ids_lit}], take: {int(take)}, bracketIds: [{enum}]) {{ heroId week matchCount winCount }} }} }} """ data = gql(token, query) return (((data or {}).get("heroStats") or {}).get("winWeek")) or [] def fetch_latest_positions_for_bracket( token: str, hero_ids: list[int], bracket: str ) -> list[dict]: """Latest-week winWeek per position; same bracket + week as headline cards.""" enum = BRACKET_ENUM[bracket] ids_lit = ", ".join(str(i) for i in hero_ids) out: list[dict] = [] for pos in POSITION_ORDER: query = f""" query {{ heroStats {{ winWeek( heroIds: [{ids_lit}] take: 1 bracketIds: [{enum}] positionIds: [{pos}] ) {{ heroId week matchCount winCount }} }} }} """ data = gql(token, query) for row in (((data or {}).get("heroStats") or {}).get("winWeek")) or []: out.append({**row, "position": pos, "bracket": bracket}) return out MATCHUP_QUERY = """ query($id: Short!, $take: Int!, $limit: Int!) { heroStats { matchUp(heroId: $id, take: $take, matchLimit: $limit) { heroId matchCountVs matchCountWith vs { heroId2 matchCount winCount synergy winsAverage } with { heroId2 matchCount winCount synergy winsAverage } } } } """ def parse_matchup_pairs(row: dict) -> tuple[list[dict], list[dict]]: """Parse raw STRATZ matchUp row into vs / with pair lists.""" vs_out: list[dict] = [] with_out: list[dict] = [] for pair in row.get("vs") or []: other = pair.get("heroId2") games = int(pair.get("matchCount") or 0) if other is None or games <= 0: continue wins = int(pair.get("winCount") or 0) vs_out.append( { "hero_id": int(other), "games": games, "wins": wins, "advantage": float(pair.get("synergy") or 0.0), "wr": float(pair.get("winsAverage") or (wins / games)), } ) for pair in row.get("with") or []: other = pair.get("heroId2") games = int(pair.get("matchCount") or 0) if other is None or games <= 0: continue wins = int(pair.get("winCount") or 0) with_out.append( { "hero_id": int(other), "games": games, "wins": wins, "synergy": float(pair.get("synergy") or 0.0), "wr": float(pair.get("winsAverage") or (wins / games)), } ) return vs_out, with_out def rank_matchup_lists( vs_out: list[dict], with_out: list[dict], take: int ) -> dict: """Sort counters / countered / synergies from parsed vs / with pairs. Positive ``advantage`` means this hero's STRATZ relative score vs the peer is favorable — not raw win-rate percentage points. A hero can appear under counters with wr < 0.5 when advantage is still positive vs baseline. """ take = max(0, int(take)) # vs advantage: positive = hero wins more vs other → counters other. # countered is the same vs list mirrored (negated advantage, 1-wr). disadvantage = [ { "hero_id": e["hero_id"], "games": e["games"], "wins": e["wins"], "advantage": -float(e["advantage"]), "wr": 1.0 - float(e["wr"]) if e["wr"] is not None else None, } for e in sorted(vs_out, key=lambda x: x["advantage"])[:take] ] counters = sorted(vs_out, key=lambda x: -x["advantage"])[:take] synergies = sorted(with_out, key=lambda x: -x["synergy"])[:take] return { "counters": counters, "countered": disadvantage, "synergies": synergies, } def fetch_matchup_tops( token: str, hero_id: int, *, take: int, match_limit: int ) -> dict: data = gql( token, MATCHUP_QUERY, {"id": hero_id, "take": take, "limit": match_limit}, ) rows = (((data or {}).get("heroStats") or {}).get("matchUp")) or [] row = rows[0] if rows else {} vs_out, with_out = parse_matchup_pairs(row) return rank_matchup_lists(vs_out, with_out, take) def annotate_matchup_cell( cell: dict, *, fetched_at: str, stale: bool = False ) -> dict: """Attach per-hero fetch metadata; preserve list payloads.""" out = { "counters": list(cell.get("counters") or []), "countered": list(cell.get("countered") or []), "synergies": list(cell.get("synergies") or []), "fetched_at": fetched_at, "stale": bool(stale), } return out def build_matchup_file_payload( by_hero: dict[str, dict], *, take: int, match_limit: int, started_at: str, finished_at: str | None = None, stats: dict | None = None, ) -> dict: return { "fetched_at": finished_at or started_at, "started_at": started_at, "finished_at": finished_at, "source": "stratz", "attribution": ATTRIBUTION, "take": take, "match_limit": match_limit, "scope": dict(MATCHUP_SCOPE), "note": MATCHUP_NOTE, "stats": stats or { "heroes": len(by_hero), "ok": 0, "failed": 0, "stale_kept": 0, }, "by_hero": by_hero, } def load_previous_matchups(path: Path) -> dict[str, dict]: if not path.is_file(): return {} try: prev = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} by_hero = prev.get("by_hero") or {} return {k: v for k, v in by_hero.items() if isinstance(v, dict)} def build_meta( weeks_rows_by_bracket: dict[str, list[dict]], position_rows_by_bracket: dict[str, list[dict]], id_to_key: dict[int, str], weeks_take: int, ) -> dict: by_hero: dict[str, dict] = {} for hid, key in id_to_key.items(): by_hero[key] = { "id": hid, "weeks": {b: [] for b in BRACKET_ORDER}, "latest": {}, "positions": {b: {} for b in BRACKET_ORDER}, } for bracket, rows in weeks_rows_by_bracket.items(): # Group by hero, sort weeks desc, keep take by_id: dict[int, list[dict]] = {} for row in rows: hid = int(row.get("heroId") or 0) if hid not in id_to_key: continue by_id.setdefault(hid, []).append(row) for hid, rows_h in by_id.items(): key = id_to_key[hid] rows_h.sort(key=lambda r: int(r.get("week") or 0), reverse=True) weeks = [] for r in rows_h[:weeks_take]: pick = int(r.get("matchCount") or 0) win = int(r.get("winCount") or 0) weeks.append( { "week": int(r.get("week") or 0), "pick": pick, "win": win, "wr": (win / pick) if pick > 0 else None, } ) by_hero[key]["weeks"][bracket] = weeks if weeks: latest = weeks[0] by_hero[key]["latest"][bracket] = _pw(latest["pick"], latest["win"]) for bracket, pos_rows in position_rows_by_bracket.items(): for row in pos_rows: hid = int(row.get("heroId") or 0) key = id_to_key.get(hid) if not key: continue pos = str(row.get("position") or "") if pos not in POSITION_ORDER: continue cell = _pw(int(row.get("matchCount") or 0), int(row.get("winCount") or 0)) by_hero[key]["positions"][bracket][pos] = cell # Totals (latest week pick sum per bracket) for pick-rate denominator totals: dict[str, dict] = {} for bracket in BRACKET_ORDER: total_pick = 0 for cell in by_hero.values(): latest = (cell.get("latest") or {}).get(bracket) or {} total_pick += int(latest.get("pick") or 0) totals[bracket] = {"pick": total_pick} # Meta board: top heroes by pick (latest week) per bracket meta_board: dict[str, list[dict]] = {} for bracket in BRACKET_ORDER: denom = max(1, int((totals.get(bracket) or {}).get("pick") or 0)) rows_board = [] for key, cell in by_hero.items(): latest = (cell.get("latest") or {}).get(bracket) if not latest or int(latest.get("pick") or 0) <= 0: continue pick = int(latest["pick"]) win = int(latest["win"]) wr = win / pick if pick else None # Same convention as OpenDota web: pick / (Σpick / 10) pr = pick / (denom / 10.0) if denom else None rows_board.append( { "key": key, "id": cell["id"], "pick": pick, "win": win, "wr": wr, "pr": pr, } ) rows_board.sort(key=lambda r: (-r["pick"], -(r["wr"] or 0), r["key"])) meta_board[bracket] = rows_board return { "fetched_at": _now_iso(), "source": "stratz", "attribution": ATTRIBUTION, "weeks_take": weeks_take, "window_label_zh": f"近 {weeks_take} 周天梯（按段位）", "latest_window_label_zh": "最近 1 周", "brackets": list(BRACKET_ORDER), "positions": list(POSITION_ORDER), "bracket_position_note": None, "totals": totals, "by_hero": by_hero, "meta_board": meta_board, } def refresh_matchup_tops( token: str, id_to_key: dict[int, str], *, take: int, match_limit: int, delay: float, out_path: Path, resume: bool, ) -> dict: """Full-refresh (default) or resume-only matchup tops. On per-hero failure, keep the previous cell and mark ``stale=True``. """ started_at = _now_iso() prev = load_previous_matchups(out_path) hero_ids = sorted(id_to_key.keys()) if resume: pending = [hid for hid in hero_ids if id_to_key[hid] not in prev] by_hero_mu: dict[str, dict] = dict(prev) print( f"resuming matchups: {len(pending)} pending / {len(hero_ids)} " f"(cached={len(prev)})", flush=True, ) else: pending = list(hero_ids) by_hero_mu = {} print( f"full matchup refresh: {len(pending)} heroes " f"(prev cached={len(prev)} as failure fallback)", flush=True, ) ok = failed = stale_kept = 0 for n, hid in enumerate(pending, start=1): key = id_to_key[hid] try: cell = fetch_matchup_tops( token, hid, take=take, match_limit=match_limit, ) by_hero_mu[key] = annotate_matchup_cell( cell, fetched_at=_now_iso(), stale=False ) ok += 1 print( f" [{n}/{len(pending)}] {key}: " f"vs={len(cell['counters'])} fear={len(cell['countered'])} " f"with={len(cell['synergies'])}", flush=True, ) except ( urllib.error.URLError, TimeoutError, RuntimeError, json.JSONDecodeError, ) as e: failed += 1 old = prev.get(key) if old: kept = annotate_matchup_cell( old, fetched_at=str(old.get("fetched_at") or started_at), stale=True, ) by_hero_mu[key] = kept stale_kept += 1 print( f" [{n}/{len(pending)}] {key} failed (kept stale): {e}", flush=True, ) else: print(f" [{n}/{len(pending)}] {key} failed: {e}", flush=True) time.sleep(delay * 2) continue finished_partial = _now_iso() payload = build_matchup_file_payload( by_hero_mu, take=take, match_limit=match_limit, started_at=started_at, finished_at=finished_partial, stats={ "heroes": len(by_hero_mu), "ok": ok, "failed": failed, "stale_kept": stale_kept, "pending_left": len(pending) - n, }, ) write_json_atomic(out_path, payload) time.sleep(delay) # Resume mode: ensure heroes already present stay; full mode already has all ok/stale. if resume: for hid in hero_ids: key = id_to_key[hid] if key not in by_hero_mu and key in prev: by_hero_mu[key] = annotate_matchup_cell( prev[key], fetched_at=str(prev[key].get("fetched_at") or started_at), stale=bool(prev[key].get("stale")), ) finished_at = _now_iso() payload = build_matchup_file_payload( by_hero_mu, take=take, match_limit=match_limit, started_at=started_at, finished_at=finished_at, stats={ "heroes": len(by_hero_mu), "ok": ok, "failed": failed, "stale_kept": stale_kept, "pending_left": 0, }, ) write_json_atomic(out_path, payload) print( f"done: matchups heroes={len(by_hero_mu)} ok={ok} failed={failed} " f"stale_kept={stale_kept} → {out_path}", flush=True, ) return payload def main() -> None: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--weeks", type=int, default=8, help="weekly buckets to keep") ap.add_argument("--delay", type=float, default=0.25) ap.add_argument("--out-meta", type=Path, default=OUT_META) ap.add_argument("--out-matchups", type=Path, default=OUT_MATCHUPS) ap.add_argument("--skip-matchups", action="store_true") ap.add_argument( "--matchups-only", action="store_true", help="skip winWeek meta; only refresh matchup tops", ) ap.add_argument( "--resume-matchups", action="store_true", help="only fetch heroes missing from existing matchup cache (interrupt resume)", ) ap.add_argument("--matchup-take", type=int, default=12) ap.add_argument("--matchup-min-games", type=int, default=50) args = ap.parse_args() token = load_token() heroes = hero_table() id_to_key = {int(h["id"]): h["key"] for h in heroes} hero_ids = sorted(id_to_key.keys()) if not args.matchups_only: print( f"fetching STRATZ meta for {len(hero_ids)} heroes, " f"{args.weeks} weeks × {len(BRACKET_ORDER)} brackets", flush=True, ) weeks_by_bracket: dict[str, list[dict]] = {} position_rows_by_bracket: dict[str, list[dict]] = {} for i, bracket in enumerate(BRACKET_ORDER, start=1): try: rows = fetch_weeks_for_bracket(token, hero_ids, bracket, args.weeks) except ( urllib.error.URLError, TimeoutError, RuntimeError, json.JSONDecodeError, ) as e: print(f" [{i}/{len(BRACKET_ORDER)}] {bracket} failed: {e}", flush=True) rows = [] weeks_by_bracket[bracket] = rows try: pos_rows = fetch_latest_positions_for_bracket(token, hero_ids, bracket) except ( urllib.error.URLError, TimeoutError, RuntimeError, json.JSONDecodeError, ) as e: print( f" [{i}/{len(BRACKET_ORDER)}] {bracket} positions failed: {e}", flush=True, ) pos_rows = [] position_rows_by_bracket[bracket] = pos_rows print( f" [{i}/{len(BRACKET_ORDER)}] {bracket}: {len(rows)} week rows, " f"{len(pos_rows)} position rows", flush=True, ) time.sleep(args.delay) meta = build_meta( weeks_by_bracket, position_rows_by_bracket, id_to_key, args.weeks ) write_json_atomic(args.out_meta, meta) print(f"wrote {args.out_meta}", flush=True) if args.skip_matchups: print("skip matchup tops", flush=True) return refresh_matchup_tops( token, id_to_key, take=args.matchup_take, match_limit=args.matchup_min_games, delay=args.delay, out_path=args.out_matchups, resume=bool(args.resume_matchups), ) if __name__ == "__main__": main()