From cd0a8a739ab7e42421a7dd3e5b691e772f0be390 Mon Sep 17 00:00:00 2001 From: Cal Corum Date: Thu, 23 Jan 2025 10:59:03 -0600 Subject: [PATCH] Catchup edits --- exceptions.py | 4 +++ helpers.py | 52 +++++++++++++++++++++++++++++++++++++ in_game/gameplay_queries.py | 51 +++++++++++++++++++++++++++++++++--- 3 files changed, 103 insertions(+), 4 deletions(-) diff --git a/exceptions.py b/exceptions.py index 0b8b38d..ca6844b 100644 --- a/exceptions.py +++ b/exceptions.py @@ -89,3 +89,7 @@ class ButtonOptionNotChosen(Exception): class MissingRoleException(GameException): pass + + +class MissingRosterException(GameException): + pass diff --git a/helpers.py b/helpers.py index 6d63afc..a208b3c 100644 --- a/helpers.py +++ b/helpers.py @@ -22,8 +22,60 @@ from in_game.gameplay_models import Team SBA_SEASON = 10 PD_SEASON = 8 +ranked_cardsets = [20, 21, 22, 17, 18, 19] LIVE_CARDSET_ID = 20 LIVE_PROMO_CARDSET_ID = 21 +MAX_CARDSET_ID = 30 +CARDSETS = { + 'Ranked': { + 'primary': ranked_cardsets, + 'human': ranked_cardsets + }, + 'Minor League': { + 'primary': [20, 8], # 1998, Mario + 'secondary': [6], # 2013 + 'human': [x for x in range(1, MAX_CARDSET_ID)] + }, + 'Major League': { + 'primary': [20, 21, 17, 18, 12, 6, 7, 8], # 1998, 1998 Promos, 2024, 24 Promos, 2008, 2013, 2012, Mario + 'secondary': [5, 3], # 2019, 2022 + 'human': ranked_cardsets + }, + 'Hall of Fame': { + 'primary': [x for x in range(1, MAX_CARDSET_ID)], + 'human': ranked_cardsets + }, + 'Flashback': { + 'primary': [5, 1, 3, 9, 8], # 2019, 2021, 2022, 2023, Mario + 'secondary': [13, 5], # 2018, 2019 + 'human': [5, 1, 3, 9, 8] # 2019, 2021, 2022, 2023 + }, + 'gauntlet-3': { + 'primary': [13], # 2018 + 'secondary': [5, 11, 9], # 2019, 2016, 2023 + 'human': [x for x in range(1, MAX_CARDSET_ID)] + }, + 'gauntlet-4': { + 'primary': [3, 6, 16], # 2022, 2013, Backyard Baseball + 'secondary': [4, 9], # 2022 Promos, 2023 + 'human': [3, 4, 6, 9, 15, 16] + }, + 'gauntlet-5': { + 'primary': [17, 8], # 2024, Mario + 'secondary': [13], # 2018 + 'human': [x for x in range(1, MAX_CARDSET_ID)] + }, + 'gauntlet-6': { + 'primary': [20, 8], # 1998, Mario + 'secondary': [12], # 2008 + 'human': [x for x in range(1, MAX_CARDSET_ID)] + }, + 'gauntlet-7': { + 'primary': [5, 23], # 2019, Brilliant Stars + 'secondary': [1], # 2021 + 'human': [x for x in range(1, MAX_CARDSET_ID)] + } +} SBA_COLOR = 'a6ce39' PD_PLAYERS = 'Paper Dynasty Players' SBA_PLAYERS_ROLE_NAME = f'Season {SBA_SEASON} Players' diff --git a/in_game/gameplay_queries.py b/in_game/gameplay_queries.py index 9391511..babcdfc 100644 --- a/in_game/gameplay_queries.py +++ b/in_game/gameplay_queries.py @@ -7,7 +7,7 @@ import pydantic from sqlalchemy import func from api_calls import db_get, db_post from sqlmodel import col -from in_game.gameplay_models import CACHE_LIMIT, BatterScouting, BatterScoutingBase, BattingCard, BattingCardBase, BattingRatings, BattingRatingsBase, Card, CardBase, Lineup, PitcherScouting, PitchingCard, PitchingCardBase, PitchingRatings, PitchingRatingsBase, Player, PlayerBase, PositionRating, PositionRatingBase, RosterLink, Session, Team, TeamBase, select, or_, Game, Play +from in_game.gameplay_models import CACHE_LIMIT, BatterScouting, BatterScoutingBase, BattingCard, BattingCardBase, BattingRatings, BattingRatingsBase, Card, CardBase, Cardset, CardsetBase, Lineup, PitcherScouting, PitchingCard, PitchingCardBase, PitchingRatings, PitchingRatingsBase, Player, PlayerBase, PositionRating, PositionRatingBase, RosterLink, Session, Team, TeamBase, select, or_, Game, Play from exceptions import DatabaseError, PositionNotFoundException, log_exception, PlayNotFoundException @@ -103,11 +103,11 @@ async def get_team_or_none( session.commit() def cache_team(json_data: dict) -> Team: - # logger.info(f'gameplay_models - get_team - cache_team - writing a team to cache: {json_data}') + logger.info(f'gameplay_queries - cache_team - writing a team to cache: {json_data}') valid_team = TeamBase.model_validate(json_data, from_attributes=True) - # logger.info(f'gameplay_models - get_team - cache_team - valid_team: {valid_team}') + logger.info(f'gameplay_queries - cache_team - valid_team: {valid_team}') db_team = Team.model_validate(valid_team) - # logger.info(f'gameplay_models - get_team - cache_team - db_team: {db_team}') + logger.info(f'gameplay_queries - cache_team - db_team: {db_team}') session.add(db_team) session.commit() session.refresh(db_team) @@ -135,6 +135,49 @@ async def get_team_or_none( for team in [x for x in t_query['teams'] if 'gauntlet' not in x['abbrev'].lower()]: return cache_team(team) + logger.warning(f'No team found') + return None + + +async def get_cardset_or_none(session: Session, cardset_id: int = None, cardset_name: str = None): + logger.info(f'Getting cardset or none / cardset_id: {cardset_id} / cardset_name: {cardset_name}') + + if cardset_id is None and cardset_name is None: + log_exception(KeyError, 'One of "cardset_id" or "cardset_name" must be included in search') + + if cardset_id is not None: + logger.info(f'Getting cardset by id: {cardset_id}') + this_cardset = session.get(Cardset, cardset_id) + else: + logger.info(f'Getting cardset by name: {cardset_name}') + this_cardset = session.exec(select(Cardset).where(func.lower(Cardset.name) == cardset_name.lower())).one_or_none() + + if this_cardset is not None: + logger.info(f'we found a cardset: {this_cardset}') + return this_cardset + + def cache_cardset(json_data: dict) -> Cardset: + logger.info(f'gameplay_queries - cache_team - writing a team to cache: {json_data}') + valid_cardset = CardsetBase.model_validate(json_data, from_attributes=True) + logger.info(f'gameplay_queries - cache_team - valid_cardset: {valid_cardset}') + db_cardset = Cardset.model_validate(valid_cardset) + logger.info(f'gameplay_queries - cache_team - db_cardset: {db_cardset}') + session.add(db_cardset) + session.commit() + session.refresh(db_cardset) + return db_cardset + + if cardset_id is not None: + c_query = await db_get('cardsets', object_id=cardset_id) + if c_query is not None: + return cache_cardset(c_query) + + elif cardset_name is not None: + c_query = await db_get('cardsets', params=[('name', cardset_name)]) + if c_query['count'] != 0: + return cache_cardset(c_query['cardsets'][0]) + + logger.warning(f'No cardset found') return None