climperor/web/serve_relations.py

"""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 — read-only browser UI.
Edit data/*.json directly, then refresh the page.
"""

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.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,
    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

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"
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_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:
            return None
        dest.write_bytes(data)
        return dest
    except (urllib.error.URLError, TimeoutError, OSError):
        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()
    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 {},
        "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 {}),
            "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) -> 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; "
            "style-src 'self' 'unsafe-inline'; "
            "script-src 'self'; "
            "media-src 'self' https://climperor.oss-cn-shanghai.aliyuncs.com",
        )
        self.end_headers()
        self.wfile.write(body)

    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; "
            "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 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("/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)}
            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/") or path.startswith("/cdn/"):
            prefix = "/portrait/" if path.startswith("/portrait/") else "/cdn/"
            key = path[len(prefix) :]
            if "/" in key or "\\" in key or not key.endswith(".png"):
                self._json(400, {"error": "bad path"})
                return
            # Prefer official Heroes-page cards; fall back to match templates.
            fpath = HERO_PORTRAITS / key
            if not fpath.is_file():
                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()
        if not str(candidate).startswith(str(WEB_DIR.resolve())) or not candidate.is_file():
            self.send_error(404)
            return
        ctype = mimetypes.guess_type(str(candidate))[0] or "application/octet-stream"
        self._send(200, candidate.read_bytes(), ctype)

    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
        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()