climperor/fetch_hero_items.py

"""Fetch OpenDota hero item popularity into data/hero_items.json.

Stores a flat Top-N of core finished items per hero (no start/early/mid/late).
Item names (Chinese): Valve dota2.com datafeed (schinese).
Item icons: Steam CDN dota_react/items/{key}.png → assets/item_icons/.

Usage:
  python fetch_hero_items.py
  python fetch_hero_items.py --force
  python fetch_hero_items.py --delay 0.3 --skip-icons
"""

from __future__ import annotations

import argparse
import json
import time
import urllib.error
from datetime import datetime, timezone
from pathlib import Path

from common import DATA, ITEM_ICONS
from grid import hero_table
from http_utils import download_icons, http_json, load_itemlist_zh

OPENDOTA = "https://api.opendota.com/api"
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_items.json"

PHASE_API = (
    "start_game_items",
    "early_game_items",
    "mid_game_items",
    "late_game_items",
)
# Crafted boots / mid-game cores (phase, treads, vanguard); skip bracer/wand.
MIN_CREATED_COST = 1400
# Non-crafted shop cores that are still finished pickups (blink / 跳刀).
ALWAYS_CORE = frozenset({"blink", "aghanims_shard"})
# Aghanim's Blessing (ultimate_scepter_2, id 271) is the activated form of
# Aghanim's Scepter (ultimate_scepter, id 108). Merge blessing counts into
# scepter so heroes show "阿哈利姆神杖" once, never "阿哈利姆福佑".
BLESSING_TO_SCEPTER = {271: 108}
TOP_N = 12
# If any upgrade's popularity >= this fraction of the intermediate's, hide it
# (e.g. AM yasha→manta). If upgrades are rare (Jugg yasha only), keep it.
UPGRADE_RATIO = 0.35


def load_item_catalog() -> tuple[dict[int, dict], dict[int, list[int]]]:
    """Return (id -> meta, id -> upgrade item ids that list it as a component)."""
    raw = http_json(ITEMS_URL)
    zh = load_itemlist_zh()
    key_to_id: dict[str, int] = {}
    out: dict[int, dict] = {}
    for key, row in raw.items():
        if not isinstance(row, dict):
            continue
        iid = row.get("id")
        if iid is None:
            continue
        key_s = str(key)
        if key_s.startswith("recipe_"):
            continue
        dname = str(row.get("dname") or key_s)
        try:
            cost = int(row.get("cost") or 0)
        except (TypeError, ValueError):
            cost = 0
        comps = [str(c) for c in (row.get("components") or []) if c]
        out[int(iid)] = {
            "key": key_s,
            "dname": dname,
            "name_loc": zh.get(int(iid)) or dname,
            "created": bool(row.get("created")),
            "cost": cost,
            "tier": row.get("tier"),
            "qual": row.get("qual"),
            "components": comps,
        }
        key_to_id[key_s] = int(iid)

    upgrades_of: dict[int, list[int]] = {iid: [] for iid in out}
    for iid, meta in out.items():
        for comp_key in meta["components"]:
            cid = key_to_id.get(comp_key)
            if cid is None or cid == iid:
                continue
            upgrades_of.setdefault(cid, []).append(iid)
    return out, upgrades_of


def is_core_finished(meta: dict) -> bool:
    """Core finished items — skip consumables, neutrals, secret-shop parts, cheap early."""
    key = meta.get("key") or ""
    if meta.get("tier") is not None:
        return False
    if meta.get("qual") == "consumable":
        return False
    cost = int(meta.get("cost") or 0)
    if cost <= 0:
        return False
    if key in ALWAYS_CORE:
        return True
    # Only recipe-assembled items (phase boots, bfury, bkb, …).
    if not meta.get("created"):
        return False
    return cost >= MIN_CREATED_COST


def merge_phase_counts(raw: dict) -> dict[int, int]:
    """Max count per item id across OpenDota phases."""
    merged: dict[int, int] = {}
    for api_key in PHASE_API:
        counts = raw.get(api_key) or {}
        if not isinstance(counts, dict):
            continue
        for sid, cnt in counts.items():
            try:
                iid = int(sid)
                c = int(cnt)
            except (TypeError, ValueError):
                continue
            if c <= 0:
                continue
            prev = merged.get(iid, 0)
            if c > prev:
                merged[iid] = c
    return merged


def fold_blessing_into_scepter(counts: dict[int, int]) -> dict[int, int]:
    """Merge Aghanim's Blessing (id 271) counts into Aghanim's Scepter (id 108).

    Blessing is the activated/synthesized form of Scepter; OpenDota records
    them under separate ids. Take max (not sum): the same game can register
    both the purchase and the activation, which would double-count.
    """
    out = dict(counts)
    for bid, sid in BLESSING_TO_SCEPTER.items():
        b = out.pop(bid, 0)
        s = out.get(sid, 0)
        if b or s:
            out[sid] = max(s, b)
    return out


def is_terminal_for_hero(
    iid: int,
    count: int,
    all_counts: dict[int, int],
    upgrades_of: dict[int, list[int]],
    ratio: float = UPGRADE_RATIO,
) -> bool:
    """True if hero rarely upgrades this item further (keep yasha for Jugg, drop for AM)."""
    parents = upgrades_of.get(iid) or []
    if not parents:
        return True
    threshold = max(1, int(count * ratio))
    for pid in parents:
        if all_counts.get(pid, 0) >= threshold:
            return False
    return True


