444 lines
18 KiB
Python
444 lines
18 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
from threading import RLock
|
||
from typing import Any
|
||
from urllib.parse import urlparse
|
||
|
||
|
||
|
||
DEFAULT_STRENGTH_STATE_LABELS: dict[str, dict[str, dict[str, str]]] = {
|
||
"regulation": {
|
||
"5x5": {"ru": "", "en": ""},
|
||
"5x4": {"ru": "PP", "en": "PP"},
|
||
"4x4": {"ru": "4 на 4", "en": "4 on 4"},
|
||
"5x3": {"ru": "5 на 3", "en": "5 on 3"},
|
||
"4x3": {"ru": "4 на 3", "en": "4 on 3"},
|
||
"3x3": {"ru": "3 на 3", "en": "3 on 3"},
|
||
},
|
||
"regular_overtime": {
|
||
"3x3": {"ru": "3 на 3", "en": "3 on 3"},
|
||
"4x3": {"ru": "4 на 3", "en": "4 on 3"},
|
||
"4x4": {"ru": "4 на 4", "en": "4 on 4"},
|
||
"5x3": {"ru": "5 на 3", "en": "5 on 3"},
|
||
"5x4": {"ru": "5 на 4", "en": "5 on 4"},
|
||
"5x5": {"ru": "5 на 5", "en": "5 on 5"},
|
||
},
|
||
"playoff_overtime": {
|
||
"5x5": {"ru": "", "en": ""},
|
||
"5x4": {"ru": "PP", "en": "PP"},
|
||
"4x4": {"ru": "4 на 4", "en": "4 on 4"},
|
||
"5x3": {"ru": "5 на 3", "en": "5 on 3"},
|
||
"4x3": {"ru": "4 на 3", "en": "4 on 3"},
|
||
"3x3": {"ru": "3 на 3", "en": "3 on 3"},
|
||
},
|
||
}
|
||
|
||
|
||
DEFAULT_PERIOD_STATUS_LABELS: dict[str, dict[str, dict[str, str]]] = {
|
||
"1": {
|
||
"compact": {"ru": "1", "en": "1"},
|
||
"short": {"ru": "1 ПЕР", "en": "P1"},
|
||
"long": {"ru": "1 период", "en": "1st period"},
|
||
},
|
||
"2": {
|
||
"compact": {"ru": "2", "en": "2"},
|
||
"short": {"ru": "2 ПЕР", "en": "P2"},
|
||
"long": {"ru": "2 период", "en": "2nd period"},
|
||
},
|
||
"3": {
|
||
"compact": {"ru": "3", "en": "3"},
|
||
"short": {"ru": "3 ПЕР", "en": "P3"},
|
||
"long": {"ru": "3 период", "en": "3rd period"},
|
||
},
|
||
"ot": {
|
||
"compact": {"ru": "ОТ", "en": "OT"},
|
||
"short": {"ru": "ОТ", "en": "OT"},
|
||
"long": {"ru": "Овертайм", "en": "Overtime"},
|
||
},
|
||
"ot_numbered": {
|
||
"compact": {"ru": "ОТ{n}", "en": "OT{n}"},
|
||
"short": {"ru": "ОТ {n}", "en": "OT {n}"},
|
||
"long": {"ru": "{n} овертайм", "en": "Overtime {n}"},
|
||
},
|
||
"so": {
|
||
"compact": {"ru": "Б", "en": "SO"},
|
||
"short": {"ru": "БУЛ", "en": "SO"},
|
||
"long": {"ru": "Буллиты", "en": "Shootout"},
|
||
},
|
||
"finished": {
|
||
"compact": {"ru": "КОН", "en": "FIN"},
|
||
"short": {"ru": "ЗАВ", "en": "FINAL"},
|
||
"long": {"ru": "Матч завершён", "en": "Final"},
|
||
},
|
||
}
|
||
|
||
|
||
DEFAULT_SCOREBOARD_TEAM_STATES: dict[str, dict[str, Any]] = {
|
||
"home_delayed_penalty": {
|
||
"ru": "Отложенный штраф",
|
||
"en": "Delayed penalty",
|
||
"input": "",
|
||
"input_title": "",
|
||
"overlay": "2",
|
||
"enabled": True,
|
||
},
|
||
"away_delayed_penalty": {
|
||
"ru": "Отложенный штраф",
|
||
"en": "Delayed penalty",
|
||
"input": "",
|
||
"input_title": "",
|
||
"overlay": "2",
|
||
"enabled": True,
|
||
},
|
||
"home_empty_net": {
|
||
"ru": "Пустые ворота",
|
||
"en": "Empty net",
|
||
"input": "",
|
||
"input_title": "",
|
||
"overlay": "3",
|
||
"enabled": True,
|
||
},
|
||
"away_empty_net": {
|
||
"ru": "Пустые ворота",
|
||
"en": "Empty net",
|
||
"input": "",
|
||
"input_title": "",
|
||
"overlay": "3",
|
||
"enabled": True,
|
||
},
|
||
}
|
||
|
||
|
||
DEFAULT_PUBLIC_SETTINGS: dict[str, Any] = {
|
||
"base_url": "https://stat2tv.khl.ru/",
|
||
"auth_mode": "auto",
|
||
"ui_language": "ru",
|
||
"fallback_language": "ru",
|
||
"store_ru": True,
|
||
"store_en": True,
|
||
"sync_interval_seconds": 300,
|
||
"request_timeout_seconds": 20,
|
||
"verify_ssl": True,
|
||
"timezone": "Europe/Moscow",
|
||
"auto_sync_on_startup": False,
|
||
"games_endpoint_template": "{tournament_id}/schedule-{tournament_id}-live.xml",
|
||
"games_auto_discovery": True,
|
||
"games_cache_seconds": 300,
|
||
"players_endpoint_template": "{tournament_id}/players-{tournament_id}.xml",
|
||
"player_countries_endpoint_template": "players-countries.xml",
|
||
"referees_endpoint_template": "{tournament_id}/referees-{tournament_id}.xml",
|
||
"match_json_endpoint_template": "{tournament_id}/json/{game_id}.json",
|
||
"match_json_en_endpoint_template": "{tournament_id}/json_en/{game_id}.json",
|
||
"shots_endpoint_template": "shots/shots-{game_id}.json",
|
||
"match_directories_cache_seconds": 300,
|
||
"standings_endpoint_template": "{tournament_id}/standings-{tournament_id}.xml",
|
||
"standings_cache_seconds": 300,
|
||
"powerplay_endpoint_template": "{tournament_id}/powerplay-{tournament_id}.xml",
|
||
"rank_endpoint_template": "{tournament_id}/rank-{tournament_id}.xml",
|
||
"navigation_endpoint_template": "{tournament_id}/navigation-{tournament_id}.xml",
|
||
"tournament_statistics_cache_seconds": 300,
|
||
"optional_resource_negative_cache_seconds": 86400,
|
||
|
||
# Operator timer defaults. Values are global, but regular-season and
|
||
# playoff overtime are configured separately.
|
||
"timer_period_minutes": 20,
|
||
"timer_regular_overtime_minutes": 5,
|
||
"timer_playoff_overtime_minutes": 20,
|
||
"timer_reset_on_period_change": True,
|
||
|
||
# Strength / power-play graphic rules.
|
||
"strength_regulation_skaters": 5,
|
||
"strength_regular_overtime_skaters": 3,
|
||
"strength_playoff_overtime_skaters": 5,
|
||
"strength_min_skaters": 3,
|
||
"strength_regulation_display": "pp",
|
||
"strength_regular_overtime_display": "numbers",
|
||
"strength_playoff_overtime_display": "pp",
|
||
"strength_powerplay_label": "PP",
|
||
"strength_shorthanded_label": "PK",
|
||
"strength_show_shorthanded": False,
|
||
"strength_state_labels": DEFAULT_STRENGTH_STATE_LABELS,
|
||
"period_status_labels": DEFAULT_PERIOD_STATUS_LABELS,
|
||
"scoreboard_team_states": DEFAULT_SCOREBOARD_TEAM_STATES,
|
||
}
|
||
|
||
DEFAULT_SECRET_SETTINGS: dict[str, str] = {
|
||
"username": "",
|
||
"password": "",
|
||
}
|
||
|
||
|
||
class HockeySettingsStore:
|
||
"""Public settings and local Stat2TV credentials.
|
||
|
||
Credentials are kept in a separate local file and are never returned by API.
|
||
Environment variables override local values:
|
||
STAT2TV_BASE_URL
|
||
STAT2TV_LOGIN
|
||
STAT2TV_PASSWORD
|
||
STAT2TV_AUTH_MODE
|
||
STAT2TV_VERIFY_SSL
|
||
"""
|
||
|
||
def __init__(self, settings_dir: Path) -> None:
|
||
self.settings_dir = Path(settings_dir)
|
||
self.settings_dir.mkdir(parents=True, exist_ok=True)
|
||
self.public_file = self.settings_dir / "hockey_api.json"
|
||
self.secret_file = self.settings_dir / "stat2tv_credentials.local.json"
|
||
self._lock = RLock()
|
||
self._ensure_files()
|
||
|
||
def _ensure_files(self) -> None:
|
||
if not self.public_file.exists():
|
||
self.public_file.write_text(
|
||
json.dumps(DEFAULT_PUBLIC_SETTINGS, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
if not self.secret_file.exists():
|
||
self.secret_file.write_text(
|
||
json.dumps(DEFAULT_SECRET_SETTINGS, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
try:
|
||
os.chmod(self.secret_file, 0o600)
|
||
except OSError:
|
||
pass
|
||
|
||
@staticmethod
|
||
def _read_json(path: Path) -> dict[str, Any]:
|
||
try:
|
||
value = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError, TypeError):
|
||
return {}
|
||
return value if isinstance(value, dict) else {}
|
||
|
||
@staticmethod
|
||
def _normalise_url(value: str) -> str:
|
||
value = (value or "").strip()
|
||
if not value:
|
||
value = DEFAULT_PUBLIC_SETTINGS["base_url"]
|
||
parsed = urlparse(value)
|
||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||
raise ValueError("Некорректный адрес Stat2TV")
|
||
return value.rstrip("/") + "/"
|
||
|
||
@staticmethod
|
||
def _language(value: Any, fallback: str) -> str:
|
||
value = str(value or "").lower().strip()
|
||
return value if value in {"ru", "en"} else fallback
|
||
|
||
def public(self) -> dict[str, Any]:
|
||
with self._lock:
|
||
raw = {**DEFAULT_PUBLIC_SETTINGS, **self._read_json(self.public_file)}
|
||
|
||
raw["base_url"] = self._normalise_url(
|
||
os.getenv("STAT2TV_BASE_URL", str(raw["base_url"]))
|
||
)
|
||
raw["auth_mode"] = str(
|
||
os.getenv("STAT2TV_AUTH_MODE", raw["auth_mode"])
|
||
).lower()
|
||
if raw["auth_mode"] not in {"auto", "basic", "digest"}:
|
||
raw["auth_mode"] = "auto"
|
||
|
||
raw["ui_language"] = self._language(raw.get("ui_language"), "ru")
|
||
raw["fallback_language"] = self._language(
|
||
raw.get("fallback_language"), "ru"
|
||
)
|
||
raw["store_ru"] = bool(raw.get("store_ru", True))
|
||
raw["store_en"] = bool(raw.get("store_en", True))
|
||
if not raw["store_ru"] and not raw["store_en"]:
|
||
raw["store_ru"] = True
|
||
raw["store_en"] = True
|
||
|
||
for key, low, high, fallback in (
|
||
("sync_interval_seconds", 30, 86400, 300),
|
||
("request_timeout_seconds", 3, 120, 20),
|
||
("games_cache_seconds", 0, 86400, 300),
|
||
("match_directories_cache_seconds", 0, 86400, 300),
|
||
("standings_cache_seconds", 0, 86400, 300),
|
||
("tournament_statistics_cache_seconds", 0, 86400, 300),
|
||
("optional_resource_negative_cache_seconds", 300, 604800, 86400),
|
||
("timer_period_minutes", 1, 60, 20),
|
||
("timer_regular_overtime_minutes", 1, 60, 5),
|
||
("timer_playoff_overtime_minutes", 1, 60, 20),
|
||
("strength_regulation_skaters", 3, 6, 5),
|
||
("strength_regular_overtime_skaters", 3, 6, 3),
|
||
("strength_playoff_overtime_skaters", 3, 6, 5),
|
||
("strength_min_skaters", 2, 5, 3),
|
||
):
|
||
try:
|
||
value = int(raw.get(key, fallback))
|
||
except (TypeError, ValueError):
|
||
value = fallback
|
||
raw[key] = max(low, min(high, value))
|
||
|
||
env_verify = os.getenv("STAT2TV_VERIFY_SSL")
|
||
if env_verify is not None:
|
||
raw["verify_ssl"] = env_verify.strip().lower() in {
|
||
"1", "true", "yes", "on"
|
||
}
|
||
else:
|
||
raw["verify_ssl"] = bool(raw.get("verify_ssl", True))
|
||
|
||
raw["timezone"] = str(raw.get("timezone") or "Europe/Moscow")
|
||
raw["auto_sync_on_startup"] = bool(raw.get("auto_sync_on_startup", False))
|
||
raw["timer_reset_on_period_change"] = bool(raw.get("timer_reset_on_period_change", True))
|
||
raw["strength_show_shorthanded"] = bool(raw.get("strength_show_shorthanded", False))
|
||
display_modes = {"pp", "numbers", "pp_numbers"}
|
||
for key, fallback in (
|
||
("strength_regulation_display", "pp"),
|
||
("strength_regular_overtime_display", "numbers"),
|
||
("strength_playoff_overtime_display", "pp"),
|
||
):
|
||
value = str(raw.get(key) or fallback).strip().lower()
|
||
raw[key] = value if value in display_modes else fallback
|
||
raw["strength_powerplay_label"] = str(raw.get("strength_powerplay_label") or "PP").strip()[:12] or "PP"
|
||
raw["strength_shorthanded_label"] = str(raw.get("strength_shorthanded_label") or "PK").strip()[:12] or "PK"
|
||
|
||
source_labels = raw.get("strength_state_labels")
|
||
source_labels = source_labels if isinstance(source_labels, dict) else {}
|
||
normalised_labels: dict[str, dict[str, dict[str, str]]] = {}
|
||
for phase, defaults in DEFAULT_STRENGTH_STATE_LABELS.items():
|
||
phase_source = source_labels.get(phase)
|
||
phase_source = phase_source if isinstance(phase_source, dict) else {}
|
||
phase_result: dict[str, dict[str, str]] = {}
|
||
for state_key, default_labels in defaults.items():
|
||
state_source = phase_source.get(state_key)
|
||
state_source = state_source if isinstance(state_source, dict) else {}
|
||
phase_result[state_key] = {
|
||
"ru": str(state_source.get("ru", default_labels["ru"]) or "").strip()[:40],
|
||
"en": str(state_source.get("en", default_labels["en"]) or "").strip()[:40],
|
||
}
|
||
normalised_labels[phase] = phase_result
|
||
raw["strength_state_labels"] = normalised_labels
|
||
|
||
source_period_labels = raw.get("period_status_labels")
|
||
source_period_labels = source_period_labels if isinstance(source_period_labels, dict) else {}
|
||
normalised_period_labels: dict[str, dict[str, dict[str, str]]] = {}
|
||
for period_key, defaults in DEFAULT_PERIOD_STATUS_LABELS.items():
|
||
source_period = source_period_labels.get(period_key)
|
||
source_period = source_period if isinstance(source_period, dict) else {}
|
||
period_result: dict[str, dict[str, str]] = {}
|
||
for format_key in ("compact", "short", "long"):
|
||
default_format = defaults[format_key]
|
||
source_format = source_period.get(format_key)
|
||
source_format = source_format if isinstance(source_format, dict) else {}
|
||
period_result[format_key] = {
|
||
"ru": str(source_format.get("ru", default_format["ru"]) or "").strip()[:64],
|
||
"en": str(source_format.get("en", default_format["en"]) or "").strip()[:64],
|
||
}
|
||
normalised_period_labels[period_key] = period_result
|
||
raw["period_status_labels"] = normalised_period_labels
|
||
|
||
source_team_states = raw.get("scoreboard_team_states")
|
||
source_team_states = source_team_states if isinstance(source_team_states, dict) else {}
|
||
normalised_team_states: dict[str, dict[str, Any]] = {}
|
||
for state_key, defaults in DEFAULT_SCOREBOARD_TEAM_STATES.items():
|
||
source = source_team_states.get(state_key)
|
||
source = source if isinstance(source, dict) else {}
|
||
overlay = str(source.get("overlay") or defaults["overlay"]).strip()
|
||
if overlay not in {"1", "2", "3", "4"}:
|
||
overlay = str(defaults["overlay"])
|
||
normalised_team_states[state_key] = {
|
||
"ru": str(source.get("ru", defaults["ru"]) or "").strip()[:48],
|
||
"en": str(source.get("en", defaults["en"]) or "").strip()[:48],
|
||
"input": str(source.get("input") or "").strip()[:300],
|
||
"input_title": str(source.get("input_title") or "").strip()[:300],
|
||
"overlay": overlay,
|
||
"enabled": bool(source.get("enabled", defaults["enabled"])),
|
||
}
|
||
raw["scoreboard_team_states"] = normalised_team_states
|
||
|
||
raw["games_endpoint_template"] = str(
|
||
raw.get("games_endpoint_template") or ""
|
||
).strip()
|
||
raw["games_auto_discovery"] = bool(
|
||
raw.get("games_auto_discovery", True)
|
||
)
|
||
for key in (
|
||
"players_endpoint_template",
|
||
"player_countries_endpoint_template",
|
||
"referees_endpoint_template",
|
||
"match_json_endpoint_template",
|
||
"match_json_en_endpoint_template",
|
||
"shots_endpoint_template",
|
||
"standings_endpoint_template",
|
||
"powerplay_endpoint_template",
|
||
"rank_endpoint_template",
|
||
"navigation_endpoint_template",
|
||
):
|
||
raw[key] = str(
|
||
raw.get(key) or DEFAULT_PUBLIC_SETTINGS[key]
|
||
).strip()
|
||
return raw
|
||
|
||
def credentials(self) -> tuple[str, str]:
|
||
with self._lock:
|
||
raw = {**DEFAULT_SECRET_SETTINGS, **self._read_json(self.secret_file)}
|
||
username = os.getenv("STAT2TV_LOGIN", str(raw.get("username") or ""))
|
||
password = os.getenv("STAT2TV_PASSWORD", str(raw.get("password") or ""))
|
||
return username.strip(), password
|
||
|
||
def status(self) -> dict[str, Any]:
|
||
settings = self.public()
|
||
username, password = self.credentials()
|
||
return {
|
||
**settings,
|
||
"credentials_configured": bool(username and password),
|
||
"username_masked": self.mask_username(username),
|
||
}
|
||
|
||
@staticmethod
|
||
def mask_username(username: str) -> str:
|
||
if not username:
|
||
return ""
|
||
if len(username) <= 2:
|
||
return "•" * len(username)
|
||
return f"{username[0]}{'•' * max(2, len(username)-2)}{username[-1]}"
|
||
|
||
def update_public(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||
allowed = set(DEFAULT_PUBLIC_SETTINGS)
|
||
with self._lock:
|
||
current = {**DEFAULT_PUBLIC_SETTINGS, **self._read_json(self.public_file)}
|
||
for key in allowed:
|
||
if key in payload:
|
||
current[key] = payload[key]
|
||
current["base_url"] = self._normalise_url(str(current["base_url"]))
|
||
current["ui_language"] = self._language(current["ui_language"], "ru")
|
||
current["fallback_language"] = self._language(
|
||
current["fallback_language"], "ru"
|
||
)
|
||
self.public_file.write_text(
|
||
json.dumps(current, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
return self.status()
|
||
|
||
def update_credentials(
|
||
self,
|
||
*,
|
||
username: str | None,
|
||
password: str | None,
|
||
clear_password: bool = False,
|
||
) -> dict[str, Any]:
|
||
with self._lock:
|
||
current = {**DEFAULT_SECRET_SETTINGS, **self._read_json(self.secret_file)}
|
||
if username is not None:
|
||
current["username"] = username.strip()
|
||
if clear_password:
|
||
current["password"] = ""
|
||
elif password not in (None, ""):
|
||
current["password"] = str(password)
|
||
self.secret_file.write_text(
|
||
json.dumps(current, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
try:
|
||
os.chmod(self.secret_file, 0o600)
|
||
except OSError:
|
||
pass
|
||
return self.status()
|