"""Fetch recent OpenDota matches per hero with final items + skill builds.

Sources (switch with --source):
  league  GET /heroes/{id}/matches (pro/league-biased; fewer requests)
  public  Scan GET /publicMatches for the hero, then match details
  both    Merge both (dedupe by match_id); each row tagged origin

Kept rows (after merge, newest first, up to --limit):
  - wins only
  - public/ladder: Legend+ only (avg_rank_tier / rank_tier >= 50)
  - league/pro: wins only (no medal filter)

Public region (--public-region, default china):
  china  Prefer Perfect World / 国服 clusters when scanning publicMatches;
         fill --limit with China pubs first, then backfill other regions if short
  any    No cluster filter (global pubs)

China clusters: classic PW ids from odota cluster→region map (regions
12/13/17/18/20/25) plus modern CJK-heavy clusters observed in publicMatches
(413/414/415/417). Each row stores `cluster` when available.

Per match: final inventory + ability_upgrades_arr resolved to ability keys.
When OpenDota has parsed the match (`purchase_log`), also store per-slot
purchase times (seconds from game start; may be negative pre-game):
  item_times / backpack_times / item_neutral_time.
Unparsed matches omit the log → times stay null (icons still shown).

Speed: --workers N concurrent match-detail fetches; stop once enough
accepted wins are collected. Optional OPENDOTA_API_KEY raises rate limits.
Heroes are still written sequentially (one JSON writer).

Incremental (default): refetch heroes missing from by_hero, or with fewer
than --limit kept matches. Use --force to refetch all.

Output: data/hero_matches.json (Climperor web site only; not used by recommend).

Usage:
  python fetch_hero_matches.py --heroes antimage
  python fetch_hero_matches.py --source both --public-region china --limit 10 --workers 6 --delay 0.05
  python fetch_hero_matches.py --force
  python fetch_hero_matches.py --enrich-item-times --heroes juggernaut
"""

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 time
import urllib.error
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone

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

OPENDOTA = "https://api.opendota.com/api"
# Consecutive OpenDota 429s before aborting the remaining hero batch.
_429_STREAK = 0
_429_STREAK_LIMIT = 3


class RateLimitTripped(Exception):
    """Enough consecutive OpenDota 429s to stop the rest of this run."""


def _opendota_json(url: str):
    """Fail-fast OpenDota JSON (no 5/10/20/40s retry chain on 429)."""
    global _429_STREAK
    try:
        data = http_json(url, retries=0)
    except urllib.error.HTTPError as e:
        if e.code == 429:
            _429_STREAK += 1
            print(
                f"HTTP 429 {url} — streak {_429_STREAK}/{_429_STREAK_LIMIT}",
                flush=True,
            )
            if _429_STREAK >= _429_STREAK_LIMIT:
                raise RateLimitTripped(
                    f"opendota 429 x{_429_STREAK}"
                ) from e
        raise
    _429_STREAK = 0
    return data
ABILITY_IDS_URL = (
    "https://raw.githubusercontent.com/odota/dotaconstants/master/build/ability_ids.json"
)
ITEMS_URL = (
    "https://raw.githubusercontent.com/odota/dotaconstants/master/build/items.json"
)
ICON_URL = (
    "https://cdn.cloudflare.steamstatic.com/apps/dota2/images/dota_react/items/{key}.png"
)
OUT = DATA / "hero_matches.json"
# Max recent matches kept per hero after merge (league + public), newest first.
DEFAULT_LIMIT = 10
DEFAULT_WORKERS = 6
PUBLIC_PAGES_CAP = 80
# Prefer deeper scan when filtering to China (hero+cluster coincidence is rarer).
PUBLIC_PAGES_CAP_CHINA = 160

# Perfect World region ids (odota region.json / Valve regions.txt).
CHINA_REGIONS = frozenset({12, 13, 17, 18, 20, 25})
# Classic PW clusters (odota cluster.json → CHINA_REGIONS) + modern observed
# 国服 clusters in publicMatches (CJK-heavy; absent from outdated cluster.json).
# Exclude 410/412 (still mapped to US East) and other non-CJK 41x samples.
CHINA_CLUSTERS = frozenset(
    {
        221,
        222,
        223,
        224,
        225,
        227,
        231,
        232,
        235,
        236,
        413,
        414,
        415,
        417,
    }
)

# OpenDota lobby_type / game_mode filters for public ranked pubs (skip bots / turbo).
BOT_LOBBY_TYPES = frozenset({3, 4, 10})  # tutorial, coop_bots, local_bots
RANKED_LOBBY_TYPES = frozenset({5, 6, 7})  # ranked solo / team / legacy ranked
EXCLUDED_GAME_MODES = frozenset({23})  # Turbo
MIN_HUMAN_PLAYERS = 10
# Legend+ (传奇及以上): rank_tier = medal*10 + stars → legend floor is 50.
MIN_RANK_TIER = 50
# Over-fetch list candidates so detail+win/rank filters can still fill --limit.
CANDIDATE_MULTIPLIER = 8


def _int_field(row: dict, key: str) -> int | None:
    raw = row.get(key)
    if raw is None:
        return None
    try:
        return int(raw)
    except (TypeError, ValueError):
        return None


