from fastapi import APIRouter, Depends, HTTPException, Query, Response from typing import List, Optional import logging import pydantic from pandas import DataFrame from ..db_engine import db, Player, model_to_dict, chunked, fn, complex_data_to_csv from ..dependencies import ( add_cache_headers, cache_result, oauth2_scheme, valid_token, PRIVATE_IN_SCHEMA, handle_db_errors, invalidate_cache, ) logger = logging.getLogger("discord_app") router = APIRouter(prefix="/api/v3/players", tags=["players"]) class PlayerModel(pydantic.BaseModel): name: str wara: float image: str image2: Optional[str] = None team_id: int season: int pitcher_injury: Optional[int] = None pos_1: str pos_2: Optional[str] = None pos_3: Optional[str] = None pos_4: Optional[str] = None pos_5: Optional[str] = None pos_6: Optional[str] = None pos_7: Optional[str] = None pos_8: Optional[str] = None vanity_card: Optional[str] = None headshot: Optional[str] = None last_game: Optional[str] = None last_game2: Optional[str] = None il_return: Optional[str] = None demotion_week: Optional[int] = None strat_code: Optional[str] = None bbref_id: Optional[str] = None injury_rating: Optional[str] = None sbaplayer_id: Optional[int] = None class PlayerList(pydantic.BaseModel): players: List[PlayerModel] @router.get("") @handle_db_errors @add_cache_headers( max_age=30 * 60 ) # 30 minutes - safe with cache invalidation on writes @cache_result(ttl=30 * 60, key_prefix="players") async def get_players( season: Optional[int], name: Optional[str] = None, team_id: list = Query(default=None), pos: list = Query(default=None), strat_code: list = Query(default=None), is_injured: Optional[bool] = None, sort: Optional[str] = None, short_output: Optional[bool] = False, csv: Optional[bool] = False, ): all_players = Player.select_season(season) if team_id is not None: all_players = all_players.where(Player.team_id << team_id) if strat_code is not None: code_list = [x.lower() for x in strat_code] all_players = all_players.where(fn.Lower(Player.strat_code) << code_list) if name is not None: all_players = all_players.where(fn.lower(Player.name) == name.lower()) if pos is not None: p_list = [x.upper() for x in pos] all_players = all_players.where( (Player.pos_1 << p_list) | (Player.pos_2 << p_list) | (Player.pos_3 << p_list) | (Player.pos_4 << p_list) | (Player.pos_5 << p_list) | (Player.pos_6 << p_list) | (Player.pos_7 << p_list) | (Player.pos_8 << p_list) ) if is_injured is not None: all_players = all_players.where(Player.il_return.is_null(False)) if sort is not None: if sort == "cost-asc": all_players = all_players.order_by(Player.wara) elif sort == "cost-desc": all_players = all_players.order_by(-Player.wara) elif sort == "name-asc": all_players = all_players.order_by(Player.name) elif sort == "name-desc": all_players = all_players.order_by(-Player.name) else: all_players = all_players.order_by(Player.id) if csv: player_list = [ [ "name", "wara", "image", "image2", "team", "season", "pitcher_injury", "pos_1", "pos_2", "pos_3", "pos_4", "pos_5", "pos_6", "pos_7", "pos_8", "last_game", "last_game2", "il_return", "demotion_week", "headshot", "vanity_card", "strat_code", "bbref_id", "injury_rating", "player_id", "sbaref_id", ] ] for line in all_players: player_list.append( [ line.name, line.wara, line.image, line.image2, line.team.abbrev, line.season, line.pitcher_injury, line.pos_1, line.pos_2, line.pos_3, line.pos_4, line.pos_5, line.pos_6, line.pos_7, line.pos_8, line.last_game, line.last_game2, line.il_return, line.demotion_week, line.headshot, line.vanity_card, line.strat_code.replace(",", "-_-") if line.strat_code is not None else "", line.bbref_id, line.injury_rating, line.id, line.sbaplayer, ] ) return_players = { "count": all_players.count(), "players": DataFrame(player_list).to_csv(header=False, index=False), "csv": True, } db.close() return Response(content=return_players["players"], media_type="text/csv") else: return_players = { "count": all_players.count(), "players": [ model_to_dict(x, recurse=not short_output) for x in all_players ], } db.close() # if csv: # return Response(content=complex_data_to_csv(return_players['players']), media_type='text/csv') return return_players @router.get("/search") @handle_db_errors @add_cache_headers( max_age=15 * 60 ) # 15 minutes - safe with cache invalidation on writes @cache_result(ttl=15 * 60, key_prefix="players-search") async def search_players( q: str = Query(..., description="Search query for player name"), season: Optional[int] = Query( default=None, description="Season to search in. Use 0 or omit for all seasons, or specific season number.", ), limit: int = Query( default=10, ge=1, le=50, description="Maximum number of results to return" ), short_output: bool = False, ): """ Real-time fuzzy search for players by name. Returns players matching the query with exact matches prioritized over partial matches. Season parameter: - Omit or use 0: Search across ALL seasons (most recent seasons prioritized) - Specific number (1-13+): Search only that season """ search_all_seasons = season is None or season == 0 if search_all_seasons: # Search across all seasons - no season filter all_players = ( Player.select() .where(fn.lower(Player.name).contains(q.lower())) .order_by(-Player.season) ) # Most recent seasons first else: # Search specific season all_players = Player.select_season(season).where( fn.lower(Player.name).contains(q.lower()) ) # Convert to list for sorting players_list = list(all_players) # Sort by relevance (exact matches first, then partial) # For all-season search, also prioritize by season (most recent first) query_lower = q.lower() exact_matches = [] partial_matches = [] for player in players_list: name_lower = player.name.lower() if name_lower == query_lower: exact_matches.append(player) elif query_lower in name_lower: partial_matches.append(player) # Sort exact and partial matches by season (most recent first) when searching all seasons if search_all_seasons: exact_matches.sort(key=lambda p: p.season, reverse=True) partial_matches.sort(key=lambda p: p.season, reverse=True) # Combine and limit results results = exact_matches + partial_matches limited_results = results[:limit] db.close() return { "count": len(limited_results), "total_matches": len(results), "all_seasons": search_all_seasons, "players": [ model_to_dict(x, recurse=not short_output) for x in limited_results ], } @router.get("/{player_id}") @handle_db_errors @add_cache_headers( max_age=30 * 60 ) # 30 minutes - safe with cache invalidation on writes @cache_result(ttl=30 * 60, key_prefix="player") async def get_one_player(player_id: int, short_output: Optional[bool] = False): this_player = Player.get_or_none(Player.id == player_id) if this_player: r_player = model_to_dict(this_player, recurse=not short_output) else: r_player = None db.close() return r_player @router.put("/{player_id}", include_in_schema=PRIVATE_IN_SCHEMA) @handle_db_errors async def put_player( player_id: int, new_player: PlayerModel, token: str = Depends(oauth2_scheme) ): if not valid_token(token): logger.warning(f"patch_player - Bad Token: {token}") raise HTTPException(status_code=401, detail="Unauthorized") if Player.get_or_none(Player.id == player_id) is None: db.close() raise HTTPException(status_code=404, detail=f"Player ID {player_id} not found") Player.update(**new_player.dict()).where(Player.id == player_id).execute() r_player = model_to_dict(Player.get_by_id(player_id)) db.close() # Invalidate player-related cache entries invalidate_cache("players*") invalidate_cache("players-search*") invalidate_cache(f"player*{player_id}*") # Invalidate team roster cache (player data affects team rosters) invalidate_cache("team-roster*") return r_player @router.patch("/{player_id}", include_in_schema=PRIVATE_IN_SCHEMA) @handle_db_errors async def patch_player( player_id: int, token: str = Depends(oauth2_scheme), name: Optional[str] = None, wara: Optional[float] = None, image: Optional[str] = None, image2: Optional[str] = None, team_id: Optional[int] = None, season: Optional[int] = None, pos_1: Optional[str] = None, pos_2: Optional[str] = None, pos_3: Optional[str] = None, pos_4: Optional[str] = None, pos_5: Optional[str] = None, pos_6: Optional[str] = None, pos_7: Optional[str] = None, pos_8: Optional[str] = None, vanity_card: Optional[str] = None, headshot: Optional[str] = None, il_return: Optional[str] = None, demotion_week: Optional[int] = None, strat_code: Optional[str] = None, bbref_id: Optional[str] = None, injury_rating: Optional[str] = None, sbaref_id: Optional[int] = None, ): if not valid_token(token): logger.warning(f"patch_player - Bad Token: {token}") raise HTTPException(status_code=401, detail="Unauthorized") if Player.get_or_none(Player.id == player_id) is None: db.close() raise HTTPException(status_code=404, detail=f"Player ID {player_id} not found") this_player = Player.get_or_none(Player.id == player_id) if this_player is None: db.close() raise HTTPException(status_code=404, detail=f"Player ID {player_id} not found") if name is not None: this_player.name = name if wara is not None: this_player.wara = wara if image is not None: this_player.image = image if image2 is not None: this_player.image2 = image2 if team_id is not None: this_player.team_id = team_id if season is not None: this_player.season = season if pos_1 is not None: this_player.pos_1 = pos_1 if pos_2 is not None: this_player.pos_2 = pos_2 if pos_3 is not None: this_player.pos_3 = pos_3 if pos_4 is not None: this_player.pos_4 = pos_4 if pos_5 is not None: this_player.pos_5 = pos_5 if pos_6 is not None: this_player.pos_6 = pos_6 if pos_7 is not None: this_player.pos_7 = pos_7 if pos_8 is not None: this_player.pos_8 = pos_8 if pos_8 is not None: this_player.pos_8 = pos_8 if vanity_card is not None: this_player.vanity_card = vanity_card if headshot is not None: this_player.headshot = headshot if il_return is not None: this_player.il_return = ( None if not il_return or il_return.lower() == "none" else il_return ) if demotion_week is not None: this_player.demotion_week = demotion_week if strat_code is not None: this_player.strat_code = strat_code if bbref_id is not None: this_player.bbref_id = bbref_id if injury_rating is not None: this_player.injury_rating = injury_rating if sbaref_id is not None: this_player.sbaplayer_id = sbaref_id if this_player.save() == 1: r_player = model_to_dict(this_player) db.close() # Invalidate player-related cache entries invalidate_cache("players*") invalidate_cache("players-search*") invalidate_cache(f"player*{player_id}*") # Invalidate team roster cache (player data affects team rosters) invalidate_cache("team-roster*") return r_player else: db.close() raise HTTPException( status_code=500, detail=f"Unable to patch player {player_id}" ) @router.post("", include_in_schema=PRIVATE_IN_SCHEMA) @handle_db_errors async def post_players(p_list: PlayerList, token: str = Depends(oauth2_scheme)): if not valid_token(token): logger.warning(f"post_players - Bad Token: {token}") raise HTTPException(status_code=401, detail="Unauthorized") new_players = [] for player in p_list.players: dupe = Player.get_or_none( Player.season == player.season, Player.name == player.name ) if dupe: db.close() raise HTTPException( status_code=500, detail=f"Player name {player.name} already in use in Season {player.season}", ) new_players.append(player.dict()) with db.atomic(): for batch in chunked(new_players, 15): Player.insert_many(batch).on_conflict_ignore().execute() db.close() # Invalidate player-related cache entries invalidate_cache("players*") invalidate_cache("players-search*") invalidate_cache("player*") # Invalidate team roster cache (new players added to teams) invalidate_cache("team-roster*") return f"Inserted {len(new_players)} players" @router.delete("/{player_id}", include_in_schema=PRIVATE_IN_SCHEMA) @handle_db_errors async def delete_player(player_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logger.warning(f"delete_player - Bad Token: {token}") raise HTTPException(status_code=401, detail="Unauthorized") this_player = Player.get_or_none(Player.id == player_id) if not this_player: db.close() raise HTTPException(status_code=404, detail=f"Player ID {player_id} not found") count = this_player.delete_instance() db.close() if count == 1: # Invalidate player-related cache entries invalidate_cache("players*") invalidate_cache("players-search*") invalidate_cache(f"player*{player_id}*") # Invalidate team roster cache (player removed from team) invalidate_cache("team-roster*") return f"Player {player_id} has been deleted" else: raise HTTPException( status_code=500, detail=f"Player {player_id} could not be deleted" )