import datetime from dataclasses import dataclass import logging from typing import Optional, Literal from api_calls import db_get from helpers import PD_SEASON PLAYER_CACHE = {} TEAM_CACHE = {} BATTINGCARD_CACHE = {} # { : { : BattingWrapper } } PITCHINGCARD_CACHE = {} # { : { : PitchingWrapper } } logger = logging.getLogger('discord_app') @dataclass class Player: player_id: int p_name: str cost: int image: str mlbclub: str franchise: str cardset: dict set_num: int rarity: dict pos_1: str description: str quantity: Optional[int] = 999 image2: Optional[str] = None pos_2: Optional[str] = None pos_3: Optional[str] = None pos_4: Optional[str] = None pos_5: Optional[str] = None pos_6: Optional[str] = None pos_7: Optional[str] = None pos_8: Optional[str] = None headshot: Optional[str] = None vanity_card: Optional[str] = None strat_code: Optional[str] = None bbref_id: Optional[str] = None fangr_id: Optional[str] = None mlbplayer: Optional[dict] = None created: datetime.datetime = datetime.datetime.now() def to_dict(self): return { 'player_id': self.player_id, 'p_name': self.p_name, 'cost': self.cost, 'image': self.image, 'mlbclub': self.mlbclub, 'franchise': self.franchise, 'cardset': self.cardset, 'set_num': self.set_num, 'rarity': self.rarity, 'pos_1': self.pos_1, 'description': self.description, 'quantity': self.quantity, 'image2': self.image2, 'pos_2': self.pos_2, 'pos_3': self.pos_3, 'pos_4': self.pos_4, 'pos_5': self.pos_5, 'pos_6': self.pos_6, 'pos_7': self.pos_7, 'pos_8': self.pos_8, 'headshot': self.headshot, 'vanity_card': self.vanity_card, 'strat_code': self.strat_code, 'bbref_id': self.bbref_id, 'fangr_id': self.fangr_id, 'mlbplayer': self.mlbplayer } @dataclass class Team: id: int abbrev: str sname: str lname: str gmid: int gmname: str gsheet: str wallet: int team_value: int collection_value: int logo: str color: str season: int event: bool career: int ranking: int has_guide: bool is_ai: bool created: datetime.datetime = datetime.datetime.now() def to_dict(self): return { 'id': self.id, 'abbrev': self.abbrev, 'sname': self.sname, 'lname': self.lname, 'gmid': self.gmid, 'gmname': self.gmname, 'gsheet': self.gsheet, 'wallet': self.wallet, 'team_value': self.team_value, 'collection_value': self.collection_value, 'logo': self.logo, 'color': self.color, 'season': self.season, 'event': self.event, 'career': self.career, 'ranking': self.ranking, 'has_guide': self.has_guide, 'is_ai': self.is_ai } @dataclass class BattingCard: player_id: int variant: int steal_low: int steal_high: int steal_auto: bool steal_jump: float bunting: str hit_and_run: str running: int offense_col: int hand: str @dataclass class CardPosition: player_id: int variant: int position: Literal['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF', 'DH'] innings: int range: int error: int arm: Optional[int] = None pb: Optional[int] = None overthrow: Optional[int] = None def to_dict(self): return_val = { 'player_id': self.player_id, 'variant': self.variant, 'position': self.position, 'innings': self.innings, 'range': self.range, 'error': self.error } if self.arm is not None: return_val['arm'] = self.arm if self.pb is not None: return_val['pb'] = self.pb if self.overthrow is not None: return_val['overthrow'] = self.overthrow return return_val @dataclass class BattingRatings: homerun: float bp_homerun: float triple: float double_three: float double_two: float double_pull: float single_two: float single_one: float single_center: float bp_single: float hbp: float walk: float strikeout: float lineout: float popout: float flyout_a: float flyout_bq: float flyout_lf_b: float flyout_rf_b: float groundout_a: float groundout_b: float groundout_c: float avg: float obp: float slg: float pull_rate: float center_rate: float slap_rate: float @dataclass class BattingWrapper: card: BattingCard ratings_vl: BattingRatings ratings_vr: BattingRatings created: datetime.datetime = datetime.datetime.now() @dataclass class PitchingCard: player_id: int variant: int balk: int wild_pitch: int hold: int starter_rating: int relief_rating: int batting: str offense_col: int hand: str closer_rating: Optional[int] = None @dataclass class PitchingRatings: homerun: float bp_homerun: float triple: float double_three: float double_two: float double_cf: float single_two: float single_one: float single_center: float bp_single: float hbp: float walk: float strikeout: float flyout_lf_b: float flyout_cf_b: float flyout_rf_b: float groundout_a: float groundout_b: float xcheck_p: float xcheck_c: float xcheck_1b: float xcheck_2b: float xcheck_3b: float xcheck_ss: float xcheck_lf: float xcheck_cf: float xcheck_rf: float avg: float obp: float slg: float @dataclass class PitchingWrapper: card: PitchingCard ratings_vl: PitchingRatings ratings_vr: PitchingRatings created: datetime.datetime = datetime.datetime.now() async def get_pd_player(player_id, as_dict: Optional[bool] = True, skip_cache: bool = False): if player_id in PLAYER_CACHE and not skip_cache: tdelta = datetime.datetime.now() - PLAYER_CACHE[player_id].created if tdelta.total_seconds() < 1209600: logger.debug(f'this_player: {PLAYER_CACHE[player_id]}') if as_dict: return PLAYER_CACHE[player_id].to_dict() else: return PLAYER_CACHE[player_id] else: logger.error(f'Refreshing player {player_id} in cache...') this_player = await db_get('players', object_id=player_id) for bad_key in ['mlbplayer', 'paperdex']: if bad_key in this_player: del this_player[bad_key] logger.debug(f'this_player: {this_player}') PLAYER_CACHE[player_id] = Player(**this_player) if as_dict: return this_player return PLAYER_CACHE[player_id] async def get_pd_team(team_or_gm_id, as_dict: bool = True, skip_cache: bool = False): if team_or_gm_id in TEAM_CACHE and not skip_cache: tdelta = datetime.datetime.now() - TEAM_CACHE[team_or_gm_id].created if tdelta.total_seconds() < 1209600: logger.info(f'this_team: {TEAM_CACHE[team_or_gm_id]}') if as_dict: return TEAM_CACHE[team_or_gm_id].to_dict() else: return TEAM_CACHE[team_or_gm_id] else: logger.error(f'Refreshing team {team_or_gm_id} in cache...') this_team = await db_get('teams', object_id=team_or_gm_id, params=[('inc_packs', False)]) if this_team is None: t_query = await db_get('teams', params=[('season', PD_SEASON), ('gm_id', team_or_gm_id)]) if t_query is None: raise KeyError(f'No team found for ID {team_or_gm_id}') this_team = t_query['teams'][0] logger.info(f'this_team: {this_team}') TEAM_CACHE[team_or_gm_id] = Team(**this_team) if as_dict: return this_team return TEAM_CACHE[team_or_gm_id] async def get_pd_battingcard(player_id: int, variant: Optional[int] = 0): if player_id in BATTINGCARD_CACHE and variant in BATTINGCARD_CACHE[player_id]: tdelta = datetime.datetime.now() - BATTINGCARD_CACHE[player_id][variant].created if tdelta.total_seconds() < 609600: logger.info(f'this_battingcard: {BATTINGCARD_CACHE[player_id][variant]}') return BATTINGCARD_CACHE[player_id][variant] vl_data, vr_data = None, None r_query = await db_get(f'battingcardratings/player/{player_id}') if r_query['count'] > 0: for row in r_query['ratings']: if row['battingcard']['variant'] == variant: if row['vs_hand'].lower() == 'r': vr_data = row elif row['vs_hand'].lower() == 'l': vl_data = row if vl_data is not None and vr_data is not None: break if None in [vl_data, vr_data]: raise KeyError(f'Could not find batting card ratings for player {player_id} variant {variant}') logger.info(f'prepping the batting card') bc_data = vl_data['battingcard'] bc_data['player_id'] = player_id del bc_data['id'], bc_data['player'] this_bc = BattingCard(**bc_data) logger.info(f'prepping the vl ratings') vl_ratings = vl_data del vl_ratings['battingcard'], vl_ratings['id'], vl_ratings['vs_hand'] this_vl = BattingRatings(**vl_ratings) logger.info(f'prepping the vl ratings') vr_ratings = vr_data del vr_ratings['battingcard'], vr_ratings['id'], vr_ratings['vs_hand'] this_vr = BattingRatings(**vr_ratings) logger.info(f'prepping the wrapper') this_wrapper = BattingWrapper( card=this_bc, ratings_vl=this_vl, ratings_vr=this_vr ) if player_id in BATTINGCARD_CACHE: BATTINGCARD_CACHE[player_id][variant] = this_wrapper else: BATTINGCARD_CACHE[player_id] = {variant: this_wrapper} return this_wrapper async def get_pd_pitchingcard(player_id: int, variant: Optional[int] = 0): if player_id in PITCHINGCARD_CACHE and variant in PITCHINGCARD_CACHE[player_id]: tdelta = datetime.datetime.now() - PITCHINGCARD_CACHE[player_id][variant].created if tdelta.total_seconds() < 609600: logger.debug(f'this_pitchingcard: {PITCHINGCARD_CACHE[player_id][variant]}') return PITCHINGCARD_CACHE[player_id][variant] vl_data, vr_data = None, None r_query = await db_get(f'pitchingcardratings/player/{player_id}') if r_query['count'] > 0: for row in r_query['ratings']: if row['pitchingcard']['variant'] == variant: if row['vs_hand'].lower() == 'r': vr_data = row elif row['vs_hand'].lower() == 'l': vl_data = row if vl_data is not None and vr_data is not None: break if None in [vl_data, vr_data]: raise KeyError(f'Could not find pitching card ratings for player {player_id} variant {variant}') logger.debug(f'prepping the pitching card') pc_data = vl_data['pitchingcard'] pc_data['player_id'] = player_id del pc_data['id'], pc_data['player'] this_pc = PitchingCard(**pc_data) logger.debug(f'prepping the vl ratings') vl_ratings = vl_data del vl_ratings['pitchingcard'], vl_ratings['id'], vl_ratings['vs_hand'] this_vl = PitchingRatings(**vl_ratings) logger.debug(f'prepping the vr ratings') vr_ratings = vr_data del vr_ratings['pitchingcard'], vr_ratings['id'], vr_ratings['vs_hand'] this_vr = PitchingRatings(**vr_ratings) logger.debug(f'prepping the wrapper') this_wrapper = PitchingWrapper( card=this_pc, ratings_vl=this_vl, ratings_vr=this_vr ) if player_id in PITCHINGCARD_CACHE: PITCHINGCARD_CACHE[player_id][variant] = this_wrapper else: PITCHINGCARD_CACHE[player_id] = {variant: this_wrapper} return this_wrapper