from __future__ import annotations import re from collections.abc import Iterable, Mapping from typing import Any def _key(value: Any) -> str: return "".join(ch for ch in str(value or "").casefold() if ch.isalnum()) def _text(value: Any) -> str: if value is None: return "" if isinstance(value, bool): return "1" if value else "0" return str(value).strip() def _number(value: Any) -> float | None: if value is None or isinstance(value, bool): return None text = _text(value).replace(",", ".") if not text: return None try: return float(text) except (TypeError, ValueError): return None def _truthy(value: Any) -> bool: return _text(value).casefold() in {"1", "true", "yes", "y", "on", "goal", "гол"} def _mapping(value: Any) -> dict[str, Any]: return dict(value) if isinstance(value, Mapping) else {} def _lowered(row: Mapping[str, Any]) -> dict[str, Any]: return {_key(key): value for key, value in row.items()} def _first(row: Mapping[str, Any], aliases: Iterable[str]) -> Any: lowered = _lowered(row) for alias in aliases: normalized = _key(alias) if normalized in lowered and lowered[normalized] not in (None, ""): return lowered[normalized] return None def _identity(value: Any) -> tuple[str, str, str]: if isinstance(value, Mapping): value_map = _mapping(value) identifier = _text(_first(value_map, ("id", "external_id", "player_id", "team_id", "club_id"))) name = _text(_first(value_map, ("name", "fullname", "full_name", "title", "team", "player"))) number = _text(_first(value_map, ("number", "jersey", "jersey_number", "num", "pn"))) return identifier, name, number return "", _text(value), "" def _period_from_key(value: Any) -> str: text = _text(value).casefold() normalized = _key(text) if normalized.isdigit(): return str(int(normalized)) if normalized in {"ot", "overtime", "extra", "extratime", "ovt", "допвремя", "от"}: return "4" if normalized in {"so", "shootout", "bullits", "bullit", "буллиты", "б"}: return "5" for prefix in ("period", "per", "p", "time", "segment"): if normalized.startswith(prefix) and normalized[len(prefix):].isdigit(): return str(int(normalized[len(prefix):])) return "" def _side_from_key(value: Any) -> str: normalized = _key(value) if normalized in {"a", "home", "host", "left", "teama", "clubа", "хозяева"}: return "home" if normalized in {"b", "away", "guest", "visitor", "right", "teamb", "гости"}: return "away" return "" def _normalised_values(values: Iterable[Any]) -> set[str]: return {_key(value) for value in values if _text(value)} def _side_for_identity( *, explicit_side: str, team_id: str, team_name: str, home_ids: set[str], away_ids: set[str], home_names: set[str], away_names: set[str], ) -> str: if explicit_side in {"home", "away"}: return explicit_side candidate_id = _key(team_id) candidate_name = _key(team_name) if candidate_id and candidate_id in home_ids: return "home" if candidate_id and candidate_id in away_ids: return "away" if candidate_name and candidate_name in home_names: return "home" if candidate_name and candidate_name in away_names: return "away" return "" def _find_container(data: Mapping[str, Any], aliases: Iterable[str]) -> Any: lowered = _lowered(data) for alias in aliases: if _key(alias) in lowered: return lowered[_key(alias)] for wrapper in ("data", "result", "statistics", "stats", "match", "game_data"): nested = lowered.get(_key(wrapper)) if isinstance(nested, Mapping): found = _find_container(nested, aliases) if found is not None: return found return None def _looks_like_event(row: Mapping[str, Any]) -> bool: keys = set(_lowered(row)) type_keys = {_key(value) for value in ("event", "type", "event_type", "code", "action", "kind", "description", "text")} person_keys = {_key(value) for value in ("player", "player_id", "scorer", "author", "team", "team_id", "club", "pn", "pname")} return bool(keys & type_keys and (keys & person_keys or "time" in keys or "moscowtime" in keys)) def _looks_like_shot(row: Mapping[str, Any]) -> bool: keys = set(_lowered(row)) x_keys = {_key(value) for value in ("x", "coord_x", "x_coord", "pos_x", "point_x", "shot_x", "sx")} y_keys = {_key(value) for value in ("y", "coord_y", "y_coord", "pos_y", "point_y", "shot_y", "sy")} if keys & x_keys and keys & y_keys: return True coords = _text(_first(row, ("coords", "coordinate", "coordinates", "point", "position", "location"))) if re.fullmatch(r"\s*-?\d+(?:[.,]\d+)?\s*[xх,:;]\s*-?\d+(?:[.,]\d+)?\s*", coords, re.I): return True for nested_key in ("coordinate", "coordinates", "coords", "point", "position", "location"): nested = _first(row, (nested_key,)) if isinstance(nested, Mapping): nested_keys = set(_lowered(nested)) if nested_keys & x_keys and nested_keys & y_keys: return True if isinstance(nested, (list, tuple)) and len(nested) >= 2: return _number(nested[0]) is not None and _number(nested[1]) is not None return False def _flatten_rows( value: Any, *, kind: str, inherited_period: str = "", inherited_side: str = "", ) -> Iterable[tuple[dict[str, Any], str, str]]: checker = _looks_like_event if kind == "event" else _looks_like_shot if isinstance(value, list): for item in value: yield from _flatten_rows(item, kind=kind, inherited_period=inherited_period, inherited_side=inherited_side) return if not isinstance(value, Mapping): return row = _mapping(value) if checker(row): yield row, inherited_period, inherited_side return for key, child in row.items(): next_period = _period_from_key(key) or inherited_period next_side = _side_from_key(key) or inherited_side if isinstance(child, (Mapping, list)): yield from _flatten_rows(child, kind=kind, inherited_period=next_period, inherited_side=next_side) def _period_sort(value: str) -> tuple[int, str]: number = _number(value) return (int(number) if number is not None else 999, str(value)) def _game_seconds(value: Any) -> int | None: match = re.fullmatch(r"\s*(\d{1,3}):(\d{2})\s*", _text(value)) if not match: return None return int(match.group(1)) * 60 + int(match.group(2)) def _infer_period(event_time: str, code: str, status: str = "") -> str: normalized_status = _key(status) normalized_code = _key(code) if normalized_code == "bu" or "shootout" in normalized_status or "буллит" in normalized_status: return "5" seconds = _game_seconds(event_time) if seconds is None: return "" if seconds < 20 * 60: return "1" if seconds < 40 * 60: return "2" if seconds < 60 * 60: return "3" if seconds < 65 * 60: return "4" return "5" PENALTY_RU = { "tripping": "Подножка", "slashing": "Удар клюшкой", "boarding": "Толчок на борт", "roughing": "Грубость", "hooking": "Задержка соперника клюшкой", "holding": "Задержка соперника", "interference": "Блокировка", "too many players on the ice": "Нарушение численного состава", "delay of game": "Задержка игры", "high-sticking": "Игра высоко поднятой клюшкой", "cross-checking": "Толчок клюшкой", "charging": "Неправильная атака", "elbowing": "Удар локтем", } def _translated_penalty_reason(value: Any, language: str) -> str: text = _text(value) if language == "en" or not text: return text return PENALTY_RU.get(text.casefold(), text) def _translate_common_text(value: Any, language: str) -> str: text = _text(value) if language == "en" or not text: return text exact = { "Commercial break": "Рекламная пауза", "The last minute of the period": "Последняя минута периода", "The last 2 minutes of the third period": "Последние 2 минуты третьего периода", } if text in exact: return exact[text] replacements = [ (r"^Team (.+) is playing in full strength$", r"Команда \1 играет в полном составе"), (r"^Delayed penalty for (.+)$", r"Отложенный штраф у команды \1"), (r"^Icing\. A face-off in the defensive zone of (.+)$", r"Проброс. Вбрасывание в зоне защиты команды \1"), (r"^Offside against the (.+)\. A face-off in the neutral zone$", r"Положение «вне игры» у команды \1. Вбрасывание в средней зоне"), (r"^(\d+) on (\d+) play$", r"Игра \1 на \2"), ] for pattern, replacement in replacements: if re.match(pattern, text, re.I): return re.sub(pattern, replacement, text, flags=re.I) return text def _event_label(code: str, description: str, language: str) -> str: normalized = _key(code or description) labels = { "go": ("Гол", "Goal"), "goal": ("Гол", "Goal"), "g": ("Гол", "Goal"), "score": ("Гол", "Goal"), "pn": ("Удаление", "Penalty"), "penalty": ("Удаление", "Penalty"), "pen": ("Удаление", "Penalty"), "bu": ("Буллит", "Shootout attempt"), "shot": ("Бросок", "Shot"), "faceoff": ("Вбрасывание", "Faceoff"), "hit": ("Силовой приём", "Hit"), "to": ("Тайм-аут", "Timeout"), "timeout": ("Тайм-аут", "Timeout"), "ga": ("Ход матча", "Game state"), "periodstart": ("Начало периода", "Period start"), "periodend": ("Конец периода", "Period end"), "gc": ("Замена вратаря", "Goalie change"), "goaliechange": ("Замена вратаря", "Goalie change"), "co": ("Комментарий", "Comment"), "ce": ("Событие", "Event"), } if normalized in labels: return labels[normalized][1 if language == "en" else 0] return description or code or ("Событие" if language != "en" else "Event") def _period_title(period: str, language: str) -> str: if period == "0": return "игры" if language != "en" else "game" if period == "4": return "овертайма" if language != "en" else "overtime" if period == "5": return "серии буллитов" if language != "en" else "shootout" return f"{period} периода" if language != "en" else f"period {period}" def _event_content(row: Mapping[str, Any], code: str, period: str, language: str) -> tuple[str, str, str]: normalized = _key(code) team = _text(_first(row, ("team_name", "club_name", "team_title", "team"))) player_number = _text(_first(row, ("pn", "player_number", "number", "jersey"))) player_name = _text(_first(row, ("pname", "player_name", "fullname", "full_name", "scorer_name"))) player = " ".join(part for part in ((f"№{player_number}" if player_number else ""), player_name) if part) if normalized in {"go", "goal", "score"}: status = _text(_first(row, ("status", "pmStatus", "strength"))) assists = [] for index in (1, 2): number = _text(_first(row, (f"assist{index}pn", f"assist{index}_number"))) name = _text(_first(row, (f"assist{index}pname", f"assist{index}_name"))) if number or name: assists.append(" ".join(part for part in ((f"№{number}" if number else ""), name) if part)) title = "Гол" if language != "en" else "Goal" description = " · ".join(part for part in (team, player, status) if part) if assists: description += (" · ассистенты: " if language != "en" else " · assists: ") + ", ".join(assists) return title, description, "goal" if normalized in {"pn", "penalty", "pen"}: minutes = _text(_first(row, ("penalty_time", "minutes", "duration"))) reason = _translated_penalty_reason(_first(row, ("penalty_reason", "reason", "description")), language) team_penalty = player_number in {"", "0"} or _key(player_name) in {"teampenalty", "командныйштраф"} title = "Командное удаление" if team_penalty and language != "en" else "Team penalty" if team_penalty else "Удаление" if language != "en" else "Penalty" description = " · ".join(part for part in (team, "" if team_penalty else player, f"{minutes} мин" if minutes and language != "en" else f"{minutes} min" if minutes else "", reason) if part) return title, description, "penalty" if normalized == "bu": skater = _text(_first(row, ("skater", "player", "pname"))) goalie = _text(_first(row, ("goalie", "goalkeeper"))) result = _text(_first(row, ("result", "status"))) if language != "en": result = {"Goal": "Гол", "Saved": "Отражён", "Missed": "Мимо"}.get(result, result) title = "Послематчевый буллит" if language != "en" else "Shootout attempt" description = " · ".join(part for part in (team, skater, goalie, result) if part) return title, description, "shootout" if normalized == "ga": status = _key(_first(row, ("status",))) start = _text(_first(row, ("key",))) in {"0", "start"} or status.startswith("start") title = ("Начало " if start else "Окончание ") + _period_title(period, language) if language == "en": title = ("Start of " if start else "End of ") + _period_title(period, language) return title, "", "period" if normalized == "gc": goalie_a = _text(_first(row, ("gkA", "goalieA"))) goalie_b = _text(_first(row, ("gkB", "goalieB"))) title = "Изменение вратаря" if language != "en" else "Goalie change" description = " · ".join(part for part in (goalie_a, goalie_b) if part) or team return title, description, "goalie" if normalized == "to": return ("Тайм-аут" if language != "en" else "Timeout"), team, "timeout" raw_text = _first(row, ("description", "desc", "text", "title", "event_name", "name")) description = _translate_common_text(raw_text, language) if normalized == "co": return ("Комментарий" if language != "en" else "Comment"), description, "comment" return _event_label(code, description, language), description, "info" def _period_from_event_text(value: Any) -> str: text = _text(value).casefold() if "овертайм" in text or "overtime" in text: return "4" if "буллит" in text or "shootout" in text: return "5" match = re.search(r"(?:статистика|statistics).*?([123])(?:-|\s)*(?:го|st|nd|rd|th)?\s*(?:период|period)", text, re.I) return match.group(1) if match else "" def parse_match_events( data: Mapping[str, Any], *, language: str = "ru", home_ids: Iterable[Any] = (), away_ids: Iterable[Any] = (), home_names: Iterable[Any] = (), away_names: Iterable[Any] = (), ) -> dict[str, Any]: container = _find_container(data, ("events", "game_events", "match_events", "timeline", "play_by_play", "playbyplay")) if container is None: return {"available": False, "segments": [], "types": [], "items": []} home_id_keys = _normalised_values(home_ids) away_id_keys = _normalised_values(away_ids) home_name_keys = _normalised_values(home_names) away_name_keys = _normalised_values(away_names) items: list[dict[str, Any]] = [] seen: set[tuple[str, ...]] = set() context_period = "" for source_index, (row, inherited_period, inherited_side) in enumerate(_flatten_rows(container, kind="event")): event_time = _text(_first(row, ("time", "clock", "event_time", "game_time", "tm", "timecode"))) # Stat2TV records may contain both `code="pn"` and `type="Penalty"`. # The compact event code is authoritative; choosing `type` first caused # real penalties to be normalised as generic/other events in some feeds. explicit_code = _text(_first(row, ("code", "event_code", "action_code", "kind_code"))) raw_type = _text(_first(row, ("type", "event_type", "event", "action", "kind", "action_type"))) code = explicit_code or raw_type status = _text(_first(row, ("status", "pmStatus"))) period = _text(_first(row, ("period", "period_id", "period_number", "num_period", "p"))) or inherited_period period_text = _first(row, ("description", "desc", "text", "title", "event_name", "name")) text_period = _period_from_event_text(period_text) if not text_period and _key(code) == "co" and "статистика матча" in _text(period_text).casefold(): text_period = "0" inherited_context = "" if _key(code) == "co" else context_period period = _period_from_key(period) or period or text_period or _infer_period(event_time, code, status) or inherited_context if period and _key(code) != "co": context_period = period score = _text(_first(row, ("score", "score_value", "current_score"))) team_value = _first(row, ("team", "club", "team_info", "club_info")) team_id, team_name, _ = _identity(team_value) team_id = team_id or _text(_first(row, ("team_id", "id_team", "club_id", "id_club"))) team_name = team_name or _text(_first(row, ("team_name", "club_name", "team_title"))) player_value = _first(row, ("player", "athlete", "scorer", "author", "participant")) player_id, player_name, player_number = _identity(player_value) player_id = player_id or _text(_first(row, ("player_id", "id_player", "athlete_id", "scorer_id"))) player_name = player_name or _text(_first(row, ("pname", "player_name", "fullname", "full_name", "scorer_name", "author_name"))) player_number = player_number or _text(_first(row, ("pn", "player_number", "number", "jersey", "jersey_number"))) explicit_side = _side_from_key(_first(row, ("side", "team_side", "team"))) or inherited_side side = _side_for_identity( explicit_side=explicit_side, team_id=team_id, team_name=team_name, home_ids=home_id_keys, away_ids=away_id_keys, home_names=home_name_keys, away_names=away_name_keys, ) title, description, category = _event_content(row, code, period, language) penalty_reason = "" penalty_minutes = "" team_penalty = False if category == "penalty": penalty_reason = _translated_penalty_reason( _first(row, ("penalty_reason", "reason", "description")), language, ) penalty_minutes = _text(_first(row, ("penalty_time", "minutes", "duration"))) team_penalty = player_number in {"", "0"} or _key(player_name) in { "teampenalty", "командныйштраф", } if not any((event_time, code, title, description, player_name, team_name, score)): continue signature = (period, event_time, code, description, team_id, team_name, player_id, player_name, player_number, score) if signature in seen: continue seen.add(signature) seconds = _game_seconds(event_time) if _key(code) == "ga": start = _text(_first(row, ("key",))) in {"0", "start"} or _key(status).startswith("start") starts = {"0": -1, "1": 0, "2": 1200, "3": 2400, "4": 3600, "5": 3900} ends = {"0": 4100, "1": 1200, "2": 2400, "3": 3600, "4": 3900, "5": 4000} seconds = (starts if start else ends).get(period, seconds if seconds is not None else -1) if seconds is None: boundary = {"1": 1200, "2": 2400, "3": 3600, "4": 3900, "5": 4000, "0": 4100}.get(period, -1) seconds = boundary - source_index / 1000 if boundary >= 0 else -source_index items.append({ "id": _text(_first(row, ("id", "event_id", "external_id"))) or str(len(items) + 1), "period": period, "time": event_time, "moscow_time": _text(_first(row, ("moscowtime", "created_at", "timestamp"))), "code": code, "type": raw_type or code, "category": category, "label": _event_label(code, description, language), "title": title, "description": description, "score": score, "side": side, "team_id": team_id, "team_name": team_name, "player_id": player_id, "player_name": player_name, "player_number": player_number, "penalty_reason": penalty_reason, "penalty_minutes": penalty_minutes, "team_penalty": team_penalty, "sort_seconds": seconds, "source_index": source_index, }) if not items: return {"available": False, "segments": [], "types": [], "items": []} periods = sorted({item["period"] for item in items if item["period"] not in {"", "0"}}, key=_period_sort) categories = [category for category in ("goal", "penalty", "shootout", "period", "timeout", "goalie", "comment", "info") if any(item["category"] == category for item in items)] return { "available": True, "segments": ["all", *periods], "types": ["all", *categories], "items": items, "count": len(items), } def _shot_coordinates(row: Mapping[str, Any]) -> tuple[float | None, float | None, str]: x = _number(_first(row, ("x", "coord_x", "x_coord", "pos_x", "point_x", "shot_x", "sx"))) y = _number(_first(row, ("y", "coord_y", "y_coord", "pos_y", "point_y", "shot_y", "sy"))) if x is not None and y is not None: return x, y, "xy" for nested_key in ("coordinate", "coordinates", "coords", "point", "position", "location"): nested = _first(row, (nested_key,)) if isinstance(nested, Mapping): x = _number(_first(nested, ("x", "coord_x", "x_coord", "pos_x"))) y = _number(_first(nested, ("y", "coord_y", "y_coord", "pos_y"))) if x is not None and y is not None: return x, y, "xy" if isinstance(nested, (list, tuple)) and len(nested) >= 2: x = _number(nested[0]) y = _number(nested[1]) if x is not None and y is not None: return x, y, "array" if isinstance(nested, str): match = re.fullmatch(r"\s*(-?\d+(?:[.,]\d+)?)\s*[xх,:;]\s*(-?\d+(?:[.,]\d+)?)\s*", nested, re.I) if match: return float(match.group(1).replace(",", ".")), float(match.group(2).replace(",", ".")), "khl1000x500" return None, None, "" def parse_shots_map( data: Mapping[str, Any], *, language: str = "ru", home_ids: Iterable[Any] = (), away_ids: Iterable[Any] = (), home_names: Iterable[Any] = (), away_names: Iterable[Any] = (), ) -> dict[str, Any]: container = _find_container(data, ("shots_map", "shotsmap", "shot_map", "shotmap", "shot_coordinates", "shotcoordinates")) if container is None: return {"available": False, "segments": [], "players": [], "items": []} home_id_keys = _normalised_values(home_ids) away_id_keys = _normalised_values(away_ids) home_name_keys = _normalised_values(home_names) away_name_keys = _normalised_values(away_names) items: list[dict[str, Any]] = [] seen: set[tuple[Any, ...]] = set() for row, inherited_period, inherited_side in _flatten_rows(container, kind="shot"): x, y, coordinate_system = _shot_coordinates(row) if x is None or y is None: continue period = _text(_first(row, ("period", "period_id", "period_number", "num_period", "p"))) or inherited_period period = _period_from_key(period) or period event_time = _text(_first(row, ("time", "clock", "event_time", "game_time", "tm"))) stime = _number(_first(row, ("stime", "seconds", "game_seconds"))) team_value = _first(row, ("team", "club", "team_info", "club_info")) team_id, team_name, _ = _identity(team_value) team_id = team_id or _text(_first(row, ("team_id", "id_team", "club_id", "id_club"))) team_name = team_name or _text(_first(row, ("team_name", "club_name", "team_title"))) player_value = _first(row, ("player", "athlete", "shooter", "author")) player_id, player_name, player_number = _identity(player_value) player_id = player_id or _text(_first(row, ("player_id", "id_player", "athlete_id", "shooter_id"))) player_name = player_name or _text(_first(row, ("player_name", "fullname", "full_name", "shooter_name", "author_name"))) player_number = player_number or _text(_first(row, ("pn", "player_number", "number", "jersey", "jersey_number"))) explicit_side = _side_from_key(_first(row, ("side", "team_side", "team"))) or inherited_side side = _side_for_identity( explicit_side=explicit_side, team_id=team_id, team_name=team_name, home_ids=home_id_keys, away_ids=away_id_keys, home_names=home_name_keys, away_names=away_name_keys, ) result_raw = _first(row, ("result", "shot_result", "outcome", "event", "type", "shot_type", "status")) result = _text(result_raw) numeric_result = _number(result_raw) normalized_result = _key(result) goal = numeric_result == 1 or _truthy(_first(row, ("goal", "is_goal", "scored", "g"))) or any(marker in normalized_result for marker in ("goal", "score", "гол")) on_target_value = _first(row, ("on_target", "is_on_target", "sog", "target")) on_target = numeric_result in {1, 2} or (_truthy(on_target_value) if on_target_value not in (None, "") else (goal or any(marker in normalized_result for marker in ("save", "saved", "ontarget", "sog", "вствор")))) blocked = numeric_result == 4 or any(marker in normalized_result for marker in ("block", "blocked", "блок")) missed = numeric_result == 3 or any(marker in normalized_result for marker in ("miss", "wide", "мимо")) if goal: result_label = "Goal" if language == "en" else "Гол" elif blocked: result_label = "Blocked" if language == "en" else "Заблокирован" elif missed: result_label = "Missed" if language == "en" else "Мимо" elif on_target: result_label = "On target" if language == "en" else "В створ" else: result_label = result or ("Shot" if language == "en" else "Бросок") signature = (period, event_time, round(x, 5), round(y, 5), side, player_number, result) if signature in seen: continue seen.add(signature) items.append({ "id": _text(_first(row, ("id", "shot_id", "event_id", "external_id"))) or str(len(items) + 1), "period": period, "time": event_time, "sort_seconds": int(stime) if stime is not None else (_game_seconds(event_time) or 0), "x": x, "y": y, "coordinate_system": coordinate_system, "rink_width": 1000 if coordinate_system == "khl1000x500" else None, "rink_height": 500 if coordinate_system == "khl1000x500" else None, "attacking_left": _text(_first(row, ("leftAB", "left_team", "attack_left"))), "side": side, "team_id": team_id, "team_name": team_name, "player_id": player_id, "player_name": player_name, "player_number": player_number, "result": result, "result_code": int(numeric_result) if numeric_result is not None else None, "result_label": result_label, "goal": goal, "on_target": on_target, "blocked": blocked, "missed": missed, }) if not items: return {"available": False, "segments": [], "players": [], "items": []} periods = sorted({item["period"] for item in items if item["period"]}, key=_period_sort) player_map: dict[str, dict[str, Any]] = {} for item in items: identifier = item["player_id"] or f"{item['side']}:{item['player_number']}" or _key(item["player_name"]) if not identifier: continue player_map.setdefault(identifier, { "id": item["player_id"], "name": item["player_name"], "number": item["player_number"], "side": item["side"], "team_id": item["team_id"], "team_name": item["team_name"], }) players = sorted(player_map.values(), key=lambda item: (0 if item.get("side") == "home" else 1, _key(item.get("name")), _key(item.get("number")))) return { "available": True, "segments": ["all", *periods], "players": players, "items": items, "count": len(items), "coordinate_system": "khl1000x500" if any(item.get("coordinate_system") == "khl1000x500" for item in items) else "auto", "rink_width": 1000, "rink_height": 500, }