#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
import json
import math
import os
import re
import subprocess
import sys
import threading
import time
import webbrowser
from collections import Counter
from datetime import datetime
from pathlib import Path
from typing import Any
import requests as std_requests
from bs4 import BeautifulSoup
try:
from curl_cffi import requests as curl_requests
except Exception:
curl_requests = None
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import HTMLResponse
import uvicorn
ROOT = Path(__file__).resolve().parent
DATA_DIR = ROOT / "data"
PLAYERS_FILE = DATA_DIR / "khl_players_all.json"
COACHES_FILE = DATA_DIR / "khl_coaches_all.json"
OFFICIALS_FILE = DATA_DIR / "khl_officials_all.json"
WEB_DIR = ROOT / "web"
INDEX_FILE = WEB_DIR / "index.html"
PLAYER_SITE_CACHE_FILE = DATA_DIR / "khl_player_site_profiles.json"
PLAYER_MATCHES_CACHE_DIR = DATA_DIR / "player_matches_cache"
PLAYER_SITE_CACHE_LOCK = threading.RLock()
_PLAYER_SITE_CACHE: dict[str, Any] | None = None
_PLAYER_SITE_FAILURES: dict[str, float] = {}
APP = FastAPI(title="KHL Data Center", version="2.5.1")
HTML = r'''
KHL Data Center
Выбери запись слева
'''
SUM_IDS = {"gp", "g", "a", "pts", "pim", "pm", "fow", "sds", "w", "l", "sop", "ga", "sv", "so", "toi", "time_on_ice", "distance_travelled"}
WEIGHTED_IDS = {"toi_avg", "sft_avg", "pim_avg"}
MAX_IDS = {"top_speed"}
STAT_ORDER = ["gp", "g", "a", "pts", "pm", "pim", "fow", "sds", "w", "l", "ga", "sv", "sv_pct", "gaa", "so", "sop", "toi", "toi_avg", "sft_avg", "pim_avg", "top_speed", "time_on_ice", "distance_travelled"]
def read_json(path: Path, default: Any) -> Any:
if not path.exists():
return default
try:
with path.open("r", encoding="utf-8") as fh:
return json.load(fh)
except (OSError, json.JSONDecodeError):
return default
KNOWN_STAGE_SEASONS: dict[int, str] = {
27: "2008/2009",
31: "2008/2009",
35: "2009/2010",
39: "2009/2010",
43: "2010/2011",
47: "2010/2011",
51: "2011/2012",
55: "2011/2012",
59: "2012/2013",
63: "2012/2013",
67: "2012/2013",
3: "2013/2014",
7: "2013/2014",
11: "2013/2014",
15: "2014/2015",
19: "2014/2015",
23: "2015/2016",
71: "2015/2016",
75: "2016/2017",
83: "2016/2017",
89: "2017/2018",
97: "2017/2018",
105: "2018/2019",
141: "2018/2019",
157: "2019/2020",
177: "2019/2020",
189: "2020/2021",
197: "2020/2021",
209: "2021/2022",
225: "2021/2022",
235: "2022/2023",
259: "2022/2023",
275: "2023/2024",
299: "2023/2024",
323: "2024/2025",
359: "2024/2025",
370: "2025/2026",
395: "2025/2026",
407: "2026/2027",
}
def stage_season(stage: dict[str, Any]) -> str:
explicit = str(stage.get("season") or "").strip()
if explicit:
return explicit
source = stage.get("source_metadata")
if isinstance(source, dict):
for key in ("season", "season_name", "season_title", "season_year", "season_years", "years", "year"):
value = source.get(key)
if value not in (None, ""):
match = re.search(r"(?:19|20)\d{2}\s*[/–—-]\s*(?:19|20)?\d{2}", str(value))
if match:
return match.group(0).replace("–", "/").replace("—", "/").replace("-", "/").replace(" ", "")
try:
stage_id = int(stage.get("id"))
except (TypeError, ValueError):
return ""
return KNOWN_STAGE_SEASONS.get(stage_id, "")
def stage_label(stage: dict[str, Any]) -> str:
season = stage_season(stage)
title = stage.get("title") or "Этап"
sid = stage.get("id")
return " · ".join(x for x in (season, title, f"ID {sid}") if x)
GENERIC_STAFF_NAMES = {
"тренер", "тренеры", "тренеры кхл", "coach", "coaches", "khl coaches",
"судья", "судьи", "судьи кхл", "official", "officials", "referee", "referees",
}
def normalize_person_text(value: Any) -> str:
return re.sub(r"\s+", " ", str(value or "")).strip(" \t\r\n,;:|—–-")
def is_generic_staff_name(value: Any) -> bool:
text = normalize_person_text(value).casefold()
if not text:
return True
if text in GENERIC_STAFF_NAMES:
return True
return text.startswith("тренеры кхл ") or text.startswith("судьи кхл ")
def is_valid_player_name(value: Any) -> bool:
text = normalize_person_text(value)
if not text:
return False
low = text.casefold()
blocked = (
"конференц", "дивизион", "игроки", "тренеры", "судьи", "статистика",
"матчи", "амплуа", "гражданство", "дата рождения", "континентальная хоккейная лига",
)
if any(x in low for x in blocked):
return False
return len(text.split()) >= 2 and bool(re.search(r"[А-Яа-яЁё]", text))
def player_sidebar_name(item: dict[str, Any], item_id: str) -> str:
"""Stable left-menu name in surname + first-name order.
New collections store profile.list_name_ru directly from KHL's player list.
Older files are repaired from structured identity fields when possible; bad
global-navigation values such as «Конференция Запад» are never displayed.
"""
identity = item.get("identity") or {}
profile = item.get("profile") or {}
direct = normalize_person_text(profile.get("list_name_ru"))
if is_valid_player_name(direct):
return direct
first = normalize_person_text(identity.get("first_name"))
last = normalize_person_text(identity.get("last_name"))
structured = f"{last} {first}".strip()
if is_valid_player_name(structured):
return structured
for candidate in (identity.get("full_name"), profile.get("name")):
text = normalize_person_text(candidate)
if is_valid_player_name(text):
return text
return f"Игрок #{item_id}"
def compose_first_last(first: Any, last: Any, middle: Any = None, *, include_middle: bool = False) -> str:
parts = [normalize_person_text(first)]
if include_middle:
parts.append(normalize_person_text(middle))
parts.append(normalize_person_text(last))
return " ".join(part for part in parts if part)
def reorder_khl_person_name(value: Any) -> str:
"""KHL обычно отдаёт персонал как Фамилия Имя [Отчество]."""
text = normalize_person_text(value)
if is_generic_staff_name(text):
return ""
parts = text.split()
if len(parts) < 2:
return text
# Для бокового списка показываем именно Имя + Фамилия.
return f"{parts[1]} {parts[0]}"
def staff_display_name(item: dict[str, Any], language: str = "ru") -> str:
profile = item.get("profile") or {}
languages = item.get("languages") or {}
lang_profile = languages.get(language) or {}
suffix = "_ru" if language == "ru" else "_en"
explicit = compose_first_last(
item.get(f"first_name{suffix}"),
item.get(f"last_name{suffix}"),
)
if explicit and not is_generic_staff_name(explicit):
return explicit
for source in (lang_profile, profile):
parts = source.get("name_parts") or {}
explicit = compose_first_last(parts.get("first_name"), parts.get("last_name"))
if explicit and not is_generic_staff_name(explicit):
return explicit
candidates = [
item.get(f"name{suffix}"),
lang_profile.get("name"),
profile.get("name_en" if language == "en" else "name"),
]
page_title = normalize_person_text(lang_profile.get("page_title") or profile.get("page_title"))
if page_title:
page_name = re.split(r",|\|| — | - ", page_title, maxsplit=1)[0]
candidates.append(page_name)
for appearance in item.get("appearances") or []:
if language and appearance.get("language") not in (None, "", language):
continue
candidates.extend((appearance.get("name"), appearance.get("title")))
if language == "ru":
candidates.append(item.get("name"))
for candidate in candidates:
result = reorder_khl_person_name(candidate)
if result:
return result
if language == "en":
return ""
return f"Персона #{item.get('id', '—')}"
def valid_person_image(value: Any) -> str:
url = str(value or "").strip()
if not url:
return ""
low = url.casefold()
bad = ("logo", "logotype", "sprite", "icon", "favicon", "banner", "advert", "sponsor", "/flags/", "flag_")
return "" if any(token in low for token in bad) else url
def aggregate_player(player: dict[str, Any], stages_by_id: dict[str, dict[str, Any]], mode: str) -> dict[str, Any]:
acc: dict[str, dict[str, Any]] = {}
stage_count = 0
for stage_id, record in (player.get("stages") or {}).items():
title = str(stages_by_id.get(str(stage_id), {}).get("title") or "").casefold()
if mode == "regular" and "регуляр" not in title:
continue
if mode == "playoff" and "плей" not in title:
continue
if mode == "hope" and "надежд" not in title:
continue
data = record.get("data") or {}
stats = data.get("stats") or []
if not stats:
continue
stage_count += 1
gp = next((float(s.get("val") or 0) for s in stats if s.get("id") == "gp"), 0.0)
for stat in stats:
sid = str(stat.get("id") or "")
try:
value = float(stat.get("val"))
except (TypeError, ValueError):
continue
if not sid:
continue
item = acc.setdefault(sid, {"id": sid, "title": stat.get("title") or sid, "sum": 0.0, "weighted": 0.0, "weight": 0.0, "max": None})
if sid in WEIGHTED_IDS:
if gp > 0:
item["weighted"] += value * gp
item["weight"] += gp
elif sid in MAX_IDS:
item["max"] = value if item["max"] is None else max(item["max"], value)
elif sid in SUM_IDS:
item["sum"] += value
stats_out = []
for sid, item in acc.items():
if sid in WEIGHTED_IDS:
if item["weight"] <= 0:
continue
value = item["weighted"] / item["weight"]
elif sid in MAX_IDS:
value = item["max"]
elif sid in SUM_IDS:
value = item["sum"]
else:
continue
if isinstance(value, float) and value.is_integer():
value = int(value)
stats_out.append({"id": sid, "title": item["title"], "val": value})
by_id = {x["id"]: x["val"] for x in stats_out}
saves = float(by_id.get("sv") or 0)
goals = float(by_id.get("ga") or 0)
toi = float(by_id.get("toi") or 0)
if saves + goals > 0:
stats_out.append({"id": "sv_pct", "title": "Процент отражённых бросков", "val": saves * 100 / (saves + goals)})
if toi > 0:
stats_out.append({"id": "gaa", "title": "Коэффициент надёжности", "val": goals * 60 / toi})
order = {key: idx for idx, key in enumerate(STAT_ORDER)}
stats_out.sort(key=lambda x: order.get(x["id"], 999))
return {"stage_count": stage_count, "stats": stats_out}
def _numeric_stage_sort(value: str) -> tuple[int, str]:
text = str(value)
return (int(text) if text.isdigit() else -1, text)
COUNTRY_BY_FLAG_CODE = {
"RU": "Россия", "BY": "Беларусь", "KZ": "Казахстан", "FI": "Финляндия",
"SE": "Швеция", "CZ": "Чехия", "SK": "Словакия", "LV": "Латвия",
"LT": "Литва", "EE": "Эстония", "US": "США", "CA": "Канада",
"DE": "Германия", "CH": "Швейцария", "AT": "Австрия", "DK": "Дания",
"NO": "Норвегия", "FR": "Франция", "SI": "Словения", "HR": "Хорватия",
"PL": "Польша", "HU": "Венгрия", "CN": "Китай", "JP": "Япония",
"KR": "Южная Корея", "GB": "Великобритания", "IT": "Италия", "UA": "Украина",
"GE": "Грузия", "AM": "Армения", "AZ": "Азербайджан", "UZ": "Узбекистан",
}
def _profile_value(profile: dict[str, Any], *keys: str) -> Any:
for key in keys:
value = profile.get(key)
if value not in (None, "", [], {}):
return value
return None
def normalize_player_profile_fields(profile: dict[str, Any]) -> dict[str, Any]:
"""Приводит разные имена полей API/KHL.ru к одному виду для интерфейса."""
aliases = {
"birthday": ("birthday", "birth_date", "date_of_birth", "birthdate", "dob"),
"country": ("country", "country_name", "nationality", "citizenship", "citizenship_name"),
"height": ("height", "height_cm", "player_height"),
"weight": ("weight", "weight_kg", "player_weight"),
"stick": ("stick", "shoots", "grip", "handedness", "shooting_side", "shooting_hand"),
}
for target, keys in aliases.items():
if profile.get(target) in (None, "", [], {}):
value = _profile_value(profile, *keys)
if value not in (None, "", [], {}):
profile[target] = value
if profile.get("country") in (None, "", [], {}):
flag = str(profile.get("flag_image_url") or profile.get("flag") or "")
match = re.search(r"/([A-Za-z]{2})\.png(?:\?|$)", flag)
if match:
country = COUNTRY_BY_FLAG_CODE.get(match.group(1).upper())
if country:
profile["country"] = country
return profile
def _load_player_site_cache() -> dict[str, Any]:
global _PLAYER_SITE_CACHE
with PLAYER_SITE_CACHE_LOCK:
if _PLAYER_SITE_CACHE is None:
raw = read_json(PLAYER_SITE_CACHE_FILE, {"profiles": {}})
if not isinstance(raw, dict):
raw = {"profiles": {}}
raw.setdefault("profiles", {})
_PLAYER_SITE_CACHE = raw
return _PLAYER_SITE_CACHE
def _save_player_site_cache() -> None:
with PLAYER_SITE_CACHE_LOCK:
cache = _load_player_site_cache()
PLAYER_SITE_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
tmp = PLAYER_SITE_CACHE_FILE.with_suffix(".json.tmp")
tmp.write_text(json.dumps(cache, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
tmp.replace(PLAYER_SITE_CACHE_FILE)
def parse_khl_player_profile_html(html: str) -> dict[str, Any]:
"""Извлекает основные антропометрические поля из персональной страницы KHL.ru."""
soup = BeautifulSoup(html or "", "html.parser")
values = [re.sub(r"\s+", " ", x).strip() for x in soup.stripped_strings]
labels = {
"дата рождения": "birthday",
"гражданство": "country",
"рост": "height",
"вес": "weight",
"хват": "stick",
"возраст": "age",
}
label_names = set(labels)
result: dict[str, Any] = {}
for idx, text in enumerate(values):
normalized = text.casefold().strip(" :")
target = labels.get(normalized)
if not target or target in result:
continue
for candidate in values[idx + 1:idx + 6]:
cand = candidate.strip()
cand_norm = cand.casefold().strip(" :")
if not cand or cand_norm in label_names:
continue
if len(cand) > 100:
break
if target in {"height", "weight", "age"}:
m = re.search(r"\b(\d{1,3})\b", cand)
if not m:
continue
result[target] = int(m.group(1))
else:
result[target] = cand
break
stick = str(result.get("stick") or "").casefold().strip()
if stick in {"l", "left", "лев", "левый"}:
result["stick"] = "левый"
elif stick in {"r", "right", "прав", "правый"}:
result["stick"] = "правый"
return result
def _download_khl_player_page(khl_id: int, matches_page: int = 1) -> str:
url = f"https://www.khl.ru/players/{khl_id}/"
params = None
if matches_page > 1:
# KHL использует стандартную битриксовую пагинацию таблицы матчей.
params = {"PAGEN_1": matches_page, "idplayer": khl_id}
headers = {
"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.7",
"Cache-Control": "no-cache",
}
if curl_requests is not None:
response = curl_requests.get(
url, params=params, headers=headers, impersonate="chrome",
timeout=8, allow_redirects=True,
)
response.raise_for_status()
return response.text
response = std_requests.get(
url, params=params,
headers={**headers, "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/150 Safari/537.36"},
timeout=8, allow_redirects=True,
)
response.raise_for_status()
return response.text
def _download_khl_player_profile(khl_id: int) -> str:
return _download_khl_player_page(khl_id, 1)
def get_khl_site_player_profile(khl_id: int) -> dict[str, Any]:
key = str(khl_id)
now = time.time()
with PLAYER_SITE_CACHE_LOCK:
cache = _load_player_site_cache()
entry = (cache.get("profiles") or {}).get(key) or {}
fetched_at = float(entry.get("fetched_at") or 0)
fields = entry.get("fields") if isinstance(entry.get("fields"), dict) else {}
ttl = 30 * 86400 if fields else 15 * 60
if fetched_at and now - fetched_at < ttl:
return dict(fields)
failed_at = _PLAYER_SITE_FAILURES.get(key, 0)
if failed_at and now - failed_at < 120:
return dict(fields)
try:
html = _download_khl_player_profile(khl_id)
fields = parse_khl_player_profile_html(html)
except Exception as exc:
_PLAYER_SITE_FAILURES[key] = now
print(f"KHL profile {khl_id}: {exc}", file=sys.stderr)
return dict(fields)
with PLAYER_SITE_CACHE_LOCK:
cache = _load_player_site_cache()
cache.setdefault("profiles", {})[key] = {
"fetched_at": now,
"fields": fields,
}
_PLAYER_SITE_FAILURES.pop(key, None)
try:
_save_player_site_cache()
except OSError as exc:
print(f"Не удалось записать кэш профиля KHL {khl_id}: {exc}", file=sys.stderr)
return dict(fields)
def _clean_match_cell(value: str) -> str:
return re.sub(r"\s+", " ", value or "").strip()
def _looks_like_match_date(value: str) -> bool:
text = _clean_match_cell(value)
return bool(
re.fullmatch(r"\d{4}-\d{2}-\d{2}", text)
or re.fullmatch(r"\d{1,2}\s+[А-Яа-яЁё]{3,}\s+\d{4}", text)
or re.fullmatch(r"\d{1,2}\.\d{1,2}\.\d{4}", text)
)
def _looks_like_score(value: str) -> bool:
return bool(re.search(r"\b\d{1,2}\s*:\s*\d{1,2}\b", _clean_match_cell(value)))
def _match_stat_headers(table_headers: list[str], stat_count: int) -> list[str]:
heads = [_clean_match_cell(x) for x in table_headers]
# Основная таблица KHL начинается с «Турнир / Команда», а матч — с Дата/Команды/Счёт.
if heads and ("турнир" in heads[0].casefold() or "команд" in heads[0].casefold()):
heads = heads[1:]
# В строке одного матча нет колонки И=1; сайт её не дублирует.
if len(heads) == stat_count + 1:
gp_idx = next((i for i, h in enumerate(heads) if h.casefold().strip() in {"и", "игры"}), -1)
if gp_idx >= 0:
heads.pop(gp_idx)
if len(heads) > stat_count:
# Сохраняем номер и последние статистические поля; лишние служебные заголовки отбрасываем.
if heads and heads[0] in {"№", "#", "N"}:
heads = [heads[0]] + heads[-(stat_count - 1):] if stat_count > 1 else [heads[0]]
else:
heads = heads[-stat_count:]
while len(heads) < stat_count:
heads.append(f"Показатель {len(heads) + 1}")
return heads[:stat_count]
def parse_khl_player_matches_html(html: str, page: int = 1) -> dict[str, Any]:
"""Разбирает строки матчей с карточки игрока KHL.ru.
На актуальном KHL матчи могут находиться в той же широкой таблице, что и сезонная
статистика, поэтому наличие отдельного thead «Дата / Команды / Счёт» не требуется.
Ищем строки по фактическому формату: дата → пара команд → счёт → статистика.
"""
soup = BeautifulSoup(html or "", "html.parser")
page_text = _clean_match_cell(soup.get_text(" ", strip=True))
total = 0
start = 0
end = 0
match = re.search(r"Игры\s+с\s+(\d+)\s+по\s+(\d+)\s+из\s+(\d+)", page_text, flags=re.I)
if match:
start, end, total = map(int, match.groups())
best_rows: list[dict[str, Any]] = []
best_headers: list[str] = []
for table_node in soup.find_all("table"):
table_headers: list[str] = []
for trh in table_node.select("thead tr"):
cells = trh.find_all(["th", "td"], recursive=False)
candidate = [_clean_match_cell(c.get_text(" ", strip=True)) for c in cells]
if len(candidate) >= len(table_headers):
table_headers = candidate
if not table_headers:
first = table_node.find("tr")
if first:
cells = first.find_all(["th", "td"], recursive=False)
if cells and any(c.name == "th" for c in cells):
table_headers = [_clean_match_cell(c.get_text(" ", strip=True)) for c in cells]
rows: list[dict[str, Any]] = []
section = ""
for tr in table_node.find_all("tr"):
cells = tr.find_all(["th", "td"], recursive=False)
if not cells:
continue
values = [_clean_match_cell(c.get_text(" ", strip=True)) for c in cells]
if not any(values):
continue
if len(cells) == 1 or (len(cells) <= 2 and cells[0].get("colspan")):
candidate = values[0]
if candidate and "игры с " not in candidate.casefold():
section = candidate
continue
# В некоторых версиях KHL первые три значения могут быть объединены иначе.
date_idx = next((i for i, value in enumerate(values[:3]) if _looks_like_match_date(value)), -1)
if date_idx < 0 or len(values) < date_idx + 3:
continue
date_value = values[date_idx]
teams_value = values[date_idx + 1]
score_value = values[date_idx + 2]
if not _looks_like_score(score_value):
continue
if not re.search(r"[-–—‐‑‒−]", teams_value) and len(teams_value.split()) < 2:
continue
stats = values[date_idx + 3:]
row_url = ""
for link in tr.find_all("a", href=True):
href = str(link.get("href") or "")
href_l = href.casefold()
if any(token in href_l for token in ("/game/", "/calendar/", "/protocol/", "/text/", "match")):
row_url = href if href.startswith("http") else f"https://www.khl.ru{href}"
break
rows.append({"section": section, "prefix": [date_value, teams_value, score_value], "stats": stats, "url": row_url})
if len(rows) > len(best_rows):
best_rows = rows
best_headers = table_headers
if not best_rows:
return {
"headers": ["Турнир", "Дата", "Команды", "Счёт"], "rows": [], "page": page,
"page_size": 30, "total": total, "pages": max(1, math.ceil(total / 30)) if total else 1,
"range_start": start, "range_end": end,
}
stat_count = max((len(row["stats"]) for row in best_rows), default=0)
stat_headers = _match_stat_headers(best_headers, stat_count)
rows_out = []
for row in best_rows:
stats = list(row["stats"])
if len(stats) < stat_count:
stats.extend([""] * (stat_count - len(stats)))
rows_out.append({
"section": row.get("section") or "—",
"values": [row.get("section") or "—", *row["prefix"], *stats[:stat_count]],
"url": row.get("url") or "",
})
if total <= 0:
total = (page - 1) * 30 + len(rows_out)
pages = max(1, math.ceil(total / 30))
if not start and rows_out:
start = (page - 1) * 30 + 1
if not end and rows_out:
end = start + len(rows_out) - 1
return {
"headers": ["Турнир", "Дата", "Команды", "Счёт", *stat_headers],
"rows": rows_out,
"page": page, "page_size": 30, "total": total, "pages": pages,
"range_start": start, "range_end": end,
}
def get_khl_site_player_matches(khl_id: int, page: int = 1) -> dict[str, Any]:
page = max(1, int(page))
PLAYER_MATCHES_CACHE_DIR.mkdir(parents=True, exist_ok=True)
cache_file = PLAYER_MATCHES_CACHE_DIR / f"{khl_id}_{page}.json"
now = time.time()
# Матчи текущего сезона могут обновляться; суток достаточно, а ручная
# перезагрузка страницы всегда сможет получить свежую страницу после удаления кэша.
try:
cached = read_json(cache_file, {})
fetched_at = float(cached.get("fetched_at") or 0) if isinstance(cached, dict) else 0
payload = cached.get("payload") if isinstance(cached, dict) else None
if fetched_at and now - fetched_at < 24 * 3600 and isinstance(payload, dict):
return payload
except Exception:
pass
html = _download_khl_player_page(khl_id, page)
payload = parse_khl_player_matches_html(html, page)
tmp = cache_file.with_suffix(".tmp")
try:
tmp.write_text(json.dumps({"fetched_at": now, "payload": payload}, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
tmp.replace(cache_file)
except OSError as exc:
print(f"Не удалось записать кэш матчей KHL {khl_id}/{page}: {exc}", file=sys.stderr)
return payload
def enrich_player_item_from_khl_site(item: dict[str, Any]) -> dict[str, Any]:
profile = item.get("_profile_full")
if not isinstance(profile, dict):
profile = dict(item.get("profile") or {})
item["_profile_full"] = profile
normalize_player_profile_fields(profile)
identity = item.get("identity") or {}
khl_id_raw = identity.get("khl_id") or profile.get("khl_id")
try:
khl_id = int(khl_id_raw)
except (TypeError, ValueError):
khl_id = 0
required = ("birthday", "country", "height", "weight", "stick")
missing = [key for key in required if profile.get(key) in (None, "", [], {})]
if khl_id > 0 and missing:
site_fields = get_khl_site_player_profile(khl_id)
for key, value in site_fields.items():
if profile.get(key) in (None, "", [], {}) and value not in (None, "", [], {}):
profile[key] = value
normalize_player_profile_fields(profile)
item["_site_profile_fields"] = sorted(site_fields)
return item
def build_full_player_profile(player: dict[str, Any]) -> tuple[dict[str, Any], str | None, bool]:
"""Собирает подробный профиль из общего объекта и самого свежего информативного этапа."""
profile = json.loads(json.dumps(player.get("profile") or {}, ensure_ascii=False))
stages = player.get("stages") or {}
latest_id: str | None = None
latest_data: dict[str, Any] = {}
for stage_id in sorted(stages, key=_numeric_stage_sort, reverse=True):
data = (stages.get(stage_id) or {}).get("data") or {}
if not isinstance(data, dict):
continue
if latest_id is None and data:
latest_id = str(stage_id)
latest_data = data
for key in (
"age", "role", "role_key", "country", "nationality", "citizenship",
"flag_image_url", "birthday", "birth_date", "date_of_birth", "height",
"weight", "stick", "shoots", "shirt_number", "team", "teams",
"seasons_count",
):
if profile.get(key) in (None, "", [], {}) and data.get(key) not in (None, "", [], {}):
profile[key] = json.loads(json.dumps(data[key], ensure_ascii=False))
identity = player.get("identity") or {}
identity_en = player.get("identity_en") or {}
profile_en = player.get("profile_en") or {}
english_name = (
profile.get("name_en")
or profile_en.get("name")
or identity_en.get("full_name")
or identity_en.get("name")
or None
)
if english_name and not profile.get("name_en"):
profile["name_en"] = english_name
normalize_player_profile_fields(profile)
role_text = str(profile.get("role") or latest_data.get("role") or "").casefold()
role_key = str(profile.get("role_key") or latest_data.get("role_key") or "").casefold()
is_goalie = "врат" in role_text or role_key in {"goalkeeper", "goalie", "gk"}
return profile, latest_id, is_goalie
class DataStore:
def __init__(self) -> None:
self.lock = threading.RLock()
self.players_data: dict[str, Any] = {}
self.coaches_data: dict[str, Any] = {}
self.officials_data: dict[str, Any] = {}
self.players: dict[str, dict[str, Any]] = {}
self.coaches: dict[str, dict[str, Any]] = {}
self.officials: dict[str, dict[str, Any]] = {}
self.player_stages: dict[str, dict[str, Any]] = {}
self.reload()
def reload(self) -> None:
with self.lock:
self.players_data = read_json(PLAYERS_FILE, {"players": [], "stages": [], "meta": {}})
self.coaches_data = read_json(COACHES_FILE, {"people": [], "meta": {}})
self.officials_data = read_json(OFFICIALS_FILE, {"people": [], "meta": {}})
self.player_stages = {}
for raw_stage in self.players_data.get("stages", []):
if raw_stage.get("id") is None:
continue
stage = json.loads(json.dumps(raw_stage, ensure_ascii=False))
if not stage.get("season"):
resolved_season = stage_season(stage)
if resolved_season:
stage["season"] = resolved_season
self.player_stages[str(stage.get("id"))] = stage
self.players = {}
for idx, p in enumerate(self.players_data.get("players", [])):
identity = p.get("identity") or {}
profile = p.get("profile") or {}
pid = str(identity.get("id") or profile.get("id") or f"index-{idx}")
self.players[pid] = p
self.coaches = {str(x.get("id")): x for x in self.coaches_data.get("people", []) if x.get("id") is not None}
self.officials = {str(x.get("id")): x for x in self.officials_data.get("people", []) if x.get("id") is not None}
def broken_player_name_count(self) -> int:
with self.lock:
return sum(1 for item_id, item in self.players.items() if player_sidebar_name(item, item_id).startswith("Игрок #"))
def counts(self) -> dict[str, int]:
with self.lock:
return {"players": len(self.players), "coaches": len(self.coaches), "officials": len(self.officials)}
def filters(self) -> dict[str, Any]:
with self.lock:
player_seasons = sorted({str(s.get("season")) for s in self.player_stages.values() if s.get("season")})
player_roles = set()
for p in self.players.values():
for r in (p.get("stages") or {}).values():
role = (r.get("data") or {}).get("role")
if role:
player_roles.add(str(role))
result: dict[str, Any] = {"players": {"seasons": player_seasons, "roles": sorted(player_roles)}}
for kind, mapping in (("coaches", self.coaches), ("officials", self.officials)):
seasons, roles, teams = set(), set(), set()
for item in mapping.values():
profile = item.get("profile") or {}
seasons.update(profile.get("seasons") or [])
for app in item.get("appearances") or []:
context = app.get("context") or {}
if context.get("season"):
seasons.add(str(context["season"]))
if context.get("role"):
roles.add(str(context["role"]))
for team in app.get("team_links") or []:
if team.get("title"):
teams.add(str(team["title"]))
result[kind] = {"seasons": sorted(seasons), "roles": sorted(roles), "teams": sorted(teams)}
return result
def mapping(self, kind: str) -> dict[str, dict[str, Any]]:
return {"players": self.players, "coaches": self.coaches, "officials": self.officials}[kind]
def search(self, kind: str, q: str, season: str, extra: str, page: int, limit: int) -> dict[str, Any]:
with self.lock:
mapping = self.mapping(kind)
q_cf = q.strip().casefold()
items = []
for item_id, item in mapping.items():
if kind == "players":
identity, profile = item.get("identity") or {}, item.get("profile") or {}
name = player_sidebar_name(item, item_id)
identity_en = item.get("identity_en") or {}
profile_en = item.get("profile_en") or {}
name_en = str(profile.get("name_en") or profile_en.get("name") or identity_en.get("full_name") or "")
khl_id = identity.get("khl_id") or profile.get("khl_id")
stages = item.get("stages") or {}
roles = {str((x.get("data") or {}).get("role")) for x in stages.values() if (x.get("data") or {}).get("role")}
profile_role = str(profile.get("role") or "").strip()
if profile_role.casefold() in {"вратарь", "защитник", "нападающий"}:
roles.add(profile_role)
seasons = {str(self.player_stages.get(str(sid), {}).get("season")) for sid in stages if self.player_stages.get(str(sid), {}).get("season")}
if season and season not in seasons:
continue
if extra and extra not in roles:
continue
image = profile.get("image") or profile.get("photo")
role = sorted(roles)[0] if roles else ""
subtitle = f"KHL {khl_id}" if khl_id else ""
count = f"{len(stages)} этапов"
searchable = f"{name} {name_en} {item_id} {khl_id or ''}".casefold()
else:
profile = item.get("profile") or {}
name = staff_display_name(item, "ru")
name_en = staff_display_name(item, "en")
original_name = str(item.get("name_ru") or profile.get("name") or item.get("name") or "")
appearances = item.get("appearances") or []
seasons = set(profile.get("seasons") or [])
roles, teams = set(), set()
for app in appearances:
ctx = app.get("context") or {}
if ctx.get("season"):
seasons.add(str(ctx["season"]))
if ctx.get("role"):
roles.add(str(ctx["role"]))
for team in app.get("team_links") or []:
if team.get("title"):
teams.add(str(team["title"]))
if season and season not in seasons:
continue
if extra and extra not in (teams if kind == "coaches" else roles):
continue
photos = profile.get("photos") or item.get("photos") or []
image = valid_person_image(photos[0]) if photos else None
role = ", ".join(sorted(roles))
subtitle = next(iter(sorted(teams)), "") if kind == "coaches" else ""
count = f"{len(appearances)} записей"
searchable = f"{name} {name_en} {original_name} {item_id} {' '.join(seasons)} {' '.join(teams)}".casefold()
if q_cf and q_cf not in searchable:
continue
items.append({"id": item_id, "name": name, "name_en": name_en, "image": image, "role": role, "subtitle": subtitle, "count": count})
items.sort(key=lambda x: x["name"].casefold())
total = len(items)
pages = max(1, math.ceil(total / limit))
page = min(max(1, page), pages)
start = (page - 1) * limit
return {"items": items[start:start + limit], "total": total, "page": page, "pages": pages}
def item(self, kind: str, item_id: str) -> dict[str, Any] | None:
with self.lock:
source = self.mapping(kind).get(item_id)
if source is None:
return None
item = json.loads(json.dumps(source, ensure_ascii=False))
if kind == "players":
for sid, record in (item.get("stages") or {}).items():
stage_meta = self.player_stages.get(str(sid), {"id": sid})
record["stage_label"] = stage_label(stage_meta)
record["stage_meta"] = json.loads(json.dumps(stage_meta, ensure_ascii=False))
full_profile, latest_stage_id, is_goalie = build_full_player_profile(item)
item["_profile_full"] = full_profile
item["_latest_stage_id"] = latest_stage_id
item["_is_goalie"] = is_goalie
item["_name_en"] = str(
full_profile.get("name_en")
or (item.get("identity_en") or {}).get("full_name")
or ""
)
computed_aggregates = {mode: aggregate_player(item, self.player_stages, mode) for mode in ("all", "regular", "playoff", "hope")}
exact_aggregates = item.get("site_aggregates") or {}
item["aggregates"] = {
mode: (json.loads(json.dumps(exact_aggregates[mode], ensure_ascii=False)) if (exact_aggregates.get(mode) or {}).get("stats") else computed_aggregates[mode])
for mode in ("all", "regular", "playoff", "hope")
}
else:
item["_display_name"] = staff_display_name(item, "ru")
item["_display_name_en"] = staff_display_name(item, "en")
return item
STORE = DataStore()
class JobManager:
def __init__(self) -> None:
self.lock = threading.Lock()
self.running = False
self.title = ""
self.message = ""
self.return_code: int | None = None
self.started_at: str | None = None
self.finished_at: str | None = None
def state(self) -> dict[str, Any]:
with self.lock:
return {"running": self.running, "title": self.title, "message": self.message, "return_code": self.return_code, "started_at": self.started_at, "finished_at": self.finished_at}
def start(self, title: str, commands: list[list[str]]) -> None:
with self.lock:
if self.running:
raise RuntimeError("Уже выполняется другое обновление")
self.running = True; self.title = title; self.message = "Запуск…"; self.return_code = None; self.started_at = datetime.now().isoformat(timespec="seconds"); self.finished_at = None
threading.Thread(target=self._run, args=(commands,), daemon=True).start()
def _run(self, commands: list[list[str]]) -> None:
code = 0
try:
for command in commands:
with self.lock:
self.message = " ".join(Path(x).name if i in (0, 1) else x for i, x in enumerate(command))
child_env = os.environ.copy()
child_env["PYTHONIOENCODING"] = "utf-8"
child_env["PYTHONUTF8"] = "1"
process = subprocess.Popen(command, cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", bufsize=1, env=child_env)
assert process.stdout is not None
for line in process.stdout:
text = line.strip()
if text:
with self.lock:
self.message = text[-500:]
code = process.wait()
if code != 0:
break
except Exception as exc:
code = 1
with self.lock:
self.message = str(exc)
finally:
STORE.reload()
with self.lock:
self.running = False; self.return_code = code; self.finished_at = datetime.now().isoformat(timespec="seconds")
self.message = "Готово" if code == 0 else f"Завершено с ошибкой {code}: {self.message}"
JOBS = JobManager()
@APP.get("/", response_class=HTMLResponse)
def root() -> str:
try:
return INDEX_FILE.read_text(encoding="utf-8")
except OSError:
return HTML
@APP.get("/api/status")
def api_status() -> dict[str, Any]:
return {
"counts": STORE.counts(),
"broken_player_names": STORE.broken_player_name_count(),
"filters": STORE.filters(),
"files": {"players": str(PLAYERS_FILE), "coaches": str(COACHES_FILE), "officials": str(OFFICIALS_FILE)},
}
@APP.get("/api/list/{kind}")
def api_list(kind: str, q: str = "", season: str = "", extra: str = "", page: int = 1, limit: int = Query(100, ge=10, le=250)) -> dict[str, Any]:
if kind not in {"players", "coaches", "officials"}:
raise HTTPException(404, "Неизвестный раздел")
return STORE.search(kind, q, season, extra, page, limit)
@APP.get("/api/item/{kind}/{item_id}")
def api_item(kind: str, item_id: str) -> dict[str, Any]:
if kind not in {"players", "coaches", "officials"}:
raise HTTPException(404, "Неизвестный раздел")
item = STORE.item(kind, item_id)
if item is None:
raise HTTPException(404, "Запись не найдена")
# Важно: карточка должна открываться только из локального JSON.
# Внешний KHL.ru больше никогда не блокирует переключение игрока.
return item
@APP.get("/api/player-extra/{item_id}")
def api_player_extra(item_id: str) -> dict[str, Any]:
source = STORE.players.get(str(item_id))
if source is None:
raise HTTPException(404, "Игрок не найден")
identity = source.get("identity") or {}
profile = source.get("profile") or {}
khl_id_raw = identity.get("khl_id") or profile.get("khl_id")
try:
khl_id = int(khl_id_raw)
except (TypeError, ValueError):
raise HTTPException(404, "У игрока нет KHL ID")
try:
fields = get_khl_site_player_profile(khl_id)
except Exception as exc:
raise HTTPException(502, f"Не удалось получить профиль KHL.ru: {exc}") from exc
return {"item_id": str(item_id), "khl_id": khl_id, "fields": fields}
@APP.get("/api/player-matches/{item_id}")
def api_player_matches(item_id: str, page: int = Query(1, ge=1, le=1000)) -> dict[str, Any]:
source = STORE.players.get(str(item_id))
if source is None:
raise HTTPException(404, "Игрок не найден")
identity = source.get("identity") or {}
profile = source.get("profile") or {}
khl_id_raw = identity.get("khl_id") or profile.get("khl_id")
try:
khl_id = int(khl_id_raw)
except (TypeError, ValueError):
raise HTTPException(404, "У игрока нет KHL ID")
try:
payload = get_khl_site_player_matches(khl_id, page)
except Exception as exc:
raise HTTPException(502, f"Не удалось получить матчи KHL.ru: {exc}") from exc
payload = dict(payload)
payload["item_id"] = str(item_id)
payload["khl_id"] = khl_id
return payload
@APP.post("/api/reload")
def api_reload() -> dict[str, Any]:
STORE.reload()
return {"ok": True, "counts": STORE.counts()}
@APP.get("/api/job")
def api_job() -> dict[str, Any]:
return JOBS.state()
@APP.post("/api/collect/{kind}")
def api_collect(kind: str) -> dict[str, Any]:
python = sys.executable
collector = str(ROOT / "khl_site_collector.py")
base_cmd = [python, collector, "--output-dir", str(DATA_DIR), "--cache-dir", str(ROOT / "cache" / "khl_site_html"), "--workers", "10"]
if kind == "players":
title, commands = "KHL.ru · обновление игроков", [base_cmd + ["--kind", "players"]]
elif kind == "coaches":
title, commands = "KHL.ru · обновление тренеров", [base_cmd + ["--kind", "coaches"]]
elif kind in {"officials-current", "staff-current"}:
title, commands = "KHL.ru · текущие судьи", [base_cmd + ["--kind", "officials", "--current-only"]]
elif kind in {"officials-history", "staff-history"}:
title, commands = "KHL.ru · все судьи", [base_cmd + ["--kind", "officials"]]
elif kind == "all":
title, commands = "KHL.ru · полное обновление", [base_cmd + ["--kind", "all"]]
else:
raise HTTPException(404, "Неизвестный тип обновления")
try:
JOBS.start(title, commands)
except RuntimeError as exc:
raise HTTPException(409, str(exc)) from exc
return {"ok": True, "job": JOBS.state()}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Единый центр данных KHL")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8765)
parser.add_argument("--no-browser", action="store_true")
return parser.parse_args()
def main() -> int:
args = parse_args()
if not args.no_browser:
threading.Timer(1.0, lambda: webbrowser.open(f"http://{args.host}:{args.port}/")).start()
uvicorn.run(APP, host=args.host, port=args.port, log_level="info")
return 0
if __name__ == "__main__":
raise SystemExit(main())