NotSoToothless
178 lines
6.7 KiB
Python
178 lines
6.7 KiB
Python
"""Raw TSS game → scoreboard model adapter.
|
|
|
|
``process_game`` receives the raw Spectra TSS payload (also persisted to disk by
|
|
``tss_ws._write_game``). Its shape is awkward for rendering: players live in a dict
|
|
keyed by uid, team identity and pvp_ratio live in a parallel ``tss`` block, and each
|
|
player carries a full ``units[]`` lineup rather than a single vehicle.
|
|
|
|
``build_scoreboard_model`` flattens all of that into a stable, renderer-friendly dict
|
|
so ``scoreboard.py`` never has to understand the raw feed. Pure (no I/O); unit-tested
|
|
against sample replays.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Optional
|
|
|
|
log = logging.getLogger("tssbot.transform")
|
|
|
|
# Imported lazily/defensively so the module is importable (and unit-testable for the
|
|
# non-translation paths) even if SHARED isn't on sys.path yet.
|
|
try:
|
|
from data_parser import LangTableReader, apply_vehicle_name_filters # type: ignore
|
|
except Exception: # pragma: no cover - exercised only when SHARED is missing
|
|
LangTableReader = None # type: ignore
|
|
|
|
def apply_vehicle_name_filters(name, strip_decorations=True): # type: ignore
|
|
return name
|
|
|
|
|
|
def _translator(lang_column: str):
|
|
"""Return a translate(internal)->human callable for the given lang column.
|
|
|
|
Falls back to identity (internal name) when data_parser is unavailable.
|
|
"""
|
|
if LangTableReader is None:
|
|
return lambda v: v
|
|
try:
|
|
reader = LangTableReader(lang_column)
|
|
except Exception as exc: # pragma: no cover
|
|
log.warning("LangTableReader(%r) failed: %s", lang_column, exc)
|
|
return lambda v: v
|
|
|
|
def _t(internal: str) -> str:
|
|
try:
|
|
translated = reader.get_translate(internal) or internal
|
|
except Exception:
|
|
translated = internal
|
|
# Strip country-leak / premium / tree decoration glyphs the PNG font can't
|
|
# render (e.g. the leading ␗ on cn_t_34_85_d_5t), matching SREBOT's renderer.
|
|
return apply_vehicle_name_filters(translated)
|
|
|
|
return _t
|
|
|
|
|
|
def _build_units(units: list[dict[str, Any]], translate, dead: set[str] | None = None) -> list[dict[str, Any]]:
|
|
"""Normalize a player's lineup: icon key (internal), translated name, used/dead flags.
|
|
|
|
Prefer an explicit per-unit ``dead``/``died`` flag if Spectra provides one; otherwise
|
|
fall back to the ``dead`` set cross-referenced from ``events.kills`` (see ``_dead_units_by_uid``).
|
|
"""
|
|
dead = dead or set()
|
|
out: list[dict[str, Any]] = []
|
|
for u in units or []:
|
|
internal = str(u.get("unit") or "").strip()
|
|
if not internal:
|
|
continue
|
|
flag = u.get("dead", u.get("died"))
|
|
is_dead = bool(flag) if flag is not None else internal in dead
|
|
out.append({
|
|
"internal": internal,
|
|
"name": translate(internal) or u.get("unit_normalized") or internal,
|
|
"used": bool(u.get("used")),
|
|
"dead": is_dead,
|
|
})
|
|
return out
|
|
|
|
|
|
def _dead_units_by_uid(game: dict[str, Any]) -> dict[str, set[str]]:
|
|
"""Map uid -> set of internal unit names that died (were destroyed in a kill event)."""
|
|
out: dict[str, set[str]] = {}
|
|
kills = ((game.get("events") or {}).get("kills")) or []
|
|
for k in kills:
|
|
uid = str(k.get("offended_uid") or "")
|
|
unit = str(k.get("offended_unit") or "").strip()
|
|
if uid and unit:
|
|
out.setdefault(uid, set()).add(unit)
|
|
return out
|
|
|
|
|
|
def _pvp_index(tss: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
"""Map uid -> {pvp_ratio, role} from the tss per-team roster blocks."""
|
|
idx: dict[str, dict[str, Any]] = {}
|
|
for slot in ("1", "2"):
|
|
team = tss.get(slot)
|
|
if not isinstance(team, dict):
|
|
continue
|
|
for entry in team.get("players") or []:
|
|
uid = str(entry.get("uid") or "")
|
|
if uid:
|
|
idx[uid] = {
|
|
"pvp_ratio": entry.get("pvp_ratio"),
|
|
"role": entry.get("role"),
|
|
}
|
|
return idx
|
|
|
|
|
|
def build_scoreboard_model(game: dict[str, Any], lang_column: str = "<English>") -> Optional[dict[str, Any]]:
|
|
"""Flatten a raw TSS game into a scoreboard render model.
|
|
|
|
Returns ``None`` if the game lacks the minimum structure (two teams w/ players).
|
|
"""
|
|
tss = game.get("tss") or {}
|
|
players_raw = game.get("players") or {}
|
|
if not isinstance(players_raw, dict):
|
|
return None
|
|
|
|
translate = _translator(lang_column)
|
|
pvp = _pvp_index(tss)
|
|
dead_units = _dead_units_by_uid(game)
|
|
|
|
teams: dict[str, dict[str, Any]] = {}
|
|
for slot in ("1", "2"):
|
|
raw_meta = tss.get(slot)
|
|
meta: dict[str, Any] = raw_meta if isinstance(raw_meta, dict) else {}
|
|
teams[slot] = {
|
|
"slot": slot,
|
|
"team_id": str(meta.get("team_id")) if meta.get("team_id") is not None else None,
|
|
"team_name": str(meta.get("team_name") or f"Team {slot}"),
|
|
"players": [],
|
|
}
|
|
|
|
for uid, p in players_raw.items():
|
|
if not isinstance(p, dict):
|
|
continue
|
|
slot = str(p.get("team") or "")
|
|
if slot not in teams:
|
|
continue
|
|
info = pvp.get(str(uid), {})
|
|
teams[slot]["players"].append({
|
|
"uid": str(uid),
|
|
"nick": str(p.get("name") or ""),
|
|
# No streamer-mode field in the TSS feed yet; renderer falls back to nick.
|
|
"fake_nick": None,
|
|
"tag": str(p.get("tag") or ""),
|
|
"country_id": p.get("country_id"),
|
|
"air_kills": int(p.get("air_kills") or 0),
|
|
"ground_kills": int(p.get("ground_kills") or 0),
|
|
"assists": int(p.get("assists") or 0),
|
|
"deaths": int(p.get("deaths") or 0),
|
|
"captures": int(p.get("captures") or 0),
|
|
"score": int(p.get("score") or 0),
|
|
"pvp_ratio": info.get("pvp_ratio"),
|
|
"role": info.get("role"),
|
|
"units": _build_units(p.get("units") or [], translate, dead_units.get(str(uid))),
|
|
})
|
|
|
|
if not teams["1"]["players"] or not teams["2"]["players"]:
|
|
return None
|
|
|
|
winner_slot = str(game.get("winner") or "")
|
|
is_draw = bool(game.get("draw"))
|
|
winner_name = teams[winner_slot]["team_name"] if winner_slot in teams and not is_draw else None
|
|
|
|
return {
|
|
"session_id": str(game.get("_id") or ""),
|
|
"utc_timestamp": int(game.get("end_ts") or 0),
|
|
"start_ts": int(game.get("start_ts") or 0),
|
|
"duration": game.get("duration"),
|
|
"map": str(game.get("mission_name") or ""),
|
|
"mission_mode": str(game.get("mission_mode") or ""),
|
|
"tournament_name": str(tss.get("tournament_name") or ""),
|
|
"bracket": str(tss.get("bracket") or ""),
|
|
"is_draw": is_draw,
|
|
"winner": winner_name,
|
|
"winner_slot": winner_slot,
|
|
"teams": [teams["1"], teams["2"]],
|
|
}
|