Files
climperor/web/serve_relations.py
T
vosonandCursor 820c3fb1f1 Ship item nicknames, Tiny fear override, and ensure-match API.
Add 冰眼/蛇矛 aliases, correct Tiny fears to Hydra's Breath, and fix production match detail POST 405.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-01 04:05:48 +08:00

1522 lines
56 KiB
Python

"""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"])
cell = {
"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 []),
}
aliases = [
str(a).strip()
for a in (row.get("aliases") or [])
if str(a).strip()
]
if aliases:
cell["aliases"] = aliases
out[key] = cell
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 ""
aliases = [
str(a).strip()
for a in (m.get("aliases") or [])
if str(a).strip()
]
if aliases:
row["aliases"] = aliases
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()