NotSoToothless
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
"""TSS win/loss standings, scoped per tournament.
|
|
|
|
SREBOT tracks squadron W/L globally. TSS records only mean something inside a
|
|
bracket, so standings are scoped to ``(tournament_id, team_id)`` and derived live
|
|
from the rows already persisted per match — no separate counter table:
|
|
|
|
* ``match_summary`` — one row per session, carries ``tournament_id``
|
|
* ``player_games_hist`` — one row per used vehicle per player, carries
|
|
``team_id``, ``session_id`` and ``victor_bool`` ('Win'/'Loss')
|
|
|
|
A draw is already written as ``victor_bool = 'Loss'`` for *both* teams (the winning
|
|
slot is empty), so counting over ``victor_bool`` treats a draw as a loss for both
|
|
sides with no special handling.
|
|
|
|
``insert_match``/``insert_player_games`` run before the scoreboard renders (see
|
|
``tss_ws._handle_game``), so the record returned here includes the match on the
|
|
board being drawn.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
from typing import Any, Optional
|
|
|
|
log = logging.getLogger("tssbot.wl")
|
|
|
|
# One outcome per session (vehicle rows are duplicated across a player's lineup, so
|
|
# collapse to a single victor_bool per session before counting — mirrors the
|
|
# DISTINCT-first rule in storage.py's docstring), then tally wins vs losses.
|
|
_STANDINGS_SQL = """
|
|
SELECT
|
|
SUM(CASE WHEN victor = 'Win' THEN 1 ELSE 0 END) AS wins,
|
|
SUM(CASE WHEN victor = 'Loss' THEN 1 ELSE 0 END) AS losses
|
|
FROM (
|
|
SELECT pgh.session_id, MAX(pgh.victor_bool) AS victor
|
|
FROM player_games_hist pgh
|
|
JOIN match_summary ms ON ms.session_id = pgh.session_id
|
|
WHERE pgh.team_id = ? AND ms.tournament_id = ?
|
|
GROUP BY pgh.session_id
|
|
)
|
|
"""
|
|
|
|
|
|
def get_tss_standings(
|
|
team_id: Optional[Any], tournament_id: Optional[Any] = None
|
|
) -> Optional[dict[str, int]]:
|
|
"""Return ``{"wins": int, "losses": int}`` for a team within one tournament.
|
|
|
|
Returns ``None`` when either id is missing, the DB is unavailable, the query
|
|
errors, or the team has no recorded matches in that tournament — in every such
|
|
case the scoreboard leaves the W/L slot blank (matching offline/sample renders).
|
|
"""
|
|
if team_id is None or tournament_id is None:
|
|
return None
|
|
try:
|
|
tid = int(team_id)
|
|
tournament = int(tournament_id)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
# Imported here so the module stays importable without STORAGE_VOL_PATH set
|
|
# (e.g. for unit tests that point _db_path at a temp database).
|
|
db_path = _db_path()
|
|
if db_path is None or not os.path.exists(db_path):
|
|
# No file → nothing to read (and don't let sqlite3.connect create a stray
|
|
# empty DB). Keeps offline/sample renders showing a blank slot.
|
|
return None
|
|
|
|
# NOT a ``mode=ro`` connection: the production DB is WAL (see storage._PRAGMAS),
|
|
# and a pure read-only open can fail when a -wal exists but the -shm wal-index
|
|
# does not (it can't be created read-only). A normal connection set to
|
|
# ``query_only`` reads WAL safely in every state while never writing. ``timeout``
|
|
# waits out a momentary write lock from the concurrent ingest path.
|
|
try:
|
|
conn = sqlite3.connect(db_path, timeout=5.0)
|
|
conn.execute("PRAGMA query_only=ON")
|
|
except sqlite3.Error:
|
|
return None
|
|
try:
|
|
row = conn.execute(_STANDINGS_SQL, (tid, tournament)).fetchone()
|
|
except sqlite3.Error as exc:
|
|
log.warning("[TSS-WL] standings query failed (team=%s tourn=%s): %s", tid, tournament, exc)
|
|
return None
|
|
finally:
|
|
conn.close()
|
|
|
|
if not row or row[0] is None:
|
|
return None
|
|
return {"wins": int(row[0] or 0), "losses": int(row[1] or 0)}
|
|
|
|
|
|
def _db_path() -> Optional[str]:
|
|
"""Resolve the battles DB path, or ``None`` if storage isn't configured."""
|
|
try:
|
|
from .storage import TSS_BATTLES_DB_PATH
|
|
except Exception:
|
|
return None
|
|
return str(TSS_BATTLES_DB_PATH)
|