Files
climperor/web/serve_relations.py
T
vosonandCursor 9c5aa5b610 Reorganize repository into pc web shared monorepo
Separate the local recognition, web publishing, and shared data paths while preserving direct script execution and existing site content.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-29 14:29:08 +08:00

992 lines
36 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 — 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 webbrowser
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,
ROOT,
STREAMER_AVATARS,
STREAMER_VIDEOS,
TEMPLATES_CDN,
UI_ICONS,
WEB_FRONTEND,
)
from shared.relations import DEFAULT_RELATIONS, load_relations
from mechanic_tags import QUERY_MECHANIC_ORDER, mechanic_query_payload
WEB_DIR = WEB_FRONTEND
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"
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": {},
}
if not STRATZ_HERO_META_PATH.is_file():
return empty
try:
raw = json.loads(STRATZ_HERO_META_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,
"source": "stratz",
"attribution": "https://stratz.com",
"take": 0,
"match_limit": 0,
"note": None,
"by_hero": {},
}
if not STRATZ_MATCHUP_TOPS_PATH.is_file():
return empty
try:
raw = json.loads(STRATZ_MATCHUP_TOPS_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"],
"take": raw.get("take") or 0,
"match_limit": raw.get("match_limit") or 0,
"note": raw.get("note"),
"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":
# Local-dev stub for the Pages Function (edge probing runs in
# production only); the empty map keeps the UI on data.json is_live.
self._json(200, {"probed_at": None, "ttl": 300, "streamers": {}})
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("/ui-icon/"):
key = path[len("/ui-icon/") :]
allowed_ui = {
"cooldown.png",
"dota2_logo.png",
"dota2_logo_wordmark.png",
"platform_douyin.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")):
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
ctype = (
"video/webm" if key.endswith(".webm") else "video/mp4"
)
self._send_file(fpath, 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 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()