from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Response from typing import Optional, List import logging import pydantic from pandas import DataFrame from ..db_engine import db, BattingStat, model_to_dict, fn, Card, Player, Current from ..dependencies import oauth2_scheme, valid_token, LOG_DATA, PRIVATE_IN_SCHEMA logging.basicConfig( filename=LOG_DATA['filename'], format=LOG_DATA['format'], level=LOG_DATA['log_level'] ) router = APIRouter( prefix='/api/v2/batstats', tags=['Pre-Season 7 Batting Stats'] ) class BatStat(pydantic.BaseModel): card_id: int team_id: int roster_num: int vs_team_id: int pos: str pa: Optional[int] = 0 ab: Optional[int] = 0 run: Optional[int] = 0 hit: Optional[int] = 0 rbi: Optional[int] = 0 double: Optional[int] = 0 triple: Optional[int] = 0 hr: Optional[int] = 0 bb: Optional[int] = 0 so: Optional[int] = 0 hbp: Optional[int] = 0 sac: Optional[int] = 0 ibb: Optional[int] = 0 gidp: Optional[int] = 0 sb: Optional[int] = 0 cs: Optional[int] = 0 bphr: Optional[int] = 0 bpfo: Optional[int] = 0 bp1b: Optional[int] = 0 bplo: Optional[int] = 0 xch: Optional[int] = 0 xhit: Optional[int] = 0 error: Optional[int] = 0 pb: Optional[int] = 0 sbc: Optional[int] = 0 csc: Optional[int] = 0 week: int season: int created: Optional[int] = int(datetime.timestamp(datetime.now())*100000) game_id: int class BattingStatModel(pydantic.BaseModel): stats: List[BatStat] class BatStatReturnList(pydantic.BaseModel): count: int stats: list[BatStat] @router.get('', response_model=BatStatReturnList) async def get_batstats( card_id: int = None, player_id: int = None, team_id: int = None, vs_team_id: int = None, week: int = None, season: int = None, week_start: int = None, week_end: int = None, created: int = None, csv: bool = None): all_stats = BattingStat.select().join(Card).join(Player) if season is not None: all_stats = all_stats.where(BattingStat.season == season) else: curr = Current.latest() all_stats = all_stats.where(BattingStat.season == curr.season) if card_id is not None: all_stats = all_stats.where(BattingStat.card_id == card_id) if player_id is not None: all_stats = all_stats.where(BattingStat.card.player.player_id == player_id) if team_id is not None: all_stats = all_stats.where(BattingStat.team_id == team_id) if vs_team_id is not None: all_stats = all_stats.where(BattingStat.vs_team_id == vs_team_id) if week is not None: all_stats = all_stats.where(BattingStat.week == week) if week_start is not None: all_stats = all_stats.where(BattingStat.week >= week_start) if week_end is not None: all_stats = all_stats.where(BattingStat.week <= week_end) if created is not None: # Convert milliseconds timestamp to datetime for PostgreSQL comparison created_dt = datetime.fromtimestamp(created / 1000) all_stats = all_stats.where(BattingStat.created == created_dt) # if all_stats.count() == 0: # db.close() # raise HTTPException(status_code=404, detail=f'No batting stats found') if csv: data_list = [['id', 'card_id', 'player_id', 'cardset', 'team', 'vs_team', 'pos', 'pa', 'ab', 'run', 'hit', 'rbi', 'double', 'triple', 'hr', 'bb', 'so', 'hbp', 'sac', 'ibb', 'gidp', 'sb', 'cs', 'bphr', 'bpfo', 'bp1b', 'bplo', 'xch', 'xhit', 'error', 'pb', 'sbc', 'csc', 'week', 'season', 'created', 'game_id', 'roster_num']] for line in all_stats: data_list.append( [ line.id, line.card.id, line.card.player.player_id, line.card.player.cardset.name, line.team.abbrev, line.vs_team.abbrev, line.pos, line.pa, line.ab, line.run, line.hit, line.rbi, line.double, line.triple, line.hr, line.bb, line.so, line.hbp, line.sac, line.ibb, line.gidp, line.sb, line.cs, line.bphr, line.bpfo, line.bp1b, line.bplo, line.xch, line.xhit, line.error, line.pb, line.sbc, line.csc, line.week, line.season, line.created, line.game_id, line.roster_num ] ) return_val = DataFrame(data_list).to_csv(header=False, index=False) db.close() return Response(content=return_val, media_type='text/csv') else: return_val = {'count': all_stats.count(), 'stats': []} for x in all_stats: return_val['stats'].append(model_to_dict(x, recurse=False)) db.close() return return_val @router.get('/player/{player_id}', response_model=BatStat) async def get_player_stats( player_id: int, team_id: int = None, vs_team_id: int = None, week_start: int = None, week_end: int = None, csv: bool = None): all_stats = (BattingStat .select(fn.COUNT(BattingStat.created).alias('game_count')) .join(Card) .group_by(BattingStat.card) .where(BattingStat.card.player == player_id)).scalar() if team_id is not None: all_stats = all_stats.where(BattingStat.team_id == team_id) if vs_team_id is not None: all_stats = all_stats.where(BattingStat.vs_team_id == vs_team_id) if week_start is not None: all_stats = all_stats.where(BattingStat.week >= week_start) if week_end is not None: all_stats = all_stats.where(BattingStat.week <= week_end) if csv: data_list = [ [ 'pa', 'ab', 'run', 'hit', 'rbi', 'double', 'triple', 'hr', 'bb', 'so', 'hbp', 'sac', 'ibb', 'gidp', 'sb', 'cs', 'bphr', 'bpfo', 'bp1b', 'bplo', 'xch', 'xhit', 'error', 'pb', 'sbc', 'csc', ],[ all_stats.pa_sum, all_stats.ab_sum, all_stats.run, all_stats.hit_sum, all_stats.rbi_sum, all_stats.double_sum, all_stats.triple_sum, all_stats.hr_sum, all_stats.bb_sum, all_stats.so_sum, all_stats.hbp_sum, all_stats.sac, all_stats.ibb_sum, all_stats.gidp_sum, all_stats.sb_sum, all_stats.cs_sum, all_stats.bphr_sum, all_stats.bpfo_sum, all_stats.bp1b_sum, all_stats.bplo_sum, all_stats.xch, all_stats.xhit_sum, all_stats.error_sum, all_stats.pb_sum, all_stats.sbc_sum, all_stats.csc_sum ] ] return_val = DataFrame(data_list).to_csv(header=False, index=False) db.close() return Response(content=return_val, media_type='text/csv') else: logging.debug(f'stat pull query: {all_stats}\n') # logging.debug(f'result 0: {all_stats[0]}\n') for x in all_stats: logging.debug(f'this_line: {model_to_dict(x)}') return_val = model_to_dict(all_stats[0]) db.close() return return_val @router.post('', include_in_schema=PRIVATE_IN_SCHEMA) async def post_batstats(stats: BattingStatModel, 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 stats. This event has been logged.' ) new_stats = [] for x in stats.stats: this_stat = BattingStat( card_id=x.card_id, team_id=x.team_id, roster_num=x.roster_num, vs_team_id=x.vs_team_id, pos=x.pos, pa=x.pa, ab=x.ab, run=x.run, hit=x.hit, rbi=x.rbi, double=x.double, triple=x.triple, hr=x.hr, bb=x.bb, so=x.so, hbp=x.hbp, sac=x.sac, ibb=x.ibb, gidp=x.gidp, sb=x.sb, cs=x.cs, bphr=x.bphr, bpfo=x.bpfo, bp1b=x.bp1b, bplo=x.bplo, xch=x.xch, xhit=x.xhit, error=x.error, pb=x.pb, sbc=x.sbc, csc=x.csc, week=x.week, season=x.season, created=x.created, game_id=x.game_id ) new_stats.append(this_stat) with db.atomic(): BattingStat.bulk_create(new_stats, batch_size=15) db.close() raise HTTPException(status_code=200, detail=f'{len(new_stats)} batting lines have been added') @router.delete('/{stat_id}', include_in_schema=PRIVATE_IN_SCHEMA) async def delete_batstat(stat_id, 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 delete stats. This event has been logged.' ) try: this_stat = BattingStat.get_by_id(stat_id) except Exception: db.close() raise HTTPException(status_code=404, detail=f'No stat found with id {stat_id}') count = this_stat.delete_instance() db.close() if count == 1: raise HTTPException(status_code=200, detail=f'Stat {stat_id} has been deleted') else: raise HTTPException(status_code=500, detail=f'Stat {stat_id} was not deleted') # @app.get('/api/v1/plays/batting') # async def get_batting_totals( # player_id: list = Query(default=None), team_id: list = Query(default=None), min_pa: Optional[int] = 1, # season: list = Query(default=None), position: list = Query(default=None), # group_by: Literal['team', 'player', 'playerteam', 'playergame', 'teamgame', 'league'] = 'player', # sort: Optional[str] = None, limit: Optional[int] = None, short_output: Optional[bool] = False): # all_stats = BattingStat.select( # BattingStat.card, BattingStat.game_id, BattingStat.team, BattingStat.vs_team, BattingStat.pos, # BattingStat.card.player.alias('player'), # fn.SUM(BattingStat.pa).alias('sum_pa'), fn.SUM(BattingStat.ab).alias('sum_ab'), # fn.SUM(BattingStat.run).alias('sum_run'), fn.SUM(BattingStat.so).alias('sum_so'), # fn.SUM(BattingStat.hit).alias('sum_hit'), fn.SUM(BattingStat.rbi).alias('sum_rbi'), # fn.SUM(BattingStat.double).alias('sum_double'), fn.SUM(BattingStat.triple).alias('sum_triple'), # fn.SUM(BattingStat.hr).alias('sum_hr'), fn.SUM(BattingStat.bb).alias('sum_bb'), # fn.SUM(BattingStat.hbp).alias('sum_hbp'), fn.SUM(BattingStat.sac).alias('sum_sac'), # fn.SUM(BattingStat.ibb).alias('sum_ibb'), fn.SUM(BattingStat.gidp).alias('sum_gidp'), # fn.SUM(BattingStat.sb).alias('sum_sb'), fn.SUM(BattingStat.cs).alias('sum_cs'), # fn.SUM(BattingStat.bphr).alias('sum_bphr'), fn.SUM(BattingStat.bpfo).alias('sum_bpfo'), # fn.SUM(BattingStat.bp1b).alias('sum_bp1b'), fn.SUM(BattingStat.bplo).alias('sum_bplo') # ).having( # fn.SUM(BattingStat.pa) >= min_pa # ).join(Card) # # if player_id is not None: # # all_players = Player.select().where(Player.id << player_id) # all_cards = Card.select().where(Card.player_id << player_id) # all_stats = all_stats.where(BattingStat.card << all_cards) # if team_id is not None: # all_teams = Team.select().where(Team.id << team_id) # all_stats = all_stats.where(BattingStat.team << all_teams) # if season is not None: # all_stats = all_stats.where(BattingStat.season << season) # if position is not None: # all_stats = all_stats.where(BattingStat.pos << position) # # if group_by == 'player': # all_stats = all_stats.group_by(SQL('player')) # elif group_by == 'playerteam': # all_stats = all_stats.group_by(SQL('player'), BattingStat.team) # elif group_by == 'playergame': # all_stats = all_stats.group_by(SQL('player'), BattingStat.game_id) # elif group_by == 'team': # all_stats = all_stats.group_by(BattingStat.team) # elif group_by == 'teamgame': # all_stats = all_stats.group_by(BattingStat.team, BattingStat.game_id) # elif group_by == 'league': # all_stats = all_stats.group_by(BattingStat.season) # # if sort == 'pa-desc': # all_stats = all_stats.order_by(SQL('sum_pa').desc()) # elif sort == 'newest': # all_stats = all_stats.order_by(-BattingStat.game_id) # elif sort == 'oldest': # all_stats = all_stats.order_by(BattingStat.game_id) # # if limit is not None: # if limit < 1: # limit = 1 # all_stats = all_stats.limit(limit) # # logging.info(f'bat_plays query: {all_stats}') # # return_stats = { # 'count': all_stats.count(), # 'stats': [{ # 'player': x.card.player_id if short_output else model_to_dict(x.card.player, recurse=False), # 'team': x.team_id if short_output else model_to_dict(x.team, recurse=False), # 'pa': x.sum_pa, # 'ab': x.sum_ab, # 'run': x.sum_run, # 'hit': x.sum_hit, # 'rbi': x.sum_rbi, # 'double': x.sum_double, # 'triple': x.sum_triple, # 'hr': x.sum_hr, # 'bb': x.sum_bb, # 'so': x.sum_so, # 'hbp': x.sum_hbp, # 'sac': x.sum_sac, # 'ibb': x.sum_ibb, # 'gidp': x.sum_gidp, # 'sb': x.sum_sb, # 'cs': x.sum_cs, # 'bphr': x.sum_bphr, # 'bpfo': x.sum_bpfo, # 'bp1b': x.sum_bp1b, # 'bplo': x.sum_bplo, # 'avg': x.sum_hit / max(x.sum_ab, 1), # 'obp': (x.sum_hit + x.sum_bb + x.sum_hbp + x.sum_ibb) / max(x.sum_pa, 1), # 'slg': (x.sum_hr * 4 + x.sum_triple * 3 + x.sum_double * 2 + # (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr)) / max(x.sum_ab, 1), # 'ops': ((x.sum_hit + x.sum_bb + x.sum_hbp + x.sum_ibb) / max(x.sum_pa, 1)) + # ((x.sum_hr * 4 + x.sum_triple * 3 + x.sum_double * 2 + # (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr)) / max(x.sum_ab, 1)), # 'woba': (.69 * x.sum_bb + .72 * x.sum_hbp + .89 * (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr) + # 1.27 * x.sum_double + 1.62 * x.sum_triple + 2.1 * x.sum_hr) / max(x.sum_pa - x.sum_ibb, 1), # 'game': x.game_id # } for x in all_stats] # } # # db.close() # return return_stats # # # @app.get('/api/v1/plays/pitching') # async def get_pitching_totals( # player_id: list = Query(default=None), team_id: list = Query(default=None), season: list = Query(default=None), # group_by: Literal['team', 'player', 'playerteam', 'playergame', 'teamgame', 'league'] = 'player', # min_pa: Optional[int] = 1, # sort: Optional[str] = None, limit: Optional[int] = None, short_output: Optional[bool] = False): # all_stats = PitchingStat.select( # PitchingStat.card, PitchingStat.team, PitchingStat.game_id, PitchingStat.vs_team, # PitchingStat.card.player.alias('player'), fn.SUM(PitchingStat.ip).alias('sum_ip'), # fn.SUM(PitchingStat.hit).alias('sum_hit'), fn.SUM(PitchingStat.run).alias('sum_run'), # fn.SUM(PitchingStat.erun).alias('sum_erun'), fn.SUM(PitchingStat.so).alias('sum_so'), # fn.SUM(PitchingStat.bb).alias('sum_bb'), fn.SUM(PitchingStat.hbp).alias('sum_hbp'), # fn.SUM(PitchingStat.wp).alias('sum_wp'), fn.SUM(PitchingStat.balk).alias('sum_balk'), # fn.SUM(PitchingStat.hr).alias('sum_hr'), fn.SUM(PitchingStat.ir).alias('sum_ir'), # fn.SUM(PitchingStat.irs).alias('sum_irs'), fn.SUM(PitchingStat.gs).alias('sum_gs'), # fn.SUM(PitchingStat.win).alias('sum_win'), fn.SUM(PitchingStat.loss).alias('sum_loss'), # fn.SUM(PitchingStat.hold).alias('sum_hold'), fn.SUM(PitchingStat.sv).alias('sum_sv'), # fn.SUM(PitchingStat.bsv).alias('sum_bsv'), fn.COUNT(PitchingStat.game_id).alias('sum_games') # ).having( # fn.SUM(PitchingStat.ip) >= max(min_pa / 3, 1) # ).join(Card) # # if player_id is not None: # all_cards = Card.select().where(Card.player_id << player_id) # all_stats = all_stats.where(PitchingStat.card << all_cards) # if team_id is not None: # all_teams = Team.select().where(Team.id << team_id) # all_stats = all_stats.where(PitchingStat.team << all_teams) # if season is not None: # all_stats = all_stats.where(PitchingStat.season << season) # # if group_by == 'player': # all_stats = all_stats.group_by(SQL('player')) # elif group_by == 'playerteam': # all_stats = all_stats.group_by(SQL('player'), PitchingStat.team) # elif group_by == 'playergame': # all_stats = all_stats.group_by(SQL('player'), PitchingStat.game_id) # elif group_by == 'team': # all_stats = all_stats.group_by(PitchingStat.team) # elif group_by == 'teamgame': # all_stats = all_stats.group_by(PitchingStat.team, PitchingStat.game_id) # elif group_by == 'league': # all_stats = all_stats.group_by(PitchingStat.season) # # if sort == 'pa-desc': # all_stats = all_stats.order_by(SQL('sum_pa').desc()) # elif sort == 'newest': # all_stats = all_stats.order_by(-PitchingStat.game_id) # elif sort == 'oldest': # all_stats = all_stats.order_by(PitchingStat.game_id) # # if limit is not None: # if limit < 1: # limit = 1 # all_stats = all_stats.limit(limit) # # logging.info(f'bat_plays query: {all_stats}') # # return_stats = { # 'count': all_stats.count(), # 'stats': [{ # 'player': x.card.player_id if short_output else model_to_dict(x.card.player, recurse=False), # 'team': x.team_id if short_output else model_to_dict(x.team, recurse=False), # 'tbf': None, # 'outs': round(x.sum_ip * 3), # 'games': x.sum_games, # 'gs': x.sum_gs, # 'win': x.sum_win, # 'loss': x.sum_loss, # 'hold': x.sum_hold, # 'save': x.sum_sv, # 'bsave': x.sum_bsv, # 'ir': x.sum_ir, # 'ir_sc': x.sum_irs, # 'runs': x.sum_run, # 'e_runs': x.sum_erun, # 'hits': x.sum_hit, # 'hr': x.sum_hr, # 'bb': x.sum_bb, # 'so': x.sum_so, # 'hbp': x.sum_hbp, # 'wp': x.sum_wp, # 'balk': x.sum_balk, # 'era': (x.sum_erun * 27) / round(x.sum_ip * 3), # 'whip': (x.sum_bb + x.sum_hit) / x.sum_ip, # 'avg': None, # 'obp': None, # 'woba': None, # 'k/9': x.sum_so * 9 / x.sum_ip, # 'bb/9': x.sum_bb * 9 / x.sum_ip, # 'k/bb': x.sum_so / max(x.sum_bb, .1), # 'game': None, # 'lob_2outs': None, # 'rbi%': None # } for x in all_stats] # } # db.close() # return return_stats