def is_legend_plus_tier(tier: int | None) -> bool:
    """True if OpenDota rank_tier / avg_rank_tier is Legend medal or higher."""
    if tier is None:
        return False
    try:
        t = int(tier)
    except (TypeError, ValueError):
        return False
    return t >= MIN_RANK_TIER


def hero_won_on_list_row(row: dict, hero_id: int | None = None) -> bool | None:
    """Infer win from list meta when radiant_win (+ slot or team) is present."""
    if "radiant_win" not in row:
        return None
    radiant_win = bool(row.get("radiant_win"))
    slot = _int_field(row, "player_slot")
    if slot is not None:
        radiant = slot < 128
        return radiant_win if radiant else (not radiant_win)
    if hero_id is None:
        return None
    radiant_team = row.get("radiant_team") or []
    dire_team = row.get("dire_team") or []
    on_radiant = False
    on_dire = False
    if isinstance(radiant_team, list):
        for hid in radiant_team:
            try:
                if int(hid) == hero_id:
                    on_radiant = True
                    break
            except (TypeError, ValueError):
                continue
    if isinstance(dire_team, list):
        for hid in dire_team:
            try:
                if int(hid) == hero_id:
                    on_dire = True
                    break
            except (TypeError, ValueError):
                continue
    if on_radiant and not on_dire:
        return radiant_win
    if on_dire and not on_radiant:
        return not radiant_win
    return None


def public_list_row_ok(row: dict, hero_id: int) -> bool:
    """True if a publicMatches list row is ranked Legend+ win for hero_id."""
    lobby = _int_field(row, "lobby_type")
    game_mode = _int_field(row, "game_mode")
    if lobby is not None and lobby in BOT_LOBBY_TYPES:
        return False
    if game_mode is not None and game_mode in EXCLUDED_GAME_MODES:
        return False
    if lobby is not None and lobby not in RANKED_LOBBY_TYPES:
        return False
    avg_rank = _int_field(row, "avg_rank_tier")
    if not is_legend_plus_tier(avg_rank):
        return False
    won = hero_won_on_list_row(row, hero_id)
    if won is False:
        return False
    return True


def match_detail_ok(match: dict, list_meta: dict | None = None) -> bool:
    """Drop bot / partial-human matches after detail fetch."""
    lobby = _int_field(match, "lobby_type")
    if lobby is None and list_meta:
        lobby = _int_field(list_meta, "lobby_type")
    game_mode = _int_field(match, "game_mode")
    if game_mode is None and list_meta:
        game_mode = _int_field(list_meta, "game_mode")
    if lobby is not None and lobby in BOT_LOBBY_TYPES:
        return False
    if game_mode is not None and game_mode in EXCLUDED_GAME_MODES:
        return False
    human = _int_field(match, "human_players")
    if human is not None and human < MIN_HUMAN_PLAYERS:
        return False
    return True


def slim_match_ok(slim: dict) -> bool:
    """Keep wins only; public/ladder also requires Legend+ rank."""
    if not isinstance(slim, dict):
        return False
    if not slim.get("won"):
        return False
    origin = str(slim.get("origin") or "")
    if origin == "public":
        tier = slim.get("rank_tier")
        if tier is None:
            tier = slim.get("avg_rank_tier")
        try:
            t_i = int(tier) if tier is not None else None
        except (TypeError, ValueError):
            t_i = None
        if not is_legend_plus_tier(t_i):
            return False
    return True


def is_china_cluster(cluster: int | None, region: int | None = None) -> bool:
    """True if cluster/region looks like Perfect World 国服."""
    if region is not None and region in CHINA_REGIONS:
        return True
    if cluster is not None and cluster in CHINA_CLUSTERS:
        return True
    return False


def load_ability_id_map() -> dict[int, str]:
    raw = http_json(ABILITY_IDS_URL)
    out: dict[int, str] = {}
    if not isinstance(raw, dict):
        return out
    for sid, key in raw.items():
        try:
            aid = int(sid)
        except (TypeError, ValueError):
            continue
        if isinstance(key, str) and key.strip():
            out[aid] = key.strip()
    return out


def load_item_id_catalog() -> dict[int, dict]:
    """id → {key, dname} from dotaconstants (English dname; UI may localize via shop)."""
    raw = http_json(ITEMS_URL)
    out: dict[int, dict] = {}
    if not isinstance(raw, dict):
        return out
    for key, row in raw.items():
        if not isinstance(row, dict) or row.get("id") is None:
            continue
        key_s = str(key)
        if key_s.startswith("recipe_"):
            continue
        try:
            iid = int(row["id"])
        except (TypeError, ValueError):
            continue
        out[iid] = {"key": key_s, "dname": str(row.get("dname") or key_s)}
    return out


def resolve_upgrades(arr: list | None, id_map: dict[int, str]) -> list[str]:
    if not isinstance(arr, list):
        return []
    keys: list[str] = []
    for raw in arr:
        try:
            aid = int(raw)
        except (TypeError, ValueError):
            continue
        key = id_map.get(aid)
        if key:
            keys.append(key)
    return keys


def item_slot_ids(player: dict, prefix: str, n: int) -> list[int]:
    out: list[int] = []
    for i in range(n):
        raw = player.get(f"{prefix}_{i}")
        try:
            iid = int(raw or 0)
        except (TypeError, ValueError):
            iid = 0
        if iid > 0:
            out.append(iid)
    return out


