75a8fc8505
All checks were successful
Build Docker Image / build (pull_request) Successful in 2m26s
Closes #76 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
309 lines
11 KiB
Python
309 lines
11 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from typing import List, Optional, Literal, Any
|
|
import copy
|
|
import logging
|
|
import pydantic
|
|
|
|
from ..db_engine import db, StratGame, Team, StratPlay, model_to_dict, chunked, fn
|
|
from ..dependencies import (
|
|
oauth2_scheme,
|
|
send_webhook_message,
|
|
update_season_pitching_stats,
|
|
valid_token,
|
|
PRIVATE_IN_SCHEMA,
|
|
handle_db_errors,
|
|
update_season_batting_stats,
|
|
)
|
|
|
|
logger = logging.getLogger("discord_app")
|
|
|
|
router = APIRouter(prefix="/api/v3/games", tags=["games"])
|
|
|
|
|
|
class GameModel(pydantic.BaseModel):
|
|
season: int
|
|
week: int
|
|
game_num: Optional[int] = None
|
|
season_type: Optional[str] = "regular"
|
|
away_team_id: int
|
|
home_team_id: int
|
|
away_score: Optional[int] = None
|
|
home_score: Optional[int] = None
|
|
away_manager_id: Optional[int] = None
|
|
home_manager_id: Optional[int] = None
|
|
scorecard_url: Optional[str] = None
|
|
|
|
|
|
class GameList(pydantic.BaseModel):
|
|
games: List[GameModel]
|
|
|
|
|
|
@router.get("")
|
|
@handle_db_errors
|
|
async def get_games(
|
|
season: list = Query(default=None),
|
|
week: list = Query(default=None),
|
|
game_num: list = Query(default=None),
|
|
season_type: Literal["regular", "post", "all"] = "all",
|
|
away_team_id: list = Query(default=None),
|
|
home_team_id: list = Query(default=None),
|
|
week_start: Optional[int] = None,
|
|
week_end: Optional[int] = None,
|
|
team1_id: list = Query(default=None),
|
|
team2_id: list = Query(default=None),
|
|
played: Optional[bool] = None,
|
|
away_manager_id: list = Query(default=None),
|
|
home_manager_id: list = Query(default=None),
|
|
manager1_id: list = Query(default=None),
|
|
manager2_id: list = Query(default=None),
|
|
division_id: Optional[int] = None,
|
|
short_output: Optional[bool] = False,
|
|
sort: Optional[str] = None,
|
|
) -> Any:
|
|
all_games = StratGame.select()
|
|
|
|
if season is not None:
|
|
all_games = all_games.where(StratGame.season << season)
|
|
if week is not None:
|
|
all_games = all_games.where(StratGame.week << week)
|
|
if game_num is not None:
|
|
all_games = all_games.where(StratGame.game_num << game_num)
|
|
if season_type != "all":
|
|
all_games = all_games.where(StratGame.season_type == season_type)
|
|
if away_team_id is not None:
|
|
all_games = all_games.where(StratGame.away_team_id << away_team_id)
|
|
if home_team_id is not None:
|
|
all_games = all_games.where(StratGame.home_team_id << home_team_id)
|
|
if week_start is not None:
|
|
all_games = all_games.where(StratGame.week >= week_start)
|
|
if week_end is not None:
|
|
all_games = all_games.where(StratGame.week <= week_end)
|
|
if team1_id is not None:
|
|
all_games = all_games.where(
|
|
(StratGame.away_team_id << team1_id) | (StratGame.home_team_id << team1_id)
|
|
)
|
|
if team2_id is not None:
|
|
all_games = all_games.where(
|
|
(StratGame.away_team_id << team2_id) | (StratGame.home_team_id << team2_id)
|
|
)
|
|
if division_id is not None:
|
|
all_teams = Team.select().where(Team.division == division_id)
|
|
all_games = all_games.where(
|
|
(StratGame.away_team << all_teams) | (StratGame.home_team << all_teams)
|
|
)
|
|
if away_manager_id is not None:
|
|
all_games = all_games.where(StratGame.away_manager_id << away_manager_id)
|
|
if home_manager_id is not None:
|
|
all_games = all_games.where(StratGame.home_manager_id << home_manager_id)
|
|
if manager1_id is not None:
|
|
all_games = all_games.where(
|
|
(StratGame.away_manager_id << manager1_id)
|
|
| (StratGame.home_manager_id << manager1_id)
|
|
)
|
|
if manager2_id is not None:
|
|
all_games = all_games.where(
|
|
(StratGame.away_manager_id << manager2_id)
|
|
| (StratGame.home_manager_id << manager2_id)
|
|
)
|
|
if played is not None:
|
|
all_games = all_games.where(StratGame.game_num.is_null(not played))
|
|
if game_num is not None:
|
|
all_games = all_games.where(StratGame.game_num << game_num)
|
|
|
|
if sort == "recent-first":
|
|
all_games = all_games.order_by(
|
|
-StratGame.season, -StratGame.week, -StratGame.game_num
|
|
)
|
|
else:
|
|
all_games = all_games.order_by(
|
|
StratGame.season, StratGame.week, StratGame.game_num
|
|
)
|
|
|
|
return_games = {
|
|
"count": all_games.count(),
|
|
"games": [model_to_dict(x, recurse=not short_output) for x in all_games],
|
|
}
|
|
db.close()
|
|
return return_games
|
|
|
|
|
|
@router.get("/{game_id}")
|
|
@handle_db_errors
|
|
async def get_one_game(game_id: int) -> Any:
|
|
this_game = StratGame.get_or_none(StratGame.id == game_id)
|
|
if not this_game:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f"StratGame ID {game_id} not found")
|
|
|
|
g_result = model_to_dict(this_game)
|
|
db.close()
|
|
return g_result
|
|
|
|
|
|
@router.patch("/{game_id}", include_in_schema=PRIVATE_IN_SCHEMA)
|
|
@handle_db_errors
|
|
async def patch_game(
|
|
game_id: int,
|
|
game_num: Optional[int] = None,
|
|
away_score: Optional[int] = None,
|
|
home_score: Optional[int] = None,
|
|
away_manager_id: Optional[int] = None,
|
|
home_manager_id: Optional[int] = None,
|
|
token: str = Depends(oauth2_scheme),
|
|
scorecard_url: Optional[str] = None,
|
|
) -> Any:
|
|
if not valid_token(token):
|
|
logger.warning(f"patch_game - Bad Token: {token}")
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
this_game = StratGame.get_or_none(StratGame.id == game_id)
|
|
if not this_game:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f"StratGame ID {game_id} not found")
|
|
|
|
if game_num is not None:
|
|
this_game.game_num = game_num
|
|
if away_score is not None:
|
|
this_game.away_score = away_score
|
|
if home_score is not None:
|
|
this_game.home_score = home_score
|
|
if away_manager_id is not None:
|
|
this_game.away_manager_id = away_manager_id
|
|
if home_manager_id is not None:
|
|
this_game.home_manager_id = home_manager_id
|
|
if scorecard_url is not None:
|
|
this_game.scorecard_url = scorecard_url
|
|
|
|
if this_game.save() == 1:
|
|
# Update batting stats for all batters in this game
|
|
try:
|
|
# Get all unique batter IDs from stratplays in this game
|
|
batter_ids = [
|
|
row.batter_id
|
|
for row in StratPlay.select(StratPlay.batter_id.distinct()).where(
|
|
StratPlay.game_id == game_id
|
|
)
|
|
]
|
|
|
|
if batter_ids:
|
|
update_season_batting_stats(batter_ids, this_game.season, db)
|
|
logger.info(
|
|
f"Updated batting stats for {len(batter_ids)} players from game {game_id}"
|
|
)
|
|
else:
|
|
logger.error(f"No batters found for game_id {game_id}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update batting stats for game {game_id}: {e}")
|
|
send_webhook_message(
|
|
f"Failed to update season batting stats after game {this_game.game_num} between {this_game.away_team.abbrev} and {this_game.home_team.abbrev}!\nError: {e}"
|
|
)
|
|
# Don't fail the patch operation if stats update fails
|
|
|
|
# Update pitching stats for all pitchers in this game
|
|
try:
|
|
# Get all unique pitcher IDs from stratplays in this game
|
|
pitcher_ids = [
|
|
row.pitcher_id
|
|
for row in StratPlay.select(StratPlay.pitcher_id.distinct()).where(
|
|
StratPlay.game_id == game_id
|
|
)
|
|
]
|
|
|
|
if pitcher_ids:
|
|
update_season_pitching_stats(pitcher_ids, this_game.season, db)
|
|
logger.info(
|
|
f"Updated pitching stats for {len(pitcher_ids)} players from game {game_id}"
|
|
)
|
|
else:
|
|
logger.error(f"No pitchers found for game_id {game_id}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update pitching stats for game {game_id}: {e}")
|
|
send_webhook_message(
|
|
f"Failed to update season pitching stats after game {this_game.game_num} between {this_game.away_team.abbrev} and {this_game.home_team.abbrev}!\nError: {e}"
|
|
)
|
|
|
|
g_result = model_to_dict(this_game)
|
|
return g_result
|
|
else:
|
|
raise HTTPException(status_code=500, detail=f"Unable to patch game {game_id}")
|
|
|
|
|
|
@router.post("/", include_in_schema=PRIVATE_IN_SCHEMA)
|
|
@handle_db_errors
|
|
async def post_games(game_list: GameList, token: str = Depends(oauth2_scheme)) -> Any:
|
|
if not valid_token(token):
|
|
logger.warning(f"post_games - Bad Token: {token}")
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
new_games = []
|
|
for x in game_list.games:
|
|
if Team.get_or_none(Team.id == x.away_team_id) is None:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"Team ID {x.away_team_id} not found"
|
|
)
|
|
if Team.get_or_none(Team.id == x.home_team_id) is None:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"Team ID {x.home_team_id} not found"
|
|
)
|
|
|
|
new_games.append(x.model_dump())
|
|
|
|
with db.atomic():
|
|
for batch in chunked(new_games, 16):
|
|
StratGame.insert_many(batch).on_conflict_ignore().execute()
|
|
db.close()
|
|
|
|
return f"Inserted {len(new_games)} games"
|
|
|
|
|
|
@router.post("/wipe/{game_id}", include_in_schema=PRIVATE_IN_SCHEMA)
|
|
@handle_db_errors
|
|
async def wipe_game(game_id: int, token: str = Depends(oauth2_scheme)) -> Any:
|
|
if not valid_token(token):
|
|
logger.warning(f"wipe_game - Bad Token: {token}")
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
this_game = StratGame.get_or_none(StratGame.id == game_id)
|
|
if not this_game:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f"StratGame ID {game_id} not found")
|
|
|
|
this_game.away_score = None
|
|
this_game.home_score = None
|
|
this_game.game_num = None
|
|
this_game.away_manager = None
|
|
this_game.home_manager = None
|
|
|
|
if this_game.save() == 1:
|
|
g_result = model_to_dict(this_game)
|
|
db.close()
|
|
return g_result
|
|
else:
|
|
db.close()
|
|
raise HTTPException(status_code=500, detail=f"Unable to wipe game {game_id}")
|
|
|
|
|
|
@router.delete("/{game_id}", include_in_schema=PRIVATE_IN_SCHEMA)
|
|
@handle_db_errors
|
|
async def delete_game(game_id: int, token: str = Depends(oauth2_scheme)) -> Any:
|
|
if not valid_token(token):
|
|
logger.warning(f"delete_game - Bad Token: {token}")
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
this_game = StratGame.get_or_none(StratGame.id == game_id)
|
|
if not this_game:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f"StratGame ID {game_id} not found")
|
|
|
|
count = this_game.delete_instance()
|
|
db.close()
|
|
|
|
if count == 1:
|
|
return f"StratGame {game_id} has been deleted"
|
|
else:
|
|
raise HTTPException(
|
|
status_code=500, detail=f"StratGame {game_id} could not be deleted"
|
|
)
|