import asyncio import datetime from typing import List import discord import logging from discord import SelectOption from discord.utils import MISSING from sqlmodel import Session from api_calls import db_delete, db_get, db_post from exceptions import ( CardNotFoundException, LegalityCheckNotRequired, LineupsMissingException, PlayNotFoundException, PositionNotFoundException, log_exception, log_errors, ) from play_lock import safe_play_lock from helpers import ( DEFENSE_NO_PITCHER_LITERAL, get_card_embeds, position_name_to_abbrev, random_insult, ) from in_game.game_helpers import legal_check from in_game.gameplay_models import Game, Lineup, Play, Team from in_game.gameplay_queries import ( get_one_lineup, get_position, get_card_or_none, ) from utilities.buttons import ask_confirm, ask_position from utilities.embeds import image_embed logger = logging.getLogger("discord_app") class DropdownOptions(discord.ui.Select): def __init__( self, option_list: list, placeholder: str = "Make your selection", min_values: int = 1, max_values: int = 1, callback=None, ): # Set the options that will be presented inside the dropdown # options = [ # discord.SelectOption(label='Red', description='Your favourite colour is red', emoji='🟥'), # discord.SelectOption(label='Green', description='Your favourite colour is green', emoji='🟩'), # discord.SelectOption(label='Blue', description='Your favourite colour is blue', emoji='🟦'), # ] # The placeholder is what will be shown when no option is chosen # The min and max values indicate we can only pick one of the three options # The options parameter defines the dropdown options. We defined this above # If a default option is set on any SelectOption, the View will not process if only the default is # selected by the user self.custom_callback = callback super().__init__( placeholder=placeholder, min_values=min_values, max_values=max_values, options=option_list, ) async def callback(self, interaction: discord.Interaction): # Use the interaction object to send a response message containing # the user's favourite colour or choice. The self object refers to the # Select object, and the values attribute gets a list of the user's # selected options. We only want the first one. # await interaction.response.send_message(f'Your favourite colour is {self.values[0]}') logger.info(f"Dropdown callback: {self.custom_callback}") await self.custom_callback(interaction, self.values) class DropdownView(discord.ui.View): """ https://discordpy.readthedocs.io/en/latest/interactions/api.html#select """ def __init__( self, dropdown_objects: list[discord.ui.Select], timeout: float = 300.0 ): super().__init__(timeout=timeout) # self.add_item(Dropdown()) for x in dropdown_objects: self.add_item(x) class SelectViewDefense(discord.ui.Select): def __init__( self, options: list, this_play: Play, base_embed: discord.Embed, session: Session, sorted_lineups: list[Lineup], responders: list[discord.User] = None, ): self.embed = base_embed self.session = session self.play = this_play self.sorted_lineups = sorted_lineups self.responders = responders super().__init__(options=options) async def callback(self, interaction: discord.Interaction): if self.responders is not None and interaction.user not in self.responders: await interaction.response.send_message( content=random_insult(), ephemeral=True, delete_after=5 ) await interaction.response.defer(thinking=True) logger.info(f"SelectViewDefense - selection: {self.values[0]}") this_lineup = self.session.get(Lineup, self.values[0]) self.embed.set_image(url=this_lineup.player.image) select_player_options = [ discord.SelectOption( label=f"{x.position} - {x.player.name}", value=f"{x.id}", default=this_lineup.position == x.position, ) for x in self.sorted_lineups ] player_dropdown = SelectViewDefense( options=select_player_options, this_play=self.play, base_embed=self.embed, session=self.session, sorted_lineups=self.sorted_lineups, responders=self.responders, ) new_view = DropdownView(dropdown_objects=[player_dropdown], timeout=60) await interaction.edit_original_response( content=None, embed=self.embed, view=new_view ) class SelectStartingPitcher(discord.ui.Select): def __init__( self, this_game: Game, this_team: Team, session: Session, league_name: str, custom_id: str = MISSING, placeholder: str | None = None, options: List[SelectOption] = ..., responders: list[discord.User] = None, ) -> None: logger.info("Inside SelectStartingPitcher init function") # Store IDs instead of objects to avoid session issues self.game_id = this_game.id self.team_id = this_team.id self.league_name = league_name self.responders = responders super().__init__(custom_id=custom_id, placeholder=placeholder, options=options) async def callback(self, interaction: discord.Interaction): if self.responders is not None and interaction.user not in self.responders: await interaction.response.send_message( content=random_insult(), ephemeral=True, delete_after=5 ) await interaction.response.defer(thinking=True) logger.info(f"SelectStartingPitcher - selection: {self.values[0]}") # Create a new session for this callback from in_game.gameplay_models import engine with Session(engine) as session: # Get fresh game and team objects this_game = session.get(Game, self.game_id) this_team = session.get(Team, self.team_id) # Get Human SP card human_sp_card = await get_card_or_none(session, card_id=self.values[0]) if human_sp_card is None: log_exception( CardNotFoundException, f"Card ID {self.values[0]} not found" ) if human_sp_card.team_id != this_team.id: logger.error( f"Card_id {self.values[0]} does not belong to {this_team.abbrev} in Game {this_game.id}" ) await interaction.channel.send( f"Uh oh. Card ID {self.values[0]} is {human_sp_card.player.name} and belongs to {human_sp_card.team.sname}. Will you double check that before we get started?" ) return await get_position(session, human_sp_card, "P") try: legal_data = await legal_check( [self.values[0]], difficulty_name=self.league_name ) if not legal_data["legal"]: await interaction.edit_original_response( content=f"It looks like this is a Ranked Legal game and {human_sp_card.player.name_with_desc} is not legal in {self.league_name} games. You can start a new game once you pick a new SP." ) return except (LegalityCheckNotRequired, Exception) as e: # Skip legality check if not required or if it fails logger.info(f"Skipping legality check: {e}") pass human_sp_lineup = Lineup( team_id=this_team.id, player_id=human_sp_card.player.id, card_id=self.values[0], position="P", batting_order=10, is_fatigued=False, game=this_game, ) session.add(human_sp_lineup) session.commit() logger.info(f"trying to delete interaction: {interaction}") try: # await interaction.delete_original_response() await interaction.edit_original_response( # content=f'The {this_team.lname} are starting **{human_sp_card.player.name_with_desc}**!\n\nRun `/set lineup` to import your lineup and `/gamestate` if you are ready to play.', content=f"The {this_team.lname} are starting **{human_sp_card.player.name_with_desc}**!", view=None, ) except Exception as e: log_exception(e, "Couldn't clean up after selecting sp") class SelectReliefPitcher(discord.ui.Select): def __init__( self, this_game: Game, this_team: Team, batting_order: int, session: Session, custom_id: str = MISSING, placeholder: str | None = None, options: List[SelectOption] = ..., responders: list[discord.User] = None, ) -> None: logger.info("Inside SelectReliefPitcher init function") self.game = this_game self.team = this_team self.batting_order = int(batting_order) self.session = session self.responders = responders super().__init__(custom_id=custom_id, placeholder=placeholder, options=options) async def callback(self, interaction: discord.Interaction): if self.responders is not None and interaction.user not in self.responders: await interaction.response.send_message( content=random_insult(), ephemeral=True, delete_after=5 ) await interaction.response.defer(thinking=True) logger.info(f"SelectReliefPitcher - selection: {self.values[0]}") this_play = self.game.current_play_or_none(self.session) if this_play is None: log_exception( PlayNotFoundException, "Could not find current play to make pitching change. If you are trying to swap SPs, run `/set pitcher` again.", ) # Get Human RP card human_rp_card = await get_card_or_none(self.session, card_id=self.values[0]) if human_rp_card is None: log_exception(CardNotFoundException, f"Card ID {self.values[0]} not found") if human_rp_card.team_id != self.team.id: logger.error( f"Card_id {self.values[0]} does not belong to {self.team.abbrev} in Game {self.game.id}" ) await interaction.channel.send( f"Uh oh. Card ID {self.values[0]} is {human_rp_card.player.name} and belongs to {human_rp_card.team.sname}. Will you double check that?" ) return # Lock the play to prevent concurrent modifications with safe_play_lock(self.session, this_play): await get_position(self.session, human_rp_card, "P") if human_rp_card.pitcherscouting.pitchingcard.relief_rating < 2: this_play.in_pow = True logger.info("De-activating the old pitcher") this_play.pitcher.active = False self.session.add(this_play.pitcher) logger.info(f"Checking for batting order != 10 ({self.batting_order})") if self.batting_order != 10: logger.info(f"Getting the player in the {self.batting_order} spot") this_lineup = get_one_lineup( self.session, self.game, self.team, active=True, batting_order=self.batting_order, ) logger.info(f"subbing lineup: {this_lineup.player.name_with_desc}") # if this_lineup != this_play.pitcher: this_lineup.active = False self.session.add(this_lineup) logger.info("Adding the RP lineup") human_rp_lineup = Lineup( team_id=self.team.id, player_id=human_rp_card.player.id, card_id=self.values[0], position="P", batting_order=self.batting_order, is_fatigued=False, game=self.game, replacing_id=this_play.pitcher.id, after_play=max(this_play.play_num - 1, 0), ) self.session.add(human_rp_lineup) logger.info("Setting new pitcher on current play") this_play.pitcher = human_rp_lineup self.session.add(this_play) logger.info("Committing changes") try: self.session.commit() except Exception as e: log_exception(e, "Couldn't commit database changes") try: logger.info("Responding to player") await interaction.edit_original_response( content=f"**{human_rp_card.player.name_with_desc}** has entered for the {human_rp_card.team.lname}!\n\nRun `/substitute` to make any other moves and `/gamestate` if you are ready to continue.", view=None, ) except Exception as e: log_exception(e, "Couldn't clean up after selecting rp") class SelectDefensiveChange(discord.ui.Select): def __init__( self, this_game: Game, this_team: Team, new_position: DEFENSE_NO_PITCHER_LITERAL, session: Session, responders: list[discord.User] = None, custom_id=MISSING, placeholder: str | None = None, options: List[SelectOption] = ..., ): logger.info("Inside SelectDefensiveChange init function") self.game = this_game self.team = this_team self.new_position = new_position self.session = session self.responders = responders super().__init__(custom_id=custom_id, placeholder=placeholder, options=options) async def callback(self, interaction: discord.Interaction): if self.responders is not None and interaction.user not in self.responders: await interaction.response.send_message( content=random_insult(), ephemeral=True, delete_after=5 ) await interaction.response.defer(thinking=True) logger.info(f"SelectDefensiveChange - selection: {self.values[0]}") this_play = self.game.current_play_or_none(self.session) if this_play is None: log_exception( PlayNotFoundException, "Could not find current play to make defensive change.", ) # Get Human Defender card human_defender_lineup = self.session.get(Lineup, self.values[0]) if human_defender_lineup is None: log_exception( LineupsMissingException, f"Lineup ID {self.values[0]} not found" ) if human_defender_lineup.team_id != self.team.id: logger.error( f"Card_id {self.values[0]} does not belong to {self.team.abbrev} in Game {self.game.id}" ) await interaction.channel.send( f"Uh oh. Card ID {self.values[0]} is {human_defender_lineup.player.name} and belongs to {human_defender_lineup.team.sname}. Will you double check that?" ) return try: other_defender = get_one_lineup( session=self.session, this_game=self.game, this_team=self.team, active=True, position=position_name_to_abbrev(self.new_position), ) logger.info( f"Existing defender found at {self.new_position}: {other_defender}" ) except Exception: logger.info(f"No existing defender found at {self.new_position}") other_defender = None human_defender_lineup.position = position_name_to_abbrev(self.new_position) self.session.add(human_defender_lineup) logger.info("Committing changes") try: self.session.commit() except Exception as e: log_exception(e, "Couldn't commit database changes") try: logger.info("Responding to player") await interaction.edit_original_response( content=f"**{human_defender_lineup.player.name_with_desc}** has moved to {self.new_position} for the {human_defender_lineup.team.lname}!\n\nRun `/substitute` to make any other moves and `/gamestate` if you are ready to continue.", view=None, ) except Exception as e: log_exception(e, "Couldn't clean up after selecting defender") if other_defender: await interaction.channel.send( f"FYI - I see {other_defender.player.name} was previously at {self.new_position}. Do not forget to move them or sub them out!" ) @log_errors class SelectSubPosition(discord.ui.Select): def __init__( self, session: Session, this_lineup: Lineup, custom_id=MISSING, placeholder=None, options: List[SelectOption] = ..., responders: list[discord.User] = None, ): self.session = session self.this_lineup = this_lineup self.responders = responders super().__init__( custom_id=custom_id, placeholder=placeholder, min_values=1, max_values=1, options=options, disabled=False, ) @log_errors async def callback(self, interaction: discord.Interaction): if self.responders is not None and interaction.user not in self.responders: await interaction.response.send_message( content=random_insult(), ephemeral=True, delete_after=5 ) logger.info(f"Setting sub position to {self.values[0]}") await interaction.edit_original_response(view=None) if self.values[0] == "PH": await interaction.channel.send( content="Their position is set to Pinch Hitter." ) return else: await get_position( self.session, self.this_lineup.card_id, position=self.values[0] ) self.this_lineup.position = self.values[0] for option in self.options: if option.value == self.values[0]: this_label = option.label self.this_lineup.position = self.values[0] self.session.add(self.this_lineup) self.session.commit() await interaction.channel.send( content=f"Their position is set to {this_label}." ) class SelectBatterSub(discord.ui.Select): def __init__( self, this_game: Game, this_team: Team, session: Session, batting_order: int, custom_id: str = MISSING, placeholder: str | None = None, options: List[SelectOption] = ..., responders: list[discord.User] = None, ): logger.info("Inside SelectBatterSub init function") self.game = this_game self.team = this_team self.session = session # self.league_name = league_name self.batting_order = batting_order self.responders = responders super().__init__( custom_id=custom_id, placeholder=placeholder, min_values=1, max_values=1, options=options, ) @log_errors async def callback(self, interaction: discord.Interaction): if self.responders is not None and interaction.user not in self.responders: await interaction.response.send_message( content=random_insult(), ephemeral=True, delete_after=5 ) await interaction.response.defer() logger.info(f"Setting batter sub to Card ID: {self.values[0]}") # Get Human batter card logger.info("Looking up card substitution") human_batter_card = await get_card_or_none(self.session, card_id=self.values[0]) if human_batter_card is None: log_exception(CardNotFoundException, f"Card ID {self.values[0]} not found") logger.info(f"human_batter_card: {human_batter_card}") logger.info("Checking team ownership of batter card") if human_batter_card.team_id != self.team.id: logger.error( f"Card_id {self.values[0]} does not belong to {self.team.abbrev} in Game {self.game.id}" ) await interaction.channel.send( f"Uh oh. Card ID {self.values[0]} is {human_batter_card.player.name} and belongs to {human_batter_card.team.sname}. Will you double check that before we get started?" ) return logger.info("Ownership is good") # legal_data = await legal_check([self.values[0]], difficulty_name=self.league_name) # if not legal_data['legal']: # await interaction.edit_original_response( # content=f'It looks like this is a Ranked Legal game and {human_batter_card.player.name_with_desc} is not legal in {self.league_name} games. You can start a new game once you pick a new SP.' # ) # return logger.info("Pulling current play to log subs") this_play = self.game.current_play_or_none(self.session) if this_play is None: log_exception(PlayNotFoundException, "Play not found during substitution") logger.info(f"this_play: {this_play}") # Lock the play to prevent concurrent modifications with safe_play_lock(self.session, this_play): last_lineup = get_one_lineup( session=self.session, this_game=self.game, this_team=self.team, active=True, batting_order=self.batting_order, ) same_position = await ask_confirm( interaction, question=f"Will **{human_batter_card.player.name}** replace {last_lineup.player.name} as the {last_lineup.position}?", label_type="yes", ) if same_position: logger.info("same_position is True") position = last_lineup.position # pos_text = '' # view = None else: logger.info("same_position is False") position = await ask_position(interaction) if position == "PH/PR": if this_play.batter == last_lineup: position = "PH" else: position = "PR" logger.info("Deactivating last_lineup") try: last_lineup.active = False self.session.add(last_lineup) self.session.flush() # Flush to ensure the change is applied logger.info(f"Set lineup ID {last_lineup.id} as inactive") except Exception as e: log_exception(e) logger.info(f"new position: {position}") if position not in ["DH", "PR", "PH"]: logger.info("go get position rating") try: await get_position(self.session, human_batter_card, position) except PositionNotFoundException: logger.error( f"Position check failed for {human_batter_card.player.name_with_desc} at {position}, rolling back session" ) self.session.rollback() await interaction.edit_original_response( content=f"Uh oh, I cannot find {position} ratings for {human_batter_card.player.name_with_desc}. Please go double-check this sub and run again.", view=None, ) return logger.info("Creating new lineup record") human_bat_lineup = Lineup( team=self.team, player=human_batter_card.player, card=human_batter_card, position=position, batting_order=self.batting_order, game=self.game, after_play=max(this_play.play_num - 1, 0), replacing_id=last_lineup.id, ) logger.info(f"adding lineup to session: {human_bat_lineup}") self.session.add(human_bat_lineup) # self.session.commit() logger.info( f"Inserted player {human_batter_card.player_id} (card {human_batter_card.id}) in the {self.batting_order} spot" ) is_pinch_runner = False if this_play.batter == last_lineup: logger.info("Setting new sub to current play batter") this_play.batter = human_bat_lineup this_play.batter_pos = position elif this_play.on_first == last_lineup: logger.info("Setting new sub to run at first - this is a pinch runner") this_play.on_first = human_bat_lineup is_pinch_runner = True elif this_play.on_second == last_lineup: logger.info("Setting new sub to run at second - this is a pinch runner") this_play.on_second = human_bat_lineup is_pinch_runner = True elif this_play.on_third == last_lineup: logger.info("Setting new sub to run at third - this is a pinch runner") this_play.on_third = human_bat_lineup is_pinch_runner = True logger.info(f"Adding play to session: {this_play}") self.session.add(this_play) self.session.commit() # If this is a pinch runner, create an entry Play record for them if is_pinch_runner: # Import inside function to avoid circular import from command_logic.logic_gameplay import create_pinch_runner_entry_play logger.info( f"Creating pinch runner entry Play for {human_bat_lineup.player.name_with_desc}" ) create_pinch_runner_entry_play( session=self.session, game=self.game, current_play=this_play, pinch_runner_lineup=human_bat_lineup, ) # if not same_position: # pos_dict_list = { # 'Pinch Hitter': 'PH', # 'Catcher': 'C', # 'First Base': '1B', # 'Second Base': '2B', # 'Third Base': '3B', # 'Shortstop': 'SS', # 'Left Field': 'LF', # 'Center Field': 'CF', # 'Right Field': 'RF', # 'Pinch Runner': 'PR', # 'Pitcher': 'P' # } # logger.info(f'Prepping sub position') # pos_text = f'What position will {human_batter_card.player.name} play?' # options=[ # SelectOption( # label=f'{pos_name}', # value=pos_dict_list[pos_name], # default=pos_name=='Pinch Hitter' # ) # for pos_name in pos_dict_list.keys() # ] # logger.info(f'options: {options}') # view = SelectSubPosition( # self.session, # this_lineup=human_bat_lineup, # placeholder='Select Position', # options=options, # responders=[interaction.user] # ) # logger.info(f'view: {view}') # logger.info(f'SelectSubPosition view is ready') logger.info("Posting final sub message") this_content = f"{human_batter_card.player.name_with_desc} has entered in the {self.batting_order} spot.\n\nIf you have additional subs to make, run `/substitution` again or run `/gamestate` to see the current plate appearance." await interaction.edit_original_response(content=this_content) class SelectPokemonEvolution(discord.ui.Select): def __init__( self, *, placeholder="Evolve the selected Pokemon", min_values=1, max_values=1, options=List[SelectOption], this_team: Team, responders: list[discord.User] = None, ): logger.info("Inside SelectPokemonEvolution init function") self.team = this_team self.responders = responders super().__init__( placeholder=placeholder, min_values=min_values, max_values=max_values, options=options, ) logger.info("init complete") async def callback(self, interaction: discord.Interaction): logger.info("entering pokemon evolution callback") if self.responders is not None and interaction.user not in self.responders: await interaction.response.send_message( content=random_insult(), ephemeral=True, delete_after=5 ) logger.info("deferring interaction") await interaction.response.defer() try: card_id = self.values[0] logger.info(f"evolving card_id: {card_id}") this_card = await db_get("cards", object_id=card_id, none_okay=False) evo_mon = await db_get( "players", object_id=this_card["player"]["fangr_id"], none_okay=False ) logger.info(f"evo mon: {card_id}") p_query = await db_post( "packs/one", payload={ "team_id": self.team.id, "pack_type_id": 4, "open_time": datetime.datetime.timestamp(datetime.datetime.now()) * 1000, }, ) pack_id = p_query["id"] logger.info(f"pack_id: {pack_id}") logger.info("Posting evolved card") await db_post( "cards", payload={ "cards": [ { "player_id": evo_mon["player_id"], "team_id": self.team.id, "pack_id": pack_id, } ] }, timeout=10, ) await interaction.edit_original_response( content=f'## {this_card["player"]["p_name"].upper()} is evolving!', embeds=await get_card_embeds(this_card), view=None, ) await db_delete("cards", object_id=card_id) embed = image_embed( image_url=evo_mon["image"], title=evo_mon["p_name"], desc=f"{evo_mon['cardset']['name']} / {evo_mon['mlbclub']}", color=evo_mon["rarity"]["color"], thumbnail_url=evo_mon["headshot"], author_name=self.team.lname, author_icon=self.team.logo, ) await asyncio.sleep(3) await interaction.channel.send( content=f'## {this_card["player"]["p_name"].upper()} evolved into {evo_mon["p_name"].upper()}!', embeds=[embed], ) except Exception as e: logger.error( f"Failed to evolve a pokemon: {e}", exc_info=True, stack_info=True ) await interaction.edit_original_response( content="Oh no, the evolution failed! Go ping the shit out of Cal so he can evolve it for you!" )