climperor/web/fetch_patches.py

"""Fetch Dota 2 patch list + per-patch details (past year) into data/patches.json.

Sources:
  - patchnoteslist (schinese): version / name / timestamp / website
  - patchnotes?version={v}&language=schinese: general_notes / items / neutral_items / heroes
  - itemlist (schinese): item + neutral ability_id -> {name_loc, icon key}
  - odota ability_ids.json: ability_id -> ability key (e.g. 5004 -> antimage_blink)
  - data/hero_abilities.json: ability key -> Chinese name_loc (fallback: odota abilities.json dname)
  - herolist (schinese): hero_id -> {name_loc, portrait key}

Only patches within the past year (default 365 days from today) are kept,
descending. Per-patch detail keeps the raw ability_id / hero_id values; a
shared `lookup` resolves only the referenced ids to {key, name_loc} so the
read-only Climperor web site can render names + icons with no runtime network calls.

Referenced item / ability icons are downloaded into assets/item_icons and
assets/ability_icons so the static export stays self-contained.

Usage:
  python fetch_patches.py
  python fetch_patches.py --days 365 --delay 0.3
  python fetch_patches.py --since 2025-07-27
  python fetch_patches.py --no-icons
  python fetch_patches.py --force            # refetch every patch detail
  python fetch_patches.py --check            # list vs cache; JSON on stdout
"""

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 time
import urllib.error
from datetime import datetime, timedelta, timezone

from shared.http_utils import download_icons, http_bytes, http_json, write_json_atomic
from shared.paths import ABILITY_ICONS, DATA, HERO_PORTRAITS, ITEM_ICONS

PATCHES_LIST_URL = "https://www.dota2.com/datafeed/patchnoteslist?language=schinese"
PATCH_NOTES_URL = "https://www.dota2.com/datafeed/patchnotes?version={version}&language=schinese"
ITEMLIST_URL = "https://www.dota2.com/datafeed/itemlist?language=schinese"
HEROLIST_URL = "https://www.dota2.com/datafeed/herolist?language=schinese"
ABILITY_IDS_URL = "https://raw.githubusercontent.com/odota/dotaconstants/master/build/ability_ids.json"
ABILITIES_URL = "https://raw.githubusercontent.com/odota/dotaconstants/master/build/abilities.json"
ITEM_ICON_URL = "https://cdn.cloudflare.steamstatic.com/apps/dota2/images/dota_react/items/{key}.png"
ABILITY_ICON_URL = "https://cdn.cloudflare.steamstatic.com/apps/dota2/images/dota_react/abilities/{key}.png"
# Hero-card CDN (same as fetch_hero_portraits). Non-hero units may only exist at half res.
HERO_CARD_URL = (
    "https://cdn.cloudflare.steamstatic.com/apps/dota2/images/dota_react/heroes/{key}.png"
)
# Prefer ability art when Valve's hero-card for a unit is tiny (spirit_bear is 128x72).
UNIT_PORTRAIT_FALLBACK_ABILITY = {"spirit_bear": "lone_druid_spirit_bear"}
PORTRAIT_TARGET_SIZE = (256, 144)

OUT = DATA / "patches.json"
HERO_ABILITIES_PATH = DATA / "hero_abilities.json"
DEFAULT_WINDOW_DAYS = 365
# Bundled shared badges — never fetched from CDN (many innate keys 404).
SKIP_ABILITY_ICON_KEYS = frozenset({"innate", "talent_tree"})
# Non-hero units that Valve places in patchnotes heroes[] under a pseudo hero_id
# absent from herolist. Resolve to a Chinese name + portrait key so the web site
# never shows #1961. Portrait files live in assets/hero_portraits/<key>.png.
UNIT_NAMES = {1961: {"key": "spirit_bear", "name_loc": "熊灵"}}


def _date(ts: int) -> str:
    return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d") if ts else ""


