185 lines
6.9 KiB
Python
185 lines
6.9 KiB
Python
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")
|