265 lines
9.7 KiB
Python
265 lines
9.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from typing import List, Optional, Literal
|
|
import logging
|
|
import pydantic
|
|
|
|
from ..db_engine import db, BattingStat, Team, Player, Current, model_to_dict, chunked, fn, per_season_weeks
|
|
from ..dependencies import oauth2_scheme, valid_token
|
|
|
|
router = APIRouter(
|
|
prefix='/api/v3/battingstats',
|
|
tags=['battingstats']
|
|
)
|
|
|
|
|
|
class BatStatModel(pydantic.BaseModel):
|
|
player_id: int
|
|
team_id: int
|
|
pos: str
|
|
pa: int
|
|
ab: int
|
|
run: int
|
|
hit: int
|
|
rbi: int
|
|
double: int
|
|
triple: int
|
|
hr: int
|
|
bb: int
|
|
so: int
|
|
hbp: int
|
|
sac: int
|
|
ibb: int
|
|
gidp: int
|
|
sb: int
|
|
cs: int
|
|
bphr: int
|
|
bpfo: int
|
|
bp1b: int
|
|
bplo: int
|
|
xba: int
|
|
xbt: int
|
|
xch: int
|
|
xhit: int
|
|
error: int
|
|
pb: int
|
|
sbc: int
|
|
csc: int
|
|
roba: int
|
|
robs: int
|
|
raa: int
|
|
rto: int
|
|
week: int
|
|
game: int
|
|
season: int
|
|
|
|
|
|
class BatStatList(pydantic.BaseModel):
|
|
count: int
|
|
stats: List[BatStatModel]
|
|
|
|
|
|
@router.get('')
|
|
async def get_batstats(
|
|
season: int, s_type: Optional[str] = None, team_abbrev: list = Query(default=None),
|
|
player_name: list = Query(default=None), player_id: list = Query(default=None),
|
|
week_start: Optional[int] = None, week_end: Optional[int] = None, game_num: list = Query(default=None),
|
|
position: list = Query(default=None), limit: Optional[int] = None, sort: Optional[str] = None,
|
|
short_output: Optional[bool] = True):
|
|
if 'post' in s_type.lower():
|
|
all_stats = BattingStat.post_season(season)
|
|
if all_stats.count() == 0:
|
|
db.close()
|
|
return {'count': 0, 'stats': []}
|
|
elif s_type.lower() in ['combined', 'total', 'all']:
|
|
all_stats = BattingStat.combined_season(season)
|
|
if all_stats.count() == 0:
|
|
db.close()
|
|
return {'count': 0, 'stats': []}
|
|
else:
|
|
all_stats = BattingStat.regular_season(season)
|
|
if all_stats.count() == 0:
|
|
db.close()
|
|
return {'count': 0, 'stats': []}
|
|
if position is not None:
|
|
all_stats = all_stats.where(BattingStat.pos << [x.upper() for x in position])
|
|
if team_abbrev is not None:
|
|
t_query = Team.select().where(Team.abbrev << [x.upper() for x in team_abbrev])
|
|
all_stats = all_stats.where(BattingStat.team << t_query)
|
|
if player_name is not None or player_id is not None:
|
|
if player_id:
|
|
all_stats = all_stats.where(BattingStat.player_id << player_id)
|
|
else:
|
|
p_query = Player.select_season(season).where(Player.name << player_name)
|
|
all_stats = all_stats.where(BattingStat.player << p_query)
|
|
if game_num:
|
|
all_stats = all_stats.where(BattingStat.game == game_num)
|
|
|
|
start = 1
|
|
end = Current.get(Current.season == season).week
|
|
if week_start is not None:
|
|
start = week_start
|
|
if week_end is not None:
|
|
end = min(week_end, end)
|
|
if start > end:
|
|
db.close()
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f'Start week {start} is after end week {end} - cannot pull stats'
|
|
)
|
|
all_stats = all_stats.where(
|
|
(BattingStat.week >= start) & (BattingStat.week <= end)
|
|
)
|
|
|
|
if limit:
|
|
all_stats = all_stats.limit(limit)
|
|
if sort:
|
|
if sort == 'newest':
|
|
all_stats = all_stats.order_by(-BattingStat.week, -BattingStat.game)
|
|
|
|
return_stats = {
|
|
'count': all_stats.count(),
|
|
'stats': [model_to_dict(x, recurse=not short_output) for x in all_stats]
|
|
}
|
|
|
|
db.close()
|
|
return return_stats
|
|
|
|
|
|
@router.get('/season/{season}')
|
|
async def get_seasonstats(
|
|
season: int, s_type: Literal['regular', 'post', 'total'] = 'regular', team_abbrev: list = Query(default=None),
|
|
team_id: list = Query(default=None), player_name: list = Query(default=None),
|
|
player_id: list = Query(default=None), full_player: Optional[bool] = False):
|
|
weeks = per_season_weeks(season, s_type)
|
|
all_stats = (
|
|
BattingStat
|
|
.select(BattingStat.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.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.so).alias('sum_so'),
|
|
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'),
|
|
fn.SUM(BattingStat.xba).alias('sum_xba'), fn.SUM(BattingStat.xbt).alias('sum_xbt'),
|
|
fn.SUM(BattingStat.xch).alias('sum_xch'), fn.SUM(BattingStat.xhit).alias('sum_xhit'),
|
|
fn.SUM(BattingStat.error).alias('sum_error'), fn.SUM(BattingStat.pb).alias('sum_pb'),
|
|
fn.SUM(BattingStat.sbc).alias('sum_sbc'), fn.SUM(BattingStat.csc).alias('sum_csc'),
|
|
fn.SUM(BattingStat.roba).alias('sum_roba'), fn.SUM(BattingStat.robs).alias('sum_robs'),
|
|
fn.SUM(BattingStat.raa).alias('sum_raa'), fn.SUM(BattingStat.rto).alias('sum_rto'))
|
|
.where((BattingStat.week >= weeks['start']) & (BattingStat.week <= weeks['end']) &
|
|
(BattingStat.season == season))
|
|
.order_by(BattingStat.player)
|
|
.group_by(BattingStat.player)
|
|
)
|
|
|
|
if team_abbrev is None and team_id is None and player_name is None and player_id is None:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f'Must include team_id/team_abbrev and/or player_name/player_id'
|
|
)
|
|
|
|
if team_id is not None:
|
|
all_teams = Team.select().where(Team.id << team_id)
|
|
all_stats = all_stats.where(BattingStat.team << all_teams)
|
|
elif team_abbrev is not None:
|
|
all_teams = Team.select().where(fn.Lower(Team.abbrev) << [x.lower() for x in team_abbrev])
|
|
all_stats = all_stats.where(BattingStat.team << all_teams)
|
|
|
|
if player_name is not None:
|
|
all_players = Player.select().where(fn.Lower(Player.name) << [x.lower() for x in player_name])
|
|
all_stats = all_stats.where(BattingStat.player << all_players)
|
|
elif player_id is not None:
|
|
all_players = Player.select().where(Player.id << player_id)
|
|
all_stats = all_stats.where(BattingStat.player << all_players)
|
|
|
|
return_stats = {
|
|
'count': all_stats.count(),
|
|
'stats': [{
|
|
'player': model_to_dict(x.player, recurse=False) if full_player else x.player_id,
|
|
'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,
|
|
'xba': x.sum_xba,
|
|
'xbt': x.sum_xbt,
|
|
'xch': x.sum_xch,
|
|
'xhit': x.sum_xhit,
|
|
'error': x.sum_error,
|
|
'pb': x.sum_pb,
|
|
'sbc': x.sum_sbc,
|
|
'csc': x.sum_csc,
|
|
'roba': x.sum_roba,
|
|
'robs': x.sum_robs,
|
|
'raa': x.sum_raa,
|
|
'rto': x.sum_rto,
|
|
} for x in all_stats]
|
|
}
|
|
db.close()
|
|
return return_stats
|
|
|
|
|
|
@router.get('/career/{player_name}')
|
|
async def get_careerstats(
|
|
s_type: Literal['regular', 'post', 'total'] = 'regular', player_name: list = Query(default=None)):
|
|
pass # Keep Career Stats table and recalculate after posting stats
|
|
|
|
|
|
@router.patch('/{stat_id}')
|
|
async def patch_batstats(stat_id: int, new_stats: BatStatModel, token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logging.warning(f'patch_batstats - Bad Token: {token}')
|
|
raise HTTPException(status_code=401, detail='Unauthorized')
|
|
|
|
if BattingStat.get_or_none(BattingStat.id == stat_id) is None:
|
|
raise HTTPException(status_code=404, detail=f'Stat ID {stat_id} not found')
|
|
|
|
BattingStat.update(**new_stats.dict()).where(BattingStat.id == stat_id).execute()
|
|
r_stat = model_to_dict(BattingStat.get_by_id(stat_id))
|
|
db.close()
|
|
return r_stat
|
|
|
|
|
|
@router.post('')
|
|
async def post_batstats(s_list: BatStatList, token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logging.warning(f'post_batstats - Bad Token: {token}')
|
|
raise HTTPException(status_code=401, detail='Unauthorized')
|
|
|
|
all_stats = []
|
|
|
|
for x in s_list.stats:
|
|
team = Team.get_or_none(Team.id == x.team_id)
|
|
this_player = Player.get_or_none(Player.id == x.player_id)
|
|
if team is None:
|
|
raise HTTPException(status_code=404, detail=f'Team ID {x.team_id} not found')
|
|
if this_player is None:
|
|
raise HTTPException(status_code=404, detail=f'Player ID {x.player_id} not found')
|
|
|
|
all_stats.append(BattingStat(**x.dict()))
|
|
|
|
with db.atomic():
|
|
BattingStat.bulk_create(all_stats, batch_size=15)
|
|
|
|
# Update career stats
|
|
|
|
return f'Added {len(all_stats)} batting lines'
|