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, Result, model_to_dict, fn, Team, DataError 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/results', tags=['results'] ) class ResultModel(pydantic.BaseModel): away_team_id: int home_team_id: int away_score: int home_score: int away_team_value: Optional[int] = None home_team_value: Optional[int] = None away_team_ranking: Optional[int] = None home_team_ranking: Optional[int] = None scorecard: str week: int season: int ranked: bool short_game: bool game_type: str @router.get('') async def get_results( away_team_id: Optional[int] = None, home_team_id: Optional[int] = None, team_one_id: Optional[int] = None, team_two_id: Optional[int] = None, away_score_min: Optional[int] = None, away_score_max: Optional[int] = None, home_score_min: Optional[int] = None, home_score_max: Optional[int] = None, bothscore_min: Optional[int] = None, bothscore_max: Optional[int] = None, season: Optional[int] = None, week: Optional[int] = None, week_start: Optional[int] = None, week_end: Optional[int] = None, ranked: Optional[bool] = None, short_game: Optional[bool] = None, game_type: Optional[str] = None, vs_ai: Optional[bool] = None, csv: Optional[bool] = None): all_results = Result.select() # if all_results.count() == 0: # db.close() # raise HTTPException(status_code=404, detail=f'There are no results to filter') if away_team_id is not None: try: this_team = Team.get_by_id(away_team_id) all_results = all_results.where(Result.away_team == this_team) except Exception: db.close() raise HTTPException(status_code=404, detail=f'No team found with id {away_team_id}') if home_team_id is not None: try: this_team = Team.get_by_id(home_team_id) all_results = all_results.where(Result.home_team == this_team) except Exception: db.close() raise HTTPException(status_code=404, detail=f'No team found with id {home_team_id}') if team_one_id is not None: try: this_team = Team.get_by_id(team_one_id) all_results = all_results.where((Result.home_team == this_team) | (Result.away_team == this_team)) except Exception: db.close() raise HTTPException(status_code=404, detail=f'No team found with id {team_one_id}') if team_two_id is not None: try: this_team = Team.get_by_id(team_two_id) all_results = all_results.where((Result.home_team == this_team) | (Result.away_team == this_team)) except Exception: db.close() raise HTTPException(status_code=404, detail=f'No team found with id {team_two_id}') if away_score_min is not None: all_results = all_results.where(Result.away_score >= away_score_min) if away_score_max is not None: all_results = all_results.where(Result.away_score <= away_score_max) if home_score_min is not None: all_results = all_results.where(Result.home_score >= home_score_min) if home_score_max is not None: all_results = all_results.where(Result.home_score <= home_score_max) if bothscore_min is not None: all_results = all_results.where((Result.home_score >= bothscore_min) & (Result.away_score >= bothscore_min)) if bothscore_max is not None: all_results = all_results.where((Result.home_score <= bothscore_max) & (Result.away_score <= bothscore_max)) if season is not None: all_results = all_results.where(Result.season == season) if week is not None: all_results = all_results.where(Result.week == week) if ranked is not None: all_results = all_results.where(Result.ranked == ranked) if short_game is not None: all_results = all_results.where(Result.short_game == short_game) if week_start is not None: all_results = all_results.where(Result.week >= week_start) if week_end is not None: all_results = all_results.where(Result.week <= week_end) if game_type is not None: all_results = all_results.where(Result.game_type == game_type) all_results = all_results.order_by(Result.id) # Not functional # if vs_ai is not None: # AwayTeam = Team.alias() # all_results = all_results.join( # Team, on=Result.home_team # ).switch(Result).join( # Team, on=(AwayTeam.id == Result.away_team).alias('a_team') # ) # # if vs_ai: # all_results = all_results.where( # (Result.home_team.is_ai == 1) | (Result.a_team.is_ai == 1) # ) # else: # all_results = all_results.where( # (Result.home_team.is_ai == 0) & (Result.a_team.is_ai == 0) # ) # logging.info(f'Result Query:\n\n{all_results}') if csv: data_list = [['id', 'away_abbrev', 'home_abbrev', 'away_score', 'home_score', 'away_tv', 'home_tv', 'game_type', 'season', 'week', 'short_game', 'ranked']] for line in all_results: data_list.append([ line.id, line.away_team.abbrev, line.home_team.abbrev, line.away_score, line.home_score, line.away_team_value, line.home_team_value, line.game_type if line.game_type else 'minor-league', line.season, line.week, line.short_game, line.ranked ]) 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_results.count(), 'results': []} for x in all_results: return_val['results'].append(model_to_dict(x)) db.close() return return_val @router.get('/{result_id}') async def get_one_results(result_id, csv: Optional[bool] = None): try: this_result = Result.get_by_id(result_id) except Exception: db.close() raise HTTPException(status_code=404, detail=f'No result found with id {result_id}') if csv: data_list = [ ['id', 'away_abbrev', 'home_abbrev', 'away_score', 'home_score', 'away_tv', 'home_tv', 'game_type', 'season', 'week', 'game_type'], [this_result.id, this_result.away_team.abbrev, this_result.away_team.abbrev, this_result.away_score, this_result.home_score, this_result.away_team_value, this_result.home_team_value, this_result.game_type if this_result.game_type else 'minor-league', this_result.season, this_result.week, this_result.game_type] ] 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 = model_to_dict(this_result) db.close() return return_val @router.get('/team/{team_id}') async def get_team_results( team_id: int, season: Optional[int] = None, week: Optional[int] = None, csv: Optional[bool] = False): all_results = Result.select().where((Result.away_team_id == team_id) | (Result.home_team_id == team_id)) try: this_team = Team.get_by_id(team_id) except Exception as e: logging.error(f'Unknown team id {team_id} trying to pull team results') raise HTTPException(404, f'Team id {team_id} not found') if season is not None: all_results = all_results.where(Result.season == season) else: all_results = all_results.where(Result.season == this_team.season) if week is not None: all_results = all_results.where(Result.week == week) r_wins, r_loss, c_wins, c_loss = 0, 0, 0, 0 for x in all_results: if x.away_team_id == team_id: if x.away_score > x.home_score: if x.ranked: r_wins += 1 else: c_wins += 1 else: if x.ranked: r_loss += 1 else: c_loss += 1 elif x.home_team_id == team_id: if x.away_score > x.home_score: if x.ranked: r_loss += 1 else: c_loss += 1 else: if x.ranked: r_wins += 1 else: c_wins += 1 if csv: data_list = [ ['team_id', 'ranked_wins', 'ranked_losses', 'casual_wins', 'casual_losses', 'team_ranking'], [team_id, r_wins, r_loss, c_wins, c_loss, this_team.ranking] ] 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 = { 'team': model_to_dict(this_team), 'ranked_wins': r_wins, 'ranked_losses': r_loss, 'casual_wins': c_wins, 'casual_losses': c_loss, } db.close() return return_val @router.post('') async def post_result(result: ResultModel, 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 results. This event has been logged.' ) this_result = Result(**result.__dict__) saved = this_result.save() if result.ranked: if not result.away_team_ranking: db.close() error = f'Ranked game did not include away team ({result.away_team_id}) ranking.' logging.error(error) raise DataError(error) if not result.home_team_ranking: db.close() error = f'Ranked game did not include home team ({result.home_team_id}) ranking.' logging.error(error) raise DataError(error) k_value = 20 if result.short_game else 60 ratio = (result.home_team_ranking - result.away_team_ranking) / 400 exp_score = 1 / (1 + (10 ** ratio)) away_win = True if result.away_score > result.home_score else False total_delta = k_value * exp_score high_delta = total_delta * exp_score if exp_score > .5 else total_delta * (1 - exp_score) low_delta = total_delta - high_delta # exp_score > .5 means away team is favorite if exp_score > .5 and away_win: final_delta = low_delta away_delta = low_delta * 3 home_delta = -low_delta elif away_win: final_delta = high_delta away_delta = high_delta * 3 home_delta = -high_delta elif exp_score <= .5 and not away_win: final_delta = low_delta away_delta = -low_delta home_delta = low_delta * 3 elif not away_win: final_delta = high_delta away_delta = -high_delta home_delta = high_delta * 3 else: final_delta = 0 away_delta = 0 home_delta = 0 logging.debug(f'/results ranking deltas\n\nk_value: {k_value} / ratio: {ratio} / ' f'exp_score: {exp_score} / away_win: {away_win} / total_delta: {total_delta} / ' f'high_delta: {high_delta} / low_delta: {low_delta} / final_delta: {final_delta} / ') away_team = Team.get_by_id(result.away_team_id) away_team.ranking += away_delta away_team.save() logging.info(f'Just updated {away_team.abbrev} ranking to {away_team.ranking}') home_team = Team.get_by_id(result.home_team_id) home_team.ranking += home_delta home_team.save() logging.info(f'Just updated {home_team.abbrev} ranking to {home_team.ranking}') if saved == 1: return_val = model_to_dict(this_result) db.close() return return_val else: db.close() raise HTTPException( status_code=418, detail='Well slap my ass and call me a teapot; I could not save that roster' ) @app.patch('/api/v1/results/{result_id}') async def patch_result( result_id, away_team_id: Optional[int] = None, home_team_id: Optional[int] = None, away_score: Optional[int] = None, home_score: Optional[int] = None, away_team_value: Optional[int] = None, home_team_value: Optional[int] = None, scorecard: Optional[str] = None, week: Optional[int] = None, season: Optional[int] = None, short_game: Optional[bool] = None, game_type: Optional[str] = None, 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 patch results. This event has been logged.' ) try: this_result = Result.get_by_id(result_id) except Exception: db.close() raise HTTPException(status_code=404, detail=f'No result found with id {result_id}') if away_team_id is not None: this_result.away_team_id = away_team_id if home_team_id is not None: this_result.home_team_id = home_team_id if away_score is not None: this_result.away_score = away_score if home_score is not None: this_result.home_score = home_score if away_team_value is not None: this_result.away_team_value = away_team_value if home_team_value is not None: this_result.home_team_value = home_team_value if scorecard is not None: this_result.scorecard = scorecard if week is not None: this_result.week = week if season is not None: this_result.season = season if game_type is not None: this_result.game_type = game_type if short_game is not None: if not short_game: this_result.short_game = None else: this_result.short_game = short_game if this_result.save() == 1: return_val = model_to_dict(this_result) db.close() return return_val else: db.close() raise HTTPException( status_code=418, detail='Well slap my ass and call me a teapot; I could not save that event' ) @app.delete('/api/v1/results/{result_id}') async def delete_result(result_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 post results. This event has been logged.' ) try: this_result = Result.get_by_id(result_id) except Exception: db.close() raise HTTPException(status_code=404, detail=f'No result found with id {result_id}') count = this_result.delete_instance() db.close() if count == 1: raise HTTPException(status_code=200, detail=f'Result {result_id} has been deleted') else: raise HTTPException(status_code=500, detail=f'Result {result_id} was not deleted')