""" Team Service Business logic for team operations: - CRUD operations - Roster management - Cache management """ import logging import copy from typing import List, Optional, Dict, Any, Literal from ..db_engine import db, Team, Manager, Division, model_to_dict, chunked, query_to_csv from .base import BaseService logger = logging.getLogger('discord_app') class TeamService(BaseService): """Service for team-related operations.""" cache_patterns = [ "teams*", "team*", "team-roster*" ] @classmethod def get_teams( cls, season: Optional[int] = None, owner_id: Optional[List[int]] = None, manager_id: Optional[List[int]] = None, team_abbrev: Optional[List[str]] = None, active_only: bool = False, short_output: bool = False, as_csv: bool = False ) -> Dict[str, Any]: """ Get teams with filtering. Args: season: Filter by season owner_id: Filter by Discord owner ID manager_id: Filter by manager IDs team_abbrev: Filter by abbreviations active_only: Exclude IL/MiL teams short_output: Exclude related data as_csv: Return as CSV Returns: Dict with count and teams list, or CSV string """ try: if season is not None: query = Team.select_season(season).order_by(Team.id.asc()) else: query = Team.select().order_by(Team.id.asc()) # Apply filters if manager_id: managers = Manager.select().where(Manager.id << manager_id) query = query.where( (Team.manager1_id << managers) | (Team.manager2_id << managers) ) if owner_id: query = query.where((Team.gmid << owner_id) | (Team.gmid2 << owner_id)) if team_abbrev: abbrev_list = [x.lower() for x in team_abbrev] query = query.where(peewee_fn.lower(Team.abbrev) << abbrev_list) if active_only: query = query.where( ~(Team.abbrev.endswith('IL')) & ~(Team.abbrev.endswith('MiL')) ) if as_csv: return query_to_csv(query, exclude=[Team.division_legacy, Team.mascot, Team.gsheet]) return { "count": query.count(), "teams": [model_to_dict(t, recurse=not short_output) for t in query] } except Exception as e: cls.handle_error(f"Error fetching teams: {e}", e) finally: cls.close_db() @classmethod def get_team(cls, team_id: int) -> Optional[Dict[str, Any]]: """Get a single team by ID.""" try: team = Team.get_or_none(Team.id == team_id) if team: return model_to_dict(team) return None except Exception as e: cls.handle_error(f"Error fetching team {team_id}: {e}", e) finally: cls.close_db() @classmethod def get_team_roster( cls, team_id: int, which: Literal['current', 'next'], sort: Optional[str] = None ) -> Dict[str, Any]: """ Get team roster with IL lists. Args: team_id: Team ID which: 'current' or 'next' week roster sort: Optional sort key Returns: Roster dict with active, short-il, long-il lists """ try: team = Team.get_by_id(team_id) if which == 'current': full_roster = team.get_this_week() else: full_roster = team.get_next_week() # Deep copy and convert to dicts result = { 'active': {'players': []}, 'shortil': {'players': []}, 'longil': {'players': []} } for section in ['active', 'shortil', 'longil']: players = copy.deepcopy(full_roster[section]['players']) result[section]['players'] = [model_to_dict(p) for p in players] # Apply sorting if sort == 'wara-desc': for section in ['active', 'shortil', 'longil']: result[section]['players'].sort(key=lambda p: p.get("wara", 0), reverse=True) return result except Exception as e: cls.handle_error(f"Error fetching roster for team {team_id}: {e}", e) finally: cls.close_db() @classmethod def update_team(cls, team_id: int, data: Dict[str, Any], token: str) -> Dict[str, Any]: """Update a team (partial update).""" cls.require_auth(token) try: team = Team.get_or_none(Team.id == team_id) if not team: from fastapi import HTTPException raise HTTPException(status_code=404, detail=f"Team ID {team_id} not found") # Apply updates for key, value in data.items(): if value is not None and hasattr(team, key): # Handle special cases if key.endswith('_id') and value == 0: setattr(team, key[:-3], None) elif key == 'division_id' and value == 0: team.division = None else: setattr(team, key, value) team.save() return cls.get_team(team_id) except Exception as e: cls.handle_error(f"Error updating team {team_id}: {e}", e) finally: cls.invalidate_related_cache(cls.cache_patterns) cls.close_db() @classmethod def create_teams(cls, teams_data: List[Dict[str, Any]], token: str) -> Dict[str, str]: """Create multiple teams.""" cls.require_auth(token) try: for team in teams_data: dupe = Team.get_or_none( Team.season == team.get("season"), Team.abbrev == team.get("abbrev") ) if dupe: from fastapi import HTTPException raise HTTPException( status_code=500, detail=f"Team {team.get('abbrev')} already exists in Season {team.get('season')}" ) # Validate foreign keys for field, model in [('manager1_id', Manager), ('manager2_id', Manager), ('division_id', Division)]: if team.get(field) and not model.get_or_none(Model.id == team[field]): from fastapi import HTTPException raise HTTPException(status_code=404, detail=f"{field} {team[field]} not found") with db.atomic(): for batch in chunked(teams_data, 15): Team.insert_many(batch).on_conflict_ignore().execute() return {"message": f"Inserted {len(teams_data)} teams"} except Exception as e: cls.handle_error(f"Error creating teams: {e}", e) finally: cls.invalidate_related_cache(cls.cache_patterns) cls.close_db() @classmethod def delete_team(cls, team_id: int, token: str) -> Dict[str, str]: """Delete a team.""" cls.require_auth(token) try: team = Team.get_or_none(Team.id == team_id) if not team: from fastapi import HTTPException raise HTTPException(status_code=404, detail=f"Team ID {team_id} not found") team.delete_instance() return {"message": f"Team {team_id} deleted"} except Exception as e: cls.handle_error(f"Error deleting team {team_id}: {e}", e) finally: cls.invalidate_related_cache(cls.cache_patterns) cls.close_db() # Fix peewee_fn reference from peewee import fn as peewee_fn