62c658fb57
Implements comprehensive automated system for weekly transaction freeze periods with priority-based contested player resolution. New Features: - Weekly freeze/thaw task (Monday 00:00 freeze, Saturday 00:00 thaw) - Priority resolution for contested transactions (worst teams get first priority) - Admin league management commands (/freeze-begin, /freeze-end, /advance-week) - Enhanced API client to handle string-based transaction IDs (moveids) - Service layer methods for transaction cancellation, unfreezing, and bulk operations - Offseason mode configuration flag to disable freeze operations Technical Changes: - api/client.py: URL-encode object_id parameter to handle colons in moveids - bot.py: Initialize and shutdown transaction freeze task - config.py: Add offseason_flag to BotConfig - services/league_service.py: Add update_current_state() for week/freeze updates - services/transaction_service.py: Add cancel/unfreeze methods with bulk support - tasks/transaction_freeze.py: Main freeze/thaw automation with error recovery - commands/admin/league_management.py: Manual admin controls for freeze system Infrastructure: - .gitlab-ci.yml and .gitlab/: GitLab CI/CD pipeline configuration - .mcp.json: MCP server configuration - Dockerfile.versioned: Versioned Docker build support - .dockerignore: Added .gitlab/ to ignore list Testing: - tests/test_tasks_transaction_freeze.py: Comprehensive freeze task tests The system uses team standings to fairly resolve contested players (multiple teams trying to acquire the same player), with worst-record teams getting priority. Includes comprehensive error handling, GM notifications, and admin reporting. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
208 lines
7.3 KiB
Python
208 lines
7.3 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
|
|
|
|
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")
|
|
|
|
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()
|
|
|
|
# Current state always has ID of 1 (single record)
|
|
current_id = 1
|
|
|
|
# Use BaseService patch method
|
|
updated_current = await self.patch(current_id, update_data)
|
|
|
|
if updated_current:
|
|
logger.info(f"Updated current state: {update_data}")
|
|
return updated_current
|
|
else:
|
|
logger.error("Failed to update current state - 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_current_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_current_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_current_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() |