def fetch_patch_list(since_ts: int) -> list[dict]:
    """Patch list filtered to timestamp >= since_ts, newest first."""
    raw = http_json(PATCHES_LIST_URL)
    patches = raw.get("patches") or []
    out: list[dict] = []
    for p in patches:
        if not isinstance(p, dict) or not p.get("patch_number"):
            continue
        ts = int(p.get("patch_timestamp") or 0)
        if ts and ts < since_ts:
            continue
        row = {
            "version": p["patch_number"],
            "name": p.get("patch_name") or p["patch_number"],
            "timestamp": ts,
            "date": _date(ts),
        }
        if p.get("patch_website"):
            row["website"] = p["patch_website"]
        out.append(row)
    out.sort(key=lambda r: r["timestamp"], reverse=True)
    return out


def fetch_patch_detail(version: str) -> dict | None:
    """Raw patch notes content for one version (general/items/neutral/heroes)."""
    try:
        raw = http_json(PATCH_NOTES_URL.format(version=version))
    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as e:
        print(f"  detail {version}: {e}", flush=True)
        return None
    if not raw or not raw.get("success", True):
        print(f"  detail {version}: datafeed reported failure", flush=True)
        return None
    ts = int(raw.get("patch_timestamp") or 0)
    return {
        "version": raw.get("patch_number") or version,
        "name": raw.get("patch_name") or version,
        "timestamp": ts,
        "general_notes": list(raw.get("general_notes") or []),
        "items": list(raw.get("items") or []),
        "neutral_items": list(raw.get("neutral_items") or []),
        "heroes": list(raw.get("heroes") or []),
    }


def load_item_index() -> dict[int, dict]:
    """itemlist id -> {key, name_loc}; key = internal name minus 'item_' prefix."""
    raw = http_json(ITEMLIST_URL)
    rows = (((raw or {}).get("result") or {}).get("data") or {}).get("itemabilities") or []
    out: dict[int, dict] = {}
    for row in rows:
        if not isinstance(row, dict) or row.get("id") is None:
            continue
        name = str(row.get("name") or "")
        key = name.removeprefix("item_")
        out[int(row["id"])] = {
            "key": key,
            "name_loc": (row.get("name_loc") or "").strip() or key,
        }
    return out


def load_ability_index() -> dict[int, str]:
    """ability_id -> ability key (e.g. 5004 -> antimage_blink)."""
    raw = http_json(ABILITY_IDS_URL)
    out: dict[int, str] = {}
    for sid, key in raw.items():
        if isinstance(key, str) and key:
            try:
                out[int(sid)] = key
            except (TypeError, ValueError):
                continue
    return out


def load_ability_names() -> dict[str, str]:
    """ability key -> Chinese name_loc (hero_abilities.json first, odota dname fallback)."""
    names: dict[str, str] = {}
    if HERO_ABILITIES_PATH.is_file():
        try:
            raw = json.loads(HERO_ABILITIES_PATH.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            raw = {}
        for cell in (raw.get("by_hero") or {}).values():
            if not isinstance(cell, dict):
                continue
            for ab in cell.get("abilities") or []:
                if isinstance(ab, dict) and ab.get("key") and ab.get("name_loc"):
                    names.setdefault(str(ab["key"]), str(ab["name_loc"]))
            for tal in cell.get("talents") or []:
                if isinstance(tal, dict) and tal.get("key") and tal.get("name_loc"):
                    names.setdefault(str(tal["key"]), str(tal["name_loc"]))
    # Fallback: odota abilities.json English dname for keys missing a Chinese name.
    try:
        ab = http_json(ABILITIES_URL)
    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError):
        ab = {}
    for key, row in ab.items():
        if isinstance(row, dict) and row.get("dname") and key not in names:
            names[key] = str(row["dname"])
    return names


def load_hero_index() -> dict[int, dict]:
    """herolist id -> {key, name_loc}; key = name minus 'npc_dota_hero_' prefix."""
    raw = http_json(HEROLIST_URL)
    heroes = (((raw or {}).get("result") or {}).get("data") or {}).get("heroes") or []
    out: dict[int, dict] = {}
    for h in heroes:
        if not isinstance(h, dict) or h.get("id") is None:
            continue
        name = str(h.get("name") or "")
        out[int(h["id"])] = {
            "key": name.removeprefix("npc_dota_hero_"),
            "name_loc": (h.get("name_loc") or "").strip() or name,
        }
    return out