def _purchase_log_entries(raw) -> list[tuple[str, int]]:
    """Chronological (key, time_sec) from OpenDota purchase_log."""
    if not isinstance(raw, list):
        return []
    out: list[tuple[str, int]] = []
    for row in raw:
        if not isinstance(row, dict):
            continue
        key = row.get("key")
        if not isinstance(key, str) or not key.strip():
            continue
        try:
            t = int(row["time"])
        except (TypeError, ValueError, KeyError):
            continue
        out.append((key.strip(), t))
    return out


def slot_purchase_times(
    item_ids: list[int],
    purchase_log,
    item_catalog: dict[int, dict],
) -> list[int | None]:
    """Map each final inventory id → last matching purchase time (sec).

    Consumes from the end of the log so duplicate keys get distinct times.
    Falls back to recipe_<key> when the finished item key is absent.
    Missing purchase_log → all nulls.
    """
    n = len(item_ids)
    if n == 0:
        return []
    entries = _purchase_log_entries(purchase_log)
    if not entries:
        return [None] * n
    used = [False] * len(entries)
    times: list[int | None] = []
    for iid in item_ids:
        meta = item_catalog.get(int(iid)) if iid else None
        if not meta:
            times.append(None)
            continue
        key = str(meta["key"])
        recipe_key = f"recipe_{key}"
        found: int | None = None
        for i in range(len(entries) - 1, -1, -1):
            if used[i]:
                continue
            if entries[i][0] == key:
                found = i
                break
        if found is None:
            for i in range(len(entries) - 1, -1, -1):
                if used[i]:
                    continue
                if entries[i][0] == recipe_key:
                    found = i
                    break
        if found is None:
            times.append(None)
        else:
            used[found] = True
            times.append(entries[found][1])
    return times


def neutral_purchase_time(
    neutral_id: int | None,
    player: dict,
    item_catalog: dict[int, dict],
) -> int | None:
    """Seconds when the final neutral was acquired, if parse data exists."""
    if not neutral_id:
        return None
    meta = item_catalog.get(int(neutral_id))
    if not meta:
        return None
    key = str(meta["key"])
    history = player.get("neutral_item_history")
    if isinstance(history, list):
        last: int | None = None
        for row in history:
            if not isinstance(row, dict):
                continue
            if row.get("item_neutral") != key:
                continue
            try:
                last = int(row["time"])
            except (TypeError, ValueError, KeyError):
                continue
        if last is not None:
            return last
    # Rare fallback: some neutrals also appear in purchase_log.
    for k, t in reversed(_purchase_log_entries(player.get("purchase_log"))):
        if k == key:
            return t
    return None


