climperor/web/fetch_hero_stats.py

"""Fetch OpenDota hero pick/win stats (all brackets) into data/hero_stats.json.

One GET /api/heroStats call covers every hero. OpenDota aggregates *recent*
matches: pub_pick equals the sum of pub_pick_trend (typically 7 daily buckets),
and bracket fields share that same recent window — not all-time / full patch.

Output is for the Climperor web site only — never merge into relations.json /
heroes.json or recommend.

Usage:
  python fetch_hero_stats.py
  python fetch_hero_stats.py --out data/hero_stats.json
"""

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import argparse
from datetime import datetime, timezone

from shared.grid import hero_table
from shared.http_utils import http_json, write_json_atomic
from shared.paths import DATA

OPENDOTA_HERO_STATS = "https://api.opendota.com/api/heroStats"
OUT = DATA / "hero_stats.json"

# OpenDota skill brackets: 1=Herald … 8=Immortal
BRACKET_ORDER = (
    "herald",
    "guardian",
    "crusader",
    "archon",
    "legend",
    "ancient",
    "divine",
    "immortal",
)
BRACKET_NUM = {name: i for i, name in enumerate(BRACKET_ORDER, start=1)}
# Fallback when a heroStats row has no pub_*_trend arrays.
DEFAULT_WINDOW_DAYS = 7


def _pw(pick: object, win: object) -> dict[str, int]:
    try:
        p = int(pick or 0)
    except (TypeError, ValueError):
        p = 0
    try:
        w = int(win or 0)
    except (TypeError, ValueError):
        w = 0
    return {"pick": max(0, p), "win": max(0, w)}


def row_to_hero_stats(row: dict) -> dict:
    brackets: dict[str, dict[str, int]] = {}
    for name, num in BRACKET_NUM.items():
        brackets[name] = _pw(row.get(f"{num}_pick"), row.get(f"{num}_win"))
    pro = _pw(row.get("pro_pick"), row.get("pro_win"))
    try:
        ban = int(row.get("pro_ban") or 0)
    except (TypeError, ValueError):
        ban = 0
    pro["ban"] = max(0, ban)
    return {
        "pub": _pw(row.get("pub_pick"), row.get("pub_win")),
        "brackets": brackets,
        "pro": pro,
        "turbo": _pw(row.get("turbo_picks"), row.get("turbo_wins")),
    }


def detect_window_days(rows: list) -> int:
    """Infer recent-window length from OpenDota pub_*_trend bucket counts."""
    for row in rows:
        if not isinstance(row, dict):
            continue
        for key in ("pub_pick_trend", "pub_win_trend"):
            trend = row.get(key)
            if isinstance(trend, list) and trend:
                return len(trend)
    return DEFAULT_WINDOW_DAYS


def sum_picks(by_hero: dict[str, dict], path: tuple[str, ...]) -> int:
    total = 0
    for cell in by_hero.values():
        cur: object = cell
        for key in path:
            if not isinstance(cur, dict):
                cur = None
                break
            cur = cur.get(key)
        if isinstance(cur, dict):
            try:
                total += int(cur.get("pick") or 0)
            except (TypeError, ValueError):
                pass
    return total


def sum_bans(by_hero: dict[str, dict], path: tuple[str, ...]) -> int:
    total = 0
    for cell in by_hero.values():
        cur: object = cell
        for key in path:
            if not isinstance(cur, dict):
                cur = None
                break
            cur = cur.get(key)
        if isinstance(cur, dict):
            try:
                total += int(cur.get("ban") or 0)
            except (TypeError, ValueError):
                pass
    return total


def build_totals(by_hero: dict[str, dict]) -> dict:
    """Aggregate pick (and pro ban) counts so the UI can derive pick/ban rates.

    Pick rate ≈ hero_pick / (sum_picks / 10) because each match contributes 10 picks.
    OpenDota exposes public ranked bans only for the pro scene (pro_ban).
    """
    brackets = {
        name: {"pick": sum_picks(by_hero, ("brackets", name))} for name in BRACKET_ORDER
    }
    return {
        "pub": {"pick": sum_picks(by_hero, ("pub",))},
        "brackets": brackets,
        "pro": {
            "pick": sum_picks(by_hero, ("pro",)),
            "ban": sum_bans(by_hero, ("pro",)),
        },
        "turbo": {"pick": sum_picks(by_hero, ("turbo",))},
    }


def build_payload(by_hero: dict[str, dict], *, window_days: int) -> dict:
    days = max(1, int(window_days))
    return {
        "fetched_at": datetime.now(timezone.utc).isoformat(),
        "source": "opendota",
        "attribution": "https://www.opendota.com",
        "window_days": days,
        "window_note": (
            f"OpenDota recent matches (~{days} days); "
            "pub_pick ≈ sum(pub_pick_trend); brackets share the same window; "
            "pick_rate = pick / (sum_picks/10); ranked ban rates not in heroStats"
        ),
        "window_label_zh": f"近约 {days} 天公开对局",
        "brackets": list(BRACKET_ORDER),
        "totals": build_totals(by_hero),
        "by_hero": by_hero,
    }


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--out", type=Path, default=OUT)
    args = ap.parse_args()

    heroes = hero_table()
    id_to_key = {int(h["id"]): h["key"] for h in heroes}

    print(f"fetching {OPENDOTA_HERO_STATS} ...", flush=True)
    raw = http_json(OPENDOTA_HERO_STATS)
    if not isinstance(raw, list):
        raise SystemExit(f"unexpected heroStats payload type: {type(raw).__name__}")
    if not raw:
        raise SystemExit("heroStats returned an empty list; keeping previous cache")

    window_days = detect_window_days(raw)
    by_hero: dict[str, dict] = {}
    unknown = 0
    for row in raw:
        if not isinstance(row, dict):
            continue
        try:
            hid = int(row.get("id"))
        except (TypeError, ValueError):
            continue
        key = id_to_key.get(hid)
        if key is None:
            unknown += 1
            continue
        by_hero[key] = row_to_hero_stats(row)

    payload = build_payload(by_hero, window_days=window_days)
    if len(by_hero) < max(100, len(heroes) * 4 // 5):
        raise SystemExit(
            f"heroStats only contained {len(by_hero)}/{len(heroes)} known heroes; "
            "keeping previous cache"
        )
    write_json_atomic(args.out, payload)
    missing = [h["key"] for h in heroes if h["key"] not in by_hero]
    print(
        f"done: {len(by_hero)} heroes -> {args.out}"
        + (f", {unknown} unknown ids" if unknown else "")
        + (f", {len(missing)} roster keys missing" if missing else ""),
        flush=True,
    )


if __name__ == "__main__":
    main()