def collect_referenced_ids(details: dict[str, dict]) -> tuple[set[int], set[int], set[int]]:
    """(item_ids, ability_ids, hero_ids) actually referenced across all details."""
    item_ids: set[int] = set()
    ability_ids: set[int] = set()
    hero_ids: set[int] = set()
    for det in details.values():
        for entry in det.get("items") or []:
            if isinstance(entry, dict):
                aid = entry.get("ability_id")
                if isinstance(aid, int) and aid > 0:
                    item_ids.add(aid)
        for entry in det.get("neutral_items") or []:
            if isinstance(entry, dict):
                aid = entry.get("ability_id")
                if isinstance(aid, int) and aid > 0:
                    item_ids.add(aid)
        for hero in det.get("heroes") or []:
            if not isinstance(hero, dict):
                continue
            hid = hero.get("hero_id")
            if isinstance(hid, int):
                hero_ids.add(hid)
            for ab in hero.get("abilities") or []:
                if isinstance(ab, dict):
                    aid = ab.get("ability_id")
                    if isinstance(aid, int) and aid > 0:
                        ability_ids.add(aid)
    return item_ids, ability_ids, hero_ids


def build_lookup(
    details: dict[str, dict],
    item_index: dict[int, dict],
    ability_index: dict[int, str],
    ability_names: dict[str, str],
    hero_index: dict[int, dict],
) -> dict:
    """Resolve only referenced ids to {key, name_loc}."""
    item_ids, ability_ids, hero_ids = collect_referenced_ids(details)
    items: dict[str, dict] = {}
    for iid in sorted(item_ids):
        meta = item_index.get(iid)
        if meta:
            items[str(iid)] = {"key": meta["key"], "name_loc": meta["name_loc"]}
    abilities: dict[str, dict] = {}
    for aid in sorted(ability_ids):
        key = ability_index.get(aid)
        if not key:
            continue
        abilities[str(aid)] = {"key": key, "name_loc": ability_names.get(key, key)}
    heroes: dict[str, dict] = {}
    for hid in sorted(hero_ids):
        meta = hero_index.get(hid)
        if meta:
            heroes[str(hid)] = {"key": meta["key"], "name_loc": meta["name_loc"]}
        elif hid in UNIT_NAMES:
            # Non-hero unit (e.g. 1961 = 熊灵): bundled portrait in hero_portraits.
            u = UNIT_NAMES[hid]
            heroes[str(hid)] = {"key": u["key"], "name_loc": u["name_loc"]}
    return {"items": items, "abilities": abilities, "heroes": heroes}


def _png_size(data: bytes) -> tuple[int, int] | None:
    if len(data) < 24 or data[:8] != b"\x89PNG\r\n\x1a\n":
        return None
    import struct

    return struct.unpack(">II", data[16:24])


