from __future__ import annotations

import json
import re
from datetime import datetime
from typing import Any

from fastapi import HTTPException
from sqlalchemy import and_, delete, inspect, select, text
from sqlalchemy.exc import SQLAlchemyError

from .auth_bridge import HockeyUser
from .database import HockeyDatabase
from .localization import stage_key
from .service import HockeyDataService
from .models import (
    Game,
    GameControlState,
    Tournament,
    MappingContextValue,
    MappingContextVariable,
    MappingSqlDataSource,
    VmixDevice,
)

_KEY_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{0,127}$")
_SOURCE_CODE_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,99}$")
_PARAM_RE = re.compile(r"(?<!:):([A-Za-z][A-Za-z0-9_]*)")
_LANG_SUFFIX_RE = re.compile(r"^(?P<base>.+)_(?P<lang>RUS|ENG)$", re.IGNORECASE)
_DANGEROUS_SQL_RE = re.compile(
    r"\b(insert|update|delete|drop|alter|create|truncate|grant|revoke|copy|call|do|execute|"
    r"vacuum|analyze|reindex|refresh|lock|set|reset|listen|notify|unlisten|prepare|deallocate|discard)\b",
    re.IGNORECASE,
)
_DANGEROUS_FUNCTION_RE = re.compile(
    r"\b(pg_sleep|pg_read_file|pg_read_binary_file|pg_ls_dir|lo_import|lo_export|dblink|"
    r"pg_terminate_backend|pg_cancel_backend)\s*\(",
    re.IGNORECASE,
)


def _utcnow() -> datetime:
    return datetime.utcnow()


