feat(tally): core model, evaluation and status formatting (#1337)
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -848,6 +848,25 @@
|
|||||||
"stack_manage": {
|
"stack_manage": {
|
||||||
"description": "Re-post your active stack embed to this channel"
|
"description": "Re-post your active stack embed to this channel"
|
||||||
},
|
},
|
||||||
|
"tally": {
|
||||||
|
"description_claim": "Track a live SQB scoreline on your current voice channel",
|
||||||
|
"description_transfer": "Transfer the active voice-channel tally to a different player",
|
||||||
|
"description_wipe": "Clear the active tally on your current voice channel",
|
||||||
|
"ign": "The player IGN to track",
|
||||||
|
"squadron_short": "The squadron short name to track",
|
||||||
|
"result_win": "Win",
|
||||||
|
"result_loss": "Loss",
|
||||||
|
"result_draw": "Draw",
|
||||||
|
"status_line": "{base}: {verb} against {opponent}",
|
||||||
|
"not_in_vc": "You must be connected to a voice channel to use this.",
|
||||||
|
"premium_required": "This is a premium feature. Use /unlock to enable it for this server.",
|
||||||
|
"need_one_input": "Provide exactly one of `ign` or `squadron_short`.",
|
||||||
|
"already_active": "A tally is already active in **{channel}** tracking **{target}**. Use /tally-transfer or /tally-wipe first.",
|
||||||
|
"claimed": "Now tracking **{target}** in **{channel}**. Status set to `0W-0L`.",
|
||||||
|
"no_active": "There is no active tally in **{channel}**.",
|
||||||
|
"transferred": "Tally in **{channel}** now tracking **{target}** (count carried over: `{base}`).",
|
||||||
|
"wiped": "Cleared the tally in **{channel}**."
|
||||||
|
},
|
||||||
"bot_status": {
|
"bot_status": {
|
||||||
"description": "View bot status: last game received and average TTL"
|
"description": "View bot status: last game received and average TTL"
|
||||||
}
|
}
|
||||||
|
|||||||
+122
@@ -0,0 +1,122 @@
|
|||||||
|
"""Live per-voice-channel SQB tally tracking.
|
||||||
|
|
||||||
|
A member connected to a voice channel claims an IGN or a squadron via the
|
||||||
|
``/tally-claim`` command. As SQB games finish involving that target, the bot
|
||||||
|
updates the voice channel's Discord status to a running scoreline such as
|
||||||
|
``2W-1L: Win against ENEMY``. The count is fresh from the moment of claim and
|
||||||
|
is independent of the cumulative standings in ``wl.py``.
|
||||||
|
|
||||||
|
State lives in an in-memory registry keyed by ``(guild_id, channel_id)``,
|
||||||
|
mirrored to ``STORAGE_DIR/TALLY.json`` so tallies survive restarts. Tallies are
|
||||||
|
wiped on a 1h idle sweep or when the voice channel empties of human members.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
from dataclasses import dataclass, field, asdict
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
from .utils import t
|
||||||
|
|
||||||
|
IDLE_TIMEOUT = 3600
|
||||||
|
RECENT_SESSIONS_MAX = 50
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Tally:
|
||||||
|
guild_id: int
|
||||||
|
channel_id: int
|
||||||
|
mode: str # "player" | "squadron"
|
||||||
|
target: str # IGN (player mode) or squadron short (squadron mode)
|
||||||
|
display_target: str
|
||||||
|
wins: int = 0
|
||||||
|
losses: int = 0
|
||||||
|
last_result_kind: Optional[str] = None # "win" | "loss" | "draw"
|
||||||
|
last_opponent: Optional[str] = None
|
||||||
|
last_update_ts: float = 0.0
|
||||||
|
claimed_by: int = 0
|
||||||
|
created_ts: float = 0.0
|
||||||
|
recent_sessions: list[str] = field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
|
def strip_tag(s: Optional[str]) -> str:
|
||||||
|
"""Strip leading/trailing non-alphanumerics (mirrors autologging._strip_tag)."""
|
||||||
|
if not s:
|
||||||
|
return ""
|
||||||
|
return re.sub(r'^[^A-Za-z0-9]+|[^A-Za-z0-9]+$', '', s) or s
|
||||||
|
|
||||||
|
|
||||||
|
def team_short(team: dict) -> str:
|
||||||
|
"""Bare short tag for a replay team dict."""
|
||||||
|
return strip_tag(team.get("squadron_short") or team.get("squadron"))
|
||||||
|
|
||||||
|
|
||||||
|
def team_index_for(tally: Tally, teams: list[dict]) -> Optional[int]:
|
||||||
|
"""Return the index of the team the tracked entity is on, else None."""
|
||||||
|
for idx, team in enumerate(teams):
|
||||||
|
if not team:
|
||||||
|
continue
|
||||||
|
if tally.mode == "squadron":
|
||||||
|
if team_short(team).lower() == strip_tag(tally.target).lower():
|
||||||
|
return idx
|
||||||
|
else: # player mode
|
||||||
|
target = tally.target.strip().lower()
|
||||||
|
for p in team.get("players", []):
|
||||||
|
if str(p.get("nick") or "").strip().lower() == target:
|
||||||
|
return idx
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def evaluate(
|
||||||
|
tally: Tally,
|
||||||
|
teams: list[dict],
|
||||||
|
winner_short: Optional[str],
|
||||||
|
is_draw: bool,
|
||||||
|
session_id: str,
|
||||||
|
) -> Optional[str]:
|
||||||
|
"""Apply a finished session to the tally, mutating it in place.
|
||||||
|
|
||||||
|
Returns the result kind applied ("win"|"loss"|"draw"), or None if the
|
||||||
|
tracked entity was not in this game or the session was already counted.
|
||||||
|
"""
|
||||||
|
if session_id and session_id in tally.recent_sessions:
|
||||||
|
return None
|
||||||
|
idx = team_index_for(tally, teams)
|
||||||
|
if idx is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
other = teams[1 - idx] if len(teams) == 2 and idx in (0, 1) else None
|
||||||
|
opponent = team_short(other) if other else None
|
||||||
|
|
||||||
|
if is_draw:
|
||||||
|
kind = "draw"
|
||||||
|
tally.losses += 1
|
||||||
|
elif winner_short and strip_tag(winner_short).lower() == team_short(teams[idx]).lower():
|
||||||
|
kind = "win"
|
||||||
|
tally.wins += 1
|
||||||
|
else:
|
||||||
|
kind = "loss"
|
||||||
|
tally.losses += 1
|
||||||
|
|
||||||
|
tally.last_result_kind = kind
|
||||||
|
tally.last_opponent = opponent
|
||||||
|
if session_id:
|
||||||
|
tally.recent_sessions.append(session_id)
|
||||||
|
if len(tally.recent_sessions) > RECENT_SESSIONS_MAX:
|
||||||
|
tally.recent_sessions = tally.recent_sessions[-RECENT_SESSIONS_MAX:]
|
||||||
|
return kind
|
||||||
|
|
||||||
|
|
||||||
|
def format_status(tally: Tally, lang: str) -> str:
|
||||||
|
"""Build the voice-channel status string for a tally."""
|
||||||
|
base = f"{tally.wins}W-{tally.losses}L"
|
||||||
|
if tally.last_result_kind and tally.last_opponent:
|
||||||
|
verb = t(lang, f"commands.tally.result_{tally.last_result_kind}")
|
||||||
|
return t(
|
||||||
|
lang,
|
||||||
|
"commands.tally.status_line",
|
||||||
|
base=base,
|
||||||
|
verb=verb,
|
||||||
|
opponent=tally.last_opponent,
|
||||||
|
)
|
||||||
|
return base
|
||||||
@@ -0,0 +1,108 @@
|
|||||||
|
"""
|
||||||
|
Pure-logic tests for BOT/tally.py (no Discord, no I/O).
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
source ../SHARED/.venv/bin/activate && python BOT/tests/test_tally_logic.py
|
||||||
|
"""
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# Set a temporary storage path before importing BOT (which imports utils)
|
||||||
|
os.environ.setdefault("STORAGE_VOL_PATH", "/tmp/tally_test_storage")
|
||||||
|
Path(os.environ["STORAGE_VOL_PATH"]).mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2])) # repo root for `BOT` package
|
||||||
|
|
||||||
|
from BOT.tally import Tally, team_index_for, evaluate, format_status, strip_tag
|
||||||
|
|
||||||
|
|
||||||
|
def _teams():
|
||||||
|
return [
|
||||||
|
{"squadron_short": "-DSPL-", "players": [{"nick": "Alpha"}, {"nick": "Bravo"}]},
|
||||||
|
{"squadron_short": "ENEMY", "players": [{"nick": "Echo"}]},
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_strip_tag():
|
||||||
|
assert strip_tag("-DSPL-") == "DSPL"
|
||||||
|
assert strip_tag("=ABC=") == "ABC"
|
||||||
|
assert strip_tag(None) == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_team_index_player_mode():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="player", target="bravo", display_target="Bravo")
|
||||||
|
assert team_index_for(t, _teams()) == 0
|
||||||
|
t2 = Tally(guild_id=1, channel_id=2, mode="player", target="nobody", display_target="nobody")
|
||||||
|
assert team_index_for(t2, _teams()) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_team_index_squadron_mode():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="ENEMY", display_target="ENEMY")
|
||||||
|
assert team_index_for(t, _teams()) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_evaluate_win_increments_and_records_opponent():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="DSPL", display_target="DSPL")
|
||||||
|
kind = evaluate(t, _teams(), winner_short="DSPL", is_draw=False, session_id="s1")
|
||||||
|
assert kind == "win"
|
||||||
|
assert (t.wins, t.losses) == (1, 0)
|
||||||
|
assert t.last_result_kind == "win"
|
||||||
|
assert t.last_opponent == "ENEMY"
|
||||||
|
|
||||||
|
|
||||||
|
def test_evaluate_loss():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="DSPL", display_target="DSPL")
|
||||||
|
kind = evaluate(t, _teams(), winner_short="ENEMY", is_draw=False, session_id="s1")
|
||||||
|
assert kind == "loss"
|
||||||
|
assert (t.wins, t.losses) == (0, 1)
|
||||||
|
assert t.last_result_kind == "loss"
|
||||||
|
|
||||||
|
|
||||||
|
def test_evaluate_draw_counts_as_loss_but_kind_draw():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="DSPL", display_target="DSPL")
|
||||||
|
kind = evaluate(t, _teams(), winner_short=None, is_draw=True, session_id="s1")
|
||||||
|
assert kind == "draw"
|
||||||
|
assert (t.wins, t.losses) == (0, 1)
|
||||||
|
assert t.last_result_kind == "draw"
|
||||||
|
|
||||||
|
|
||||||
|
def test_evaluate_dedup_same_session():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="DSPL", display_target="DSPL")
|
||||||
|
assert evaluate(t, _teams(), "DSPL", False, "s1") == "win"
|
||||||
|
assert evaluate(t, _teams(), "DSPL", False, "s1") is None
|
||||||
|
assert (t.wins, t.losses) == (1, 0)
|
||||||
|
|
||||||
|
|
||||||
|
def test_evaluate_not_involved_returns_none():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="OTHER", display_target="OTHER")
|
||||||
|
assert evaluate(t, _teams(), "DSPL", False, "s1") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_format_status_fresh():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="DSPL", display_target="DSPL")
|
||||||
|
assert format_status(t, "en") == "0W-0L"
|
||||||
|
|
||||||
|
|
||||||
|
def test_format_status_after_win():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="DSPL", display_target="DSPL")
|
||||||
|
evaluate(t, _teams(), "DSPL", False, "s1")
|
||||||
|
assert format_status(t, "en") == "1W-0L: Win against ENEMY"
|
||||||
|
|
||||||
|
|
||||||
|
def test_format_status_after_draw():
|
||||||
|
t = Tally(guild_id=1, channel_id=2, mode="squadron", target="DSPL", display_target="DSPL")
|
||||||
|
evaluate(t, _teams(), None, True, "s1")
|
||||||
|
assert format_status(t, "en") == "0W-1L: Draw against ENEMY"
|
||||||
|
|
||||||
|
|
||||||
|
def _run():
|
||||||
|
fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
|
||||||
|
for fn in fns:
|
||||||
|
fn()
|
||||||
|
print(f"ok {fn.__name__}")
|
||||||
|
print(f"\nALL {len(fns)} TESTS PASSED")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
_run()
|
||||||
Reference in New Issue
Block a user