def core_from_popularity(
    raw: dict,
    catalog: dict[int, dict],
    upgrades_of: dict[int, list[int]],
    n: int = TOP_N,
) -> list[dict]:
    """Merge phases → core items → drop intermediates that this hero upgrades."""
    all_counts = fold_blessing_into_scepter(merge_phase_counts(raw))
    merged: dict[int, int] = {}
    for iid, c in all_counts.items():
        meta = catalog.get(iid)
        if meta is None or not is_core_finished(meta):
            continue
        if not is_terminal_for_hero(iid, c, all_counts, upgrades_of):
            continue
        merged[iid] = c
    ranked = sorted(merged.items(), key=lambda t: (-t[1], t[0]))
    return [{"id": iid, "count": c} for iid, c in ranked[:n]]


def fetch_popularity(hero_id: int) -> dict:
    return http_json(f"{OPENDOTA}/heroes/{hero_id}/itemPopularity")  # type: ignore[return-value]


def build_payload(by_hero: dict, catalog: dict[int, dict], used_ids: set[int]) -> dict:
    items_out = {
        str(iid): {
            "key": catalog[iid]["key"],
            "dname": catalog[iid]["dname"],
            "name_loc": catalog[iid]["name_loc"],
        }
        for iid in sorted(used_ids)
        if iid in catalog
    }
    return {
        "meta": {
            "source": "opendota+valve",
            "attribution": "https://www.opendota.com ; https://www.dota2.com",
            "fetched_at": datetime.now(timezone.utc).isoformat(),
            "top_n": TOP_N,
            "mode": "core_finished_terminal",
            "upgrade_ratio": UPGRADE_RATIO,
            "icons": "steam_cdn",
        },
        "items": items_out,
        "by_hero": by_hero,
    }


def collect_used_ids(by_hero: dict) -> set[int]:
    used: set[int] = set()
    for cell in by_hero.values():
        if not isinstance(cell, list):
            continue
        for row in cell:
            if isinstance(row, dict) and "id" in row:
                used.add(int(row["id"]))
    return used


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--delay", type=float, default=0.3)
    ap.add_argument("--out", type=Path, default=OUT)
    ap.add_argument("--skip-icons", action="store_true")
    ap.add_argument("--force-icons", action="store_true")
    ap.add_argument(
        "--force",
        action="store_true",
        help="Refetch all heroes (ignore existing by_hero cache)",
    )
    args = ap.parse_args()

    print("fetching item catalog (dotaconstants + Valve schinese)...", flush=True)
    catalog, upgrades_of = load_item_catalog()
    print(f"  {len(catalog)} items", flush=True)

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

    by_hero: dict[str, list] = {}
    if args.out.is_file() and not args.force:
        try:
            prev = json.loads(args.out.read_text(encoding="utf-8"))
            for k, cell in (prev.get("by_hero") or {}).items():
                if isinstance(cell, list):
                    by_hero[str(k)] = cell
            print(f"resuming with {len(by_hero)} heroes already cached", flush=True)
        except (OSError, json.JSONDecodeError):
            pass
    elif args.force:
        print("force: refetching all heroes", flush=True)

    pending = [k for k in sorted(by_key) if k not in by_hero]
    print(f"fetching {len(pending)} / {len(by_key)} heroes -> {args.out}", flush=True)

    used_ids = collect_used_ids(by_hero)

    for n, key in enumerate(pending, start=1):
        hid = by_key[key]
        try:
            raw = fetch_popularity(hid)
        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, TypeError) as e:
            print(f"  [{n}/{len(pending)}] {key} ({hid}) failed: {e}", flush=True)
            time.sleep(max(args.delay, 0.1) * 2)
            continue

        core = core_from_popularity(raw, catalog, upgrades_of)
        by_hero[key] = core
        used_ids.update(int(row["id"]) for row in core if isinstance(row, dict) and "id" in row)
        print(f"  [{n}/{len(pending)}] {key}: {len(core)} core items", flush=True)

        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(
            json.dumps(build_payload(by_hero, catalog, used_ids), ensure_ascii=False, indent=2)
            + "\n",
            encoding="utf-8",
        )
        time.sleep(args.delay)

    used_ids = collect_used_ids(by_hero)
    payload = build_payload(by_hero, catalog, used_ids)
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )

    if not args.skip_icons:
        keys = {catalog[iid]["key"] for iid in used_ids if iid in catalog}
        print(f"downloading {len(keys)} item icons from Steam CDN -> {ITEM_ICONS}", flush=True)
        saved, skipped, fail = download_icons(keys, ICON_URL, ITEM_ICONS, force=args.force_icons)
        print(f"  icons saved={saved} skipped={skipped} fail={fail}", flush=True)

    missing = [k for k in by_key if k not in by_hero]
    print(
        f"done: {len(by_hero)} heroes, {len(payload['items'])} items"
        + (f", {len(missing)} still missing" if missing else ""),
        flush=True,
    )


if __name__ == "__main__":
    main()