"""Shared HTTP helpers: fetch JSON/bytes, download icons, load Valve datafeeds. Used by all fetch_*.py scripts and serve_relations.py so HTTP logic, User-Agent, timeout, and retry conventions live in exactly one place. """ from __future__ import annotations import json import time import urllib.error import urllib.request from pathlib import Path UA = "climperor" DEFAULT_TIMEOUT = 60 HEROES_URL = "https://www.dota2.com/datafeed/herolist?language={lang}" ITEMLIST_URL = "https://www.dota2.com/datafeed/itemlist?language={lang}" def http_json(url: str, *, timeout: int = DEFAULT_TIMEOUT) -> dict | list: req = urllib.request.Request(url, headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode()) def http_bytes(url: str, *, timeout: int = DEFAULT_TIMEOUT) -> bytes: req = urllib.request.Request(url, headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.read() def download_icons( keys, url_template: str, dest_dir: Path, *, force: bool = False, delay: float = 0.0, min_size: int = 32, skip_keys: frozenset[str] | set[str] | None = None, ) -> tuple[int, int, int]: """Download PNG icons from a CDN. Returns (saved, skipped_existing, fail). keys — iterable of template substitution values (hero/item/ability keys). url_template — e.g. "https://cdn.../abilities/{key}.png". dest_dir — target directory (created if missing). force — re-download even if the file exists. delay — seconds to sleep between requests (rate limiting). min_size — files smaller than this are treated as empty and re-downloaded. skip_keys — keys to ignore entirely (e.g. bundled icons that 404 on CDN). """ dest_dir.mkdir(parents=True, exist_ok=True) skip = set(skip_keys or ()) saved = skipped = fail = 0 for key in sorted(keys): if key in skip or "/" in key or "\\" in key or ".." in key: continue dest = dest_dir / f"{key}.png" if dest.is_file() and dest.stat().st_size >= min_size and not force: skipped += 1 continue try: data = http_bytes(url_template.format(key=key)) if not data or len(data) < min_size: raise ValueError("empty icon") dest.write_bytes(data) saved += 1 print(f" icon saved {key}.png ({len(data)} bytes)", flush=True) except (urllib.error.URLError, TimeoutError, ValueError, OSError) as e: print(f" icon {key}: {e}", flush=True) fail += 1 if delay > 0: time.sleep(delay) return saved, skipped, fail def load_itemlist(lang: str = "schinese") -> dict[int, dict[str, str]]: """Valve datafeed item list: item id -> {name_loc, name}.""" raw = http_json(ITEMLIST_URL.format(lang=lang)) rows = (((raw or {}).get("result") or {}).get("data") or {}).get("itemabilities") or [] out: dict[int, dict[str, str]] = {} for row in rows: if not isinstance(row, dict): continue iid = row.get("id") if iid is None: continue out[int(iid)] = { "name_loc": (row.get("name_loc") or "").strip(), "name": (row.get("name") or "").strip(), } return out def load_itemlist_zh() -> dict[int, str]: """Convenience: item id -> Chinese localized name.""" return {iid: v["name_loc"] for iid, v in load_itemlist().items() if v["name_loc"]} def fetch_hero_list(lang: str = "schinese") -> list[dict]: """Hero list from Valve's datafeed (id, key, name_loc, primary_attr).""" data = http_json(HEROES_URL.format(lang=lang)) heroes = data.get("result", {}).get("data", {}).get("heroes") or data.get("heroes") if not heroes: raise SystemExit("hero list came back empty") return heroes def fetch_hero_keys() -> list[str]: """Hero keys (e.g. antimage, earthshaker) from the English datafeed.""" heroes = fetch_hero_list(lang="english") keys = [] for h in heroes: name = h.get("name") or "" keys.append(name.removeprefix("npc_dota_hero_")) return keys