"""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 = "") -> 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_id": tss.get("tournament_id"), "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"]], }