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("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("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("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("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("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("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("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("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