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 credentials_source(self) -> str: """Return where Stat2TV credentials come from without exposing secrets.""" if os.getenv("STAT2TV_LOGIN") is not None or os.getenv("STAT2TV_PASSWORD") is not None: return "env" raw = self._read_json(self.secret_file) if str(raw.get("username") or "").strip() or str(raw.get("password") or ""): return "local_file" return "missing" def status(self) -> dict[str, Any]: settings = self.public() username, password = self.credentials() return { **settings, "credentials_configured": bool(username and password), "credentials_source": self.credentials_source(), "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()