"""Local dev server for the Climperor web site (web/relations/). Usage: python serve_relations.py python serve_relations.py --port 8765 Hero relations, rankings, streamers, mechanics, items, patches, players — read-only browser UI. Edit data/*.json directly, then refresh the page. Player pages: GET /api/players/{account_id}[/{match_id}] from pc/player_pages/; POST /api/players/enrich and /api/players/ensure-match (local OpenDota backfill). """ 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 mimetypes import threading import time import webbrowser from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse import urllib.parse import urllib.error from shared.grid import ATTR_ORDER, hero_table from shared.hero_tags import TAG_ORDER, tags_for_hero from shared.http_utils import http_bytes from shared.paths import ( ABILITY_ICONS, ABILITY_VIDEOS, ATTR_ICONS, DATA, HERO_PORTRAITS, ITEM_CAT_ICONS, ITEM_ICONS, PC_DIR, PC_PLAYER_PAGES, RANK_ICONS, ROLE_ICONS, ROOT, STREAMER_AVATARS, STREAMER_VIDEOS, TEMPLATES_CDN, UI_ICONS, WEB_FRONTEND, ) from shared.relations import DEFAULT_RELATIONS, load_relations from fetch_streamer_live import probe_streamers from mechanic_tags import QUERY_MECHANIC_ORDER, mechanic_query_payload from steam_auth import ( cookie_header as _steam_cookie_header, create_session_token as _steam_create_token, env_secrets as _steam_env_secrets, fetch_steam_persona as _steam_fetch_persona, read_cookie as _steam_read_cookie, steam_id64_to_account_id as _steam_id64_to_account, steam_login_redirect_url as _steam_login_url, verify_session_token as _steam_verify_token, verify_steam_openid as _steam_verify_openid, ) sys.path.insert(0, str(PC_DIR)) from common import load_config as _load_pc_config # noqa: E402 from player_pages import ( # noqa: E402 enrich_profile_recent as _enrich_profile_recent, get_profile_for_web as _get_profile_for_web, ensure_match_detail as _ensure_match_detail, ) WEB_DIR = WEB_FRONTEND MOBILE_DEMAND_PATH = ROOT / "web" / ".refresh" / "mobile_demand.json" _MOBILE_DEMAND_LOCK = threading.Lock() _LIVE_STATUS_LOCK = threading.Lock() _LIVE_STATUS_TTL = 60 _LIVE_STATUS_CACHE: tuple[float, dict] | None = None def _read_mobile_demand_count() -> int: try: if not MOBILE_DEMAND_PATH.is_file(): return 0 data = json.loads(MOBILE_DEMAND_PATH.read_text(encoding="utf-8")) n = int(data.get("count", 0)) return n if n >= 0 else 0 except (OSError, ValueError, TypeError, json.JSONDecodeError): return 0 def _inc_mobile_demand_count() -> int: with _MOBILE_DEMAND_LOCK: n = _read_mobile_demand_count() + 1 MOBILE_DEMAND_PATH.parent.mkdir(parents=True, exist_ok=True) MOBILE_DEMAND_PATH.write_text( json.dumps({"count": n}, ensure_ascii=False) + "\n", encoding="utf-8", ) return n def _probe_live_status() -> dict: """Probe local streamer rooms without mutating streamers.json.""" global _LIVE_STATUS_CACHE now = time.monotonic() cached = _LIVE_STATUS_CACHE if cached and now - cached[0] < _LIVE_STATUS_TTL: return cached[1] with _LIVE_STATUS_LOCK: now = time.monotonic() cached = _LIVE_STATUS_CACHE if cached and now - cached[0] < _LIVE_STATUS_TTL: return cached[1] try: payload = json.loads(STREAMERS_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): result = { "probed_at": datetime.now(timezone.utc).isoformat(), "ttl": _LIVE_STATUS_TTL, "streamers": {}, } _LIVE_STATUS_CACHE = (now, result) return result rows = payload.get("streamers") if not isinstance(rows, list): rows = [] for row in rows: if isinstance(row, dict) and row.get("live_url"): # A failed probe is unknown and must not preserve a stale live badge. row["is_live"] = False row.pop("live_probed_at", None) probe_streamers(payload) streamers = {} for row in rows: if not isinstance(row, dict) or not row.get("live_url"): continue sid = str(row.get("id") or "").strip() if not sid: continue cell = {"is_live": row.get("is_live") is True} if not row.get("live_probed_at"): cell["stale"] = True streamers[sid] = cell result = { "probed_at": datetime.now(timezone.utc).isoformat(), "ttl": _LIVE_STATUS_TTL, "streamers": streamers, } _LIVE_STATUS_CACHE = (time.monotonic(), result) return result GRID_ORDER_PATH = DATA / "hero_grid_order.json" HERO_ITEMS_PATH = DATA / "hero_items.json" HERO_STATS_PATH = DATA / "hero_stats.json" HERO_MATCHES_PATH = DATA / "hero_matches.json" HERO_ITEM_FEARS_PATH = DATA / "hero_item_fears.json" HERO_ABILITIES_PATH = DATA / "hero_abilities.json" ITEM_SHOP_PATH = DATA / "item_shop.json" ITEMS_META_PATH = DATA / "items_meta.json" PATCHES_PATH = DATA / "patches.json" PATCH_SUMMARIES_PATH = DATA / "patch_summaries.json" LEADERBOARDS_PATH = DATA / "leaderboards.json" PRO_MATCHES_PATH = DATA / "pro_matches.json" STREAMERS_PATH = DATA / "streamers.json" STRATZ_HERO_META_PATH = DATA / "stratz_hero_meta.json" STRATZ_MATCHUP_TOPS_PATH = DATA / "stratz_matchup_tops.json" _LEGACY_DATA = ROOT / "data" def _first_existing(*candidates: Path) -> Path | None: for p in candidates: if p.is_file(): return p return None ATTR_COLS = {"str": 6, "agi": 6, "int": 6, "all": 4} ATTR_LABELS = {"str": "力量", "agi": "敏捷", "int": "智力", "all": "全才"} ABILITY_ICON_URL = ( "https://cdn.cloudflare.steamstatic.com/apps/dota2/images/dota_react/abilities/{key}.png" ) # Shared innate badge (official-style gold droplet); UI falls back here when CDN 404s. INNATE_ICON_NAME = "innate" # Bundled talent-tree trigger icon (not fetched from CDN). TALENT_TREE_ICON_NAME = "talent_tree" def load_hero_items() -> dict: """Cached OpenDota item popularity (see fetch_hero_items.py).""" empty = {"meta": {}, "items": {}, "by_hero": {}} if not HERO_ITEMS_PATH.is_file(): return empty try: raw = json.loads(HERO_ITEMS_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty return { "meta": dict(raw.get("meta") or {}), "items": dict(raw.get("items") or {}), "by_hero": dict(raw.get("by_hero") or {}), } def load_hero_matches() -> dict: """Cached recent matches + builds (see fetch_hero_matches.py). Preview only.""" empty: dict = {"meta": {}, "items": {}, "by_hero": {}} if not HERO_MATCHES_PATH.is_file(): return empty try: raw = json.loads(HERO_MATCHES_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty by_hero = raw.get("by_hero") if not isinstance(by_hero, dict): by_hero = {} return { "meta": dict(raw.get("meta") or {}), "items": dict(raw.get("items") or {}), "by_hero": by_hero, } def load_hero_stats() -> dict: """Cached OpenDota bracket pick/win (see fetch_hero_stats.py). Preview only.""" empty: dict = { "fetched_at": None, "source": "opendota", "attribution": "https://www.opendota.com", "window_days": 7, "window_note": None, "window_label_zh": "近约 7 天公开对局", "brackets": [], "totals": {}, "by_hero": {}, } if not HERO_STATS_PATH.is_file(): return empty try: raw = json.loads(HERO_STATS_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty try: window_days = int(raw.get("window_days") or 7) except (TypeError, ValueError): window_days = 7 window_days = max(1, window_days) label = raw.get("window_label_zh") or f"近约 {window_days} 天公开对局" return { "fetched_at": raw.get("fetched_at"), "source": raw.get("source") or "opendota", "attribution": raw.get("attribution") or "https://www.opendota.com", "window_days": window_days, "window_note": raw.get("window_note"), "window_label_zh": label, "brackets": list(raw.get("brackets") or []), "totals": dict(raw.get("totals") or {}), "by_hero": dict(raw.get("by_hero") or {}), } def load_hero_item_fears() -> dict: """Cached rule-based items that counter each hero (see item_fears.py).""" empty = {"meta": {}, "items": {}, "by_hero": {}} if not HERO_ITEM_FEARS_PATH.is_file(): return empty try: raw = json.loads(HERO_ITEM_FEARS_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty return { "meta": dict(raw.get("meta") or {}), "items": dict(raw.get("items") or {}), "by_hero": dict(raw.get("by_hero") or {}), } def load_items_meta_index() -> dict: """key → slim item row (desc/tags/cost) for hero-page inspect lookups.""" out: dict[str, dict] = {} if not ITEMS_META_PATH.is_file(): return out try: raw = json.loads(ITEMS_META_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return out for row in (raw.get("items") or {}).values(): if not isinstance(row, dict) or not row.get("key"): continue key = str(row["key"]) out[key] = { "key": key, "name_loc": row.get("name_loc") or row.get("dname") or key, "cost": row.get("cost"), "desc_loc": row.get("desc_loc") or "", "tags": list(row.get("tags") or []), } return out def load_item_shop() -> dict: """Full shop catalog for the Items page (see fetch_item_shop.py).""" empty = { "meta": {}, "basic": {"sections": []}, "upgraded": {"sections": []}, "items": {}, } if not ITEM_SHOP_PATH.is_file(): return empty try: raw = json.loads(ITEM_SHOP_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty items = dict(raw.get("items") or {}) # Merge mechanism tags / short desc from items_meta when present. if ITEMS_META_PATH.is_file(): try: meta = json.loads(ITEMS_META_PATH.read_text(encoding="utf-8")) by_key = {} for row in (meta.get("items") or {}).values(): if isinstance(row, dict) and row.get("key"): by_key[row["key"]] = row for key, row in items.items(): m = by_key.get(key) if not m: continue row = dict(row) row["tags"] = list(m.get("tags") or []) row["desc_loc"] = m.get("desc_loc") or "" items[key] = row except (OSError, json.JSONDecodeError): pass return { "meta": dict(raw.get("meta") or {}), "basic": dict(raw.get("basic") or {"sections": []}), "upgraded": dict(raw.get("upgraded") or {"sections": []}), "items": items, } def load_patches() -> dict: """Patch list + per-patch details + id lookup (see fetch_patches.py). Returns {patches, lookup, details}; empty-shaped when the file is missing so the web UI degrades to "no data" instead of crashing. """ empty = {"patches": [], "lookup": {}, "details": {}} if not PATCHES_PATH.is_file(): return empty try: raw = json.loads(PATCHES_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty return { "patches": list(raw.get("patches") or []), "lookup": dict(raw.get("lookup") or {}), "details": dict(raw.get("details") or {}), } def load_patch_summaries() -> dict: """Hand-authored patch readings (see web/data/patch_summaries.json). Returns {by_version: {version: summary}}; empty when missing/invalid. Only the latest version is expected to have an entry. """ empty: dict = {"by_version": {}} if not PATCH_SUMMARIES_PATH.is_file(): return empty try: raw = json.loads(PATCH_SUMMARIES_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty if not isinstance(raw, dict): return empty by_version = raw.get("by_version") if not isinstance(by_version, dict): return empty out: dict[str, dict] = {} for ver, cell in by_version.items(): if not isinstance(ver, str) or not isinstance(cell, dict): continue out[ver] = cell return {"by_version": out} def load_leaderboards() -> dict: """Valve Immortal division Top 100 (see fetch_leaderboards.py). Preview only.""" empty: dict = { "fetched_at": None, "source": "valve", "attribution": "https://www.dota2.com/leaderboards", "note": None, "default_region": "china", "region_order": ["china", "europe", "americas", "se_asia"], "regions": {}, } if not LEADERBOARDS_PATH.is_file(): return empty try: raw = json.loads(LEADERBOARDS_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty if not isinstance(raw, dict): return empty regions = raw.get("regions") if isinstance(raw.get("regions"), dict) else {} order = raw.get("region_order") if not isinstance(order, list) or not order: order = list(empty["region_order"]) return { "fetched_at": raw.get("fetched_at"), "source": raw.get("source") or "valve", "attribution": raw.get("attribution") or empty["attribution"], "note": raw.get("note"), "default_region": raw.get("default_region") or "china", "region_order": [str(x) for x in order], "regions": regions, } def load_pro_matches() -> dict: """Pro-player recent matches (see fetch_pro_matches.py). Preview only.""" empty: dict = { "meta": {}, "items": {}, "pros": {}, "by_pro": {}, "by_hero": {}, } if not PRO_MATCHES_PATH.is_file(): return empty try: raw = json.loads(PRO_MATCHES_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty if not isinstance(raw, dict): return empty return { "meta": dict(raw.get("meta") or {}), "items": dict(raw.get("items") or {}), "pros": dict(raw.get("pros") or {}), "by_pro": dict(raw.get("by_pro") or {}), "by_hero": dict(raw.get("by_hero") or {}), } def load_streamers() -> dict: """Manual streamer directory + Douyin profile enrichment (fetch_streamers.py). Web「主播」only — do not merge into relations/heroes or recommend. """ empty: dict = { "fetched_at": None, "source": "manual+douyin", "platform_meta": { "douyin": { "label_zh": "抖音", "icon": "ui-icon/platform_douyin.png", } }, "streamers": [], } if not STREAMERS_PATH.is_file(): return empty try: raw = json.loads(STREAMERS_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty if not isinstance(raw, dict): return empty rows = raw.get("streamers") if not isinstance(rows, list): rows = [] platform_meta = raw.get("platform_meta") if not isinstance(platform_meta, dict): platform_meta = dict(empty["platform_meta"]) else: platform_meta = { **empty["platform_meta"], **{k: v for k, v in platform_meta.items() if isinstance(v, dict)}, } return { "fetched_at": raw.get("fetched_at"), "source": raw.get("source") or "manual+douyin", "platform_meta": platform_meta, "streamers": [r for r in rows if isinstance(r, dict)], } def load_stratz_hero_meta() -> dict: """STRATZ weekly WR/pick by bracket + positions (see fetch_stratz_meta.py). Web only.""" empty: dict = { "fetched_at": None, "source": "stratz", "attribution": "https://stratz.com", "weeks_take": 0, "window_label_zh": None, "brackets": [], "positions": [], "bracket_position_note": None, "totals": {}, "by_hero": {}, "meta_board": {}, } path = _first_existing(STRATZ_HERO_META_PATH, _LEGACY_DATA / "stratz_hero_meta.json") if path is None: 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 return { "fetched_at": raw.get("fetched_at"), "source": raw.get("source") or "stratz", "attribution": raw.get("attribution") or empty["attribution"], "weeks_take": raw.get("weeks_take") or 0, "window_label_zh": raw.get("window_label_zh"), "brackets": list(raw.get("brackets") or []), "positions": list(raw.get("positions") or []), "bracket_position_note": raw.get("bracket_position_note"), "totals": dict(raw.get("totals") or {}), "by_hero": dict(raw.get("by_hero") or {}), "meta_board": dict(raw.get("meta_board") or {}), } def load_stratz_matchup_tops() -> dict: """STRATZ vs/with top lists per hero (see fetch_stratz_meta.py). Web only.""" empty: dict = { "fetched_at": None, "started_at": None, "finished_at": None, "source": "stratz", "attribution": "https://stratz.com", "take": 0, "match_limit": 0, "scope": { "kind": "global_aggregate", "label_zh": "全局聚合（未按段位 / 分路 / 周过滤）", }, "note": None, "stats": {}, "by_hero": {}, } path = _first_existing( STRATZ_MATCHUP_TOPS_PATH, _LEGACY_DATA / "stratz_matchup_tops.json" ) if path is None: 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 scope = raw.get("scope") if isinstance(raw.get("scope"), dict) else empty["scope"] return { "fetched_at": raw.get("fetched_at"), "started_at": raw.get("started_at"), "finished_at": raw.get("finished_at"), "source": raw.get("source") or "stratz", "attribution": raw.get("attribution") or empty["attribution"], "take": raw.get("take") or 0, "match_limit": raw.get("match_limit") or 0, "scope": scope, "note": raw.get("note"), "stats": dict(raw.get("stats") or {}), "by_hero": dict(raw.get("by_hero") or {}), } def load_hero_abilities() -> dict: """Slim per-hero abilities / Aghs upgrades / talents for the web hero pane.""" empty = {"meta": {}, "by_hero": {}} if not HERO_ABILITIES_PATH.is_file(): return empty try: raw = json.loads(HERO_ABILITIES_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty by_hero: dict[str, dict] = {} for key, cell in (raw.get("by_hero") or {}).items(): if not isinstance(cell, dict): continue abs_out = [] for ab in cell.get("abilities") or []: if not isinstance(ab, dict) or not ab.get("key"): continue specials = [ {"label": str(s["label"]), "value": str(s["value"])} for s in (ab.get("specials") or []) if isinstance(s, dict) and s.get("label") and s.get("value") ] abs_out.append( { "key": ab["key"], "name_loc": ab.get("name_loc") or ab["key"], "desc_loc": ab.get("desc_loc") or "", "shard_loc": ab.get("shard_loc") or "", "scepter_loc": ab.get("scepter_loc") or "", "has_shard": bool(ab.get("has_shard")), "has_scepter": bool(ab.get("has_scepter")), "granted_by_shard": bool(ab.get("granted_by_shard")), "granted_by_scepter": bool(ab.get("granted_by_scepter")), "dispellable": ab.get("dispellable") or "none", "is_innate": bool(ab.get("is_innate")), "target_label": ab.get("target_label") or "", "affects_label": ab.get("affects_label") or "", "damage_label": ab.get("damage_label") or "", "immunity_label": ab.get("immunity_label") or "", "cast_range": ab.get("cast_range") or "", "cast_point": ab.get("cast_point") or "", "channel_time": ab.get("channel_time") or "", "cooldown": ab.get("cooldown") or "", "mana_cost": ab.get("mana_cost") or "", "specials": specials, "lore_loc": ab.get("lore_loc") or "", "tags": [ str(t) for t in (ab.get("tags") or []) if isinstance(t, str) and t in QUERY_MECHANIC_ORDER ], } ) talents_out = [] for tal in cell.get("talents") or []: if not isinstance(tal, dict) or not tal.get("key"): continue talents_out.append( { "key": tal["key"], "name_loc": tal.get("name_loc") or tal["key"], "level": int(tal.get("level") or 0), "side": tal.get("side") or "left", } ) by_hero[str(key)] = { "abilities": abs_out, "talents": talents_out, } return { "meta": dict(raw.get("meta") or {}), "by_hero": by_hero, } def ensure_ability_icon(key: str) -> Path | None: """Return local ability icon path, downloading from Steam CDN on miss. ``innate.png`` / ``talent_tree.png`` are bundled shared badges (not fetched from CDN). Many innate ability keys 404 on Steam ``dota_react/abilities``; the web UI falls back to ``innate.png`` when a per-ability icon is missing. """ if not key or "/" in key or "\\" in key or ".." in key: return None ABILITY_ICONS.mkdir(parents=True, exist_ok=True) dest = ABILITY_ICONS / f"{key}.png" if dest.is_file() and dest.stat().st_size >= 32: return dest if key in (INNATE_ICON_NAME, TALENT_TREE_ICON_NAME): return None try: data = http_bytes(ABILITY_ICON_URL.format(key=key), timeout=30) if not data or len(data) < 32: raise OSError("empty ability icon") dest.write_bytes(data) return dest except (urllib.error.URLError, TimeoutError, OSError): # CDN 404s are common for innates; serve the bundled badge when present. innate = ABILITY_ICONS / f"{INNATE_ICON_NAME}.png" if innate.is_file() and innate.stat().st_size >= 32: return innate return None def load_grid_order() -> dict[str, list[str]]: """Column membership + order for the pick grid (see data/hero_grid_order.json).""" if not GRID_ORDER_PATH.is_file(): return {} raw = json.loads(GRID_ORDER_PATH.read_text(encoding="utf-8")) out: dict[str, list[str]] = {} for attr in ATTR_ORDER: keys = raw.get(attr) or [] if isinstance(keys, list): out[attr] = [str(k) for k in keys if isinstance(k, str) and not k.startswith("_")] return out def build_payload() -> dict: heroes = hero_table() by_key = {} slim = [] for h in heroes: roles = list(h.get("roles") or []) tags = list(h.get("tags") or []) or tags_for_hero(h["key"], roles) row = { "id": int(h["id"]), "key": h["key"], "name": h.get("name") or h["key"], "name_loc": h.get("name_loc") or h["key"], "aliases": list(h.get("aliases") or []), "abbr": [str(a).lower() for a in (h.get("abbr") or []) if str(a).strip()], "attr": h.get("attr") or "all", "roles": roles, "tags": tags, } # Level-1 strip + combat stats (from OpenDota via fetch_cdn_templates.py). for key in ( "base_str", "str_gain", "base_agi", "agi_gain", "base_int", "int_gain", "health", "mana", "health_regen", "mana_regen", "armor", "damage_min", "damage_max", "move_speed", "attack_range", "attack_rate", "projectile_speed", "magic_resist", "turn_rate", "vision_day", "vision_night", ): if key in h and h[key] is not None: row[key] = h[key] slim.append(row) by_key[row["key"]] = row grid = load_grid_order() by_attr: dict[str, list[dict]] = {a: [] for a in ATTR_ORDER} placed: set[str] = set() for attr in ATTR_ORDER: for key in grid.get(attr) or []: row = by_key.get(key) if row is None or key in placed: continue by_attr[attr].append(row) placed.add(key) # Heroes missing from the order file → append under their primary attr. rest = [r for r in slim if r["key"] not in placed] rest.sort(key=lambda r: r["id"]) for row in rest: attr = row["attr"] if row["attr"] in by_attr else "all" by_attr[attr].append(row) slim_ordered: list[dict] = [] for attr in ATTR_ORDER: slim_ordered.extend(by_attr[attr]) rel = load_relations() items = load_hero_items() hero_stats = load_hero_stats() hero_matches = load_hero_matches() fears = load_hero_item_fears() abilities = load_hero_abilities() shop = load_item_shop() items_meta = load_items_meta_index() patches_data = load_patches() patch_summaries = load_patch_summaries() leaderboards = load_leaderboards() pro_matches = load_pro_matches() streamers = load_streamers() stratz_meta = load_stratz_hero_meta() stratz_matchups = load_stratz_matchup_tops() # Attach craft graph from shop catalog when available. for key, row in (shop.get("items") or {}).items(): if not isinstance(row, dict): continue cell = items_meta.setdefault(key, {"key": key}) cell.setdefault("name_loc", row.get("name_loc") or key) if row.get("cost") is not None: cell["cost"] = row.get("cost") if row.get("desc_loc"): cell["desc_loc"] = row.get("desc_loc") cell["components"] = list(row.get("components") or []) cell["builds_into"] = list(row.get("builds_into") or []) return { "heroes": slim_ordered, "by_attr": by_attr, "attr_order": list(ATTR_ORDER), "attr_cols": ATTR_COLS, "attr_labels": ATTR_LABELS, "tag_order": list(TAG_ORDER), "relations": rel, "hero_items": items, "hero_stats": hero_stats, "hero_matches": hero_matches, "hero_item_fears": fears, "hero_abilities": abilities, "item_shop": shop, "items_meta": items_meta, "patches": patches_data.get("patches") or [], "patch_lookup": patches_data.get("lookup") or {}, "patch_details": patches_data.get("details") or {}, "patch_summaries": patch_summaries, "leaderboards": leaderboards, "pro_matches": pro_matches, "streamers": streamers, "stratz_hero_meta": stratz_meta, "stratz_matchup_tops": stratz_matchups, "mechanic_query": mechanic_query_payload(), "meta": { "relations_path": str(DEFAULT_RELATIONS.relative_to(ROOT)).replace("\\", "/"), "grid_order_path": str(GRID_ORDER_PATH.relative_to(ROOT)).replace("\\", "/"), "hero_items_path": str(HERO_ITEMS_PATH.relative_to(ROOT)).replace("\\", "/"), "hero_stats_path": str(HERO_STATS_PATH.relative_to(ROOT)).replace("\\", "/"), "hero_matches_path": str(HERO_MATCHES_PATH.relative_to(ROOT)).replace( "\\", "/" ), "hero_item_fears_path": str(HERO_ITEM_FEARS_PATH.relative_to(ROOT)).replace( "\\", "/" ), "hero_abilities_path": str(HERO_ABILITIES_PATH.relative_to(ROOT)).replace( "\\", "/" ), "item_shop_path": str(ITEM_SHOP_PATH.relative_to(ROOT)).replace("\\", "/"), "leaderboards_path": str(LEADERBOARDS_PATH.relative_to(ROOT)).replace( "\\", "/" ), "streamers_path": str(STREAMERS_PATH.relative_to(ROOT)).replace("\\", "/"), "stratz_hero_meta_path": str(STRATZ_HERO_META_PATH.relative_to(ROOT)).replace( "\\", "/" ), "stratz_matchup_tops_path": str( STRATZ_MATCHUP_TOPS_PATH.relative_to(ROOT) ).replace("\\", "/"), "source": (rel.get("meta") or {}).get("source"), "counters": len(rel.get("counters") or []), "synergies": len(rel.get("synergies") or []), "item_heroes": len(items.get("by_hero") or {}), "stats_heroes": len(hero_stats.get("by_hero") or {}), "match_heroes": len(hero_matches.get("by_hero") or {}), "fear_heroes": len(fears.get("by_hero") or {}), "ability_heroes": len(abilities.get("by_hero") or {}), "shop_items": len(shop.get("items") or {}), "patches": len(patches_data.get("patches") or []), "patch_details": len(patches_data.get("details") or {}), "patch_summaries": len(patch_summaries.get("by_version") or {}), "leaderboard_regions": len(leaderboards.get("regions") or {}), "pro_match_players": len(pro_matches.get("by_pro") or {}), "pro_match_heroes": len(pro_matches.get("by_hero") or {}), "streamers": len(streamers.get("streamers") or []), "stratz_meta_heroes": len(stratz_meta.get("by_hero") or {}), "stratz_matchup_heroes": len(stratz_matchups.get("by_hero") or {}), }, } class Handler(BaseHTTPRequestHandler): server_version = "ClimperorRelations/2.0" def log_message(self, fmt: str, *args) -> None: print(f"[relations] {self.address_string()} {fmt % args}") def _send( self, code: int, body: bytes, content_type: str, *, extra_headers: dict[str, str] | None = None, ) -> None: self.send_response(code) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header( "Content-Security-Policy", "default-src 'self'; " "img-src 'self' data: https://climperor.oss-cn-shanghai.aliyuncs.com " "https://avatars.steamstatic.com; " "style-src 'self' 'unsafe-inline'; " "script-src 'self'; " "media-src 'self' https://climperor.oss-cn-shanghai.aliyuncs.com", ) if extra_headers: for k, v in extra_headers.items(): self.send_header(k, v) self.end_headers() if body: self.wfile.write(body) def _redirect(self, location: str, *, set_cookie: str | None = None) -> None: headers = {"Location": location} if set_cookie: headers["Set-Cookie"] = set_cookie self._send(302, b"", "text/plain; charset=utf-8", extra_headers=headers) def _request_origin(self) -> str: host = self.headers.get("Host") or "127.0.0.1:8765" # Local serve is always http. return f"http://{host}" def _serve_steam_auth_get(self, path: str) -> bool: """Handle Steam auth GETs. Returns True if handled.""" api_key, session_secret = _steam_env_secrets() origin = self._request_origin() if path == "/api/auth/steam": if not api_key or not session_secret: self._send(503, b"Steam login not configured", "text/plain; charset=utf-8") return True self._redirect(_steam_login_url(origin)) return True if path == "/api/auth/steam/callback": if not api_key or not session_secret: self._redirect(f"{origin}/?auth=unconfigured") return True qs = urllib.parse.parse_qs(urlparse(self.path).query) ok, steamid = _steam_verify_openid(qs) if not ok or not steamid: self._redirect(f"{origin}/?auth=denied") return True account_id = _steam_id64_to_account(steamid) if not account_id: self._redirect(f"{origin}/?auth=error") return True persona = _steam_fetch_persona(api_key, steamid) token = _steam_create_token( session_secret, { "steamid": steamid, "account_id": account_id, "personaname": persona.get("personaname"), "avatar": persona.get("avatar"), }, ) self._redirect( f"{origin}/home", set_cookie=_steam_cookie_header(token, secure=False), ) return True if path == "/api/auth/me": token = _steam_read_cookie(self.headers.get("Cookie")) session = _steam_verify_token(session_secret, token) if session_secret else None if not session: self._json(200, {"authenticated": False}) return True self._json( 200, { "authenticated": True, "steamid": session.get("steamid"), "account_id": session.get("account_id"), "personaname": session.get("personaname"), "avatar": session.get("avatar"), }, ) return True if path == "/api/auth/logout": self._redirect( f"{origin}/", set_cookie=_steam_cookie_header("", clear=True, secure=False), ) return True return False def _send_file(self, fpath: Path, content_type: str) -> None: """Stream a file with optional HTTP Range (needed for HTML5 video seek).""" size = fpath.stat().st_size range_hdr = self.headers.get("Range") or self.headers.get("range") start, end = 0, size - 1 code = 200 if range_hdr and range_hdr.startswith("bytes=") and size > 0: spec = range_hdr[len("bytes=") :].strip() if "," not in spec: left, _, right = spec.partition("-") try: if left == "" and right: # suffix bytes: bytes=-N suffix = int(right) start = max(0, size - suffix) else: start = int(left) if left else 0 end = int(right) if right else size - 1 if start < 0 or end >= size or start > end: self.send_response(416) self.send_header("Content-Range", f"bytes */{size}") self.end_headers() return code = 206 except ValueError: start, end = 0, size - 1 code = 200 length = end - start + 1 self.send_response(code) self.send_header("Content-Type", content_type) self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Length", str(length)) if code == 206: self.send_header("Content-Range", f"bytes {start}-{end}/{size}") self.send_header("Cache-Control", "no-store") self.send_header( "Content-Security-Policy", "default-src 'self'; " "img-src 'self' data: https://climperor.oss-cn-shanghai.aliyuncs.com " "https://avatars.steamstatic.com; " "style-src 'self' 'unsafe-inline'; " "script-src 'self'; " "media-src 'self' https://climperor.oss-cn-shanghai.aliyuncs.com", ) self.end_headers() with fpath.open("rb") as fh: fh.seek(start) remaining = length while remaining > 0: chunk = fh.read(min(1 << 20, remaining)) if not chunk: break self.wfile.write(chunk) remaining -= len(chunk) def _json(self, code: int, obj: object) -> None: self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8") def do_GET(self) -> None: # noqa: N802 path = urlparse(self.path).path if path.startswith("/api/auth/"): if self._serve_steam_auth_get(path): return if path in ("/", "/index.html"): index = WEB_DIR / "index.html" if not index.is_file(): self._json(500, {"error": "web/relations/index.html missing"}) return self._send(200, index.read_bytes(), "text/html; charset=utf-8") return if path in ("/api/data", "/data.json"): self._json(200, build_payload()) return if path == "/api/live-status": self._json(200, _probe_live_status()) return if path == "/api/mobile-demand": self._json(200, {"count": _read_mobile_demand_count()}) return if path.startswith("/api/players/"): self._serve_player_api(path) return if path.startswith("/attr/"): key = path[len("/attr/") :] if key not in ("str.png", "agi.png", "int.png", "all.png"): self._json(400, {"error": "bad attr icon"}) return fpath = ATTR_ICONS / key if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/rank/"): key = path[len("/rank/") :] allowed = {f"rank_icon_{i}.png" for i in range(1, 9)} allowed |= {f"rank_star_{i}.png" for i in range(1, 6)} if key not in allowed: self._json(400, {"error": "bad rank icon"}) return fpath = RANK_ICONS / key if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/role-icon/"): key = path[len("/role-icon/") :] allowed_role_icons = { "Carry.png", "Support.png", "Pusher.png", "Escape.png", "Nuker.png", "Initiator.png", "Disabler.png", "Durable.png", } if key not in allowed_role_icons: self._json(400, {"error": "bad role icon"}) return fpath = ROLE_ICONS / key if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/ui-icon/"): key = path[len("/ui-icon/") :] allowed_ui = { "cooldown.png", "dota2_logo.png", "dota2_logo_wordmark.png", "platform_douyin.png", "platform_bilibili.png", # Hero combat stats (dota2.com.cn/herostatic/stats) "icon_damage.png", "icon_attack_time.png", "icon_attack_range.png", "icon_projectile_speed.png", "icon_armor.png", "icon_magic_resist.png", "icon_movement_speed.png", "icon_turn_rate.png", "icon_vision.png", } if key not in allowed_ui: self._json(400, {"error": "bad ui icon"}) return fpath = UI_ICONS / key if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/streamer-avatar/"): key = path[len("/streamer-avatar/") :] if "/" in key or "\\" in key or ".." in key: self._json(400, {"error": "bad path"}) return if not (key.endswith(".jpg") or key.endswith(".jpeg") or key.endswith(".png") or key.endswith(".webp")): self._json(400, {"error": "bad path"}) return fpath = STREAMER_AVATARS / key if not fpath.is_file(): self.send_error(404) return ctype = mimetypes.guess_type(key)[0] or "image/jpeg" self._send(200, fpath.read_bytes(), ctype) return if path.startswith("/streamer-video/"): key = path[len("/streamer-video/") :] if "/" in key or "\\" in key or ".." in key: self._json(400, {"error": "bad path"}) return if not ( key.endswith(".mp4") or key.endswith(".webm") or key.endswith(".jpg") or key.endswith(".jpeg") or key.endswith(".webp") or key.endswith(".png") ): self._json(400, {"error": "bad path"}) return if key.startswith("_"): self.send_error(404) return fpath = STREAMER_VIDEOS / key if not fpath.is_file(): self.send_error(404) return if key.endswith(".webm"): ctype = "video/webm" elif key.endswith(".mp4"): ctype = "video/mp4" else: ctype = mimetypes.guess_type(key)[0] or "image/jpeg" if ctype.startswith("video/"): self._send_file(fpath, ctype) else: self._send(200, fpath.read_bytes(), ctype) return if path.startswith("/portrait/"): key = path[len("/portrait/") :] if "/" in key or "\\" in key or not key.endswith(".png"): self._json(400, {"error": "bad path"}) return # Wide Heroes-page cards only — never serve match templates here. fpath = HERO_PORTRAITS / key if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/cdn/"): key = path[len("/cdn/") :] if "/" in key or "\\" in key or not key.endswith(".png"): self._json(400, {"error": "bad path"}) return fpath = TEMPLATES_CDN / key if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/item-cat/"): key = path[len("/item-cat/") :] if "/" in key or "\\" in key or not key.endswith(".png"): self._json(400, {"error": "bad path"}) return if not key.startswith("itemcat_"): self._json(400, {"error": "bad path"}) return fpath = ITEM_CAT_ICONS / key if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/item/"): key = path[len("/item/") :] if "/" in key or "\\" in key or not key.endswith(".png"): self._json(400, {"error": "bad path"}) return fpath = ITEM_ICONS / key # All recipe scrolls share recipe.png on Steam CDN. if not fpath.is_file() and key.startswith("recipe_"): fpath = ITEM_ICONS / "recipe.png" if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/ability/"): key = path[len("/ability/") :] if "/" in key or "\\" in key or not key.endswith(".png"): self._json(400, {"error": "bad path"}) return stem = key[: -len(".png")] fpath = ensure_ability_icon(stem) if fpath is None: self.send_error(404) return self._send(200, fpath.read_bytes(), "image/png") return if path.startswith("/ability-video/"): rel_path = path[len("/ability-video/") :] parts = rel_path.split("/") if len(parts) != 2: self._json(400, {"error": "bad path"}) return hero, fname = parts if ( not hero or ".." in hero or "\\" in hero or ".." in fname or "\\" in fname ): self._json(400, {"error": "bad path"}) return if fname.endswith(".webm"): ctype = "video/webm" elif fname.endswith(".mp4"): ctype = "video/mp4" else: self._json(400, {"error": "bad path"}) return fpath = ABILITY_VIDEOS / hero / fname if not fpath.is_file(): self.send_error(404) return self._send(200, fpath.read_bytes(), ctype) return rel = path.lstrip("/") candidate = (WEB_DIR / rel).resolve() web_root = WEB_DIR.resolve() if str(candidate).startswith(str(web_root)) and candidate.is_file(): ctype = mimetypes.guess_type(str(candidate))[0] or "application/octet-stream" self._send(200, candidate.read_bytes(), ctype) return # History SPA fallback: /heroes/axe → index.html (client router). spa_pages = { "home", "heroes", "rankings", "matches", "streamers", "trends", "mechanics", "items", "patches", "players", } first = path.strip("/").split("/", 1)[0] if path.strip("/") else "" if first in spa_pages: index = WEB_DIR / "index.html" if index.is_file(): self._send(200, index.read_bytes(), "text/html; charset=utf-8") return self.send_error(404) def _serve_player_api(self, path: str) -> None: """GET /api/players/{account_id}[/match_id] from pc/player_pages/.""" parts = [p for p in path.strip("/").split("/") if p] # ["api", "players", account_id] or + match_id if len(parts) < 3 or parts[0] != "api" or parts[1] != "players": self._json(400, {"error": "bad players path"}) return account_id = parts[2] if account_id == "me" and len(parts) == 3: api_key, session_secret = _steam_env_secrets() token = _steam_read_cookie(self.headers.get("Cookie")) session = ( _steam_verify_token(session_secret, token) if session_secret else None ) if not session or not session.get("account_id"): self._json(401, {"authenticated": False}) return aid = int(session["account_id"]) try: cfg = _load_pc_config() # Cache-first (TTL); cold miss syncs once, stale refresh is async. profile = _get_profile_for_web(cfg, aid, force=False, include_gsi=True) except Exception as e: # noqa: BLE001 self._json(500, {"error": "enrich failed", "detail": str(e)}) return if not profile.get("personaname") and session.get("personaname"): profile["personaname"] = session.get("personaname") if not profile.get("avatar") and session.get("avatar"): profile["avatar"] = session.get("avatar") profile["authenticated"] = True self._json(200, profile) return if not account_id.isdigit(): self._json(400, {"error": "bad account_id"}) return if len(parts) == 3: fpath = PC_PLAYER_PAGES / account_id / "profile.json" if not fpath.is_file(): self._json(404, {"error": "profile not found", "account_id": account_id}) return try: self._json(200, json.loads(fpath.read_text(encoding="utf-8"))) except (OSError, ValueError) as e: self._json(500, {"error": str(e)}) return if len(parts) == 4: match_id = parts[3] if not match_id.isdigit(): self._json(400, {"error": "bad match_id"}) return fpath = PC_PLAYER_PAGES / account_id / "matches" / f"{match_id}.json" if not fpath.is_file(): self._json( 404, { "error": "match not found", "account_id": account_id, "match_id": match_id, }, ) return try: self._json(200, json.loads(fpath.read_text(encoding="utf-8"))) except (OSError, ValueError) as e: self._json(500, {"error": str(e)}) return self._json(400, {"error": "bad players path"}) def _read_json_body(self) -> dict | None: length = int(self.headers.get("Content-Length", 0) or 0) raw = self.rfile.read(length) if length else b"{}" try: body = json.loads(raw.decode("utf-8")) except (ValueError, UnicodeDecodeError): return None return body if isinstance(body, dict) else {} def do_POST(self) -> None: # noqa: N802 path = urlparse(self.path).path if path == "/api/mobile-demand": self._json(200, {"count": _inc_mobile_demand_count(), "voted": True}) return if path == "/api/auth/logout": origin = self._request_origin() self._redirect( f"{origin}/", set_cookie=_steam_cookie_header("", clear=True, secure=False), ) return if path == "/api/players/publish": # Local dev: no OSS write; PC already wrote pc/player_pages/. body = self._read_json_body() if body is None: self._json(400, {"error": "invalid json"}) return account_id = body.get("account_id") match_id = body.get("match_id") self._json( 200, { "ok": True, "local": True, "message": "local publish no-op; files under pc/player_pages/", "account_id": account_id, "match_id": match_id, }, ) return if path == "/api/players/enrich": body = self._read_json_body() if body is None: self._json(400, {"error": "invalid json"}) return account_id = body.get("account_id") try: aid = int(account_id) except (TypeError, ValueError): aid = 0 if aid <= 0: self._json(400, {"error": "account_id required"}) return include_gsi = body.get("include_gsi") if include_gsi is None: include_gsi = True force = bool(body.get("force", True)) try: cfg = _load_pc_config() if force: profile = _enrich_profile_recent( cfg, aid, include_gsi=bool(include_gsi) ) else: profile = _get_profile_for_web( cfg, aid, force=False, include_gsi=bool(include_gsi) ) except Exception as e: # noqa: BLE001 self._json(500, {"error": "enrich failed", "detail": str(e)}) return if profile.get("error"): self._json(400, profile) return self._json(200, {"ok": True, "profile": profile}) return if path == "/api/players/ensure-match": body = self._read_json_body() if body is None: self._json(400, {"error": "invalid json"}) return try: aid = int(body.get("account_id")) mid = int(body.get("match_id")) except (TypeError, ValueError): aid, mid = 0, 0 if aid <= 0 or mid <= 0: self._json(400, {"error": "account_id and match_id required"}) return try: cfg = _load_pc_config() detail, err = _ensure_match_detail(cfg, aid, mid) except Exception as e: # noqa: BLE001 self._json(500, {"error": "ensure-match failed", "detail": str(e)}) return if err or not detail: self._json( 404 if err else 500, {"error": err or "unknown", "account_id": aid, "match_id": mid}, ) return self._json(200, {"ok": True, "match": detail}) return self.send_error(404) def main() -> None: ap = argparse.ArgumentParser(description="Preview qualitative hero relations (read-only)") ap.add_argument("--port", type=int, default=8765) ap.add_argument("--host", default="127.0.0.1") ap.add_argument("--no-browser", action="store_true") args = ap.parse_args() if not (WEB_DIR / "index.html").is_file(): raise SystemExit(f"missing UI: {WEB_DIR / 'index.html'}") if not TEMPLATES_CDN.is_dir(): raise SystemExit("templates/cdn missing — run python fetch_cdn_templates.py") httpd = ThreadingHTTPServer((args.host, args.port), Handler) url = f"http://{args.host}:{args.port}/" print(f"Climperor web: {url}") print(f"edit JSON then refresh: {DEFAULT_RELATIONS}") if not args.no_browser: try: webbrowser.open(url) except Exception: # noqa: BLE001 pass try: httpd.serve_forever() except KeyboardInterrupt: print("\nstopped") httpd.server_close() if __name__ == "__main__": main()