from fastapi import APIRouter, Depends, HTTPException, Query, Response from typing import Literal, Optional, List import logging import pandas as pd import pydantic from pydantic import validator, root_validator from ..db_engine import db, BattingCardRatings, model_to_dict, chunked, BattingCard, Player, query_to_csv, Team from ..dependencies import oauth2_scheme, valid_token, LOG_DATA logging.basicConfig( filename=LOG_DATA['filename'], format=LOG_DATA['format'], level=LOG_DATA['log_level'] ) router = APIRouter( prefix='/api/v2/battingcardratings', tags=['battingcardratings'] ) class BattingCardRatingsModel(pydantic.BaseModel): battingcard_id: int vs_hand: Literal['R', 'L', 'vR', 'vL'] homerun: float = 0.0 bp_homerun: float = 0.0 triple: float = 0.0 double_three: float = 0.0 double_two: float = 0.0 double_pull: float = 0.0 single_two: float = 0.0 single_one: float = 0.0 single_center: float = 0.0 bp_single: float = 0.0 hbp: float = 0.0 walk: float = 0.0 strikeout: float = 0.0 lineout: float = 0.0 popout: float = 0.0 flyout_a: float = 0.0 flyout_bq: float = 0.0 flyout_lf_b: float = 0.0 flyout_rf_b: float = 0.0 groundout_a: float = 0.0 groundout_b: float = 0.0 groundout_c: float = 0.0 avg: float = 0.0 obp: float = 0.0 slg: float = 0.0 pull_rate: float = 0.0 center_rate: float = 0.0 slap_rate: float = 0.0 @validator("avg", always=True) def avg_validator(cls, v, values, **kwargs): return (values['homerun'] + values['bp_homerun'] / 2 + values['triple'] + values['double_three'] + values['double_two'] + values['double_pull'] + values['single_two'] + values['single_one'] + values['single_center'] + values['bp_single'] / 2) / 108 @validator("obp", always=True) def obp_validator(cls, v, values, **kwargs): return ((values['hbp'] + values['walk']) / 108) + values['avg'] @validator("slg", always=True) def slg_validator(cls, v, values, **kwargs): return (values['homerun'] * 4 + values['bp_homerun'] * 2 + values['triple'] * 3 + values['double_three'] * 2 + values['double_two'] * 2 + values['double_pull'] * 2 + values['single_two'] + values['single_one'] + values['single_center'] + values['bp_single'] / 2) / 108 @root_validator def validate_chance_total(cls, values): total_chances = ( values['homerun'] + values['bp_homerun'] + values['triple'] + values['double_three'] + values['double_two'] + values['double_pull'] + values['single_two'] + values['single_one'] + values['single_center'] + values['bp_single'] + values['hbp'] + values['walk'] + values['strikeout'] + values['lineout'] + values['popout'] + values['flyout_a'] + values['flyout_bq'] + values['flyout_lf_b'] + values['flyout_rf_b'] + values['groundout_a'] + values['groundout_b'] + values['groundout_c']) if round(total_chances) != 108: raise ValueError("Must have exactly 108 chances on the card") return values class RatingsList(pydantic.BaseModel): ratings: List[BattingCardRatingsModel] @router.get('') async def get_card_ratings( team_id: int, ts: str, battingcard_id: list = Query(default=None), cardset_id: list = Query(default=None), vs_hand: Literal['R', 'L', 'vR', 'vL'] = None, short_output: bool = False, csv: bool = False): this_team = Team.get_or_none(Team.id == team_id) logging.debug(f'Team: {this_team} / has_guide: {this_team.has_guide}') if this_team is None or ts != this_team.team_hash() or this_team.has_guide != 1: logging.warning(f'Team_id {team_id} attempted to pull ratings') db.close() raise HTTPException( status_code=401, detail='You are not authorized to pull card ratings.' ) # elif not valid_token(token): # logging.warning(f'Bad Token: {token}') # db.close() # raise HTTPException( # status_code=401, # detail='You are not authorized to pull card ratings.' # ) all_ratings = BattingCardRatings.select() if battingcard_id is not None: all_ratings = all_ratings.where(BattingCardRatings.battingcard_id << battingcard_id) if vs_hand is not None: all_ratings = all_ratings.where(BattingCardRatings.vs_hand == vs_hand[-1]) if cardset_id is not None: set_players = Player.select(Player.player_id).where(Player.cardset_id << cardset_id) set_cards = BattingCard.select(BattingCard.id).where(BattingCard.player << set_players) all_ratings = all_ratings.where(BattingCardRatings.battingcard << set_cards) if csv: # return_val = query_to_csv(all_ratings) return_vals = [model_to_dict(x) for x in all_ratings] for x in return_vals: x.update(x['battingcard']) x['player_id'] = x['battingcard']['player']['player_id'] del x['battingcard'] del x['player'] db.close() return Response(content=pd.DataFrame(return_vals).to_csv(index=False), media_type='text/csv') else: return_val = {'count': all_ratings.count(), 'ratings': [ model_to_dict(x, recurse=not short_output) for x in all_ratings ]} db.close() return return_val @router.get('/scouting') async def get_card_scouting(team_id: int, ts: str, cardset_id: list = Query(default=None)): this_team = Team.get_or_none(Team.id == team_id) logging.debug(f'Team: {this_team} / has_guide: {this_team.has_guide}') if this_team is None or ts != this_team.team_hash() or this_team.has_guide != 1: logging.warning(f'Team_id {team_id} attempted to pull ratings') db.close() return 'Your team does not have the ratings guide enabled. If you have purchased a copy ping Cal to ' \ 'make sure it is enabled on your team. If you are interested you can pick it up here (thank you!): ' \ 'https://ko-fi.com/manticorum/shop' all_ratings = BattingCardRatings.select() if cardset_id is not None: set_players = Player.select(Player.player_id).where(Player.cardset_id << cardset_id) set_cards = BattingCard.select(BattingCard.id).where(BattingCard.player << set_players) all_ratings = all_ratings.where(BattingCardRatings.battingcard << set_cards) vl_query = all_ratings.where(BattingCardRatings.vs_hand == 'L') vr_query = all_ratings.where(BattingCardRatings.vs_hand == 'R') vl_vals = [model_to_dict(x) for x in vl_query] for x in vl_vals: x.update(x['battingcard']) x['player_id'] = x['battingcard']['player']['player_id'] x['player_name'] = x['battingcard']['player']['p_name'] x['rarity'] = x['battingcard']['player']['rarity']['name'] x['cardset_id'] = x['battingcard']['player']['cardset']['id'] x['cardset_name'] = x['battingcard']['player']['cardset']['name'] del x['battingcard'] del x['player'] vr_vals = [model_to_dict(x) for x in vr_query] for x in vr_vals: x['player_id'] = x['battingcard']['player']['player_id'] del x['battingcard'] vl = pd.DataFrame(vl_vals) vr = pd.DataFrame(vr_vals) db.close() output = pd.merge(vl, vr, on='player_id', suffixes=('_vl', '_vr')) first = ['player_id', 'player_name', 'cardset_name', 'rarity', 'hand', 'variant'] exclude = first + ['id_vl', 'id_vr', 'vs_hand_vl', 'vs_hand_vr'] output = output[first + [col for col in output.columns if col not in exclude]].sort_values(by=['player_id']) # output = output.sort_values(by=['player_id']) return Response(content=pd.DataFrame(output).to_csv(index=False), media_type='text/csv') @router.get('/{ratings_id}') async def get_one_rating(ratings_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'Bad Token: {token}') db.close() raise HTTPException( status_code=401, detail='You are not authorized to pull card ratings.' ) this_rating = BattingCardRatings.get_or_none(BattingCardRatings.id == ratings_id) if this_rating is None: db.close() raise HTTPException(status_code=404, detail=f'BattingCardRating id {ratings_id} not found') r_data = model_to_dict(this_rating) db.close() return r_data @router.get('/player/{player_id}') async def get_player_ratings( player_id: int, variant: list = Query(default=None), short_output: bool = False, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'Bad Token: {token}') db.close() raise HTTPException( status_code=401, detail='You are not authorized to pull card ratings.' ) all_cards = BattingCard.select().where(BattingCard.player_id == player_id).order_by(BattingCard.variant) if variant is not None: all_cards = all_cards.where(BattingCard.variant << variant) all_ratings = BattingCardRatings.select().where(BattingCardRatings.battingcard << all_cards) return_val = {'count': all_ratings.count(), 'ratings': [ model_to_dict(x, recurse=not short_output) for x in all_ratings ]} db.close() return return_val @router.put('') async def put_ratings(ratings: RatingsList, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'Bad Token: {token}') db.close() raise HTTPException( status_code=401, detail='You are not authorized to post card ratings.' ) new_ratings = [] updates = 0 for x in ratings.ratings: try: BattingCardRatings.get( (BattingCardRatings.battingcard_id == x.battingcard_id) & (BattingCardRatings.vs_hand == x.vs_hand) ) updates += BattingCardRatings.update(x.dict()).where( (BattingCardRatings.battingcard_id == x.battingcard_id) & (BattingCardRatings.vs_hand == x.vs_hand) ).execute() except BattingCardRatings.DoesNotExist: new_ratings.append(x.dict()) with db.atomic(): for batch in chunked(new_ratings, 30): BattingCardRatings.insert_many(batch).on_conflict_replace().execute() db.close() return f'Updated ratings: {updates}; new ratings: {len(new_ratings)}' @router.delete('/{ratings_id}') async def delete_rating( ratings_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'Bad Token: {token}') db.close() raise HTTPException( status_code=401, detail='You are not authorized to post card ratings.' ) this_rating = BattingCardRatings.get_or_none(BattingCardRatings.id == ratings_id) if this_rating is None: db.close() raise HTTPException(status_code=404, detail=f'BattingCardRating id {ratings_id} not found') count = this_rating.delete_instance() db.close() if count == 1: return f'Rating {this_rating} has been deleted' else: raise HTTPException(status_code=500, detail=f'Rating {this_rating} could not be deleted')