def extract_player_row(
    match: dict,
    hero_id: int,
    *,
    origin: str,
    id_map: dict[int, str],
    item_catalog: dict[int, dict] | None = None,
    list_meta: dict | None = None,
) -> dict | None:
    """Build one slim match row for the player on hero_id, or None."""
    players = match.get("players")
    if not isinstance(players, list):
        return None
    player = None
    for p in players:
        if not isinstance(p, dict):
            continue
        try:
            if int(p.get("hero_id") or 0) == hero_id:
                player = p
                break
        except (TypeError, ValueError):
            continue
    if player is None:
        return None

    try:
        match_id = int(match.get("match_id") or 0)
    except (TypeError, ValueError):
        return None
    if match_id <= 0:
        return None

    slot = int(player.get("player_slot") or 0)
    radiant = slot < 128
    radiant_win = bool(match.get("radiant_win"))
    won = radiant_win if radiant else not radiant_win

    start_time = match.get("start_time")
    if start_time is None and list_meta:
        start_time = list_meta.get("start_time")
    try:
        start_time_i = int(start_time) if start_time is not None else None
    except (TypeError, ValueError):
        start_time_i = None

    try:
        duration = int(match.get("duration") or 0)
    except (TypeError, ValueError):
        duration = 0

    league_name = None
    if list_meta and list_meta.get("league_name"):
        league_name = str(list_meta["league_name"])
    league = match.get("league")
    if not league_name and isinstance(league, dict) and league.get("name"):
        league_name = str(league["name"])

    avg_rank = None
    if list_meta and list_meta.get("avg_rank_tier") is not None:
        try:
            avg_rank = int(list_meta["avg_rank_tier"])
        except (TypeError, ValueError):
            avg_rank = None

    account_id = player.get("account_id")
    try:
        account_id_i = int(account_id) if account_id is not None else None
    except (TypeError, ValueError):
        account_id_i = None

    personaname = player.get("personaname")
    if isinstance(personaname, str):
        personaname = personaname.strip() or None
    else:
        personaname = None
    # Pro/registered name when present (league matches).
    pro_name = player.get("name")
    if isinstance(pro_name, str):
        pro_name = pro_name.strip() or None
    else:
        pro_name = None

    rank_tier = player.get("rank_tier")
    try:
        rank_tier_i = int(rank_tier) if rank_tier is not None else None
    except (TypeError, ValueError):
        rank_tier_i = None
    if rank_tier_i is not None and rank_tier_i <= 0:
        rank_tier_i = None

    leaderboard_rank = player.get("leaderboard_rank")
    try:
        leaderboard_rank_i = (
            int(leaderboard_rank) if leaderboard_rank is not None else None
        )
    except (TypeError, ValueError):
        leaderboard_rank_i = None
    if leaderboard_rank_i is not None and leaderboard_rank_i <= 0:
        leaderboard_rank_i = None

    neutral = player.get("item_neutral")
    try:
        neutral_i = int(neutral) if neutral else None
    except (TypeError, ValueError):
        neutral_i = None
    if neutral_i is not None and neutral_i <= 0:
        neutral_i = None

    def _stat(key: str) -> int:
        try:
            return int(player.get(key) or 0)
        except (TypeError, ValueError):
            return 0

    cluster = match.get("cluster")
    if cluster is None and list_meta:
        cluster = list_meta.get("cluster")
    try:
        cluster_i = int(cluster) if cluster is not None else None
    except (TypeError, ValueError):
        cluster_i = None

    region = match.get("region")
    try:
        region_i = int(region) if region is not None else None
    except (TypeError, ValueError):
        region_i = None

    lobby_type = _int_field(match, "lobby_type")
    if lobby_type is None and list_meta:
        lobby_type = _int_field(list_meta, "lobby_type")
    game_mode = _int_field(match, "game_mode")
    if game_mode is None and list_meta:
        game_mode = _int_field(list_meta, "game_mode")
    human_players = _int_field(match, "human_players")

    items = item_slot_ids(player, "item", 6)
    backpack = item_slot_ids(player, "backpack", 3)
    catalog = item_catalog or {}
    # Shared consume so a duplicated key across inv+backpack does not
    # double-claim the same purchase_log entry.
    shared_times = slot_purchase_times(
        items + backpack, player.get("purchase_log"), catalog
    )
    item_times = shared_times[: len(items)]
    backpack_times = shared_times[len(items) :]
    neut_time = neutral_purchase_time(neutral_i, player, catalog)

    out = {
        "match_id": match_id,
        "origin": origin,
        "start_time": start_time_i,
        "duration": duration,
        "won": won,
        "kills": _stat("kills"),
        "deaths": _stat("deaths"),
        "assists": _stat("assists"),
        "account_id": account_id_i,
        "personaname": personaname,
        "name": pro_name,
        "rank_tier": rank_tier_i,
        "leaderboard_rank": leaderboard_rank_i,
        "league_name": league_name,
        "avg_rank_tier": avg_rank,
        "cluster": cluster_i,
        "region": region_i,
        "lobby_type": lobby_type,
        "game_mode": game_mode,
        "human_players": human_players,
        "items": items,
        "backpack": backpack,
        "item_neutral": neutral_i,
        "item_times": item_times,
        "backpack_times": backpack_times,
        "item_neutral_time": neut_time,
        "ability_upgrades": resolve_upgrades(
            player.get("ability_upgrades_arr"), id_map
        ),
    }
    # Convenience label for UI (pro name > persona); never an account id.
    display = pro_name or personaname
    if display:
        out["display_name"] = display
    return out


def opendota_api_key() -> str:
    """Optional key raises OpenDota rate limits (env OPENDOTA_API_KEY)."""
    return (os.environ.get("OPENDOTA_API_KEY") or "").strip()


def opendota_url(path: str) -> str:
    """Build api.opendota.com URL; append api_key query when configured."""
    base = f"{OPENDOTA}{path}"
    key = opendota_api_key()
    if not key:
        return base
    sep = "&" if "?" in base else "?"
    return f"{base}{sep}api_key={key}"


