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