from __future__ import annotations import base64 import io import json import re import zipfile from copy import deepcopy from pathlib import Path from typing import Any, Callable, TYPE_CHECKING from fastapi import APIRouter, Body, Depends, HTTPException, Query from fastapi.responses import JSONResponse from broadcast_settings import settings as settings_lib from broadcast_settings.joins import flatten_dict from broadcast_settings.transform import apply_configured_json if TYPE_CHECKING: from .service import HockeyDataService SUPPORTED_SOURCE_TYPES = { "info", "scoreboard", "home_lineup", "away_lineup", "officials", "penalties", "events", "full_match", } SOURCE_LABELS = { "info": "Информация о матче", "scoreboard": "Верхний счёт", "home_lineup": "Состав хозяев", "away_lineup": "Состав гостей", "officials": "Судьи", "penalties": "Активные удаления", "events": "События матча", "full_match": "Полный матч", } def _read_json(path: Path, fallback: dict[str, Any]) -> dict[str, Any]: try: data = json.loads(path.read_text(encoding="utf-8")) return data if isinstance(data, dict) else deepcopy(fallback) except Exception: return deepcopy(fallback) def configure_hockey_vmix_settings(base_dir: Path) -> None: default_settings = { "version": 1, "active_vmix_json": "scoreboard", "selection": {}, } default_vmix_json = _read_json( base_dir / "hockey_data" / "default_vmix_json.json", {"version": 1, "configs": []}, ) default_vmix_functions = _read_json( base_dir / "hockey_data" / "default_vmix_functions.json", {"version": 1, "categories": []}, ) settings_lib.configure( project_dir=base_dir, app_dir=base_dir, settings_dir=base_dir / "settings", settings_file="vmix_settings.json", vmix_json_file="vmix_json.json", vmix_functions_file="vmix_functions.json", default_settings=default_settings, default_vmix_json=default_vmix_json, default_vmix_functions=default_vmix_functions, ) settings_lib.ensure_settings_files() # Existing projects may have an older config file. Normalize it on startup. settings_lib.write_vmix_json(normalize_hockey_vmix_json(settings_lib.read_vmix_json())) def _safe_key(value: Any, fallback: str = "json") -> str: text = re.sub(r"[^A-Za-z0-9_]+", "_", str(value or "").strip()).strip("_") return text or fallback def endpoint_for_key(key: str) -> str: return f"/vmix/hockey/channel/{{channel}}/{{language}}/custom/{_safe_key(key)}.json" def normalize_hockey_vmix_json(data: Any) -> dict[str, Any]: source = data if isinstance(data, dict) else {} configs = source.get("configs") if isinstance(source.get("configs"), list) else [] normalized: list[dict[str, Any]] = [] used: set[str] = set() for index, raw in enumerate(configs, start=1): if not isinstance(raw, dict): continue cfg = deepcopy(raw) base_key = _safe_key(cfg.get("key"), f"json_{index}") key = base_key suffix = 2 while key in used: key = f"{base_key}_{suffix}" suffix += 1 used.add(key) source_type = str(cfg.get("source_type") or "scoreboard").strip().lower() if source_type not in SUPPORTED_SOURCE_TYPES: cfg.setdefault("imported_source_type", source_type) source_type = "scoreboard" cfg.update( { "key": key, "title": str(cfg.get("title") or key), "description": str(cfg.get("description") or ""), "endpoint": endpoint_for_key(key), "source_type": source_type, "root_key": str(cfg.get("root_key") or "rows"), "output_mode": "all" if str(cfg.get("output_mode") or "columns").lower() == "all" else "columns", "default_limit": max(0, int(cfg.get("default_limit") or 0)), "enabled": cfg.get("enabled", True) is not False, "columns": cfg.get("columns") if isinstance(cfg.get("columns"), list) else [], "joins": cfg.get("joins") if isinstance(cfg.get("joins"), list) else [], } ) normalized.append(cfg) return {"version": int(source.get("version") or 1), "configs": normalized} class HockeyVmixPortable: def __init__(self, service: "HockeyDataService") -> None: self.service = service @staticmethod def _data_language(language: str) -> str: return "en" if language == "en" else "ru" def session_or_error(self, token: str) -> dict[str, Any]: session = self.service.get_session(str(token or "").strip()) if session is None: raise KeyError("Сессия не найдена") return session def _context(self, token: str, language: str) -> dict[str, Any]: session = self.session_or_error(token) data_language = self._data_language(language) game_id = str(session.get("game_external_id") or "") tournament_id = str(session.get("tournament_external_id") or "") details = self.service.game_details(game_id, language=data_language) if game_id else None tournament = self.service.tournament(tournament_id, language=data_language) if tournament_id else None control = self.service.game_control(game_id, language=data_language) if game_id else None return { "session": session, "language": language, "data_language": data_language, "details": details or {}, "tournament": tournament or {}, "control": control or {}, "game": (details or {}).get("game") or {}, } @staticmethod def _timer_text(control: dict[str, Any]) -> tuple[str, dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any]]: shootout = control.get("shootout") or {} timer_rules = control.get("timer_rules") or {} strength = control.get("strength") or {} timer_state = control.get("timers") or {} main_timer = timer_state.get("main_timer") or {} current_ms = max( 0, int(float(main_timer.get("current_ms", timer_rules.get("duration_ms", 20 * 60 * 1000)) or 0)), ) total_seconds = current_ms // 1000 return ( f"{total_seconds // 60:02d}:{total_seconds % 60:02d}", shootout, timer_rules, strength, timer_state, ) def info_payload(self, token: str, language: str, context: dict[str, Any] | None = None) -> dict[str, Any]: ctx = context or self._context(token, language) game = ctx["game"] return { "session_token": token, # legacy field "channel_key": token, "language": language, "tournament": ctx["tournament"], "game": game, "game_id": ctx["session"].get("game_external_id", ""), "home_team": (game.get("home") or {}).get("name", ""), "away_team": (game.get("away") or {}).get("name", ""), "arena": game.get("arena", ""), "date": game.get("date", ""), "start_time": game.get("time", ""), "referees": ctx["details"].get("referees", []), "control": ctx["control"], } def scoreboard_payload(self, token: str, language: str, context: dict[str, Any] | None = None) -> dict[str, Any]: ctx = context or self._context(token, language) game = ctx["game"] control = ctx["control"] timer_text, shootout, timer_rules, strength, timer_state = self._timer_text(control) main_timer = timer_state.get("main_timer") or {} flags = control.get("flags") or {} return { "session_token": token, # legacy field "channel_key": token, "language": language, "game_id": ctx["session"].get("game_external_id", ""), "home_score": (game.get("home") or {}).get("score", 0), "away_score": (game.get("away") or {}).get("score", 0), "period": control.get("current_period", "1"), "time": timer_text, "timer_running": bool(main_timer.get("running", False)), "timer_paused": bool(main_timer.get("paused", True)), "timer_default_minutes": timer_rules.get("minutes", 20), "timer_phase": timer_rules.get("phase", "regulation"), "strength": strength.get("strength_label", ""), "strength_phase": strength.get("phase", "regulation"), "strength_advantage_side": strength.get("advantage_side", ""), "strength_state_key": strength.get("state_key", ""), "strength_state_label": strength.get("state_label", ""), "home_strength": strength.get("home_skaters", 0), "away_strength": strength.get("away_skaters", 0), "home_powerplay_label": strength.get("home_label", ""), "away_powerplay_label": strength.get("away_label", ""), "home_delayed_penalty": bool(flags.get("home_delayed_penalty", False)), "away_delayed_penalty": bool(flags.get("away_delayed_penalty", False)), "home_empty_net": bool(flags.get("home_empty_net", False)), "away_empty_net": bool(flags.get("away_empty_net", False)), "status": game.get("status", "not_loaded"), "shootout_home": (shootout.get("home") or {}).get("goals", 0), "shootout_away": (shootout.get("away") or {}).get("goals", 0), "shootout_attempts": shootout.get("attempts", []), } def line_up_payload(self, token: str, language: str, side: str, context: dict[str, Any] | None = None) -> dict[str, Any]: ctx = context or self._context(token, language) items = ((ctx["game"].get(side) or {}).get("players") or []) return {"session_token": token, "channel_key": token, "language": language, "items": items} def officials_payload(self, token: str, language: str, context: dict[str, Any] | None = None) -> dict[str, Any]: ctx = context or self._context(token, language) return {"session_token": token, "channel_key": token, "language": language, "items": ctx["details"].get("referees", [])} def penalties_payload(self, token: str, language: str, context: dict[str, Any] | None = None) -> dict[str, Any]: ctx = context or self._context(token, language) timer_state = (ctx["control"].get("timers") or {}) board = timer_state.get("penalty_board") or {} raw_items = board.get("penalties") items = [item for item in raw_items if isinstance(item, dict)] if isinstance(raw_items, list) else [] def _remaining(item: dict[str, Any]) -> float: try: return float(item.get("remainingMs", item.get("remaining_ms", 0)) or 0) except (TypeError, ValueError): return 0.0 home_items = sorted( [item for item in items if str(item.get("side") or (item.get("player") or {}).get("side") or "").lower() == "home"], key=lambda item: (_remaining(item), float(item.get("createdAt", 0) or 0)), ) away_items = sorted( [item for item in items if str(item.get("side") or (item.get("player") or {}).get("side") or "").lower() == "away"], key=lambda item: (_remaining(item), float(item.get("createdAt", 0) or 0)), ) return { "session_token": token, # legacy field "channel_key": token, "language": language, "game_id": ctx["session"].get("game_external_id", ""), "home": home_items, "away": away_items, "home_next": home_items[0] if home_items else {}, "away_next": away_items[0] if away_items else {}, "items": items, "strength": ctx["control"].get("strength") or {}, "timer_rules": ctx["control"].get("timer_rules") or {}, "updated_at": timer_state.get("updated_at", ""), } def events_payload(self, token: str, language: str, context: dict[str, Any] | None = None) -> dict[str, Any]: ctx = context or self._context(token, language) details = ctx["details"] items = details.get("events") if not isinstance(items, list): items = details.get("play_by_play") if not isinstance(items, list): items = [] return {"session_token": token, "channel_key": token, "language": language, "items": items} def full_match_payload(self, token: str, language: str, context: dict[str, Any] | None = None) -> dict[str, Any]: ctx = context or self._context(token, language) return { "session": ctx["session"], "language": language, "tournament": ctx["tournament"], "details": ctx["details"], "control": ctx["control"], "scoreboard": self.scoreboard_payload(token, language, context=ctx), } def source_payload(self, token: str, language: str, source_type: str, context: dict[str, Any] | None = None) -> Any: source_type = str(source_type or "scoreboard").strip().lower() if source_type not in SUPPORTED_SOURCE_TYPES: raise ValueError(f"Неизвестный хоккейный источник: {source_type}") ctx = context or self._context(token, language) if source_type == "info": return self.info_payload(token, language, context=ctx) if source_type == "scoreboard": return self.scoreboard_payload(token, language, context=ctx) if source_type == "home_lineup": return self.line_up_payload(token, language, "home", context=ctx) if source_type == "away_lineup": return self.line_up_payload(token, language, "away", context=ctx) if source_type == "officials": return self.officials_payload(token, language, context=ctx) if source_type == "penalties": return self.penalties_payload(token, language, context=ctx) if source_type == "events": return self.events_payload(token, language, context=ctx) return self.full_match_payload(token, language, context=ctx) def source_rows(self, token: str, language: str, source_type: str, context: dict[str, Any] | None = None) -> list[dict[str, Any]]: payload = self.source_payload(token, language, source_type, context=context) if isinstance(payload, list): return [item for item in payload if isinstance(item, dict)] if not isinstance(payload, dict): return [{"value": payload}] if source_type in {"home_lineup", "away_lineup", "officials", "penalties", "events"}: items = payload.get("items") if isinstance(items, list): return [item for item in items if isinstance(item, dict)] return [payload] def source_fields(self, token: str, language: str, source_type: str) -> dict[str, Any]: rows = self.source_rows(token, language, source_type) keys: list[str] = [] for row in rows[:30]: for key, value in flatten_dict(row).items(): if isinstance(value, (dict, list)): continue key = str(key) if key not in keys: keys.append(key) return {"ok": True, "source_type": source_type, "count": len(rows), "fields": keys} def configured_output(self, token: str, language: str, json_key: str, limit: int | None = None) -> list[dict[str, Any]]: config = settings_lib.get_vmix_config(json_key) context = self._context(token, language) cache: dict[str, list[dict[str, Any]]] = {} def rows_for(source_type: str) -> list[dict[str, Any]]: source_type = str(source_type or "scoreboard").strip().lower() if source_type not in cache: cache[source_type] = self.source_rows(token, language, source_type, context=context) return deepcopy(cache[source_type]) def source_resolver(join: dict[str, Any]) -> Any: source_type = str( join.get("source_type") or join.get("source") or join.get("type") or "scoreboard" ).strip().lower() return rows_for(source_type) rows = rows_for(str(config.get("source_type") or "scoreboard")) return apply_configured_json(rows, config, limit=limit, source_resolver=source_resolver) def _extract_import_json_from_zip(file_name: str, content: bytes) -> tuple[dict[str, Any], list[str]]: try: archive = zipfile.ZipFile(io.BytesIO(content)) except zipfile.BadZipFile as exc: raise ValueError("ZIP-файл не открывается или повреждён") from exc names = [name for name in archive.namelist() if not name.endswith("/")] json_names = [name for name in names if name.lower().endswith(".json")] if not json_names: raise ValueError("В ZIP не найдено JSON-файлов с настройками") def load_member(name: str) -> Any: with archive.open(name) as file: return json.loads(file.read().decode("utf-8-sig")) preferred_vmix = next( (name for name in json_names if name.lower().replace("\\", "/").endswith("settings/vmix_json.json")), None, ) or next((name for name in json_names if name.lower().endswith("vmix_json.json")), None) preferred_settings = next( (name for name in json_names if name.lower().replace("\\", "/").endswith("settings/settings.json")), None, ) or next((name for name in json_names if name.lower().endswith("settings.json")), None) result: dict[str, Any] = {"_import_archive": file_name} used: list[str] = [] if preferred_settings: result["settings"] = load_member(preferred_settings) used.append(preferred_settings) if preferred_vmix: result["vmix_json"] = load_member(preferred_vmix) if isinstance(result["vmix_json"], dict): result["configs"] = result["vmix_json"].get("configs") used.append(preferred_vmix) if used: result["_import_used_files"] = used return result, used files: dict[str, Any] = {} for name in json_names[:50]: lowered = name.lower() if any(part in lowered for part in ("cache/", "__pycache__", "response_", "scores")): continue try: files[name] = load_member(name) used.append(name) except Exception: continue if not files: raise ValueError("JSON-файлы найдены, но настройки прочитать не удалось") return {"_import_archive": file_name, "files": files, "_import_used_files": used}, used def create_hockey_vmix_settings_router( portable: HockeyVmixPortable, *, admin_dependency: Callable[..., Any], ) -> APIRouter: router = APIRouter( prefix="/api/hockey/vmix", tags=["Hockey vMix settings"], dependencies=[Depends(admin_dependency)], ) @router.get("/settings") async def get_settings() -> dict[str, Any]: return { "settings": settings_lib.read_settings(), "vmix_json": normalize_hockey_vmix_json(settings_lib.read_vmix_json()), "vmix_functions": settings_lib.read_vmix_functions(), "source_types": [ {"value": key, "label": SOURCE_LABELS[key]} for key in SOURCE_LABELS ], "meta": settings_lib.settings_meta(), } @router.put("/configs") async def save_configs(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: raw = payload.get("vmix_json", payload) if not isinstance(raw, dict): raise HTTPException(status_code=400, detail="vmix_json должен быть объектом") normalized = normalize_hockey_vmix_json(raw) return {"vmix_json": settings_lib.write_vmix_json(normalized), "meta": settings_lib.settings_meta()} @router.get("/source-fields") async def source_fields( token: str = Query(..., min_length=4), language: str = Query("ru", pattern="^(ru|en|bilingual)$"), source_type: str = Query("scoreboard"), ) -> dict[str, Any]: try: return portable.source_fields(token, language, source_type) except (KeyError, ValueError) as exc: raise HTTPException(status_code=404 if isinstance(exc, KeyError) else 400, detail=str(exc)) from exc @router.get("/preview/{json_key}") async def preview( json_key: str, token: str = Query(..., min_length=4), language: str = Query("ru", pattern="^(ru|en|bilingual)$"), limit: int | None = Query(None, ge=0, le=1000), ) -> JSONResponse: try: return JSONResponse(portable.configured_output(token, language, json_key, limit=limit)) except (KeyError, ValueError) as exc: raise HTTPException(status_code=404 if isinstance(exc, KeyError) else 400, detail=str(exc)) from exc @router.get("/export") async def export_settings() -> dict[str, Any]: return { "format": "hockey-vmix-settings", "version": 1, "vmix_json": normalize_hockey_vmix_json(settings_lib.read_vmix_json()), } @router.post("/import/preview-file") async def import_preview_file(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: file_name = str(payload.get("file_name") or payload.get("name") or "settings.json") content_b64 = payload.get("content_base64") or payload.get("base64") if not content_b64: raise HTTPException(status_code=400, detail="Нужно передать файл в content_base64") try: content = base64.b64decode(str(content_b64), validate=False) if file_name.lower().endswith(".zip"): data, _ = _extract_import_json_from_zip(file_name, content) else: data = json.loads(content.decode("utf-8-sig")) preview_value = settings_lib.preview_settings_import(data, file_name=file_name) preview_value["raw_import"] = data return preview_value except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=400, detail=f"Не удалось прочитать настройки: {exc}") from exc @router.post("/import/apply") async def import_apply(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: preview_value = payload.get("preview") or payload if not isinstance(preview_value, dict): raise HTTPException(status_code=400, detail="Нужно передать preview") try: result = settings_lib.apply_settings_import( preview=preview_value, selected_indexes=payload.get("selected_indexes") if isinstance(payload.get("selected_indexes"), list) else None, import_general=False, import_vmix=True, mode=str(payload.get("mode") or "add_new"), ) normalized = normalize_hockey_vmix_json(result.get("vmix_json") or settings_lib.read_vmix_json()) saved = settings_lib.write_vmix_json(normalized) return {**result, "vmix_json": saved, "meta": settings_lib.settings_meta()} except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return router