def _cover_resize_png(data: bytes, size: tuple[int, int] = PORTRAIT_TARGET_SIZE) -> bytes:
    """Center-crop / scale image bytes to a 16:9 hero-card PNG."""
    import cv2
    import numpy as np

    arr = np.frombuffer(data, dtype=np.uint8)
    img = cv2.imdecode(arr, cv2.IMREAD_UNCHANGED)
    if img is None:
        raise ValueError("cv2 could not decode image")
    if img.ndim == 2:
        img = cv2.cvtColor(img, cv2.COLOR_GRAY2BGR)
    elif img.shape[2] == 4:
        img = cv2.cvtColor(img, cv2.COLOR_BGRA2BGR)
    h, w = img.shape[:2]
    tw, th = size
    scale = max(tw / w, th / h)
    nw, nh = max(tw, int(round(w * scale))), max(th, int(round(h * scale)))
    resized = cv2.resize(img, (nw, nh), interpolation=cv2.INTER_CUBIC)
    x0 = max(0, (nw - tw) // 2)
    y0 = max(0, (nh - th) // 2)
    crop = resized[y0 : y0 + th, x0 : x0 + tw]
    ok, buf = cv2.imencode(".png", crop)
    if not ok:
        raise ValueError("cv2 could not encode png")
    return buf.tobytes()


def download_unit_portraits(*, delay: float) -> None:
    """Fetch portraits for UNIT_NAMES into hero_portraits (not in heroes.json).

    Valve ships some unit cards at half resolution (spirit_bear 128x72). When a
    fallback ability icon is configured, use that art cover-cropped to 256x144
    so the patch page matches normal hero cards.
    """
    HERO_PORTRAITS.mkdir(parents=True, exist_ok=True)
    ok = skip = fail = 0
    for unit in UNIT_NAMES.values():
        key = unit["key"]
        out = HERO_PORTRAITS / f"{key}.png"
        if out.is_file():
            dims = _png_size(out.read_bytes())
            if dims == PORTRAIT_TARGET_SIZE:
                skip += 1
                continue
        try:
            fb = UNIT_PORTRAIT_FALLBACK_ABILITY.get(key)
            if fb:
                raw = http_bytes(ABILITY_ICON_URL.format(key=fb), timeout=30)
                data = _cover_resize_png(raw)
                src = f"ability:{fb}"
            else:
                raw = http_bytes(HERO_CARD_URL.format(key=key), timeout=30)
                dims = _png_size(raw)
                data = (
                    _cover_resize_png(raw)
                    if dims and dims != PORTRAIT_TARGET_SIZE
                    else raw
                )
                src = "hero-card"
            out.write_bytes(data)
            ok += 1
            print(f"  unit portrait {key} <- {src} ({len(data)} bytes)", flush=True)
            time.sleep(max(delay, 0.05))
        except Exception as e:  # noqa: BLE001
            fail += 1
            print(f"  FAIL unit portrait {key}: {e}", flush=True)
    print(
        f"  unit portraits saved={ok} skipped={skip} fail={fail} -> {HERO_PORTRAITS}",
        flush=True,
    )


def download_referenced_icons(lookup: dict, *, delay: float) -> None:
    """Pull referenced item + ability icons into assets/ (skip existing)."""
    item_keys = {v["key"] for v in (lookup.get("items") or {}).values() if v.get("key")}
    # Recipe scrolls share one generic icon on Steam CDN.
    recipe_keys = {k for k in item_keys if k.startswith("recipe_")}
    item_keys -= recipe_keys
    item_keys.add("recipe")
    print(f"downloading {len(item_keys)} item icons -> {ITEM_ICONS}", flush=True)
    saved, skipped, fail = download_icons(item_keys, ITEM_ICON_URL, ITEM_ICONS, delay=delay)
    print(f"  item icons saved={saved} skipped={skipped} fail={fail}", flush=True)

    ability_keys = {
        v["key"]
        for v in (lookup.get("abilities") or {}).values()
        if v.get("key") and v["key"] not in SKIP_ABILITY_ICON_KEYS
    }
    print(f"downloading {len(ability_keys)} ability icons -> {ABILITY_ICONS}", flush=True)
    saved, skipped, fail = download_icons(
        ability_keys, ABILITY_ICON_URL, ABILITY_ICONS, delay=delay, skip_keys=SKIP_ABILITY_ICON_KEYS
    )
    print(f"  ability icons saved={saved} skipped={skipped} fail={fail}", flush=True)

    # Pseudo-heroes in patchnotes (e.g. 熊灵) are absent from herolist / heroes.json.
    print("downloading unit portraits for patch lookup ...", flush=True)
    download_unit_portraits(delay=delay)


def load_existing_details() -> dict[str, dict]:
    if not OUT.is_file():
        return {}
    try:
        raw = json.loads(OUT.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    out: dict[str, dict] = {}
    for v, det in (raw.get("details") or {}).items():
        if isinstance(det, dict):
            out[str(v)] = det
    return out


def check_for_new_patches(*, days: int, since: str | None) -> dict:
    """Compare Valve patch list to local details cache; do not write files.

    Returns a JSON-serializable dict with has_new / new_versions / latest / …
    """
    if since:
        since_dt = datetime.strptime(since, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    else:
        since_dt = datetime.now(tz=timezone.utc) - timedelta(days=days)
    since_ts = int(since_dt.timestamp())
    patches = fetch_patch_list(since_ts)
    cached = load_existing_details()
    new_versions = [p["version"] for p in patches if p["version"] not in cached]
    latest = patches[0]["version"] if patches else None
    return {
        "has_new": bool(new_versions),
        "new_versions": new_versions,
        "latest": latest,
        "window_count": len(patches),
        "cached_details": len(cached),
        "since": since_dt.date().isoformat(),
    }


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=DEFAULT_WINDOW_DAYS, help="window in days from today")
    ap.add_argument("--since", type=str, default=None, help="override cutoff (YYYY-MM-DD, UTC)")
    ap.add_argument("--delay", type=float, default=0.3, help="seconds between detail/icon requests")
    ap.add_argument("--no-icons", action="store_true", help="skip icon downloads")
    ap.add_argument("--force", action="store_true", help="refetch every patch detail")
    ap.add_argument(
        "--check",
        action="store_true",
        help="only compare patch list to local details; print JSON to stdout",
    )
    args = ap.parse_args()

    if args.check:
        result = check_for_new_patches(days=args.days, since=args.since)
        print(json.dumps(result, ensure_ascii=False), flush=True)
        return

    if args.since:
        since_dt = datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    else:
        since_dt = datetime.now(tz=timezone.utc) - timedelta(days=args.days)
    since_ts = int(since_dt.timestamp())

    print(f"fetching patch list (since {since_dt.date()})...", flush=True)
    patches = fetch_patch_list(since_ts)
    print(f"  {len(patches)} patches in window", flush=True)
    if not patches:
        raise SystemExit(
            "patch list returned no versions in the requested window; "
            "keeping previous cache"
        )

    cached = {} if args.force else load_existing_details()
    details: dict[str, dict] = {}
    print("fetching per-patch details...", flush=True)
    for n, p in enumerate(patches, start=1):
        version = p["version"]
        det = cached.get(version)
        if det is None:
            det = fetch_patch_detail(version)
            time.sleep(max(args.delay, 0.05))
        if det is None:
            print(f"  [{n}/{len(patches)}] {version}: skipped (no detail)", flush=True)
            continue
        details[version] = det
        print(f"  [{n}/{len(patches)}] {version}: {len(det.get('heroes') or [])} heroes", flush=True)

    print("building lookup (itemlist + ability_ids + hero_abilities + herolist)...", flush=True)
    item_index = load_item_index()
    ability_index = load_ability_index()
    ability_names = load_ability_names()
    hero_index = load_hero_index()
    lookup = build_lookup(details, item_index, ability_index, ability_names, hero_index)
    print(
        f"  lookup: items={len(lookup['items'])} abilities={len(lookup['abilities'])} heroes={len(lookup['heroes'])}",
        flush=True,
    )

    if not args.no_icons:
        download_referenced_icons(lookup, delay=args.delay)

    payload = {
        "meta": {
            "source": PATCHES_LIST_URL,
            "fetched_at": datetime.now(tz=timezone.utc).isoformat(),
            "window_days": args.days,
            "since": since_dt.date().isoformat(),
            "count": len(patches),
            "details_count": len(details),
            "lookup_counts": {
                "items": len(lookup["items"]),
                "abilities": len(lookup["abilities"]),
                "heroes": len(lookup["heroes"]),
            },
        },
        "patches": patches,
        "lookup": lookup,
        "details": details,
    }
    write_json_atomic(OUT, payload)
    print(
        f"saved {len(patches)} patches, {len(details)} details -> {OUT}",
        flush=True,
    )


if __name__ == "__main__":
    main()