561 lines
20 KiB
Python
561 lines
20 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from collections import OrderedDict, defaultdict
|
||
from typing import Any
|
||
from xml.etree import ElementTree as ET
|
||
|
||
from .match_parser import local_name
|
||
|
||
|
||
_SYSTEM_KEYS = {
|
||
"id",
|
||
"external_id",
|
||
"idtournament",
|
||
"id_tournament",
|
||
"tournament_id",
|
||
"tournamentid",
|
||
"gendate",
|
||
"gen_date",
|
||
"generated_at",
|
||
"date",
|
||
"language",
|
||
"lang",
|
||
"type",
|
||
"category",
|
||
"category_id",
|
||
}
|
||
|
||
_NAME_KEYS_RU = (
|
||
"name",
|
||
"fullname",
|
||
"full_name",
|
||
"player",
|
||
"player_name",
|
||
"fio",
|
||
"title",
|
||
)
|
||
_NAME_KEYS_EN = (
|
||
"nameen",
|
||
"name_en",
|
||
"fullnameen",
|
||
"fullname_en",
|
||
"playeren",
|
||
"player_en",
|
||
"titleen",
|
||
"title_en",
|
||
)
|
||
_TEAM_KEYS_RU = (
|
||
"team",
|
||
"team_name",
|
||
"teamname",
|
||
"team_title",
|
||
"teamtitle",
|
||
"team_short_name",
|
||
"teamshortname",
|
||
"team_name_ru",
|
||
"team_name_rus",
|
||
"team_ru",
|
||
"club",
|
||
"club_name",
|
||
"clubname",
|
||
"club_title",
|
||
"clubtitle",
|
||
"club_short_name",
|
||
"clubshortname",
|
||
"club_name_ru",
|
||
"club_name_rus",
|
||
"club_ru",
|
||
)
|
||
_TEAM_KEYS_EN = (
|
||
"teamen",
|
||
"team_en",
|
||
"team_name_en",
|
||
"teamnameen",
|
||
"team_title_en",
|
||
"teamtitleen",
|
||
"team_short_name_en",
|
||
"teamshortnameen",
|
||
"cluben",
|
||
"club_en",
|
||
"club_name_en",
|
||
"clubnameen",
|
||
"club_title_en",
|
||
"clubtitleen",
|
||
"club_short_name_en",
|
||
"clubshortnameen",
|
||
)
|
||
_TEAM_ID_KEYS = (
|
||
"team_id",
|
||
"teamid",
|
||
"club_id",
|
||
"clubid",
|
||
"team_entry_id",
|
||
"teamentryid",
|
||
"club_entry_id",
|
||
"clubentryid",
|
||
"entry_id",
|
||
"entryid",
|
||
)
|
||
_PLAYER_ID_KEYS = (
|
||
"player_id",
|
||
"playerid",
|
||
"person_id",
|
||
"personid",
|
||
"athlete_id",
|
||
"athleteid",
|
||
)
|
||
_RANK_KEYS = ("rank", "position", "place", "pos")
|
||
_NUMBER_KEYS = ("number", "num", "jersey", "jersey_number", "shirt_number")
|
||
_PERIOD_KEYS_RU = (
|
||
"period",
|
||
"period_ru",
|
||
"period_name",
|
||
"periodname",
|
||
"stage",
|
||
"stage_name",
|
||
)
|
||
_PERIOD_KEYS_EN = (
|
||
"perioden",
|
||
"period_en",
|
||
"period_name_en",
|
||
"periodnameen",
|
||
"stageen",
|
||
"stage_en",
|
||
"stage_name_en",
|
||
)
|
||
|
||
_LABELS: dict[str, tuple[str, str]] = {
|
||
"rank": ("Место", "Rank"),
|
||
"position": ("Место", "Position"),
|
||
"place": ("Место", "Place"),
|
||
"pos": ("Место", "Position"),
|
||
"number": ("№", "No."),
|
||
"num": ("№", "No."),
|
||
"jersey": ("№", "No."),
|
||
"jersey_number": ("№", "No."),
|
||
"gp": ("И", "GP"),
|
||
"games": ("И", "GP"),
|
||
"games_played": ("И", "GP"),
|
||
"g": ("Г", "G"),
|
||
"goals": ("Голы", "Goals"),
|
||
"a": ("П", "A"),
|
||
"assists": ("Передачи", "Assists"),
|
||
"pts": ("О", "PTS"),
|
||
"points": ("Очки", "Points"),
|
||
"plus_minus": ("+/-", "+/-"),
|
||
"plusminus": ("+/-", "+/-"),
|
||
"pim": ("Штр", "PIM"),
|
||
"penalty_minutes": ("Штраф", "PIM"),
|
||
"ppg": ("ГБ", "PPG"),
|
||
"power_play_goals": ("Голы в большинстве", "Power-play goals"),
|
||
"ppa": ("ПБ", "PPA"),
|
||
"power_play_assists": ("Передачи в большинстве", "Power-play assists"),
|
||
"ppp": ("ОБ", "PPP"),
|
||
"power_play_points": ("Очки в большинстве", "Power-play points"),
|
||
"ppo": ("Попытки", "Opportunities"),
|
||
"power_play_opportunities": ("Большинств", "Power-play opportunities"),
|
||
"pp_pct": ("Реализация, %", "Power play, %"),
|
||
"pp_percent": ("Реализация, %", "Power play, %"),
|
||
"power_play_pct": ("Реализация, %", "Power play, %"),
|
||
"power_play_percent": ("Реализация, %", "Power play, %"),
|
||
"shg": ("ГМ", "SHG"),
|
||
"short_handed_goals": ("Голы в меньшинстве", "Short-handed goals"),
|
||
"shga": ("Пропущено в меньшинстве", "SHGA"),
|
||
"pk": ("Нейтрализации", "Penalty kills"),
|
||
"pko": ("Меньшинств", "Times short-handed"),
|
||
"pk_pct": ("Надёжность меньшинства, %", "Penalty kill, %"),
|
||
"penalty_kill_pct": ("Надёжность меньшинства, %", "Penalty kill, %"),
|
||
"toi": ("Время", "TOI"),
|
||
"time_on_ice": ("Время на льду", "Time on ice"),
|
||
"avg_toi": ("Среднее время", "Average TOI"),
|
||
"shots": ("Броски", "Shots"),
|
||
"sog": ("Броски в створ", "SOG"),
|
||
"wins": ("Победы", "Wins"),
|
||
"losses": ("Поражения", "Losses"),
|
||
"save_pct": ("% отражённых", "Save %"),
|
||
"gaa": ("КН", "GAA"),
|
||
}
|
||
|
||
|
||
def _text(value: Any) -> str:
|
||
return str(value or "").strip()
|
||
|
||
|
||
def _key(value: Any) -> str:
|
||
text = local_name(str(value or "")).strip()
|
||
text = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", text)
|
||
text = re.sub(r"[^A-Za-zА-Яа-я0-9]+", "_", text)
|
||
return text.strip("_").lower()
|
||
|
||
|
||
def _element_fields(node: ET.Element) -> OrderedDict[str, str]:
|
||
result: OrderedDict[str, str] = OrderedDict()
|
||
for raw_key, raw_value in node.attrib.items():
|
||
key = _key(raw_key)
|
||
if key:
|
||
result[key] = _text(raw_value)
|
||
|
||
# Keep parent context for nested identity objects such as
|
||
# <Player><Team><Id>...</Id><Name>...</Name></Team></Player>.
|
||
# Without this, the team's generic Id/Name would overwrite the player's.
|
||
parent_by_id: dict[int, ET.Element] = {}
|
||
for parent in node.iter():
|
||
for child in list(parent):
|
||
parent_by_id[id(child)] = parent
|
||
|
||
identity_containers = {"team", "club", "player", "person", "athlete"}
|
||
for child in node.iter():
|
||
if child is node or list(child):
|
||
continue
|
||
child_attrs = {_key(k): _text(v) for k, v in child.attrib.items()}
|
||
field_name = child_attrs.get("name") or child_attrs.get("key")
|
||
field_value = child_attrs.get("value")
|
||
if field_name and field_value not in (None, ""):
|
||
result[_key(field_name)] = field_value
|
||
continue
|
||
|
||
value = _text(child.text) or child_attrs.get("value", "")
|
||
key = _key(child.tag)
|
||
if not key or not value:
|
||
continue
|
||
|
||
parent = parent_by_id.get(id(child))
|
||
parent_key = _key(parent.tag) if parent is not None else ""
|
||
if parent_key in identity_containers and key in {
|
||
"id", "external_id", "name", "name_en", "title", "title_en",
|
||
"short_name", "short_name_en", "number",
|
||
}:
|
||
# Direct player fields keep their ordinary keys. Nested team/club
|
||
# fields receive a prefix so they cannot overwrite player identity.
|
||
if parent is node and parent_key in {"player", "person", "athlete"}:
|
||
result.setdefault(key, value)
|
||
else:
|
||
result[f"{parent_key}_{key}"] = value
|
||
continue
|
||
|
||
# First scalar wins. This is important when different nested objects use
|
||
# the same generic tag names.
|
||
result.setdefault(key, value)
|
||
return result
|
||
|
||
|
||
def _first(fields: dict[str, str], keys: tuple[str, ...]) -> str:
|
||
for key in keys:
|
||
value = _text(fields.get(key))
|
||
if value:
|
||
return value
|
||
return ""
|
||
|
||
|
||
def _section_label(node: ET.Element, language: str, fallback: str) -> tuple[str, dict[str, str]]:
|
||
fields = {_key(key): _text(value) for key, value in node.attrib.items()}
|
||
ru = _first(fields, ("name", "title", "label", "category", "type"))
|
||
en = _first(fields, ("nameen", "name_en", "titleen", "title_en", "labelen", "label_en"))
|
||
tag = local_name(node.tag)
|
||
fallback_ru = {
|
||
"powerplay": "Большинство",
|
||
"rank": "Рейтинг",
|
||
"ranking": "Рейтинг",
|
||
"players": "Игроки",
|
||
"teams": "Команды",
|
||
}.get(_key(tag), fallback)
|
||
fallback_en = {
|
||
"powerplay": "Power play",
|
||
"rank": "Ranking",
|
||
"ranking": "Ranking",
|
||
"players": "Players",
|
||
"teams": "Teams",
|
||
}.get(_key(tag), fallback)
|
||
ru = ru or fallback_ru
|
||
en = en or fallback_en
|
||
return (en if language == "en" and en else ru or en, {"ru": ru, "en": en})
|
||
|
||
|
||
def _section_period(node: ET.Element) -> dict[str, str]:
|
||
fields = {_key(key): _text(value) for key, value in node.attrib.items()}
|
||
period_ru = _first(fields, _PERIOD_KEYS_RU)
|
||
period_en = _first(fields, _PERIOD_KEYS_EN)
|
||
# Most Stat2TV feeds expose only one language-neutral `period` value.
|
||
period_ru = period_ru or period_en
|
||
period_en = period_en or period_ru
|
||
return {"ru": period_ru, "en": period_en}
|
||
|
||
|
||
def _label_with_period(label: str, period: str, language: str) -> str:
|
||
label = _text(label)
|
||
period = _text(period)
|
||
if not period:
|
||
return label
|
||
|
||
# Do not repeat the period when the feed already included it in `name`.
|
||
if _key(period) and _key(period) in _key(label):
|
||
return label
|
||
|
||
period_word = "period" if language == "en" else "период"
|
||
return f"{label} · {period_word} {period}".strip()
|
||
|
||
|
||
def _leaf_row_candidates(root: ET.Element) -> list[tuple[ET.Element, ET.Element, OrderedDict[str, str]]]:
|
||
parent_by_id: dict[int, ET.Element] = {}
|
||
for parent in root.iter():
|
||
for child in list(parent):
|
||
parent_by_id[id(child)] = parent
|
||
|
||
candidates: list[tuple[ET.Element, ET.Element, OrderedDict[str, str]]] = []
|
||
technical_tags = {"stat", "field", "value", "param", "parameter", "property"}
|
||
technical_keys = {"name", "key", "label", "value"}
|
||
for node in root.iter():
|
||
if node is root:
|
||
continue
|
||
parent_node = parent_by_id.get(id(node))
|
||
node_tag = _key(node.tag)
|
||
parent_tag = _key(parent_node.tag) if parent_node is not None else ""
|
||
if node_tag in {"team", "club"} and parent_tag in {"player", "person", "athlete"}:
|
||
continue
|
||
fields = _element_fields(node)
|
||
if len(fields) < 2:
|
||
continue
|
||
|
||
# Nested <Stat name="GP" value="10"/> nodes describe a parent row and
|
||
# must not be rendered as standalone records.
|
||
if _key(node.tag) in technical_tags and set(fields).issubset(technical_keys):
|
||
continue
|
||
|
||
nested_data_rows = 0
|
||
for child in list(node):
|
||
child_fields = _element_fields(child)
|
||
child_tag = _key(child.tag)
|
||
if child_tag in technical_tags and set(child_fields).issubset(technical_keys):
|
||
continue
|
||
if _key(node.tag) in {"player", "person", "athlete"} and child_tag in {
|
||
"team", "club", "stats", "statistics",
|
||
}:
|
||
continue
|
||
if len(child_fields) >= 2 and (
|
||
_first(child_fields, _NAME_KEYS_RU + _NAME_KEYS_EN + _TEAM_KEYS_RU + _TEAM_KEYS_EN)
|
||
or _first(child_fields, _RANK_KEYS)
|
||
):
|
||
nested_data_rows += 1
|
||
if nested_data_rows:
|
||
continue
|
||
parent = parent_by_id.get(id(node), root)
|
||
candidates.append((parent, node, fields))
|
||
return candidates
|
||
|
||
|
||
def _column_label(key: str, language: str) -> str:
|
||
labels = _LABELS.get(key)
|
||
if labels:
|
||
return labels[1] if language == "en" else labels[0]
|
||
return key.replace("_", " ").strip().title()
|
||
|
||
|
||
def _column_format(key: str) -> str:
|
||
if "pct" in key or "percent" in key:
|
||
return "percent"
|
||
if key in {"toi", "avg_toi", "time", "time_on_ice"} or "time" in key:
|
||
return "time"
|
||
return "text"
|
||
|
||
|
||
def parse_tournament_statistics_xml(
|
||
content: bytes | str,
|
||
*,
|
||
resource_type: str,
|
||
language: str = "ru",
|
||
) -> dict[str, Any]:
|
||
raw = content.encode("utf-8") if isinstance(content, str) else content
|
||
root = ET.fromstring(raw)
|
||
kind = _key(resource_type)
|
||
root_fields = {_key(key): _text(value) for key, value in root.attrib.items()}
|
||
|
||
grouped: dict[int, list[tuple[ET.Element, OrderedDict[str, str]]]] = defaultdict(list)
|
||
parents: dict[int, ET.Element] = {}
|
||
for parent, node, fields in _leaf_row_candidates(root):
|
||
grouped[id(parent)].append((node, fields))
|
||
parents[id(parent)] = parent
|
||
|
||
sections: list[dict[str, Any]] = []
|
||
seen_signatures: set[tuple[str, ...]] = set()
|
||
used_section_ids: set[str] = set()
|
||
for section_index, (parent_id, entries) in enumerate(grouped.items(), start=1):
|
||
parent = parents[parent_id]
|
||
row_payloads: list[dict[str, Any]] = []
|
||
column_order: OrderedDict[str, None] = OrderedDict()
|
||
|
||
for row_index, (node, fields) in enumerate(entries, start=1):
|
||
name_ru = _first(fields, _NAME_KEYS_RU)
|
||
name_en = _first(fields, _NAME_KEYS_EN)
|
||
team_ru = _first(fields, _TEAM_KEYS_RU)
|
||
team_en = _first(fields, _TEAM_KEYS_EN)
|
||
|
||
# In team-based XML the generic name attribute is the team name.
|
||
if kind == "powerplay" and not team_ru and not team_en:
|
||
team_ru, team_en = name_ru, name_en
|
||
name_ru = name_en = ""
|
||
|
||
identity_keys = set(
|
||
_NAME_KEYS_RU + _NAME_KEYS_EN + _TEAM_KEYS_RU + _TEAM_KEYS_EN
|
||
+ _TEAM_ID_KEYS + _PLAYER_ID_KEYS + _RANK_KEYS + _NUMBER_KEYS
|
||
)
|
||
values: OrderedDict[str, str] = OrderedDict()
|
||
for key, value in fields.items():
|
||
if key in identity_keys or key in _SYSTEM_KEYS:
|
||
continue
|
||
if value == "":
|
||
continue
|
||
values[key] = value
|
||
column_order.setdefault(key, None)
|
||
|
||
rank = _first(fields, _RANK_KEYS)
|
||
number = _first(fields, _NUMBER_KEYS)
|
||
team_id = _first(fields, _TEAM_ID_KEYS)
|
||
player_id = _first(fields, _PLAYER_ID_KEYS)
|
||
display_name = (
|
||
team_en if language == "en" and team_en else team_ru or team_en
|
||
) if kind == "powerplay" else (
|
||
name_en if language == "en" and name_en else name_ru or name_en
|
||
)
|
||
display_team = team_en if language == "en" and team_en else team_ru or team_en
|
||
|
||
# Keep unknown but useful rows. Completely empty technical nodes are ignored.
|
||
if not (display_name or display_team or rank or number or values):
|
||
continue
|
||
row_payloads.append(
|
||
{
|
||
"id": _first(fields, ("id", "external_id")) or player_id or f"row-{row_index}",
|
||
"player_id": player_id or (
|
||
_first(fields, ("id", "external_id")) if kind == "rank" else ""
|
||
),
|
||
"team_id": team_id,
|
||
"rank": rank,
|
||
"number": number,
|
||
"name": display_name,
|
||
"names": {"ru": name_ru, "en": name_en},
|
||
"team": display_team,
|
||
"team_names": {"ru": team_ru, "en": team_en},
|
||
"values": dict(values),
|
||
"source_tag": local_name(node.tag),
|
||
}
|
||
)
|
||
|
||
if not row_payloads:
|
||
continue
|
||
signature = tuple(
|
||
sorted(
|
||
f"{row.get('id')}|{row.get('rank')}|{row.get('name')}|{row.get('team')}"
|
||
for row in row_payloads
|
||
)
|
||
)
|
||
if signature in seen_signatures:
|
||
continue
|
||
seen_signatures.add(signature)
|
||
|
||
fallback = "Power play" if kind == "powerplay" else "Ranking"
|
||
label, names = _section_label(parent, language, fallback)
|
||
period_names = _section_period(parent)
|
||
if kind == "rank" and (period_names["ru"] or period_names["en"]):
|
||
names = {
|
||
"ru": _label_with_period(names.get("ru", ""), period_names["ru"], "ru"),
|
||
"en": _label_with_period(names.get("en", ""), period_names["en"], "en"),
|
||
}
|
||
label = names["en"] if language == "en" else names["ru"]
|
||
|
||
# Some Stat2TV feeds expose categories as repeated nodes without an id,
|
||
# for example <Category name="PP">, <Category name="GF">. Using only
|
||
# the parent tag made every UI tab receive the same id ("category"), so
|
||
# all tabs looked active and clicks could not switch the table. Prefer
|
||
# the XML id, then the category label, and always make the result unique.
|
||
base_section_id = (
|
||
_text(parent.attrib.get("id"))
|
||
or _key(label)
|
||
or _key(parent.tag)
|
||
or f"section-{section_index}"
|
||
)
|
||
section_id = base_section_id
|
||
duplicate_index = 2
|
||
while section_id in used_section_ids:
|
||
section_id = f"{base_section_id}-{duplicate_index}"
|
||
duplicate_index += 1
|
||
used_section_ids.add(section_id)
|
||
|
||
columns = [
|
||
{
|
||
"key": key,
|
||
"label": _column_label(key, language),
|
||
"format": _column_format(key),
|
||
}
|
||
for key in column_order
|
||
]
|
||
sections.append(
|
||
{
|
||
"id": section_id,
|
||
"label": label,
|
||
"names": names,
|
||
"period": period_names["en"] if language == "en" else period_names["ru"],
|
||
"periods": period_names,
|
||
"rows": row_payloads,
|
||
"columns": columns,
|
||
}
|
||
)
|
||
|
||
# Some feeds use the root itself as a single record. Preserve it as a fallback.
|
||
if not sections:
|
||
root_values = _element_fields(root)
|
||
if len(root_values) >= 2:
|
||
values = {
|
||
key: value for key, value in root_values.items()
|
||
if key not in _SYSTEM_KEYS and value != ""
|
||
}
|
||
if values:
|
||
fallback = "Power play" if kind == "powerplay" else "Ranking"
|
||
label, names = _section_label(root, language, fallback)
|
||
sections.append(
|
||
{
|
||
"id": _key(root.tag) or kind,
|
||
"label": label,
|
||
"names": names,
|
||
"rows": [{
|
||
"id": _first(root_values, ("id", "external_id")) or "row-1",
|
||
"player_id": _first(root_values, _PLAYER_ID_KEYS),
|
||
"team_id": _first(root_values, _TEAM_ID_KEYS),
|
||
"rank": _first(root_values, _RANK_KEYS),
|
||
"number": _first(root_values, _NUMBER_KEYS),
|
||
"name": _first(root_values, _NAME_KEYS_RU),
|
||
"names": {
|
||
"ru": _first(root_values, _NAME_KEYS_RU),
|
||
"en": _first(root_values, _NAME_KEYS_EN),
|
||
},
|
||
"team": _first(root_values, _TEAM_KEYS_RU),
|
||
"team_names": {
|
||
"ru": _first(root_values, _TEAM_KEYS_RU),
|
||
"en": _first(root_values, _TEAM_KEYS_EN),
|
||
},
|
||
"values": values,
|
||
"source_tag": local_name(root.tag),
|
||
}],
|
||
"columns": [
|
||
{"key": key, "label": _column_label(key, language), "format": _column_format(key)}
|
||
for key in values
|
||
],
|
||
}
|
||
)
|
||
|
||
tournament_id = _first(
|
||
root_fields,
|
||
("idtournament", "id_tournament", "tournament_id", "tournamentid", "tournament"),
|
||
)
|
||
generated_at = _first(
|
||
root_fields,
|
||
("gendate", "gen_date", "generated_at", "generated", "updated_at", "date"),
|
||
)
|
||
return {
|
||
"resource_type": kind,
|
||
"root_tag": local_name(root.tag),
|
||
"tournament_id": tournament_id,
|
||
"generated_at": generated_at,
|
||
"available": any(section.get("rows") for section in sections),
|
||
"sections": sections,
|
||
}
|