61f36353d8
Closes #91 - league_service.get_current_state(): @cached_single_item(ttl=60) — 60s Redis cache - standings_service.get_league_standings(): in-memory dict cache with 10-minute TTL keyed by season - player_service.get_free_agents(): @cached_api_call(ttl=300) — 5-minute Redis cache - dice/rolls.py _get_channel_embed_color(): in-memory dict cache keyed by channel_id with 5-minute TTL, matching the autocomplete.py pattern from PR #100 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
244 lines
8.0 KiB
Python
244 lines
8.0 KiB
Python
"""
|
|
League service for Discord Bot v2.0
|
|
|
|
Handles league-wide operations including current state, standings, and season information.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional, List, Dict, Any
|
|
|
|
from config import get_config
|
|
from services.base_service import BaseService
|
|
from models.current import Current
|
|
from exceptions import APIException
|
|
from utils.decorators import cached_single_item
|
|
|
|
logger = logging.getLogger(f"{__name__}.LeagueService")
|
|
|
|
|
|
class LeagueService(BaseService[Current]):
|
|
"""
|
|
Service for league-wide operations.
|
|
|
|
Features:
|
|
- Current league state retrieval
|
|
- Season standings
|
|
- League-wide statistics
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""Initialize league service."""
|
|
super().__init__(Current, "current")
|
|
logger.debug("LeagueService initialized")
|
|
|
|
@cached_single_item(ttl=60)
|
|
async def get_current_state(self) -> Optional[Current]:
|
|
"""
|
|
Get the current league state including week, season, and settings.
|
|
|
|
Returns:
|
|
Current league state or None if not available
|
|
"""
|
|
try:
|
|
client = await self.get_client()
|
|
data = await client.get("current")
|
|
|
|
if data:
|
|
current = Current.from_api_data(data)
|
|
logger.debug(
|
|
f"Retrieved current state: Week {current.week}, Season {current.season}"
|
|
)
|
|
return current
|
|
|
|
logger.debug("No current state data found")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get current league state: {e}")
|
|
return None
|
|
|
|
async def update_current_state(
|
|
self, week: Optional[int] = None, freeze: Optional[bool] = None
|
|
) -> Optional[Current]:
|
|
"""
|
|
Update current league state (week and/or freeze status).
|
|
|
|
This is typically used by automated tasks to increment the week
|
|
and toggle freeze status during weekly operations.
|
|
|
|
Args:
|
|
week: New week number (None to leave unchanged)
|
|
freeze: New freeze status (None to leave unchanged)
|
|
|
|
Returns:
|
|
Updated Current object or None if update failed
|
|
|
|
Raises:
|
|
APIException: If the update operation fails
|
|
"""
|
|
try:
|
|
# Build update data
|
|
update_data = {}
|
|
if week is not None:
|
|
update_data["week"] = week
|
|
if freeze is not None:
|
|
update_data["freeze"] = freeze
|
|
|
|
if not update_data:
|
|
logger.warning("update_current_state called with no updates")
|
|
return await self.get_current_state()
|
|
|
|
# Get the current state to find its actual ID
|
|
# (Current table has one row per season, NOT a single row with id=1)
|
|
current = await self.get_current_state()
|
|
if not current:
|
|
logger.error(
|
|
"Cannot update current state - unable to fetch current state"
|
|
)
|
|
return None
|
|
|
|
current_id = current.id
|
|
logger.debug(
|
|
f"Updating current state id={current_id} (season {current.season})"
|
|
)
|
|
|
|
# Use BaseService patch method
|
|
updated_current = await self.patch(
|
|
current_id, update_data, use_query_params=True
|
|
)
|
|
|
|
if updated_current:
|
|
logger.info(f"Updated current state id={current_id}: {update_data}")
|
|
return updated_current
|
|
else:
|
|
logger.error(
|
|
f"Failed to update current state id={current_id} - patch returned None"
|
|
)
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating current state: {e}")
|
|
raise APIException(f"Failed to update current state: {e}")
|
|
|
|
async def get_standings(
|
|
self, season: Optional[int] = None
|
|
) -> Optional[List[Dict[str, Any]]]:
|
|
"""
|
|
Get league standings for a season.
|
|
|
|
Args:
|
|
season: Season number (defaults to current season)
|
|
|
|
Returns:
|
|
List of standings data or None if not available
|
|
"""
|
|
try:
|
|
season = season or get_config().sba_season
|
|
client = await self.get_client()
|
|
data = await client.get("standings", params=[("season", str(season))])
|
|
|
|
if data and isinstance(data, list):
|
|
logger.debug(
|
|
f"Retrieved standings for season {season}: {len(data)} teams"
|
|
)
|
|
return data
|
|
elif data and isinstance(data, dict):
|
|
# Handle case where API returns a dict with standings array
|
|
standings_data = data.get("standings", data.get("items", []))
|
|
if standings_data:
|
|
logger.debug(
|
|
f"Retrieved standings for season {season}: {len(standings_data)} teams"
|
|
)
|
|
return standings_data
|
|
|
|
logger.debug(f"No standings data found for season {season}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get standings for season {season}: {e}")
|
|
return None
|
|
|
|
async def get_division_standings(
|
|
self, division_id: int, season: Optional[int] = None
|
|
) -> Optional[List[Dict[str, Any]]]:
|
|
"""
|
|
Get standings for a specific division.
|
|
|
|
Args:
|
|
division_id: Division identifier
|
|
season: Season number (defaults to current season)
|
|
|
|
Returns:
|
|
List of division standings or None if not available
|
|
"""
|
|
try:
|
|
season = season or get_config().sba_season
|
|
client = await self.get_client()
|
|
data = await client.get(
|
|
f"standings/division/{division_id}", params=[("season", str(season))]
|
|
)
|
|
|
|
if data and isinstance(data, list):
|
|
logger.debug(
|
|
f"Retrieved division {division_id} standings for season {season}: {len(data)} teams"
|
|
)
|
|
return data
|
|
|
|
logger.debug(
|
|
f"No division standings found for division {division_id}, season {season}"
|
|
)
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get division {division_id} standings: {e}")
|
|
return None
|
|
|
|
async def get_league_leaders(
|
|
self, stat_type: str = "batting", season: Optional[int] = None, limit: int = 10
|
|
) -> Optional[List[Dict[str, Any]]]:
|
|
"""
|
|
Get league leaders for a specific statistic category.
|
|
|
|
Args:
|
|
stat_type: Type of stats ('batting', 'pitching', 'fielding')
|
|
season: Season number (defaults to current season)
|
|
limit: Number of leaders to return
|
|
|
|
Returns:
|
|
List of league leaders or None if not available
|
|
"""
|
|
try:
|
|
season = season or get_config().sba_season
|
|
client = await self.get_client()
|
|
|
|
params = [("season", str(season)), ("limit", str(limit))]
|
|
|
|
data = await client.get(f"leaders/{stat_type}", params=params)
|
|
|
|
if data:
|
|
# Handle different response formats
|
|
if isinstance(data, list):
|
|
leaders = data
|
|
elif isinstance(data, dict):
|
|
leaders = data.get(
|
|
"leaders", data.get("items", data.get("results", []))
|
|
)
|
|
else:
|
|
leaders = []
|
|
|
|
logger.debug(
|
|
f"Retrieved {stat_type} leaders for season {season}: {len(leaders)} players"
|
|
)
|
|
return leaders[:limit] # Ensure we don't exceed limit
|
|
|
|
logger.debug(f"No {stat_type} leaders found for season {season}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get {stat_type} leaders for season {season}: {e}")
|
|
return None
|
|
|
|
|
|
# Global service instance
|
|
league_service = LeagueService()
|