mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-04 03:52:34 +00:00
agessaman
7aba432b54
S4. BaseCommand and the 40 bundled overrides that took no argument now take `message: MeshMessage | None = None`, matching joke and rain, which already use the message. None of the 40 bodies read it, so their help text is unchanged. help and command_manager keep their TypeError retry, since local plugins may still implement get_help_text(self). For bundled commands the retry no longer runs: `help <command>` calls get_help_text(message) once instead of failing and calling it again. A new test pins the signature for every bundled command class. The offline smoke run against the real config produces the same help text for all 46 commands.
298 lines
13 KiB
Python
298 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
World Cup command for the MeshCore Bot.
|
|
|
|
Provides FIFA World Cup coverage (men's fifa.world / women's fifa.wwc) via the ESPN API,
|
|
but only while a tournament is actually in progress (determined from the ESPN tournament
|
|
calendar). Supports:
|
|
|
|
wc / worldcup -> today's scores, including live results
|
|
wc group <X> -> standings for group X
|
|
wc <nation> -> that nation's matches in the current tournament
|
|
|
|
World Cup data is also reachable year-round through the `sports` command
|
|
(e.g. `sports fifa`, `sports brazil`); this command is the season-gated, purpose-built
|
|
front end with live scores, standings, and nation lookups.
|
|
"""
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from ..clients.espn_client import ESPNClient
|
|
from ..clients.sports_mappings import SPORT_EMOJIS
|
|
from ..clients.worldcup_data import WorldCupData
|
|
from ..models import MeshMessage
|
|
from ..utils import espn_dates_for_local_day, filter_events_local_day, get_config_timezone
|
|
from .base_command import BaseCommand
|
|
|
|
if TYPE_CHECKING:
|
|
from ..core import MeshCoreBot
|
|
|
|
SOCCER_EMOJI = SPORT_EMOJIS.get("soccer", "⚽")
|
|
FINAL_STATUSES = {"STATUS_FINAL", "STATUS_FULL_TIME", "STATUS_FINAL_PEN", "STATUS_POSTPONED"}
|
|
|
|
|
|
class WorldCupCommand(BaseCommand):
|
|
"""Season-gated FIFA World Cup scores, standings, and nation lookups."""
|
|
|
|
# Plugin metadata
|
|
# Read-only informational output; safe for scheduled {cmd:...} rendering.
|
|
render_safe = True
|
|
name = "worldcup"
|
|
honors_skip_channel_check = False
|
|
enabled_attr = "worldcup_enabled"
|
|
keywords = ["wc", "worldcup"]
|
|
description = "FIFA World Cup scores, standings, and nation results (in-season only)"
|
|
category = "sports"
|
|
cooldown_seconds = 3
|
|
requires_internet = True
|
|
|
|
# Documentation
|
|
short_description = "FIFA World Cup scores (in-season)"
|
|
usage = "wc [group <X> | <nation>]"
|
|
examples = ["wc", "wc group a", "wc brazil"]
|
|
parameters = [
|
|
{"name": "group <X>", "description": "Standings for a group (e.g. group a)"},
|
|
{"name": "nation", "description": "A nation's matches (e.g. brazil, usa)"},
|
|
]
|
|
|
|
def __init__(self, bot: "MeshCoreBot"):
|
|
"""Initialize the World Cup command with an ESPN client and data helper."""
|
|
super().__init__(bot)
|
|
self.url_timeout = self.get_config_value("Worldcup_Command", "api_timeout", fallback=10, value_type="int")
|
|
|
|
# Standard enabled flag
|
|
self.worldcup_enabled = self.get_config_value("Worldcup_Command", "enabled", fallback=True, value_type="bool")
|
|
|
|
cache_ttl_minutes = self.get_config_value("Worldcup_Command", "cache_ttl_minutes", fallback=360, value_type="int")
|
|
|
|
self.espn_client: ESPNClient = ESPNClient(logger=self.logger, timeout=self.url_timeout)
|
|
self.wc_data = WorldCupData(self.espn_client, logger=self.logger, cache_ttl=cache_ttl_minutes * 60)
|
|
|
|
def matches_keyword(self, message: MeshMessage) -> bool:
|
|
"""Match only when a keyword is the first word (mirrors SportsCommand)."""
|
|
if not self.keywords:
|
|
return False
|
|
|
|
def _matches(content_lower: str) -> bool:
|
|
words = content_lower.split()
|
|
if not words:
|
|
return False
|
|
return any(words[0] == keyword.lower() for keyword in self.keywords)
|
|
|
|
return self._cleaned_content_matches(message, _matches)
|
|
|
|
def get_help_text(self, message: MeshMessage | None = None) -> str:
|
|
return self.translate("commands.worldcup.help")
|
|
|
|
# ------------------------------------------------------------------ helpers
|
|
|
|
def _pack_lines(self, lines: list[str], max_len: int, max_chunks: int = 3) -> list[str]:
|
|
"""Greedily pack lines into up to max_chunks messages within max_len UTF-8 bytes."""
|
|
chunks: list[str] = []
|
|
current = ""
|
|
for line in lines:
|
|
candidate = line if not current else f"{current}\n{line}"
|
|
if len(candidate.encode("utf-8")) <= max_len:
|
|
current = candidate
|
|
continue
|
|
# Line doesn't fit in current chunk; flush and start a new one
|
|
if current:
|
|
chunks.append(current)
|
|
if len(chunks) >= max_chunks:
|
|
return chunks
|
|
current = ""
|
|
if len(line.encode("utf-8")) <= max_len:
|
|
current = line
|
|
else:
|
|
# Single oversized line: hard-truncate on a byte boundary
|
|
truncated = line.encode("utf-8")[:max_len].decode("utf-8", "ignore")
|
|
chunks.append(truncated)
|
|
if len(chunks) >= max_chunks:
|
|
return chunks
|
|
if current and len(chunks) < max_chunks:
|
|
chunks.append(current)
|
|
return chunks[:max_chunks]
|
|
|
|
@staticmethod
|
|
def _today_sort_key(game: dict) -> tuple[int, float]:
|
|
"""Order today's games: live first, then completed results, then upcoming."""
|
|
status = game.get("status", "")
|
|
ts = game.get("event_timestamp") or 0
|
|
if game.get("timestamp", 0) < 0:
|
|
return (0, ts) # live / halftime
|
|
if status in ("STATUS_FINAL", "STATUS_FULL_TIME", "STATUS_FINAL_PEN", "STATUS_POSTPONED"):
|
|
return (1, ts) # completed results
|
|
return (2, ts) # scheduled later
|
|
|
|
@staticmethod
|
|
def _dedupe_games(games: list[dict]) -> list[dict]:
|
|
"""Drop duplicate events (the team-schedule endpoint can repeat today's game)."""
|
|
seen = set()
|
|
unique = []
|
|
for game in games:
|
|
key = game.get("id") or game.get("formatted")
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
unique.append(game)
|
|
return unique
|
|
|
|
@staticmethod
|
|
def _select_nation_games(games: list[dict], in_group_stage: bool) -> list[dict]:
|
|
"""Pick which of a nation's matches to show so the reply fits one message.
|
|
|
|
During the group stage, show all of the nation's (known) group matches. Afterward,
|
|
show just the last result (or a live game) plus the next scheduled fixture.
|
|
"""
|
|
if in_group_stage:
|
|
return games
|
|
|
|
live = [g for g in games if g.get("timestamp", 0) < 0]
|
|
completed = sorted(
|
|
[g for g in games if g.get("status") in FINAL_STATUSES],
|
|
key=lambda g: g.get("event_timestamp") or 0,
|
|
reverse=True,
|
|
)
|
|
upcoming = sorted(
|
|
[g for g in games if g.get("status") == "STATUS_SCHEDULED"],
|
|
key=lambda g: g.get("event_timestamp") or 0,
|
|
)
|
|
selected: list[dict] = []
|
|
if live:
|
|
selected.extend(live)
|
|
elif completed:
|
|
selected.append(completed[0])
|
|
if upcoming:
|
|
selected.append(upcoming[0])
|
|
return selected
|
|
|
|
def _format_games(self, games: list[dict]) -> list[str]:
|
|
"""Format parsed game dicts into '⚽ ...' lines (emoji de-duplicated)."""
|
|
lines = []
|
|
for game in games:
|
|
formatted = game.get("formatted", "").strip()
|
|
if formatted and formatted[0] in SPORT_EMOJIS.values():
|
|
formatted = formatted[1:].strip()
|
|
lines.append(f"{SOCCER_EMOJI} {formatted}")
|
|
return lines
|
|
|
|
async def _refresh_live(self, games: list[dict], league: str) -> list[dict]:
|
|
"""Refresh any live games with up-to-date scoreboard data (for stale schedules)."""
|
|
for i, game in enumerate(games):
|
|
if game.get("timestamp", 0) < 0 and game.get("id"):
|
|
try:
|
|
live = await self.espn_client.fetch_live_event_data(game["id"], "soccer", league)
|
|
if live:
|
|
updated = self.espn_client.parse_game_event_with_timestamp(live, "", "soccer", league)
|
|
if updated:
|
|
games[i] = updated
|
|
except Exception as e:
|
|
self.logger.warning(f"World Cup live refresh failed for {game.get('id')}: {e}")
|
|
return games
|
|
|
|
# ----------------------------------------------------------------- handlers
|
|
|
|
async def _handle_today(self, message: MeshMessage, league: str) -> bool:
|
|
"""Send today's scores (live first), chunked across up to 3 messages."""
|
|
local_tz, _ = get_config_timezone(self.bot.config, self.logger)
|
|
start_date, end_date, local_start_ts, local_end_ts = espn_dates_for_local_day(local_tz)
|
|
data = await self.espn_client.fetch_scoreboard_with_calendar(
|
|
"soccer", league, start_date=start_date, end_date=end_date,
|
|
)
|
|
games = data.get("events", []) if data else []
|
|
games = filter_events_local_day(games, local_start_ts, local_end_ts)
|
|
if not games:
|
|
return await self.send_response(message, self.translate("commands.worldcup.no_games_today"))
|
|
|
|
games.sort(key=self._today_sort_key)
|
|
lines = self._format_games(games)
|
|
chunks = self._pack_lines(lines, self.get_max_message_length(message), max_chunks=3)
|
|
return await self.send_response_chunked(message, chunks)
|
|
|
|
async def _handle_group(self, message: MeshMessage, league: str, letter: str) -> bool:
|
|
"""Send standings for a single group, or list groups if no letter given."""
|
|
groups = await self.espn_client.fetch_standings("soccer", league)
|
|
if not groups:
|
|
return await self.send_response(message, self.translate("commands.worldcup.no_games_today"))
|
|
|
|
if not letter:
|
|
available = ", ".join(
|
|
g["group_name"].replace("Group ", "").strip() for g in groups if g.get("group_name")
|
|
)
|
|
return await self.send_response(message, self.translate("commands.worldcup.groups_list", groups=available))
|
|
|
|
letter = letter.strip().lower()
|
|
match = next(
|
|
(g for g in groups if g.get("group_name", "").strip().lower().endswith(f" {letter}")),
|
|
None,
|
|
)
|
|
if not match:
|
|
return await self.send_response(message, self.translate("commands.worldcup.group_not_found", group=letter.upper()))
|
|
|
|
lines = [f"{SOCCER_EMOJI} {match['group_name']}"]
|
|
for e in match["entries"]:
|
|
lines.append(f"{e['rank']}.{e['abbr']} {e['pts']}p {e['gd']:+d} ({e['w']}-{e['d']}-{e['l']})")
|
|
chunks = self._pack_lines(lines, self.get_max_message_length(message), max_chunks=2)
|
|
return await self.send_response_chunked(message, chunks)
|
|
|
|
async def _handle_nation(self, message: MeshMessage, active: dict, name: str) -> bool:
|
|
"""Send a nation's matches in the current tournament (results + upcoming fixtures)."""
|
|
league = active["league"]
|
|
team_info = await self.wc_data.resolve_nation(name, league)
|
|
if not team_info:
|
|
return await self.send_response(message, self.translate("commands.worldcup.nation_not_found", nation=name))
|
|
|
|
# Prefer the dated scoreboard range: it includes known future fixtures, which the
|
|
# team /schedule endpoint omits for tournament national teams. Fall back to the
|
|
# schedule endpoint if the tournament window is unavailable.
|
|
start, end = active.get("start_date"), active.get("end_date")
|
|
if start and end:
|
|
games = await self.espn_client.fetch_team_fixtures("soccer", league, team_info["team_id"], start, end)
|
|
else:
|
|
games = await self.espn_client.fetch_team_schedule("soccer", league, team_info["team_id"])
|
|
|
|
if not games:
|
|
return await self.send_response(message, self.translate("commands.worldcup.no_games_nation", nation=name))
|
|
|
|
games = self._dedupe_games(games)
|
|
games = await self._refresh_live(games, league)
|
|
games = self._select_nation_games(games, active.get("in_group_stage", False))
|
|
games.sort(key=self._today_sort_key)
|
|
lines = self._format_games(games)
|
|
chunks = self._pack_lines(lines, self.get_max_message_length(message), max_chunks=1)
|
|
return await self.send_response_chunked(message, chunks)
|
|
|
|
async def execute(self, message: MeshMessage) -> bool:
|
|
"""Main entry point: gate on season, then dispatch on subcommand."""
|
|
try:
|
|
self.record_execution(message.sender_id)
|
|
|
|
content = message.content.strip()
|
|
if content.startswith("!"):
|
|
content = content[1:].strip()
|
|
|
|
# Drop the leading keyword, keep the remainder as the argument string
|
|
parts = content.split(" ", 1)
|
|
args = parts[1].strip() if len(parts) > 1 else ""
|
|
|
|
active = await self.wc_data.get_active_tournament()
|
|
if not active:
|
|
return await self.send_response(message, self.translate("commands.worldcup.not_in_season"))
|
|
|
|
league = active["league"]
|
|
|
|
if not args:
|
|
return await self._handle_today(message, league)
|
|
|
|
lowered = args.lower()
|
|
if lowered == "group" or lowered == "groups":
|
|
return await self._handle_group(message, league, "")
|
|
if lowered.startswith("group "):
|
|
return await self._handle_group(message, league, args[6:].strip())
|
|
|
|
return await self._handle_nation(message, active, args)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error in worldcup execute: {e}")
|
|
return await self.send_response(message, self.translate("commands.worldcup.error_fetching"))
|