DEFAULT_CONTEXT_VARIABLES: tuple[dict[str, Any], ...] = (
    {"key": "game_id", "label": "Текущий матч", "category": "Система", "entity_type": "match", "scope": "match", "source_type": "system"},
    {"key": "tournament_id", "label": "Текущий турнир", "category": "Система", "entity_type": "tournament", "scope": "match", "source_type": "system"},
    {"key": "team1_id", "label": "Команда 1 / хозяева", "category": "Система", "entity_type": "team", "scope": "match", "source_type": "system"},
    {"key": "team2_id", "label": "Команда 2 / гости", "category": "Система", "entity_type": "team", "scope": "match", "source_type": "system"},
    {"key": "account_id", "label": "Аккаунт оператора", "category": "Система", "entity_type": "account", "scope": "account", "source_type": "system"},
    {"key": "device_id", "label": "Активный Agent", "category": "Система", "entity_type": "device", "scope": "session", "source_type": "system"},
    {"key": "assignment_id", "label": "Текущее назначение Agent", "category": "Система", "entity_type": "assignment", "scope": "session", "source_type": "system"},
    {"key": "active_tab", "label": "Активная вкладка интерфейса", "category": "Система", "entity_type": "tab", "scope": "session", "source_type": "selection"},

    {"key": "selected_team_id", "label": "Выбранная команда", "category": "Выделенные", "entity_type": "team", "scope": "match", "source_type": "selection"},
    {"key": "selected_player_id", "label": "Выбранный игрок", "category": "Выделенные", "entity_type": "player", "scope": "match", "source_type": "selection"},
    {"key": "selected_player_team_id", "label": "Команда выбранного игрока", "category": "Выделенные", "entity_type": "team", "scope": "match", "source_type": "selection"},
    {"key": "player_team1_id", "label": "Выбранный игрок команды 1", "category": "Выделенные", "entity_type": "player", "scope": "match", "source_type": "selection"},
    {"key": "player_team2_id", "label": "Выбранный игрок команды 2", "category": "Выделенные", "entity_type": "player", "scope": "match", "source_type": "selection"},
    {"key": "selected_goalkeeper_id", "label": "Выбранный вратарь", "category": "Выделенные", "entity_type": "player", "scope": "match", "source_type": "selection"},
    {"key": "selected_coach_id", "label": "Выбранный тренер", "category": "Выделенные", "entity_type": "coach", "scope": "match", "source_type": "selection"},
    {"key": "selected_referee_id", "label": "Выбранный судья", "category": "Выделенные", "entity_type": "referee", "scope": "match", "source_type": "selection"},
    {"key": "selected_goal_id", "label": "Выбранный гол", "category": "Выделенные", "entity_type": "goal", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_id", "label": "Выбранное удаление", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_player_id", "label": "Игрок выбранного удаления (external ID)", "category": "Выделенные", "entity_type": "player", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_player_db_id", "label": "Игрок выбранного удаления (DB ID)", "category": "Выделенные", "entity_type": "player", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_team_penalty", "label": "Выбранное удаление командное (1/0)", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection", "value_type": "text"},
    {"key": "selected_penalty_team_id", "label": "Команда выбранного удаления", "category": "Выделенные", "entity_type": "team", "scope": "match", "source_type": "selection"},

    {"key": "selected_home_penalty_id", "label": "Удаление левой команды — ID удаления", "category": "Удаления — левая команда", "entity_type": "penalty", "scope": "match_shared", "source_type": "selection"},
    {"key": "selected_home_penalty_player_id", "label": "Удаление левой команды — игрок external ID", "category": "Удаления — левая команда", "entity_type": "player", "scope": "match_shared", "source_type": "selection"},
    {"key": "selected_home_penalty_player_db_id", "label": "Удаление левой команды — игрок DB ID", "category": "Удаления — левая команда", "entity_type": "player", "scope": "match_shared", "source_type": "selection"},
    {"key": "selected_home_penalty_team_penalty", "label": "Удаление левой команды — командное (1/0)", "category": "Удаления — левая команда", "entity_type": "penalty", "scope": "match_shared", "source_type": "selection", "value_type": "text"},
    {"key": "selected_away_penalty_id", "label": "Удаление правой команды — ID удаления", "category": "Удаления — правая команда", "entity_type": "penalty", "scope": "match_shared", "source_type": "selection"},
    {"key": "selected_away_penalty_player_id", "label": "Удаление правой команды — игрок external ID", "category": "Удаления — правая команда", "entity_type": "player", "scope": "match_shared", "source_type": "selection"},
    {"key": "selected_away_penalty_player_db_id", "label": "Удаление правой команды — игрок DB ID", "category": "Удаления — правая команда", "entity_type": "player", "scope": "match_shared", "source_type": "selection"},
    {"key": "selected_away_penalty_team_penalty", "label": "Удаление правой команды — командное (1/0)", "category": "Удаления — правая команда", "entity_type": "penalty", "scope": "match_shared", "source_type": "selection", "value_type": "text"},

    {"key": "active_home_penalty_id", "label": "Активное удаление левой команды — ID", "category": "Активные удаления — левая команда", "entity_type": "penalty", "scope": "match_shared", "source_type": "selection"},
    {"key": "active_home_penalty_player_id", "label": "Активное удаление левой команды — игрок external ID", "category": "Активные удаления — левая команда", "entity_type": "player", "scope": "match_shared", "source_type": "selection"},
    {"key": "active_home_penalty_player_db_id", "label": "Активное удаление левой команды — игрок DB ID", "category": "Активные удаления — левая команда", "entity_type": "player", "scope": "match_shared", "source_type": "selection"},
    {"key": "active_home_penalty_team_penalty", "label": "Активное удаление левой команды — командное (1/0)", "category": "Активные удаления — левая команда", "entity_type": "penalty", "scope": "match_shared", "source_type": "selection", "value_type": "text"},
    {"key": "active_away_penalty_id", "label": "Активное удаление правой команды — ID", "category": "Активные удаления — правая команда", "entity_type": "penalty", "scope": "match_shared", "source_type": "selection"},
    {"key": "active_away_penalty_player_id", "label": "Активное удаление правой команды — игрок external ID", "category": "Активные удаления — правая команда", "entity_type": "player", "scope": "match_shared", "source_type": "selection"},
    {"key": "active_away_penalty_player_db_id", "label": "Активное удаление правой команды — игрок DB ID", "category": "Активные удаления — правая команда", "entity_type": "player", "scope": "match_shared", "source_type": "selection"},
    {"key": "active_away_penalty_team_penalty", "label": "Активное удаление правой команды — командное (1/0)", "category": "Активные удаления — правая команда", "entity_type": "penalty", "scope": "match_shared", "source_type": "selection", "value_type": "text"},
    {"key": "selected_penalty_infraction_id", "label": "Нарушение выбранного удаления", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_preset_id", "label": "Длительность выбранного удаления", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_side", "label": "Сторона выбранного удаления", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_event_time", "label": "Время события выбранного удаления", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_status", "label": "Статус выбранного удаления", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_remaining_ms", "label": "Остаток выбранного удаления, мс", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection"},
    {"key": "selected_penalty_duration_ms", "label": "Длительность выбранного удаления, мс", "category": "Выделенные", "entity_type": "penalty", "scope": "match", "source_type": "selection"},
    {"key": "selected_event_id", "label": "Выбранное событие", "category": "Выделенные", "entity_type": "event", "scope": "match", "source_type": "selection"},
    {"key": "selected_stat_player_id", "label": "Игрок для статистики", "category": "Выделенные", "entity_type": "player", "scope": "match", "source_type": "selection"},
    {"key": "selected_stat_team_id", "label": "Команда для статистики", "category": "Выделенные", "entity_type": "team", "scope": "match", "source_type": "selection"},

    {"key": "highlighted_game_id", "label": "Выделенный матч", "category": "Дополнительные", "entity_type": "match", "scope": "session", "source_type": "manual"},
    {"key": "compare_player1_id", "label": "Игрок сравнения 1", "category": "Сравнение", "entity_type": "player", "scope": "session", "source_type": "manual"},
    {"key": "compare_player2_id", "label": "Игрок сравнения 2", "category": "Сравнение", "entity_type": "player", "scope": "session", "source_type": "manual"},
    {"key": "compare_team1_id", "label": "Команда сравнения 1", "category": "Сравнение", "entity_type": "team", "scope": "session", "source_type": "manual"},
    {"key": "compare_team2_id", "label": "Команда сравнения 2", "category": "Сравнение", "entity_type": "team", "scope": "session", "source_type": "manual"},
 )


# Side-specific penalty selections and active-penalty mirrors are shared inside
# one match. This lets the operator select a HOME/AWAY penalty while an admin
# watches Mapping from another account without getting a private copy.
MATCH_SHARED_CONTEXT_KEYS: frozenset[str] = frozenset({
    "selected_home_penalty_id",
    "selected_home_penalty_player_id",
    "selected_home_penalty_player_db_id",
    "selected_home_penalty_team_penalty",
    "selected_away_penalty_id",
    "selected_away_penalty_player_id",
    "selected_away_penalty_player_db_id",
    "selected_away_penalty_team_penalty",
    "active_home_penalty_id",
    "active_home_penalty_player_id",
    "active_home_penalty_player_db_id",
    "active_home_penalty_team_penalty",
    "active_away_penalty_id",
    "active_away_penalty_player_id",
    "active_away_penalty_player_db_id",
    "active_away_penalty_team_penalty",
})


DEFAULT_SQL_SOURCES: tuple[dict[str, Any], ...] = (
    {
        "code": "game",
        "name": "Матч и команды",
        "category": "Матч",
        "description": "Основные данные текущего матча. Параметр :game_id берётся из контекста.",
        "sql_text": """SELECT
    g.external_id AS \"id\",
    g.tournament_external_id AS \"tournament_id\",
    CAST(g.game_date AS TEXT) AS \"date\",
    g.start_time AS \"time\",
    g.status AS \"status\",
    g.arena_ru AS \"venue\",
    g.home_team_external_id AS \"home.id\",
    g.home_team_name_ru AS \"home.name\",
    g.home_team_short_ru AS \"home.short_name\",
    g.home_team_city_ru AS \"home.city\",
    g.home_score AS \"home.score\",
    g.away_team_external_id AS \"away.id\",
    g.away_team_name_ru AS \"away.name\",
    g.away_team_short_ru AS \"away.short_name\",
    g.away_team_city_ru AS \"away.city\",
    g.away_score AS \"away.score\"
FROM hockey_games g
WHERE g.external_id = :game_id
LIMIT 1""",
        "field_metadata": {
            "id": {"label": "ID матча"}, "tournament_id": {"label": "ID турнира"},
            "date": {"label": "Дата"}, "time": {"label": "Время начала"}, "status": {"label": "Статус"}, "venue": {"label": "Арена / площадка"},
            "home.id": {"label": "ID команды", "category": "Хозяева"}, "home.name": {"label": "Название команды", "category": "Хозяева"},
            "home.short_name": {"label": "Короткое название", "category": "Хозяева"}, "home.city": {"label": "Город", "category": "Хозяева"}, "home.score": {"label": "Счёт", "category": "Хозяева", "kind": "number"}, "home.logo": {"label": "Логотип", "category": "Хозяева", "kind": "image"},
            "away.id": {"label": "ID команды", "category": "Гости"}, "away.name": {"label": "Название команды", "category": "Гости"},
            "away.short_name": {"label": "Короткое название", "category": "Гости"}, "away.city": {"label": "Город", "category": "Гости"}, "away.score": {"label": "Счёт", "category": "Гости", "kind": "number"}, "away.logo": {"label": "Логотип", "category": "Гости", "kind": "image"},
        },
    },
    {
        "code": "tournament",
        "name": "Турнир",
        "category": "Турнир",
        "description": "Данные выбранного турнира.",
        "sql_text": """SELECT
    t.external_id AS \"id\",
    t.name_ru AS \"name\",
    t.common_name_ru AS \"common_name\",
    t.season AS \"season\",
    t.season_part AS \"stage\",
    t.level AS \"level\"
FROM hockey_tournaments t
WHERE t.external_id = :tournament_id
LIMIT 1""",
        "field_metadata": {"id": {"label": "ID турнира"}, "name": {"label": "Название турнира"}, "common_name": {"label": "Короткое название"}, "season": {"label": "Сезон"}, "stage": {"label": "Этап"}, "level": {"label": "Лига"}},
    },
    {
        "code": "selected_player",
        "name": "Выбранный игрок",
        "category": "Выбранный игрок",
        "description": "Карточка игрока по :selected_player_id. До выбора игрока источник возвращает пустой результат.",
        "sql_text": """SELECT
    p.external_id AS \"id\",
    p.full_name_ru AS \"full_name\",
    p.last_name_ru AS \"last_name\",
    p.first_name_ru AS \"first_name\",
    COALESCE(pt.jersey_number, '') AS \"number\",
    p.position_ru AS \"position\",
    p.photo_url AS \"photo\",
    COALESCE(pt.club_external_id, '') AS \"team_id\",
    COALESCE(pt.team_name_ru, '') AS \"team_name\"
FROM hockey_players p
LEFT JOIN hockey_player_tournaments pt
  ON pt.player_id = p.id AND pt.tournament_external_id = :tournament_id
WHERE p.external_id = :selected_player_id
ORDER BY pt.active DESC
LIMIT 1""",
        "field_metadata": {"id": {"label": "ID игрока"}, "full_name": {"label": "Фамилия + Имя"}, "last_name": {"label": "Фамилия"}, "first_name": {"label": "Имя"}, "number": {"label": "Номер"}, "position": {"label": "Амплуа"}, "photo": {"label": "Фото", "kind": "image"}, "team_id": {"label": "ID команды"}, "team_name": {"label": "Команда"}},
    },
    {
        "code": "highlighted_game",
        "name": "Выделенный матч",
        "category": "Выделенный матч",
        "description": "Любой дополнительный матч по пользовательской переменной :highlighted_game_id.",
        "sql_text": """SELECT
    g.external_id AS \"id\",
    CAST(g.game_date AS TEXT) AS \"date\",
    g.start_time AS \"time\",
    g.home_team_name_ru AS \"home.name\",
    g.home_score AS \"home.score\",
    g.away_team_name_ru AS \"away.name\",
    g.away_score AS \"away.score\"
FROM hockey_games g
WHERE g.external_id = :highlighted_game_id
LIMIT 1""",
        "field_metadata": {},
    },
)


class MappingDataService:
    """Dynamic context variables and read-only SQL data sources for vMix mapping."""

    def __init__(self, database: HockeyDatabase, settings: Any | None = None) -> None:
        self.database = database
        self.settings = settings
        self._database_schema_cache: dict[str, Any] | None = None

    def _current_language(self, supplied: dict[str, Any] | None = None) -> str:
        supplied = supplied or {}
        raw = str(supplied.get("ui_language") or supplied.get("language") or "").strip().lower()
        if raw not in {"ru", "en"} and self.settings is not None:
            try:
                raw = str(self.settings.public().get("ui_language") or "ru").strip().lower()
            except Exception:
                raw = "ru"
        return raw if raw in {"ru", "en"} else "ru"

    def ensure_defaults(self) -> None:
        now = _utcnow()
        with self.database.session() as session:
            existing_var_rows = {row.key: row for row in session.scalars(select(MappingContextVariable))}
            existing_vars = set(existing_var_rows)
            # Build78 initially stored automatic HOME/AWAY penalty identifiers as
            # account-scoped match values. Migrate existing databases in-place so
            # an operator can set a penalty and an admin can immediately see it in Mapping.
            for key in MATCH_SHARED_CONTEXT_KEYS:
                row = existing_var_rows.get(key)
                if row is not None and row.scope != "match_shared":
                    row.scope = "match_shared"
                    row.updated_by = "system"
                    row.updated_at = now
            for index, item in enumerate(DEFAULT_CONTEXT_VARIABLES):
                if item["key"] in existing_vars:
                    continue
                session.add(MappingContextVariable(
                    key=item["key"], label=item["label"], category=item.get("category", "Контекст"),
                    description=item.get("description", ""), value_type=item.get("value_type", "id"),
                    entity_type=item.get("entity_type", ""), scope=item.get("scope", "match"),
                    source_type=item.get("source_type", "manual"), default_value=item.get("default_value", ""),
                    is_system=item.get("source_type") == "system", enabled=True, sort_order=index,
                    created_by="system", updated_by="system", created_at=now, updated_at=now,
                ))
            existing_sources = {row.code for row in session.scalars(select(MappingSqlDataSource))}
            for index, item in enumerate(DEFAULT_SQL_SOURCES):
                if item["code"] in existing_sources:
                    continue
                session.add(MappingSqlDataSource(
                    code=item["code"], name=item["name"], category=item.get("category", "Данные"),
                    description=item.get("description", ""), sql_text=item["sql_text"],
                    field_metadata_json=json.dumps(item.get("field_metadata", {}), ensure_ascii=False),
                    enabled=True, sort_order=index, created_by="system", updated_by="system",
                    created_at=now, updated_at=now,
                ))

    @staticmethod
    def _variable_payload(row: MappingContextVariable) -> dict[str, Any]:
        return {
            "id": row.id, "key": row.key, "label": row.label, "category": row.category,
            "description": row.description, "value_type": row.value_type, "entity_type": row.entity_type,
            "scope": row.scope, "source_type": row.source_type, "default_value": row.default_value,
            "is_system": bool(row.is_system), "enabled": bool(row.enabled), "sort_order": row.sort_order,
        }

    @staticmethod
    def _source_payload(row: MappingSqlDataSource, *, include_sql: bool = True) -> dict[str, Any]:
        try:
            metadata = json.loads(row.field_metadata_json or "{}")
        except Exception:
            metadata = {}
        result = {
            "id": row.id, "code": row.code, "name": row.name, "category": row.category,
            "description": row.description, "enabled": bool(row.enabled),
            "auto_refresh_enabled": bool(getattr(row, "auto_refresh_enabled", False)),
            "refresh_interval_ms": max(1000, int(getattr(row, "refresh_interval_ms", 1000) or 1000)),
            "sort_order": row.sort_order,
            "field_metadata": metadata if isinstance(metadata, dict) else {},
        }
        if include_sql:
            result["sql_text"] = row.sql_text
        return result

    def list_variables(self) -> dict[str, Any]:
        self.ensure_defaults()
        with self.database.session() as session:
            rows = list(session.scalars(select(MappingContextVariable).order_by(MappingContextVariable.sort_order, MappingContextVariable.id)))
        return {"variables": [self._variable_payload(row) for row in rows]}

    def create_variable(self, payload: Any, user: HockeyUser) -> dict[str, Any]:
        self.ensure_defaults()
        key = str(payload.key or "").strip()
        if not _KEY_RE.fullmatch(key):
            raise HTTPException(status_code=422, detail="Ключ: латиница/цифры/._-, начинаться должен с буквы")
        if str(payload.source_type or "manual") == "system":
            raise HTTPException(status_code=422, detail="Пользовательская переменная не может иметь источник SYSTEM")
        now = _utcnow()
        with self.database.session() as session:
            if session.scalar(select(MappingContextVariable).where(MappingContextVariable.key == key)) is not None:
                raise HTTPException(status_code=409, detail="Переменная с таким ключом уже существует")
            row = MappingContextVariable(
                key=key, label=str(payload.label or key)[:200], category=str(payload.category or "Пользовательские")[:100],
                description=str(payload.description or "")[:4000], value_type=str(payload.value_type or "id")[:32],
                entity_type=str(payload.entity_type or "")[:32], scope=str(payload.scope or "match")[:32],
                source_type=str(payload.source_type or "manual")[:32], default_value=str(payload.default_value or "")[:4000],
                is_system=False, enabled=bool(payload.enabled), sort_order=int(payload.sort_order or 1000),
                created_by=user.login, updated_by=user.login, created_at=now, updated_at=now,
            )
            session.add(row); session.flush(); result = self._variable_payload(row)
        return result

    def update_variable(self, variable_id: int, payload: Any, user: HockeyUser) -> dict[str, Any]:
        self.ensure_defaults()
        with self.database.session() as session:
            row = session.get(MappingContextVariable, variable_id)
            if row is None:
                raise HTTPException(status_code=404, detail="Переменная не найдена")
            if row.is_system:
                # System keys are part of the runtime contract; labels can be edited but the technical contract cannot.
                row.label = str(payload.label or row.label)[:200]
                row.category = str(payload.category or row.category)[:100]
                row.description = str(payload.description or row.description)[:4000]
                row.enabled = bool(payload.enabled)
            else:
                key = str(payload.key or row.key).strip()
                if not _KEY_RE.fullmatch(key):
                    raise HTTPException(status_code=422, detail="Некорректный ключ переменной")
                duplicate = session.scalar(select(MappingContextVariable).where(and_(MappingContextVariable.key == key, MappingContextVariable.id != row.id)))
                if duplicate is not None:
                    raise HTTPException(status_code=409, detail="Переменная с таким ключом уже существует")
                row.key = key
                row.label = str(payload.label or key)[:200]
                row.category = str(payload.category or "Пользовательские")[:100]
                row.description = str(payload.description or "")[:4000]
                row.value_type = str(payload.value_type or "id")[:32]
                row.entity_type = str(payload.entity_type or "")[:32]
                row.scope = str(payload.scope or "match")[:32]
                row.source_type = str(payload.source_type or "manual")[:32]
                row.default_value = str(payload.default_value or "")[:4000]
                row.enabled = bool(payload.enabled)
                row.sort_order = int(payload.sort_order or row.sort_order or 1000)
            row.updated_by = user.login; row.updated_at = _utcnow(); session.flush(); result = self._variable_payload(row)
        return result

    def delete_variable(self, variable_id: int) -> dict[str, Any]:
        self.ensure_defaults()
        with self.database.session() as session:
            row = session.get(MappingContextVariable, variable_id)
            if row is None:
                raise HTTPException(status_code=404, detail="Переменная не найдена")
            if row.is_system:
                raise HTTPException(status_code=409, detail="Системную переменную удалить нельзя")
            session.execute(delete(MappingContextValue).where(MappingContextValue.variable_id == variable_id))
            session.delete(row)
        return {"ok": True, "id": variable_id}

    def list_sources(self) -> dict[str, Any]:
        self.ensure_defaults()
        with self.database.session() as session:
            rows = list(session.scalars(select(MappingSqlDataSource).order_by(MappingSqlDataSource.sort_order, MappingSqlDataSource.id)))
        return {"sources": [self._source_payload(row) for row in rows]}

    def database_schema_reference(self) -> dict[str, Any]:
        """Return a compact read-only reference of hockey database tables.

        This is intentionally metadata-only: table/column names and SQL types,
        never row data or connection credentials. The SQL editor uses it as an
        expandable cheat sheet so operators do not have to inspect PostgreSQL
        manually just to remember a table name.
        """
        if self._database_schema_cache is not None:
            return self._database_schema_cache
        try:
            inspector = inspect(self.database.engine)
            table_names = sorted(
                name for name in inspector.get_table_names()
                if str(name).startswith("hockey_")
            )
            tables: list[dict[str, Any]] = []
            for table_name in table_names:
                columns = []
                for column in inspector.get_columns(table_name):
                    columns.append({
                        "name": str(column.get("name") or ""),
                        "type": str(column.get("type") or ""),
                        "nullable": bool(column.get("nullable", True)),
                    })
                tables.append({
                    "name": table_name,
                    "column_count": len(columns),
                    "columns": columns,
                })
            payload = {"tables": tables, "table_count": len(tables)}
            self._database_schema_cache = payload
            return payload
        except SQLAlchemyError as exc:
            return {"tables": [], "table_count": 0, "error": str(exc)}
        except Exception as exc:
            return {"tables": [], "table_count": 0, "error": str(exc)}

    def create_source(self, payload: Any, user: HockeyUser) -> dict[str, Any]:
        self.ensure_defaults()
        code = str(payload.code or "").strip()
        if not _SOURCE_CODE_RE.fullmatch(code):
            raise HTTPException(status_code=422, detail="Код источника: латиница, цифры и _, начинаться должен с буквы")
        sql_text = str(payload.sql_text or "").strip()
        self.validate_sql(sql_text)
        now = _utcnow()
        with self.database.session() as session:
            if session.scalar(select(MappingSqlDataSource).where(MappingSqlDataSource.code == code)) is not None:
                raise HTTPException(status_code=409, detail="SQL-источник с таким кодом уже существует")
            row = MappingSqlDataSource(
                code=code, name=str(payload.name or code)[:200], category=str(payload.category or "Данные")[:100],
                description=str(payload.description or "")[:4000], sql_text=sql_text,
                field_metadata_json=json.dumps(payload.field_metadata or {}, ensure_ascii=False),
                enabled=bool(payload.enabled),
                auto_refresh_enabled=bool(getattr(payload, "auto_refresh_enabled", False)),
                refresh_interval_ms=max(1000, min(3600000, int(getattr(payload, "refresh_interval_ms", 1000) or 1000))),
                sort_order=int(payload.sort_order or 1000),
                created_by=user.login, updated_by=user.login, created_at=now, updated_at=now,
            )
            session.add(row); session.flush(); result = self._source_payload(row)
        return result

    def update_source(self, source_id: int, payload: Any, user: HockeyUser) -> dict[str, Any]:
        self.ensure_defaults()
        sql_text = str(payload.sql_text or "").strip(); self.validate_sql(sql_text)
        with self.database.session() as session:
            row = session.get(MappingSqlDataSource, source_id)
            if row is None: raise HTTPException(status_code=404, detail="SQL-источник не найден")
            code = str(payload.code or row.code).strip()
            if not _SOURCE_CODE_RE.fullmatch(code): raise HTTPException(status_code=422, detail="Некорректный код источника")
            duplicate = session.scalar(select(MappingSqlDataSource).where(and_(MappingSqlDataSource.code == code, MappingSqlDataSource.id != row.id)))
            if duplicate is not None: raise HTTPException(status_code=409, detail="SQL-источник с таким кодом уже существует")
            row.code=code; row.name=str(payload.name or code)[:200]; row.category=str(payload.category or "Данные")[:100]
            row.description=str(payload.description or "")[:4000]; row.sql_text=sql_text
            row.field_metadata_json=json.dumps(payload.field_metadata or {}, ensure_ascii=False)
            row.enabled=bool(payload.enabled)
            row.auto_refresh_enabled=bool(getattr(payload, "auto_refresh_enabled", False))
            row.refresh_interval_ms=max(1000, min(3600000, int(getattr(payload, "refresh_interval_ms", 1000) or 1000)))
            row.sort_order=int(payload.sort_order or row.sort_order or 1000)
            row.updated_by=user.login; row.updated_at=_utcnow(); session.flush(); result=self._source_payload(row)
        return result

    def delete_source(self, source_id: int) -> dict[str, Any]:
        self.ensure_defaults()
        with self.database.session() as session:
            row=session.get(MappingSqlDataSource, source_id)
            if row is None: raise HTTPException(status_code=404, detail="SQL-источник не найден")
            session.delete(row)
        return {"ok": True, "id": source_id}

    @staticmethod
    def _normalise_sql(sql_text: str) -> str:
        """Remove SQL comments outside quoted strings and allow one trailing semicolon.

        The stored editor text is not changed; normalisation happens only for validation/execution.
        Multiple statements remain forbidden.
        """
        source = str(sql_text or "")
        out: list[str] = []
        i = 0
        quote = ""
        while i < len(source):
            ch = source[i]
            nxt = source[i + 1] if i + 1 < len(source) else ""
            if quote:
                out.append(ch)
                if ch == quote:
                    if nxt == quote:
                        out.append(nxt); i += 2; continue
                    quote = ""
                i += 1
                continue
            if ch in {"'", '"'}:
                quote = ch; out.append(ch); i += 1; continue
            if ch == "-" and nxt == "-":
                i += 2
                while i < len(source) and source[i] not in "\r\n": i += 1
                out.append("\n")
                continue
            if ch == "/" and nxt == "*":
                end = source.find("*/", i + 2)
                if end < 0:
                    raise HTTPException(status_code=422, detail="Незакрытый SQL-комментарий")
                out.append(" "); i = end + 2; continue
            out.append(ch); i += 1
        sql = "".join(out).strip()
        if not sql:
            raise HTTPException(status_code=422, detail="SQL пуст")
        if sql.endswith(";"):
            sql = sql[:-1].rstrip()
        # A semicolon that remains outside comments/quotes means more than one statement.
        quote = ""; i = 0
        while i < len(sql):
            ch = sql[i]; nxt = sql[i + 1] if i + 1 < len(sql) else ""
            if quote:
                if ch == quote:
                    if nxt == quote: i += 2; continue
                    quote = ""
                i += 1; continue
            if ch in {"'", '"'}: quote = ch
            elif ch == ";":
                raise HTTPException(status_code=422, detail="Разрешён только один read-only SELECT")
            i += 1
        return sql

    @classmethod
    def validate_sql(cls, sql_text: str) -> set[str]:
        sql = cls._normalise_sql(sql_text)
        if not re.match(r"^(select|with)\b", sql, re.IGNORECASE):
            raise HTTPException(status_code=422, detail="Разрешены только SELECT / WITH ... SELECT")
        if _DANGEROUS_SQL_RE.search(sql) or _DANGEROUS_FUNCTION_RE.search(sql):
            raise HTTPException(status_code=422, detail="В SQL обнаружена запрещённая операция")
        return cls._find_context_params(sql)

    @staticmethod
    def _find_context_params(sql: str) -> set[str]:
        """Find :context parameters only in SQL code, never inside quoted strings/identifiers.

        This keeps PostgreSQL formats such as 'HH24:MI:SS' and casts such as
        value::text from being mistaken for Mapping context variables.
        """
        params: set[str] = set()
        i = 0
        quote = ""
        while i < len(sql):
            ch = sql[i]
            nxt = sql[i + 1] if i + 1 < len(sql) else ""
            if quote:
                if ch == quote:
                    if nxt == quote:
                        i += 2
                        continue
                    quote = ""
                i += 1
                continue
            if ch in {"'", '"'}:
                quote = ch
                i += 1
                continue
            if ch == ":":
                # PostgreSQL cast operator :: is not a bind parameter.
                prev = sql[i - 1] if i > 0 else ""
                if prev == ":" or nxt == ":" or not (nxt.isalpha()):
                    i += 1
                    continue
                j = i + 2
                while j < len(sql) and (sql[j].isalnum() or sql[j] == "_"):
                    j += 1
                params.add(sql[i + 1:j])
                i = j
                continue
            i += 1
        return params

    @staticmethod
    def _scope_key(variable: MappingContextVariable, user: HockeyUser, context: dict[str, str]) -> str:
        scope = (variable.scope or "match").lower()
        game_id = context.get("game_id", "")
        token = context.get("session_token", "")
        if scope == "global": return "global"
        if scope == "account": return f"account:{user.id}"
        if scope in {"session", "temporary"}: return f"session:{token or user.id}"
        if scope in {"match_shared", "shared_match"}: return f"match:{game_id or '-'}"
        return f"match:{game_id or '-'}:account:{user.id}"

    def resolve_context(self, user: HockeyUser, supplied: dict[str, Any] | None = None) -> tuple[dict[str, str], list[dict[str, Any]]]:
        self.ensure_defaults()
        supplied = supplied or {}
        context: dict[str, str] = {str(k): str(v) for k, v in supplied.items() if v is not None and str(v) != ""}
        context.setdefault("account_id", str(user.id))
        context.setdefault("ui_language", self._current_language(supplied))
        game_id = context.get("game_id", "")
        if game_id:
            with self.database.session() as session:
                game = session.scalar(select(Game).where(Game.external_id == game_id))
                if game is not None:
                    context.setdefault("tournament_id", game.tournament_external_id or "")
                    context.setdefault("team1_id", game.home_team_external_id or "")
                    context.setdefault("team2_id", game.away_team_external_id or "")
        if not context.get("device_id"):
            with self.database.session() as session:
                device = session.scalar(select(VmixDevice).where(and_(VmixDevice.wfl_user_id == user.id, VmixDevice.is_active_for_account.is_(True))))
                if device is not None:
                    context["device_id"] = device.device_uuid
                    if device.current_assignment_key: context["assignment_id"] = device.current_assignment_key
                    if not context.get("game_id") and device.current_match_external_id: context["game_id"] = device.current_match_external_id

        # Parameterised lower-panel controls are match-scoped system context.
        # SQL bind names cannot contain dots, therefore panel.stat_period.value
        # becomes :panel_stat_period_value while Mapping keeps the dotted key.
        runtime_variable_payloads: list[dict[str, Any]] = []
        resolved_game_id = str(context.get("game_id") or "").strip()
        if resolved_game_id:
            with self.database.session() as session:
                game = session.scalar(select(Game).where(Game.external_id == resolved_game_id))
                if game is not None:
                    context.setdefault("tournament_id", game.tournament_external_id or "")
                    context.setdefault("team1_id", game.home_team_external_id or "")
                    context.setdefault("team2_id", game.away_team_external_id or "")
                control = session.scalar(select(GameControlState).where(GameControlState.game_external_id == resolved_game_id))
            if game is not None:
                score_control = HockeyDataService._score_control_payload(game, control)
                context.setdefault("score_home", str(score_control["home"]))
                context.setdefault("score_away", str(score_control["away"]))
                context.setdefault("score_external_home", str(score_control["external_home"]))
                context.setdefault("score_external_away", str(score_control["external_away"]))
                context.setdefault("score_manual", "1" if score_control["manual"] else "0")
            if control is not None:
                runtime_values = HockeyDataService._runtime_values_from_row(control)
                player_panel_labels: dict[str, str] = {}
                for meta_key, meta_value in runtime_values.items():
                    meta_match = re.match(r"^player_select\.([A-Za-z0-9_]+)\._label$", str(meta_key))
                    if meta_match:
                        player_panel_labels[meta_match.group(1)] = str(meta_value or meta_match.group(1))
                for raw_key, raw_value in runtime_values.items():
                    raw_key_text = str(raw_key)
                    player_match = re.match(r"^player_select\.([A-Za-z0-9_]+)\.(.+)$", raw_key_text)
                    if player_match and player_match.group(2).startswith("_"):
                        continue
                    sql_key = re.sub(r"[^A-Za-z0-9_]+", "_", raw_key_text).strip("_")[:120]
                    if not sql_key or not re.match(r"^[A-Za-z]", sql_key):
                        sql_key = f"panel_{sql_key}"[:120]
                    context.setdefault(sql_key, str(raw_value))
                    if player_match:
                        panel_id, field_key = player_match.groups()
                        slot_match = re.match(r"player(\d+)_(id|db_id|team_id|side|number|name)$", field_key)
                        field_labels = {
                            "id": "external ID", "db_id": "DB ID", "team_id": "ID команды",
                            "side": "сторона", "number": "номер", "name": "имя",
                        }
                        label = field_key
                        if slot_match:
                            slot_no, field_name = slot_match.groups()
                            label = f"Игрок {slot_no} — {field_labels.get(field_name, field_name)}"
                        category = f"Игровые выборки · {player_panel_labels.get(panel_id, panel_id)}"
                        description = f"Игрок, перетащенный оператором в блок {player_panel_labels.get(panel_id, panel_id)}."
                    else:
                        label = raw_key_text
                        category = "Нижняя панель"
                        description = f"Runtime-параметр нижней панели: {raw_key_text}"
                    runtime_variable_payloads.append({
                        "key": sql_key,
                        "label": label,
                        "category": category,
                        "description": description,
                        "scope": "match",
                        "source_type": "system",
                        "is_system": True,
                        "value": str(raw_value),
                    })
        with self.database.session() as session:
            variables = list(session.scalars(select(MappingContextVariable).where(MappingContextVariable.enabled.is_(True)).order_by(MappingContextVariable.sort_order, MappingContextVariable.id)))
            for variable in variables:
                if variable.source_type == "system":
                    continue
                scope_key = self._scope_key(variable, user, context)
                value = session.scalar(select(MappingContextValue).where(and_(MappingContextValue.variable_id == variable.id, MappingContextValue.scope_key == scope_key)))
                if value is not None and value.value_text != "" and variable.key not in context: context[variable.key] = value.value_text
                elif variable.default_value and variable.key not in context: context[variable.key] = variable.default_value
        payload = []
        for variable in variables:
            item = self._variable_payload(variable); item["value"] = context.get(variable.key, ""); payload.append(item)
        payload.extend(runtime_variable_payloads)
        return context, payload

    def set_context_value(self, key: str, value: Any, user: HockeyUser, supplied: dict[str, Any] | None = None) -> dict[str, Any]:
        self.ensure_defaults()
        context, _ = self.resolve_context(user, supplied)
        with self.database.session() as session:
            variable = session.scalar(select(MappingContextVariable).where(MappingContextVariable.key == key))
            if variable is None:
                # BUILD86: be defensive with long-lived production databases.
                # Penalty selection keys were added over several builds; if a row
                # is missing for any reason, restore the known default contract
                # immediately instead of breaking an operator click with 404.
                default_index = next((i for i, item in enumerate(DEFAULT_CONTEXT_VARIABLES) if item.get("key") == key), -1)
                default_item = DEFAULT_CONTEXT_VARIABLES[default_index] if default_index >= 0 else None
                if default_item is not None:
                    now = _utcnow()
                    variable = MappingContextVariable(
                        key=default_item["key"],
                        label=default_item["label"],
                        category=default_item.get("category", "Контекст"),
                        description=default_item.get("description", ""),
                        value_type=default_item.get("value_type", "id"),
                        entity_type=default_item.get("entity_type", ""),
                        scope=default_item.get("scope", "match"),
                        source_type=default_item.get("source_type", "manual"),
                        default_value=default_item.get("default_value", ""),
                        is_system=default_item.get("source_type") == "system",
                        enabled=True,
                        sort_order=default_index,
                        created_by="system",
                        updated_by="system",
                        created_at=now,
                        updated_at=now,
                    )
                    session.add(variable)
                    session.flush()
                else:
                    raise HTTPException(status_code=404, detail=f"Переменная не найдена: {key}")
            if variable.source_type == "system" or variable.is_system:
                raise HTTPException(status_code=409, detail=f"Системная переменная задаётся программой: {key}")
            scope_key = self._scope_key(variable, user, context)
            row = session.scalar(select(MappingContextValue).where(and_(MappingContextValue.variable_id == variable.id, MappingContextValue.scope_key == scope_key)))
            if row is None:
                row = MappingContextValue(variable_id=variable.id, scope_key=scope_key, wfl_user_id=user.id,
                    operator_session_token=context.get("session_token", ""), game_external_id=context.get("game_id", ""),
                    value_text=str(value if value is not None else ""), updated_by=user.login, updated_at=_utcnow())
                session.add(row)
            else:
                row.value_text=str(value if value is not None else ""); row.updated_by=user.login; row.updated_at=_utcnow()
            session.flush()
        return {"ok": True, "key": key, "value": str(value if value is not None else ""), "scope_key": scope_key}

    @staticmethod
    def _pretty_label(column: str) -> str:
        leaf = str(column).split(".")[-1].replace("_", " ").strip()
        return leaf[:1].upper() + leaf[1:] if leaf else str(column)

    @staticmethod
    def _infer_kind(column: str, value: Any) -> str:
        name = column.lower()
        if any(token in name for token in ("photo", "logo", "image", "source")): return "image"
        if isinstance(value, (int, float)): return "number"
        return "text"

    def _run_source(self, source: MappingSqlDataSource, context: dict[str, str], *, max_rows: int = 25) -> dict[str, Any]:
        params = self.validate_sql(source.sql_text)
        sql_text = self._normalise_sql(source.sql_text)
        missing = sorted(name for name in params if not context.get(name))
        if missing:
            return {"ok": True, "skipped": True, "missing": missing, "columns": [], "rows": [], "row_count": 0}
        bound = {name: context.get(name, "") for name in params}
        try:
            with self.database.engine.connect() as connection:
                transaction = connection.begin()
                try:
                    if connection.dialect.name == "postgresql":
                        connection.execute(text("SET TRANSACTION READ ONLY"))
                        connection.execute(text("SET LOCAL statement_timeout = '2000ms'"))
                    result = connection.execute(text(sql_text), bound)
                    columns = list(result.keys())
                    rows = [dict(row) for row in result.mappings().fetchmany(max_rows)]
                finally:
                    transaction.rollback()
        except SQLAlchemyError as error:
            return {"ok": False, "error": str(error).splitlines()[0][:500], "columns": [], "rows": [], "row_count": 0}
        serial_rows = []
        for row in rows:
            serial_rows.append({key: (value.isoformat() if hasattr(value, "isoformat") else value) for key, value in row.items()})
        return {"ok": True, "skipped": False, "columns": columns, "rows": serial_rows, "row_count": len(serial_rows)}

    def preview_sql(self, sql_text: str, user: HockeyUser, supplied: dict[str, Any] | None = None) -> dict[str, Any]:
        self.ensure_defaults(); context, variables = self.resolve_context(user, supplied)
        class _PreviewSource:
            pass
        source = _PreviewSource()
        source.sql_text = str(sql_text or "")
        result = self._run_source(source, context, max_rows=250)
        return {"context": context, "variables": variables, **result}

    def preview_source(self, source_id: int, user: HockeyUser, supplied: dict[str, Any] | None = None) -> dict[str, Any]:
        self.ensure_defaults(); context, variables = self.resolve_context(user, supplied)
        with self.database.session() as session:
            source = session.get(MappingSqlDataSource, source_id)
            if source is None: raise HTTPException(status_code=404, detail="SQL-источник не найден")
            payload = self._source_payload(source)
            result = self._run_source(source, context, max_rows=250)
        return {"source": payload, "context": context, "variables": variables, **result}

    @staticmethod
    def _table_row_label(row: dict[str, Any], index: int) -> str:
        preferred = (
            "full_name", "name", "title", "team_name", "player_name", "last_name",
            "id", "external_id", "number",
        )
        for key in preferred:
            value = row.get(key)
            if value not in (None, ""):
                text_value = str(value)
                if len(text_value) > 46:
                    text_value = text_value[:43] + "…"
                return f"Строка {index} · {text_value}"
        for value in row.values():
            if value not in (None, ""):
                text_value = str(value)
                if len(text_value) > 46:
                    text_value = text_value[:43] + "…"
                return f"Строка {index} · {text_value}"
        return f"Строка {index}"

    @staticmethod
    def _language_pairs(columns: list[str]) -> dict[str, dict[str, str]]:
        pairs: dict[str, dict[str, str]] = {}
        for column in columns:
            match = _LANG_SUFFIX_RE.match(str(column))
            if not match:
                continue
            base = match.group("base")
            lang = match.group("lang").upper()
            canonical = base.casefold()
            entry = pairs.setdefault(canonical, {"base": base})
            entry[lang] = str(column)
        return {key: value for key, value in pairs.items() if value.get("RUS") and value.get("ENG")}

    @staticmethod
    def _localized_value(row: dict[str, Any], pair: dict[str, str], language: str) -> tuple[Any, str]:
        primary = pair.get("RUS" if language == "ru" else "ENG", "")
        fallback = pair.get("ENG" if language == "ru" else "RUS", "")
        value = row.get(primary, "") if primary else ""
        if value in (None, "") and fallback:
            value = row.get(fallback, "")
            primary = fallback
        return value, primary

    def _live_control_items(self, context: dict[str, str], language: str) -> list[dict[str, Any]]:
        game_id = str(context.get("game_id") or "").strip()
        if not game_id:
            return []
        settings = self.settings.public() if self.settings is not None else {}
        with self.database.session() as session:
            game = session.scalar(select(Game).where(Game.external_id == game_id))
            if game is None:
                return []
            tournament = session.scalar(
                select(Tournament).where(Tournament.external_id == game.tournament_external_id)
            )
            control = session.scalar(
                select(GameControlState).where(GameControlState.game_external_id == game_id)
            )
        current_period = str(getattr(control, "current_period", "1") or "1")
        stage = stage_key(
            str(getattr(tournament, "season_part", "") or ""),
            str(getattr(tournament, "round_code", "") or ""),
        ) if tournament is not None else "other"
        timer_state = (
            HockeyDataService._timer_state_from_row(control)
            if control is not None else HockeyDataService._default_game_timer_state()
        )
        strength = HockeyDataService._strength_payload(
            settings,
            stage=stage,
            current_period=current_period,
            timer_state=timer_state,
            language=language,
        )
        flags = HockeyDataService._runtime_flags_from_row(control) if control is not None else {}
        period_status = HockeyDataService._period_status_payload(
            settings,
            current_period=current_period,
            language=language,
        )
        result: list[dict[str, Any]] = []
        score_control = HockeyDataService._score_control_payload(game, control)
        score_fields = (
            ("home", "Счёт HOME · LIVE", score_control["home"], "Оперативный счёт HOME. После первого ручного +/- становится главным до синхронизации."),
            ("away", "Счёт AWAY · LIVE", score_control["away"], "Оперативный счёт AWAY. После первого ручного +/- становится главным до синхронизации."),
            ("external_home", "Счёт HOME · SQL/API", score_control["external_home"], "Последний счёт HOME из базы/API; используется для первоначальной и ручной повторной синхронизации."),
            ("external_away", "Счёт AWAY · SQL/API", score_control["external_away"], "Последний счёт AWAY из базы/API; используется для первоначальной и ручной повторной синхронизации."),
            ("manual", "Ручной режим счёта", "1" if score_control["manual"] else "0", "1 — LIVE счёт зафиксирован оператором; 0 — LIVE следует SQL/API."),
        )
        for key, label, value, description in score_fields:
            result.append({
                "key": f"score.{key}",
                "label": label,
                "category": "Матч · Оперативный счёт",
                "value": value,
                "kind": "number" if key != "manual" else "text",
                "source_code": "live_control",
                "description": description,
            })
        period_fields = (
            ("key", "Ключ периода", "text"),
            ("number", "Номер периода / овертайма", "number"),
            ("label", "Статус периода · основной", "text"),
            ("compact", "Статус периода · компактный", "text"),
            ("short", "Статус периода · короткий", "text"),
            ("long", "Статус периода · полный", "text"),
            ("ru", "Статус периода · RU", "text"),
            ("en", "Статус периода · EN", "text"),
        )
        for key, label, kind in period_fields:
            result.append({
                "key": f"period.{key}",
                "label": label,
                "category": "Матч · Период",
                "value": period_status.get(key, ""),
                "kind": kind,
                "source_code": "live_control",
                "description": "Настраиваемая подпись текущего периода из раздела «Таймеры и численные составы».",
            })
        formats = period_status.get("formats") if isinstance(period_status.get("formats"), dict) else {}
        for fmt in ("compact", "short", "long"):
            fmt_values = formats.get(fmt) if isinstance(formats.get(fmt), dict) else {}
            for lang in ("ru", "en"):
                result.append({
                    "key": f"period.{lang}.{fmt}",
                    "label": f"{fmt.upper()} · {lang.upper()}",
                    "category": "Матч · Период · Форматы",
                    "value": fmt_values.get(lang, ""),
                    "kind": "text",
                    "source_code": "live_control",
                    "description": "Конкретный языковой вариант статуса периода, не зависящий от языка Mapping.",
                })
        strength_fields = (
            ("state_key", "Состояние", "text"),
            ("state_label", "Подпись численного состава", "text"),
            ("strength_label", "Численный состав HOME×AWAY", "text"),
            ("advantage_numbers", "Состав большинства", "text"),
            ("home_label", "Подпись HOME", "text"),
            ("away_label", "Подпись AWAY", "text"),
            ("home_skaters", "Игроков HOME", "number"),
            ("away_skaters", "Игроков AWAY", "number"),
            ("home_penalties", "Удалений HOME", "number"),
            ("away_penalties", "Удалений AWAY", "number"),
            ("advantage_side", "Сторона большинства", "text"),
        )
        for key, label, kind in strength_fields:
            result.append({
                "key": f"strength.{key}",
                "label": label,
                "category": "Матч · Численный состав",
                "value": strength.get(key, ""),
                "kind": kind,
                "source_code": "live_control",
                "description": "Рассчитано из текущих удалений с подписями из настроек «Таймеры и численные составы».",
            })
        for key, value in flags.items():
            result.append({
                "key": f"match_flag.{key}",
                "label": key,
                "category": "Матч · Флаги оператора",
                "value": "1" if value else "0",
                "kind": "text",
                "source_code": "live_control",
                "description": "Текущее состояние операторской кнопки матча.",
            })
        runtime_values = HockeyDataService._runtime_values_from_row(control) if control is not None else {}
        player_panel_labels: dict[str, str] = {}
        for meta_key, meta_value in runtime_values.items():
            meta_match = re.match(r"^player_select\.([A-Za-z0-9_]+)\._label$", str(meta_key))
            if meta_match:
                player_panel_labels[meta_match.group(1)] = str(meta_value or meta_match.group(1))
        for key, value in runtime_values.items():
            mapping_key = str(key)
            if mapping_key.startswith("panel."):
                result.append({
                    "key": mapping_key,
                    "label": mapping_key,
                    "category": "Нижняя панель · Параметры",
                    "value": value,
                    "kind": "text",
                    "source_code": "live_control",
                    "description": "Выбранное значение параметризованной кнопки нижней панели.",
                })
                continue
            player_match = re.match(r"^player_select\.([A-Za-z0-9_]+)\.(.+)$", mapping_key)
            if not player_match:
                continue
            panel_id, field_key = player_match.groups()
            if field_key.startswith("_"):
                continue
            slot_match = re.match(r"player(\d+)_(id|db_id|team_id|side|number|name)$", field_key)
            field_labels = {
                "id": "external ID", "db_id": "DB ID", "team_id": "ID команды",
                "side": "сторона", "number": "номер", "name": "имя",
            }
            label = field_key
            if slot_match:
                slot_no, field_name = slot_match.groups()
                label = f"Игрок {slot_no} — {field_labels.get(field_name, field_name)}"
            panel_label = player_panel_labels.get(panel_id, panel_id)
            result.append({
                "key": mapping_key,
                "label": label,
                "category": f"Игровые выборки · {panel_label}",
                "value": value,
                "kind": "text",
                "source_code": "live_control",
                "description": f"Значение слота блока игроков «{panel_label}».",
            })
        return result

    def data_catalog(
        self,
        user: HockeyUser,
        supplied: dict[str, Any] | None = None,
        *,
        source_codes: set[str] | None = None,
    ) -> dict[str, Any]:
        self.ensure_defaults()
        language = self._current_language(supplied)
        context, variables = self.resolve_context(user, {**(supplied or {}), "ui_language": language})
        items: list[dict[str, Any]] = []
        tables: list[dict[str, Any]] = []
        # Context variables are usable directly as mapping values too.
        items.extend(self._live_control_items(context, language))
        for variable in variables:
            items.append({
                "key": f"context.{variable['key']}", "label": variable["label"], "category": f"Контекст · {variable['category']}",
                "value": variable.get("value", ""), "kind": "text", "source_code": "context", "variable_key": variable["key"],
                "description": variable.get("description", ""),
            })
        source_results = []
        with self.database.session() as session:
            sources = list(session.scalars(select(MappingSqlDataSource).where(MappingSqlDataSource.enabled.is_(True)).order_by(MappingSqlDataSource.sort_order, MappingSqlDataSource.id)))
            if source_codes is not None:
                wanted = {str(code) for code in source_codes}
                sources = [source for source in sources if source.code in wanted]
            for source in sources:
                payload = self._source_payload(source)
                # Mapping may address a concrete row/cell, therefore the catalog keeps a
                # larger read-only window than the compact SQL preview. 250 rows covers
                # rosters, standings and most hockey tables without changing the SQL.
                result = self._run_source(source, context, max_rows=250)
                source_results.append({"id": source.id, "code": source.code, "name": source.name, "ok": result.get("ok"), "skipped": result.get("skipped", False), "missing": result.get("missing", []), "row_count": result.get("row_count", 0), "error": result.get("error", "")})
                rows = result.get("rows") or []
                columns = [str(value) for value in (result.get("columns") or [])]
                metadata = payload.get("field_metadata", {})
                metadata = metadata if isinstance(metadata, dict) else {}
                first = rows[0] if rows else {}
                language_pairs = self._language_pairs(columns)
                paired_raw = {name for pair in language_pairs.values() for name in (pair.get("RUS", ""), pair.get("ENG", "")) if name}

                # Raw SQL columns remain available for explicit mappings, even when they
                # participate in an automatic _RUS/_ENG pair.
                raw_column_info: dict[str, dict[str, Any]] = {}
                for column in columns:
                    meta = metadata.get(column, {}) if isinstance(metadata.get(column, {}), dict) else {}
                    value = first.get(column, "")
                    info = {
                        "key": column,
                        "label": str(meta.get("label") or self._pretty_label(column)),
                        "kind": str(meta.get("kind") or self._infer_kind(column, value)),
                        "description": str(meta.get("description") or ""),
                        "localized": False,
                    }
                    raw_column_info[column] = info
                    items.append({
                        "key": f"{source.code}.{column}", "label": info["label"],
                        "category": str(meta.get("category") or source.category or source.name), "value": value,
                        "kind": info["kind"], "source_code": source.code, "source_name": source.name,
                        "column": column, "row_index": 1, "description": info["description"] or source.description or "",
                        "language_explicit": True if column in paired_raw else False,
                    })

                # Visible table columns collapse paired Foo_RUS/Foo_ENG columns into one
                # virtual Foo column. A saved mapping points at source.Foo and is resolved
                # according to the current UI language every time it is applied.
                visible_columns: list[dict[str, Any]] = []
                seen_pair_bases: set[str] = set()
                for column in columns:
                    match = _LANG_SUFFIX_RE.match(column)
                    pair = language_pairs.get(match.group("base").casefold()) if match else None
                    if pair:
                        base = str(pair["base"])
                        canonical = base.casefold()
                        if canonical in seen_pair_bases:
                            continue
                        seen_pair_bases.add(canonical)
                        chosen_value, chosen_column = self._localized_value(first, pair, language)
                        base_meta = metadata.get(base, {}) if isinstance(metadata.get(base, {}), dict) else {}
                        chosen_meta = metadata.get(chosen_column, {}) if isinstance(metadata.get(chosen_column, {}), dict) else {}
                        meta = {**chosen_meta, **base_meta}
                        kind = str(meta.get("kind") or self._infer_kind(base, chosen_value))
                        label = str(meta.get("label") or self._pretty_label(base))
                        description = str(meta.get("description") or f"Автовыбор языка: {pair['RUS']} / {pair['ENG']}")
                        info = {
                            "key": base, "label": label, "kind": kind, "description": description,
                            "localized": True, "language": language,
                            "rus_column": pair["RUS"], "eng_column": pair["ENG"], "resolved_column": chosen_column,
                        }
                        visible_columns.append(info)
                        items.append({
                            "key": f"{source.code}.{base}", "label": label,
                            "category": str(meta.get("category") or source.category or source.name),
                            "value": chosen_value, "kind": kind, "source_code": source.code, "source_name": source.name,
                            "column": base, "row_index": 1, "description": description,
                            "localized": True, "language": language,
                            "rus_column": pair["RUS"], "eng_column": pair["ENG"], "resolved_column": chosen_column,
                        })
                    else:
                        visible_columns.append(raw_column_info[column])

                table_rows = []
                for row_index, row in enumerate(rows, start=1):
                    row_label = self._table_row_label(row, row_index)
                    # Add virtual localized values to the row payload used by the preview/search.
                    row_values = dict(row)
                    for pair in language_pairs.values():
                        base = str(pair["base"])
                        chosen_value, _ = self._localized_value(row, pair, language)
                        row_values[base] = chosen_value
                    table_rows.append({"index": row_index, "label": row_label, "values": row_values})

                    # Raw cell keys remain supported for explicit language selection.
                    for column in columns:
                        info = raw_column_info[column]
                        value = row.get(column, "")
                        items.append({
                            "key": f"{source.code}.row.{row_index}.{column}", "label": info["label"],
                            "category": f"{source.name} · {row_label}", "value": value, "kind": info["kind"],
                            "source_code": source.code, "source_name": source.name, "column": column,
                            "row_index": row_index, "row_label": row_label, "table_cell": True,
                            "description": info.get("description") or source.description or "",
                            "language_explicit": True if column in paired_raw else False,
                        })

                    # And expose one virtual cell for every paired language column.
                    for pair in language_pairs.values():
                        base = str(pair["base"])
                        chosen_value, chosen_column = self._localized_value(row, pair, language)
                        base_meta = metadata.get(base, {}) if isinstance(metadata.get(base, {}), dict) else {}
                        chosen_meta = metadata.get(chosen_column, {}) if isinstance(metadata.get(chosen_column, {}), dict) else {}
                        meta = {**chosen_meta, **base_meta}
                        kind = str(meta.get("kind") or self._infer_kind(base, chosen_value))
                        label = str(meta.get("label") or self._pretty_label(base))
                        description = str(meta.get("description") or f"Автовыбор языка: {pair['RUS']} / {pair['ENG']}")
                        items.append({
                            "key": f"{source.code}.row.{row_index}.{base}", "label": label,
                            "category": f"{source.name} · {row_label}", "value": chosen_value, "kind": kind,
                            "source_code": source.code, "source_name": source.name, "column": base,
                            "row_index": row_index, "row_label": row_label, "table_cell": True,
                            "description": description, "localized": True, "language": language,
                            "rus_column": pair["RUS"], "eng_column": pair["ENG"], "resolved_column": chosen_column,
                        })

                tables.append({
                    "id": source.id, "code": source.code, "name": source.name,
                    "category": source.category, "description": source.description,
                    "columns": visible_columns, "raw_columns": [raw_column_info[column] for column in columns],
                    "rows": table_rows, "row_count": len(table_rows), "ok": result.get("ok", False),
                    "skipped": result.get("skipped", False), "missing": result.get("missing", []),
                    "error": result.get("error", ""), "language": language,
                    "localized_pairs": [
                        {"base": pair["base"], "rus": pair["RUS"], "eng": pair["ENG"]}
                        for pair in language_pairs.values()
                    ],
                })
        game_label = context.get("game_id", "") or "Матч не выбран"
        return {
            "context": context, "variables": variables, "items": items, "tables": tables,
            "sources": source_results, "game_label": game_label, "language": language,
        }