def fetch_match(match_id: int) -> dict | None:
    try:
        raw = _opendota_json(opendota_url(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 league_match_ids(hero_id: int, limit: int) -> list[dict]:
    """Return win list metas from /heroes/{id}/matches (up to candidate cap)."""
    try:
        raw = _opendota_json(opendota_url(f"/heroes/{hero_id}/matches"))
    except RateLimitTripped:
        raise
    except (
        urllib.error.HTTPError,
        urllib.error.URLError,
        TimeoutError,
        json.JSONDecodeError,
        OSError,
    ):
        return []
    if not isinstance(raw, list):
        return []
    want = max(limit * CANDIDATE_MULTIPLIER, limit)
    out: list[dict] = []
    for row in raw:
        if not isinstance(row, dict):
            continue
        try:
            mid = int(row.get("match_id") or 0)
        except (TypeError, ValueError):
            continue
        if mid <= 0:
            continue
        won = hero_won_on_list_row(row, hero_id)
        if won is False:
            continue
        out.append(row)
        if len(out) >= want:
            break
    return out


def _hero_in_public_row(row: dict, hero_id: int) -> bool:
    radiant = row.get("radiant_team") or []
    dire = row.get("dire_team") or []
    for side in (radiant, dire):
        if not isinstance(side, list):
            continue
        for hid in side:
            try:
                if int(hid) == hero_id:
                    return True
            except (TypeError, ValueError):
                continue
    return False


def public_match_ids(
    hero_id: int,
    limit: int,
    *,
    delay: float,
    public_region: str = "china",
) -> list[dict]:
    """Scan publicMatches for Legend+ ranked wins containing hero_id.

    When public_region=china, collect China-cluster hits first; if fewer than
    the candidate cap, backfill with other-region pubs from the same scan.
    """
    prefer_china = public_region == "china"
    pages_cap = PUBLIC_PAGES_CAP_CHINA if prefer_china else PUBLIC_PAGES_CAP
    want = max(limit * CANDIDATE_MULTIPLIER, limit)
    china_out: list[dict] = []
    other_out: list[dict] = []
    seen: set[int] = set()
    less_than: int | None = None

    def _enough() -> bool:
        if prefer_china:
            return len(china_out) >= want
        return len(china_out) + len(other_out) >= want

    for _page in range(pages_cap):
        if _enough():
            break
        url = opendota_url("/publicMatches")
        if less_than is not None:
            # opendota_url may already have ?api_key=
            sep = "&" if "?" in url else "?"
            url = f"{url}{sep}less_than_match_id={less_than}"
        try:
            raw = _opendota_json(url)
        except RateLimitTripped:
            raise
        except (
            urllib.error.HTTPError,
            urllib.error.URLError,
            TimeoutError,
            json.JSONDecodeError,
            OSError,
        ):
            break
        if not isinstance(raw, list) or not raw:
            break
        page_min: int | None = None
        for row in raw:
            if not isinstance(row, dict):
                continue
            try:
                mid = int(row.get("match_id") or 0)
                duration = int(row.get("duration") or 0)
            except (TypeError, ValueError):
                continue
            if mid <= 0:
                continue
            if page_min is None or mid < page_min:
                page_min = mid
            if duration <= 0:
                continue
            if mid in seen:
                continue
            if not _hero_in_public_row(row, hero_id):
                continue
            if not public_list_row_ok(row, hero_id):
                continue
            try:
                cluster_i = int(row["cluster"]) if row.get("cluster") is not None else None
            except (TypeError, ValueError):
                cluster_i = None
            seen.add(mid)
            if is_china_cluster(cluster_i):
                china_out.append(row)
            else:
                other_out.append(row)
            if _enough():
                break
        if page_min is None:
            break
        if less_than is not None and page_min >= less_than:
            break
        less_than = page_min
        if delay > 0:
            time.sleep(delay)

    if prefer_china:
        out = china_out[:want]
        if len(out) < want:
            need = want - len(out)
            out.extend(other_out[:need])
        return out
    merged = china_out + other_out
    return merged[:want]


def fill_matches(
    hero_id: int,
    list_rows: list[dict],
    *,
    origin: str,
    id_map: dict[int, str],
    item_catalog: dict[int, dict],
    delay: float,
    workers: int = 1,
    need: int | None = None,
) -> list[dict]:
    """Fetch details; keep only slim_match_ok rows; stop once `need` accepted."""
    candidates: list[tuple[int, dict]] = []
    for row in list_rows:
        try:
            mid = int(row.get("match_id") or 0)
        except (TypeError, ValueError):
            continue
        if mid <= 0:
            continue
        candidates.append((mid, row))

    if not candidates:
        return []

    target = need if need is not None and need > 0 else len(candidates)
    workers_n = max(1, int(workers))
    matches: list[dict] = []

    def _process(mid: int, row: dict) -> dict | None:
        detail = fetch_match(mid)
        if not detail:
            return None
        if origin == "public" and not match_detail_ok(detail, row):
            return None
        slim = extract_player_row(
            detail,
            hero_id,
            origin=origin,
            id_map=id_map,
            item_catalog=item_catalog,
            list_meta=row,
        )
        if not slim or not slim_match_ok(slim):
            return None
        return slim

    if workers_n <= 1:
        for mid, row in candidates:
            if len(matches) >= target:
                break
            slim = _process(mid, row)
            if slim:
                matches.append(slim)
            if delay > 0:
                time.sleep(delay)
        return matches

    chunk = max(workers_n * 2, workers_n)
    idx = 0
    while idx < len(candidates) and len(matches) < target:
        batch = candidates[idx : idx + chunk]
        idx += len(batch)
        with ThreadPoolExecutor(max_workers=workers_n) as pool:
            futs = [pool.submit(_process, mid, row) for mid, row in batch]
            for fut in as_completed(futs):
                try:
                    slim = fut.result()
                except Exception:
                    slim = None
                if slim:
                    matches.append(slim)
        if delay > 0:
            time.sleep(delay)
    return matches


def fetch_hero_matches(
    hero_id: int,
    *,
    source: str,
    limit: int,
    id_map: dict[int, str],
    item_catalog: dict[int, dict],
    delay: float,
    public_region: str = "china",
    workers: int = 1,
) -> list[dict]:
    by_id: dict[int, dict] = {}
    workers_n = max(1, int(workers))

    if source in ("league", "both"):
        league_rows = league_match_ids(hero_id, limit)
        if delay > 0:
            time.sleep(delay)
        for slim in fill_matches(
            hero_id,
            league_rows,
            origin="league",
            id_map=id_map,
            item_catalog=item_catalog,
            delay=delay,
            workers=workers_n,
            need=limit,
        ):
            by_id[slim["match_id"]] = slim

    if source in ("public", "both") and len(by_id) < limit:
        pub_rows = public_match_ids(
            hero_id, limit, delay=delay, public_region=public_region
        )
        need_more = limit - len(by_id)
        for slim in fill_matches(
            hero_id,
            pub_rows,
            origin="public",
            id_map=id_map,
            item_catalog=item_catalog,
            delay=delay,
            workers=workers_n,
            need=need_more,
        ):
            mid = slim["match_id"]
            if mid not in by_id:
                by_id[mid] = slim

    rows = list(by_id.values())
    rows.sort(key=lambda r: int(r.get("start_time") or 0), reverse=True)
    return rows[:limit]


def hero_cell_needs_fetch(cell, limit: int) -> bool:
    """True if hero is missing, empty, or has fewer than limit kept matches."""
    if cell is None:
        return True
    if isinstance(cell, list):
        matches = cell
    elif isinstance(cell, dict):
        matches = cell.get("matches")
        if not isinstance(matches, list):
            return True
    else:
        return True
    return len(matches) < limit


def load_existing(path: Path) -> dict:
    empty = {"meta": {}, "by_hero": {}}
    if not path.is_file():
        return empty
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return empty
    if not isinstance(raw, dict):
        return empty
    by_hero = raw.get("by_hero")
    if not isinstance(by_hero, dict):
        by_hero = {}
    return {"meta": dict(raw.get("meta") or {}), "by_hero": by_hero}


def collect_item_ids(by_hero: dict) -> set[int]:
    used: set[int] = set()
    for cell in by_hero.values():
        matches = []
        if isinstance(cell, dict):
            matches = cell.get("matches") or []
        elif isinstance(cell, list):
            matches = cell
        if not isinstance(matches, list):
            continue
        for row in matches:
            if not isinstance(row, dict):
                continue
            for key in ("items", "backpack"):
                for raw in row.get(key) or []:
                    try:
                        iid = int(raw)
                    except (TypeError, ValueError):
                        continue
                    if iid > 0:
                        used.add(iid)
            neut = row.get("item_neutral")
            if neut is not None:
                try:
                    nid = int(neut)
                except (TypeError, ValueError):
                    continue
                if nid > 0:
                    used.add(nid)
    return used


def write_out(
    path: Path,
    by_hero: dict,
    *,
    source: str,
    limit: int,
    item_catalog: dict[int, dict],
    public_region: str = "china",
) -> None:
    used = collect_item_ids(by_hero)
    items_out = {
        str(iid): {
            "key": item_catalog[iid]["key"],
            "dname": item_catalog[iid]["dname"],
            "name_loc": item_catalog[iid]["dname"],
        }
        for iid in sorted(used)
        if iid in item_catalog
    }
    payload = {
        "meta": {
            "source": "opendota",
            "attribution": "https://www.opendota.com",
            "fetched_at": datetime.now(timezone.utc).isoformat(),
            "match_source": source,
            "public_region": public_region,
            "limit": limit,
            "wins_only": True,
            "min_rank_tier": MIN_RANK_TIER,
            "min_bracket": "legend",
        },
        "items": items_out,
        "by_hero": by_hero,
    }
    write_json_atomic(path, payload)


def needs_player_enrich(row: dict) -> bool:
    """True if row is missing nickname / rank fields from newer schema."""
    if not isinstance(row, dict):
        return False
    if "personaname" not in row and "rank_tier" not in row:
        return True
    return False


def needs_item_times_enrich(row: dict) -> bool:
    """True if row lacks item_times schema (nulls still count as enriched)."""
    if not isinstance(row, dict):
        return False
    return "item_times" not in row


def enrich_player_fields(
    path: Path,
    *,
    hero_keys: list[str],
    by_key: dict,
    id_map: dict[int, str],
    item_catalog: dict[int, dict],
    delay: float,
    source: str,
    limit: int,
) -> None:
    """Re-fetch match details to fill personaname / rank_tier on existing rows."""
    existing = load_existing(path)
    by_hero: dict = dict(existing.get("by_hero") or {})
    meta = existing.get("meta") or {}
    match_source = str(meta.get("match_source") or source)
    try:
        limit_i = int(meta.get("limit") or limit)
    except (TypeError, ValueError):
        limit_i = limit

    total = 0
    updated = 0
    for hi, key in enumerate(hero_keys, 1):
        cell = by_hero.get(key)
        if not isinstance(cell, dict):
            continue
        matches = cell.get("matches")
        if not isinstance(matches, list) or not matches:
            continue
        hero = by_key.get(key)
        if not hero:
            continue
        hid = int(hero["id"])
        dirty = False
        for i, row in enumerate(matches):
            if not isinstance(row, dict):
                continue
            if not needs_player_enrich(row):
                continue
            mid = row.get("match_id")
            try:
                mid_i = int(mid)
            except (TypeError, ValueError):
                continue
            total += 1
            detail = fetch_match(mid_i)
            if delay > 0:
                time.sleep(delay)
            if not detail:
                # Still mark schema keys so we do not retry forever.
                row["personaname"] = row.get("personaname")
                row["name"] = row.get("name")
                row["rank_tier"] = row.get("rank_tier")
                row["leaderboard_rank"] = row.get("leaderboard_rank")
                dirty = True
                continue
            slim = extract_player_row(
                detail,
                hid,
                origin=str(row.get("origin") or "league"),
                id_map=id_map,
                item_catalog=item_catalog,
                list_meta=row,
            )
            if not slim:
                row["personaname"] = None
                row["name"] = None
                row["rank_tier"] = None
                row["leaderboard_rank"] = None
                dirty = True
                continue
            for field in (
                "personaname",
                "name",
                "rank_tier",
                "leaderboard_rank",
                "account_id",
            ):
                row[field] = slim.get(field)
            display = slim.get("display_name") or slim.get("name") or slim.get(
                "personaname"
            )
            if display:
                row["display_name"] = display
            else:
                row.pop("display_name", None)
            matches[i] = row
            updated += 1
            dirty = True
        if dirty:
            by_hero[key] = {"matches": matches}
            write_out(
                path,
                by_hero,
                source=match_source,
                limit=limit_i,
                item_catalog=item_catalog,
                public_region=str(meta.get("public_region") or "china"),
            )
            print(f"[{hi}/{len(hero_keys)}] {key} enriched (saved)", flush=True)
        else:
            print(f"[{hi}/{len(hero_keys)}] {key} ok", flush=True)
    print(f"enrich done checked={total} updated={updated} → {path}")


def enrich_item_times(
    path: Path,
    *,
    hero_keys: list[str],
    by_key: dict,
    id_map: dict[int, str],
    item_catalog: dict[int, dict],
    delay: float,
    source: str,
    limit: int,
) -> None:
    """Re-fetch match details to fill item_times / backpack_times / item_neutral_time."""
    existing = load_existing(path)
    by_hero: dict = dict(existing.get("by_hero") or {})
    meta = existing.get("meta") or {}
    match_source = str(meta.get("match_source") or source)
    try:
        limit_i = int(meta.get("limit") or limit)
    except (TypeError, ValueError):
        limit_i = limit
    public_region = str(meta.get("public_region") or "china")

    total = 0
    updated = 0
    with_times = 0
    for hi, key in enumerate(hero_keys, 1):
        cell = by_hero.get(key)
        if not isinstance(cell, dict):
            continue
        matches = cell.get("matches")
        if not isinstance(matches, list) or not matches:
            continue
        hero = by_key.get(key)
        if not hero:
            continue
        hid = int(hero["id"])
        dirty = False
        for i, row in enumerate(matches):
            if not isinstance(row, dict):
                continue
            if not needs_item_times_enrich(row):
                continue
            mid = row.get("match_id")
            try:
                mid_i = int(mid)
            except (TypeError, ValueError):
                continue
            total += 1
            detail = fetch_match(mid_i)
            if delay > 0:
                time.sleep(delay)

            def _null_times(target: dict) -> None:
                n_items = (
                    len(target.get("items") or [])
                    if isinstance(target.get("items"), list)
                    else 0
                )
                n_bp = (
                    len(target.get("backpack") or [])
                    if isinstance(target.get("backpack"), list)
                    else 0
                )
                target["item_times"] = [None] * n_items
                target["backpack_times"] = [None] * n_bp
                target["item_neutral_time"] = None

            if not detail:
                _null_times(row)
                dirty = True
                continue
            slim = extract_player_row(
                detail,
                hid,
                origin=str(row.get("origin") or "league"),
                id_map=id_map,
                item_catalog=item_catalog,
                list_meta=row,
            )
            if not slim:
                _null_times(row)
                dirty = True
                continue
            for field in (
                "items",
                "backpack",
                "item_neutral",
                "item_times",
                "backpack_times",
                "item_neutral_time",
            ):
                row[field] = slim.get(field)
            times = slim.get("item_times") or []
            bp_times = slim.get("backpack_times") or []
            if (
                any(t is not None for t in times)
                or any(t is not None for t in bp_times)
                or slim.get("item_neutral_time") is not None
            ):
                with_times += 1
            matches[i] = row
            updated += 1
            dirty = True
        if dirty:
            by_hero[key] = {"matches": matches}
            write_out(
                path,
                by_hero,
                source=match_source,
                limit=limit_i,
                item_catalog=item_catalog,
                public_region=public_region,
            )
            print(
                f"[{hi}/{len(hero_keys)}] {key} item-times enriched (saved)",
                flush=True,
            )
        else:
            print(f"[{hi}/{len(hero_keys)}] {key} ok", flush=True)
    print(
        f"enrich-item-times done checked={total} updated={updated} "
        f"with_times={with_times} → {path}"
    )


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument(
        "--source",
        choices=("league", "public", "both"),
        default="league",
        help="Match list source (default: league)",
    )
    ap.add_argument(
        "--limit",
        type=int,
        default=DEFAULT_LIMIT,
        help=f"Recent matches kept per hero after merge, newest first (default: {DEFAULT_LIMIT})",
    )
    ap.add_argument(
        "--heroes",
        default="",
        help="Comma-separated hero keys to fetch (default: all)",
    )
    ap.add_argument(
        "--delay",
        type=float,
        default=None,
        help="Sleep between list pages / detail batches (default: 0.05 with "
        "--workers>1, else 0.35)",
    )
    ap.add_argument(
        "--workers",
        type=int,
        default=DEFAULT_WORKERS,
        help=f"Concurrent match-detail fetches per hero (default: {DEFAULT_WORKERS})",
    )
    ap.add_argument("--out", type=Path, default=OUT)
    ap.add_argument(
        "--force",
        action="store_true",
        help="Refetch heroes even if already present with enough matches",
    )
    ap.add_argument(
        "--enrich-players",
        action="store_true",
        help="Re-fetch match details to fill personaname/rank_tier on existing rows",
    )
    ap.add_argument(
        "--enrich-item-times",
        action="store_true",
        help="Re-fetch match details to fill item_times/backpack_times on existing rows",
    )
    ap.add_argument(
        "--public-region",
        choices=("china", "any"),
        default="china",
        help=(
            "Public match region preference (default: china). "
            "china = Perfect World clusters first, backfill others if short; "
            "any = no cluster filter"
        ),
    )
    ap.add_argument("--skip-icons", action="store_true")
    ap.add_argument("--force-icons", action="store_true")
    args = ap.parse_args()
    limit = max(1, int(args.limit))
    public_region = str(args.public_region)
    workers = max(1, int(args.workers))
    if args.delay is None:
        delay = 0.05 if workers > 1 else 0.35
    else:
        delay = max(0.0, float(args.delay))

    heroes = hero_table()
    by_key = {h["key"]: h for h in heroes}
    want_keys: list[str]
    if args.heroes.strip():
        want_keys = []
        for part in args.heroes.split(","):
            key = part.strip()
            if not key:
                continue
            if key not in by_key:
                raise SystemExit(f"unknown hero key: {key}")
            want_keys.append(key)
    else:
        want_keys = sorted(by_key.keys())

    if args.enrich_players or args.enrich_item_times:
        print("loading ability_ids + items catalog…")
        id_map = load_ability_id_map()
        item_catalog = load_item_id_catalog()
        print(f"ability_ids={len(id_map)} items={len(item_catalog)}")
        if args.enrich_item_times:
            enrich_item_times(
                args.out,
                hero_keys=want_keys,
                by_key=by_key,
                id_map=id_map,
                item_catalog=item_catalog,
                delay=delay,
                source=args.source,
                limit=limit,
            )
        if args.enrich_players:
            enrich_player_fields(
                args.out,
                hero_keys=want_keys,
                by_key=by_key,
                id_map=id_map,
                item_catalog=item_catalog,
                delay=delay,
                source=args.source,
                limit=limit,
            )
        return

    existing = load_existing(args.out)
    by_hero: dict = dict(existing.get("by_hero") or {})
    if args.force:
        for key in want_keys:
            by_hero.pop(key, None)

    pending = [
        k
        for k in want_keys
        if args.force or hero_cell_needs_fetch(by_hero.get(k), limit)
    ]
    # Drop stale short cells so refetch replaces them.
    for key in pending:
        by_hero.pop(key, None)

    api_key_note = "yes" if opendota_api_key() else "no"
    print(
        f"heroes={len(want_keys)} pending={len(pending)} "
        f"source={args.source} public_region={public_region} "
        f"limit={limit} workers={workers} delay={delay} "
        f"api_key={api_key_note} out={args.out}"
    )
    if not pending:
        print(
            "nothing to do (use --force to refetch, "
            "--enrich-players, or --enrich-item-times)"
        )
        return

    print("loading ability_ids + items catalog…")
    id_map = load_ability_id_map()
    item_catalog = load_item_id_catalog()
    print(f"ability_ids={len(id_map)} items={len(item_catalog)}")

    for i, key in enumerate(pending, 1):
        hero = by_key[key]
        hid = int(hero["id"])
        print(f"[{i}/{len(pending)}] {key} id={hid} …", flush=True)
        try:
            matches = fetch_hero_matches(
                hid,
                source=args.source,
                limit=limit,
                id_map=id_map,
                item_catalog=item_catalog,
                delay=delay,
                public_region=public_region,
                workers=workers,
            )
        except RateLimitTripped as exc:
            print(
                f"  rate-limited; keeping prior cache for remaining ({exc})",
                flush=True,
            )
            prior = existing.get("by_hero") if isinstance(existing, dict) else {}
            if not isinstance(prior, dict):
                prior = {}
            for rest_key in pending[i - 1 :]:
                prev = prior.get(rest_key)
                if prev is not None:
                    by_hero[rest_key] = prev
            write_out(
                args.out,
                by_hero,
                source=args.source,
                limit=limit,
                item_catalog=item_catalog,
                public_region=public_region,
            )
            break
        except (
            urllib.error.HTTPError,
            urllib.error.URLError,
            TimeoutError,
            OSError,
            json.JSONDecodeError,
        ) as exc:
            print(f"  FAIL {exc}")
            continue
        by_hero[key] = {"matches": matches}
        write_out(
            args.out,
            by_hero,
            source=args.source,
            limit=limit,
            item_catalog=item_catalog,
            public_region=public_region,
        )
        n_pub = sum(1 for m in matches if m.get("origin") == "public")
        n_china = sum(
            1
            for m in matches
            if m.get("origin") == "public"
            and is_china_cluster(m.get("cluster"), m.get("region"))
        )
        print(
            f"  matches={len(matches)} public={n_pub} china={n_china} (saved)",
            flush=True,
        )

    if not args.skip_icons:
        keys: list[str] = []
        try:
            written = json.loads(args.out.read_text(encoding="utf-8"))
            keys = sorted(
                {
                    str(r["key"])
                    for r in (written.get("items") or {}).values()
                    if isinstance(r, dict) and r.get("key")
                }
            )
        except (OSError, json.JSONDecodeError):
            keys = []
        if keys:
            print(f"downloading item icons ({len(keys)})…")
            saved, skipped, fail = download_icons(
                keys,
                ICON_URL,
                ITEM_ICONS,
                force=args.force_icons,
                delay=min(0.15, max(0.0, delay)),
            )
            print(f"icons saved={saved} skipped={skipped} fail={fail}")

    print(f"done → {args.out} heroes={len(by_hero)}")


if __name__ == "__main__":
    main()