645 lines
30 KiB
Python
645 lines
30 KiB
Python
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,
|
||
}
|