from __future__ import annotations import json from dataclasses import dataclass, field from datetime import date from typing import Any from xml.etree import ElementTree as ET from .match_parser import local_name, parse_date_value from .iso_country_codes import iso2_from_code, iso3_from_code, khl_country_codes def _json_object(content: bytes | str) -> dict[str, Any]: text = ( content.decode("utf-8-sig") if isinstance(content, bytes) else str(content).lstrip("\ufeff") ) value = json.loads(text) if not isinstance(value, dict): raise ValueError("Карточка матча Stat2TV должна быть JSON-объектом") return value def _raw_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":")) def _text(value: Any) -> str: return str(value or "").strip() def _truthy(value: Any) -> bool: return _text(value).lower() in {"1", "true", "yes", "y", "on"} def _node_values(node: ET.Element) -> dict[str, str]: values = {local_name(str(key)).lower(): _text(value) for key, value in node.attrib.items()} for child in list(node): key = local_name(child.tag).lower() text = _text(child.text) if text and key not in values: values[key] = text for attr_key, attr_value in child.attrib.items(): compound = f"{key}_{local_name(str(attr_key)).lower()}" values.setdefault(compound, _text(attr_value)) return values def _first(values: dict[str, str], *keys: str) -> str: for key in keys: value = _text(values.get(key.lower())) if value: return value return "" def _integer(value: Any) -> int: text = _text(value).replace(",", ".") if not text: return 0 try: return max(0, int(round(float(text)))) except (TypeError, ValueError): digits = "".join(character for character in text if character.isdigit()) return int(digits) if digits else 0 def country_flag_emoji(code: str) -> str: value = iso2_from_code(_text(code)) if len(value) != 2: return "" return "".join(chr(127397 + ord(character)) for character in value) def _split_full_name(value: str) -> tuple[str, str]: parts = [part for part in _text(value).split() if part] if not parts: return "", "" return parts[0], " ".join(parts[1:]) def canonical_player_role(value: Any) -> str: role = _text(value).lower() if role in {"g", "в", "вр", "goalie", "goalkeeper"}: return "goalkeeper" if role in {"d", "з", "защ", "defender", "defence", "defense"}: return "defender" if role in {"f", "н", "нап", "forward"}: return "forward" return role def player_position_labels(role: str) -> tuple[str, str]: return { "goalkeeper": ("ВР", "G"), "defender": ("ЗАЩ", "D"), "forward": ("НАП", "F"), }.get(role, (role.upper(), role.upper())) def canonical_captain_role(value: Any) -> str: role = _text(value).lower() if role in {"к", "c", "captain"}: return "captain" if role in {"а", "a", "assistant", "alternate"}: return "assistant" return "" @dataclass(slots=True) class ParsedPlayerProfile: external_id: str first_name_ru: str = "" first_name_en: str = "" last_name_ru: str = "" last_name_en: str = "" full_name_ru: str = "" full_name_en: str = "" position_ru: str = "" position_en: str = "" birth_date: date | None = None height_cm: int = 0 weight_kg: int = 0 stick: str = "" country_external_id: str = "" country_code: str = "" active: bool = True photo_url: str = "" raw_payload: str = "" @dataclass(slots=True) class ParsedRefereeProfile: external_id: str first_name_ru: str = "" first_name_en: str = "" last_name_ru: str = "" last_name_en: str = "" full_name_ru: str = "" full_name_en: str = "" position_code: str = "" number: str = "" active: bool = True birth_date: date | None = None country_ru: str = "" country_en: str = "" country_code: str = "" town_ru: str = "" town_en: str = "" raw_payload: str = "" @dataclass(slots=True) class ParsedCountry: external_id: str iso2: str = "" iso3: str = "" name_ru: str = "" name_en: str = "" flag_emoji: str = "" active: bool = True raw_payload: str = "" @dataclass(slots=True) class ParsedCountries: countries: dict[str, ParsedCountry] = field(default_factory=dict) player_country: dict[str, str] = field(default_factory=dict) @dataclass(slots=True) class ParsedCoachProfile: external_id: str first_name_ru: str = "" first_name_en: str = "" last_name_ru: str = "" last_name_en: str = "" full_name_ru: str = "" full_name_en: str = "" birth_date: date | None = None country_external_id: str = "" country_code: str = "" photo_url: str = "" active: bool = True team_external_id: str = "" team_name_ru: str = "" team_name_en: str = "" role_ru: str = "Главный тренер" role_en: str = "Head coach" raw_payload: str = "" @dataclass(slots=True) class ParsedRosterEntry: player_external_id: str team_external_id: str team_entry_external_id: str side: str number: str role: str position_ru: str position_en: str line_number: str captain_role: str lineup_status: str name_ru: str name_en: str raw_payload: str @dataclass(slots=True) class ParsedOfficialAssignment: referee_external_id: str role: str slot: int number: str name_ru: str name_en: str raw_payload: str @dataclass(slots=True) class ParsedMatchDetails: game_external_id: str game_ru: dict[str, Any] game_en: dict[str, Any] teams_ru: dict[str, Any] teams_en: dict[str, Any] rosters: list[ParsedRosterEntry] = field(default_factory=list) officials: list[ParsedOfficialAssignment] = field(default_factory=list) def parse_players_xml(content: bytes | str) -> dict[str, ParsedPlayerProfile]: raw = content.encode("utf-8") if isinstance(content, str) else content root = ET.fromstring(raw) result: dict[str, ParsedPlayerProfile] = {} for node in root.iter(): tag = local_name(node.tag).lower() if tag not in {"playerstats", "player", "playerstat", "athlete"}: continue values = _node_values(node) external_id = _first(values, "id", "playerid", "player_id", "idplayer") if not external_id: continue role = canonical_player_role(_first(values, "pos", "position", "role", "amplua")) position_ru, position_en = player_position_labels(role) full_name_ru = _first(values, "name", "fullname", "full_name", "fio") full_name_en = _first(values, "nameen", "name_en", "fullnameen", "full_name_en") last_name_ru, first_name_ru = _split_full_name(full_name_ru) last_name_en, first_name_en = _split_full_name(full_name_en) raw_country_code = _first( values, "countrycode", "country_code", "iso2", "iso3", "countryiso2", "nationalitycode", "nationality_code", ).upper() country_code = iso2_from_code(raw_country_code) or raw_country_code country_external_id = _first( values, "countryid", "country_id", "idcountry", "nationalityid", "nationality_id", "country", ) result[external_id] = ParsedPlayerProfile( external_id=external_id, first_name_ru=_first(values, "firstname", "first_name", "namefirst") or first_name_ru, first_name_en=_first(values, "firstnameen", "firstname_en", "first_name_en") or first_name_en, last_name_ru=_first(values, "lastname", "last_name", "namelast") or last_name_ru, last_name_en=_first(values, "lastnameen", "lastname_en", "last_name_en") or last_name_en, full_name_ru=full_name_ru, full_name_en=full_name_en, position_ru=_first(values, "positionru", "position_ru", "posname") or position_ru, position_en=_first(values, "positionen", "position_en", "posnameen") or position_en, birth_date=parse_date_value(_first(values, "birthdate", "birth_date", "birthday", "datebirth", "dob")), height_cm=_integer(_first(values, "height", "heightcm", "height_cm", "rost")), weight_kg=_integer(_first(values, "weight", "weightkg", "weight_kg", "ves")), stick=_first(values, "stick", "grip", "shoots", "shoot", "hand", "catch"), country_external_id=country_external_id, country_code=country_code, active=not _first(values, "isactive", "active") or _truthy(_first(values, "isactive", "active")), photo_url=_first(values, "photo", "photourl", "photo_url", "image", "imageurl"), raw_payload=_raw_json(values), ) return result def parse_coaches_xml(content: bytes | str) -> dict[str, ParsedCoachProfile]: raw = content.encode("utf-8") if isinstance(content, str) else content root = ET.fromstring(raw) result: dict[str, ParsedCoachProfile] = {} for node in root.iter(): tag = local_name(node.tag).lower() if tag not in {"coach", "trainer", "headcoach", "coachstats"}: continue values = _node_values(node) external_id = _first(values, "id", "coachid", "coach_id", "idcoach") full_name_ru = _first(values, "name", "fullname", "full_name", "fio") full_name_en = _first(values, "nameen", "name_en", "fullnameen", "full_name_en") if not external_id: external_id = _first(values, "teamid", "clubid", "club_id") + ":" + (full_name_ru or full_name_en) external_id = external_id.strip(":") if not external_id or not (full_name_ru or full_name_en): continue last_name_ru, first_name_ru = _split_full_name(full_name_ru) last_name_en, first_name_en = _split_full_name(full_name_en) result[external_id] = ParsedCoachProfile( external_id=external_id[:64], first_name_ru=_first(values, "firstname", "first_name") or first_name_ru, first_name_en=_first(values, "firstnameen", "first_name_en") or first_name_en, last_name_ru=_first(values, "lastname", "last_name") or last_name_ru, last_name_en=_first(values, "lastnameen", "last_name_en") or last_name_en, full_name_ru=full_name_ru, full_name_en=full_name_en, birth_date=parse_date_value(_first(values, "birthdate", "birth_date", "birthday", "dob")), country_external_id=_first(values, "countryid", "country_id", "idcountry", "country"), country_code=(iso2_from_code(_first(values, "countrycode", "country_code", "iso2", "iso3")) or _first(values, "countrycode", "country_code", "iso2", "iso3").upper()), photo_url=_first(values, "photo", "photourl", "photo_url"), active=not _first(values, "isactive", "active") or _truthy(_first(values, "isactive", "active")), team_external_id=_first(values, "clubid", "club_id", "teamid", "team_id"), team_name_ru=_first(values, "teamname", "clubname", "team_name", "club_name"), team_name_en=_first(values, "teamnameen", "clubnameen", "team_name_en", "club_name_en"), role_ru=_first(values, "roleru", "role_ru", "positionru") or "Главный тренер", role_en=_first(values, "roleen", "role_en", "positionen") or "Head coach", raw_payload=_raw_json(values), ) return result def parse_player_countries_xml(content: bytes | str) -> ParsedCountries: """Parse the global player-country feed across known Stat2TV XML layouts. The feed has appeared both as a flat player mapping and as nested country/player groups. We collect bilingual labels from ``label xml:lang`` nodes, recognise ISO-2/ISO-3 aliases, and create minimal country records for mapping-only codes. """ raw = content.encode("utf-8") if isinstance(content, str) else content root = ET.fromstring(raw) parsed = ParsedCountries() # Current Stat2TV layout: # # It is a player-country mapping, not a standalone country directory. Build # the directory while walking those rows so names and flags are not lost. for player_node in root.iter(): if local_name(player_node.tag).lower() != "player": continue values = _node_values(player_node) player_id = _first(values, "playerid", "player_id", "idplayer", "personid", "person_id", "id") country_id = _first(values, "countryid", "country_id", "idcountry", "cid") if not player_id or not country_id or country_id == "0": continue name_ru = _first(values, "countryname", "country_name", "countrynameru", "country_name_ru") name_en = _first(values, "countrynameen", "country_name_en") raw_code = _first(values, "countrycode", "country_code", "iso", "iso2", "iso3") iso2 = iso2_from_code(raw_code) iso3 = iso3_from_code(raw_code) if not iso2 and not iso3: iso2, iso3 = khl_country_codes(country_id) parsed.player_country[player_id] = country_id existing = parsed.countries.get(country_id) parsed.countries[country_id] = ParsedCountry( external_id=country_id[:64], iso2=(iso2 or (existing.iso2 if existing else ""))[:2], iso3=(iso3 or (existing.iso3 if existing else ""))[:3], name_ru=name_ru or (existing.name_ru if existing else ""), name_en=name_en or (existing.name_en if existing else ""), flag_emoji=country_flag_emoji(iso2 or (existing.iso2 if existing else "")), active=True, raw_payload=_raw_json(values), ) def labels(node: ET.Element) -> tuple[str, str]: ru = en = "" for child in node.iter(): if local_name(child.tag).lower() not in {"label", "name", "countryname"}: continue text = (child.text or "").strip() lang = (child.attrib.get("{http://www.w3.org/XML/1998/namespace}lang") or child.attrib.get("lang") or "").lower() if lang.startswith("ru") and text: ru = ru or text elif lang.startswith("en") and text: en = en or text return ru, en def country_parts(values: dict[str, str], node: ET.Element) -> tuple[str, str, str, str, str]: raw_country = _first(values, "country", "countrycode", "country_code", "iso", "iso2", "iso3", "code") iso2 = _first(values, "iso2", "alpha2", "code2", "countryiso2").upper() iso3 = _first(values, "iso3", "alpha3", "code3", "countryiso3").upper() if not iso2 and len(raw_country) == 2 and raw_country.isalpha(): iso2 = raw_country.upper() if not iso3 and len(raw_country) == 3 and raw_country.isalpha(): iso3 = raw_country.upper() if not iso2 and iso3: iso2 = iso2_from_code(iso3) label_ru, label_en = labels(node) name_ru = _first(values, "nameru", "name_ru", "countryname", "country_name", "name") or label_ru name_en = _first(values, "nameen", "name_en", "countrynameen", "country_name_en") or label_en external_id = _first(values, "countryid", "country_id", "idcountry", "cid") or iso3 or iso2 or raw_country return external_id, iso2, iso3, name_ru, name_en country_stack: list[str] = [] def walk(node: ET.Element, inherited_country: str = "") -> None: tag = local_name(node.tag).lower() values = _node_values(node) external_id, iso2, iso3, name_ru, name_en = country_parts(values, node) explicit_country = _first(values, "countryid", "country_id", "idcountry", "country", "countrycode", "country_code", "iso2", "iso3") looks_country = tag in {"country", "nation", "nationality", "countryitem", "country_item"} if tag == "item" and (iso2 or iso3) and (name_ru or name_en): looks_country = True current_country = inherited_country if looks_country: external_id = _first(values, "countryid", "country_id", "idcountry", "cid", "id") or external_id if looks_country and external_id: current_country = external_id parsed.countries[external_id] = ParsedCountry( external_id=external_id[:64], iso2=iso2[:2], iso3=iso3[:3], name_ru=name_ru, name_en=name_en, flag_emoji=country_flag_emoji(iso2), active=not _first(values, "isactive", "active") or _truthy(_first(values, "isactive", "active")), raw_payload=_raw_json(values), ) player_id = _first(values, "playerid", "player_id", "idplayer", "personid", "person_id") if tag in {"player", "playercountry", "player_country", "person"} and not player_id: player_id = _first(values, "id") if tag == "item" and not looks_country and not player_id and (explicit_country or inherited_country): player_id = _first(values, "id") mapping = explicit_country or current_country if player_id and mapping and str(mapping).strip() != "0": parsed.player_country[player_id] = mapping for child in list(node): walk(child, current_country) walk(root) # Resolve aliases and create records for codes that only occur in mappings. alias: dict[str, str] = {} for key, country in parsed.countries.items(): for code in (key, country.iso2, country.iso3): if code: alias[str(code).upper()] = key for player_id, mapping in list(parsed.player_country.items()): if str(mapping).strip() == "0": parsed.player_country.pop(player_id, None) continue canonical = alias.get(str(mapping).upper(), mapping) parsed.player_country[player_id] = canonical if canonical in parsed.countries: continue code = str(mapping).upper() iso2 = iso2_from_code(code) iso3 = code if len(code) == 3 and code.isalpha() else "" if not iso2 and not iso3: iso2, iso3 = khl_country_codes(str(canonical)) parsed.countries[canonical] = ParsedCountry( external_id=str(canonical)[:64], iso2=iso2, iso3=iso3, flag_emoji=country_flag_emoji(iso2), raw_payload=_raw_json({"code": mapping}), ) return parsed def parse_referees_xml(content: bytes | str) -> dict[str, ParsedRefereeProfile]: raw = content.encode("utf-8") if isinstance(content, str) else content root = ET.fromstring(raw) result: dict[str, ParsedRefereeProfile] = {} for node in root.iter(): if local_name(node.tag).lower() not in {"referee", "official"}: continue attrs = _node_values(node) external_id = _first(attrs, "id", "refereeid", "referee_id", "idofficial") if not external_id: continue full_name_ru = attrs.get("name", "") full_name_en = attrs.get("nameen", "") last_name_ru, first_name_ru = _split_full_name(full_name_ru) last_name_en, first_name_en = _split_full_name(full_name_en) result[external_id] = ParsedRefereeProfile( external_id=external_id, first_name_ru=attrs.get("firstname") or first_name_ru, first_name_en=attrs.get("firstnameen") or first_name_en, last_name_ru=attrs.get("lastname") or last_name_ru, last_name_en=attrs.get("lastnameen") or last_name_en, full_name_ru=full_name_ru, full_name_en=full_name_en, position_code=attrs.get("pos") or attrs.get("lastpos", ""), number=attrs.get("jn", ""), active=_truthy(attrs.get("isactive", "Y")), birth_date=parse_date_value(attrs.get("birthdate", "")), country_ru=_first(attrs, "country", "countryru", "country_ru"), country_en=_first(attrs, "countryen", "country_en"), country_code=(iso2_from_code(_first(attrs, "countrycode", "country_code", "iso2", "iso3")) or _first(attrs, "countrycode", "country_code", "iso2", "iso3").upper()), town_ru=attrs.get("town", ""), town_en=attrs.get("townen", ""), raw_payload=_raw_json(attrs), ) return result def parse_match_details_json( content_ru: bytes | str, content_en: bytes | str | None = None, *, expected_game_id: str = "", ) -> ParsedMatchDetails: data_ru = _json_object(content_ru) data_en = _json_object(content_en) if content_en else data_ru game_ru = data_ru.get("game") if isinstance(data_ru.get("game"), dict) else {} game_en = data_en.get("game") if isinstance(data_en.get("game"), dict) else {} game_id = _text(game_ru.get("idschedule") or game_en.get("idschedule")) if not game_id: raise ValueError("В карточке Stat2TV отсутствует id матча") if expected_game_id and game_id != _text(expected_game_id): raise ValueError( f"Stat2TV вернул карточку матча {game_id} вместо {expected_game_id}" ) players_ru = data_ru.get("players") if isinstance(data_ru.get("players"), dict) else {} players_en = data_en.get("players") if isinstance(data_en.get("players"), dict) else {} rosters: list[ParsedRosterEntry] = [] for source_side, side in (("A", "home"), ("B", "away")): rows_ru = players_ru.get(source_side) if isinstance(players_ru.get(source_side), dict) else {} rows_en = players_en.get(source_side) if isinstance(players_en.get(source_side), dict) else {} numbers = list(dict.fromkeys([*rows_ru.keys(), *rows_en.keys()])) suffix = "A" if source_side == "A" else "B" team_external_id = _text(game_ru.get(f"idclub{suffix}") or game_en.get(f"idclub{suffix}")) team_entry_external_id = _text(game_ru.get(f"idteam{suffix}") or game_en.get(f"idteam{suffix}")) for number in numbers: row_ru = rows_ru.get(number) if isinstance(rows_ru.get(number), dict) else {} row_en = rows_en.get(number) if isinstance(rows_en.get(number), dict) else {} player_id = _text(row_ru.get("id") or row_en.get("id")) if not player_id: continue role = canonical_player_role(row_en.get("ps") or row_ru.get("ps")) position_ru, position_en = player_position_labels(role) played = row_ru.get("pig", row_en.get("pig", 0)) rosters.append( ParsedRosterEntry( player_external_id=player_id, team_external_id=team_external_id, team_entry_external_id=team_entry_external_id, side=side, number=_text(number), role=role, position_ru=position_ru, position_en=position_en, line_number=_text(row_ru.get("line") or row_en.get("line")), captain_role=canonical_captain_role( row_ru.get("ca") or row_en.get("ca") ), lineup_status="active" if _truthy(played) else "reserve", name_ru=_text(row_ru.get("name") or row_en.get("name")), name_en=_text(row_en.get("name") or row_ru.get("name")), raw_payload=_raw_json({"ru": row_ru, "en": row_en}), ) ) officials: list[ParsedOfficialAssignment] = [] for prefix, role, slot in ( ("mref1", "head", 1), ("mref2", "head", 2), ("lref1", "linesman", 1), ("lref2", "linesman", 2), ): referee_id = _text(game_ru.get(f"{prefix}_id") or game_en.get(f"{prefix}_id")) if not referee_id: continue officials.append( ParsedOfficialAssignment( referee_external_id=referee_id, role=role, slot=slot, number=_text(game_ru.get(f"{prefix}_num") or game_en.get(f"{prefix}_num")), name_ru=_text(game_ru.get(prefix) or game_en.get(prefix)), name_en=_text(game_en.get(prefix) or game_ru.get(prefix)), raw_payload=_raw_json( { "id": referee_id, "role": role, "slot": slot, "number": game_ru.get(f"{prefix}_num") or game_en.get(f"{prefix}_num"), "name_ru": game_ru.get(prefix), "name_en": game_en.get(prefix), } ), ) ) teams_ru = data_ru.get("teams") if isinstance(data_ru.get("teams"), dict) else {} teams_en = data_en.get("teams") if isinstance(data_en.get("teams"), dict) else {} return ParsedMatchDetails( game_external_id=game_id, game_ru=game_ru, game_en=game_en, teams_ru=teams_ru, teams_en=teams_en, rosters=rosters, officials=officials, ) def fallback_player_profile(entry: ParsedRosterEntry) -> ParsedPlayerProfile: last_name_ru, first_name_ru = _split_full_name(entry.name_ru) last_name_en, first_name_en = _split_full_name(entry.name_en) return ParsedPlayerProfile( external_id=entry.player_external_id, first_name_ru=first_name_ru, first_name_en=first_name_en, last_name_ru=last_name_ru, last_name_en=last_name_en, full_name_ru=entry.name_ru, full_name_en=entry.name_en, position_ru=entry.position_ru, position_en=entry.position_en, raw_payload=entry.raw_payload, ) def fallback_referee_profile( assignment: ParsedOfficialAssignment, ) -> ParsedRefereeProfile: last_name_ru, first_name_ru = _split_full_name(assignment.name_ru) last_name_en, first_name_en = _split_full_name(assignment.name_en) return ParsedRefereeProfile( external_id=assignment.referee_external_id, first_name_ru=first_name_ru, first_name_en=first_name_en, last_name_ru=last_name_ru, last_name_en=last_name_en, full_name_ru=assignment.name_ru, full_name_en=assignment.name_en, number=assignment.number, raw_payload=assignment.raw_payload, )