from __future__ import annotations import json from dataclasses import dataclass from typing import Any, Iterable @dataclass(slots=True) class ParsedShots: home: int away: int strategy: str def _json_value(content: bytes | str) -> Any: text = content.decode("utf-8-sig") if isinstance(content, bytes) else str(content).lstrip("\ufeff") return json.loads(text) def _key(value: Any) -> str: return "".join(ch for ch in str(value or "").casefold() if ch.isalnum()) def _number(value: Any) -> int | None: if isinstance(value, bool) or value is None or value == "": return None try: number = float(str(value).replace(",", ".")) except (TypeError, ValueError): return None if number < 0: return None return int(number) def _dict_numeric_count(row: Any) -> int | None: if isinstance(row, list): return len(row) if not isinstance(row, dict): return _number(row) preferred = ( "shots", "shot_count", "shots_count", "total_shots", "total", "count", "attempts", "shotattempts", "shot_attempts", "value", ) lowered = {_key(k): v for k, v in row.items()} for candidate in preferred: value = lowered.get(_key(candidate)) number = _number(value) if number is not None: return number return None def _side_value(root: Any, aliases: Iterable[str]) -> int | None: if not isinstance(root, dict): return None lowered = {_key(k): v for k, v in root.items()} for alias in aliases: if _key(alias) not in lowered: continue raw_value = lowered[_key(alias)] number = _dict_numeric_count(raw_value) if number is None and isinstance(raw_value, dict) and raw_value: keys = [str(key).strip() for key in raw_value] if all(key.isdigit() for key in keys): number = len(raw_value) if number is not None: return number return None def _row_team_values(row: dict[str, Any]) -> list[str]: values: list[str] = [] for key, value in row.items(): norm = _key(key) if norm in { "team", "teamid", "idteam", "club", "clubid", "idclub", "side", "teamside", "teamname", "clubname", "teamtitle", }: if isinstance(value, dict): values.extend(str(item) for item in value.values() if not isinstance(item, (dict, list))) else: values.append(str(value)) return values def _matches_team(row: dict[str, Any], ids: set[str], names: set[str], side_aliases: set[str]) -> bool: candidates = {_key(value) for value in _row_team_values(row) if value not in (None, "")} return bool(candidates & ids or candidates & names or candidates & side_aliases) def _walk_lists(value: Any) -> Iterable[list[Any]]: if isinstance(value, list): yield value for item in value: yield from _walk_lists(item) elif isinstance(value, dict): for item in value.values(): yield from _walk_lists(item) def parse_shots_json( content: bytes | str, *, home_ids: Iterable[Any] = (), away_ids: Iterable[Any] = (), home_names: Iterable[Any] = (), away_names: Iterable[Any] = (), ) -> ParsedShots: """Extract home/away shot attempts from multiple Stat2TV JSON shapes. The optional endpoint differs between leagues. Some feeds expose totals by side, some expose a teams list, while others expose one object per shot. This parser deliberately accepts all three forms and returns no value when the payload cannot be mapped unambiguously. """ data = _json_value(content) containers: list[Any] = [data] if isinstance(data, dict): for key in ("data", "result", "statistics", "stats", "shots", "totals"): value = data.get(key) if isinstance(value, (dict, list)): containers.append(value) for root in containers: home = _side_value(root, ("home", "teamA", "A", "left", "host", "homeTeam")) away = _side_value(root, ("away", "teamB", "B", "right", "visitor", "guest", "awayTeam")) if home is not None and away is not None: return ParsedShots(home=home, away=away, strategy="direct-sides") home_id_keys = {_key(value) for value in home_ids if str(value or "").strip()} away_id_keys = {_key(value) for value in away_ids if str(value or "").strip()} home_name_keys = {_key(value) for value in home_names if str(value or "").strip()} away_name_keys = {_key(value) for value in away_names if str(value or "").strip()} home_side_aliases = {_key(value) for value in ("home", "A", "left", "host")} away_side_aliases = {_key(value) for value in ("away", "B", "right", "visitor", "guest")} # Team-total rows. for rows in _walk_lists(data): mapped_home: int | None = None mapped_away: int | None = None for item in rows: if not isinstance(item, dict): continue count = _dict_numeric_count(item) if count is None: continue if _matches_team(item, home_id_keys, home_name_keys, home_side_aliases): mapped_home = count elif _matches_team(item, away_id_keys, away_name_keys, away_side_aliases): mapped_away = count if mapped_home is not None and mapped_away is not None: return ParsedShots(home=mapped_home, away=mapped_away, strategy="team-totals") # Event rows: each mapped row is one shot unless it explicitly carries a # count greater than one. for rows in _walk_lists(data): home_count = 0 away_count = 0 mapped = 0 for item in rows: if not isinstance(item, dict): continue multiplier = _dict_numeric_count(item) # Event payloads commonly include a generic id/value; those are not # totals. Count one record unless a shot-specific count is present. specific_count = None lowered = {_key(k): v for k, v in item.items()} for key in ("shotcount", "shotscount", "attempts", "shotattempts"): specific_count = _number(lowered.get(key)) if specific_count is not None: break amount = specific_count if specific_count is not None else 1 if _matches_team(item, home_id_keys, home_name_keys, home_side_aliases): home_count += amount mapped += 1 elif _matches_team(item, away_id_keys, away_name_keys, away_side_aliases): away_count += amount mapped += 1 if mapped and home_count + away_count: return ParsedShots(home=home_count, away=away_count, strategy="shot-events") raise ValueError("Не удалось определить броски хозяев и гостей в shots JSON")