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 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_game_lineups, 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(f'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(f'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 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(f'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(f'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(f'Setting new pitcher on current play') this_play.pitcher = human_rp_lineup self.session.add(this_play) logger.info(f'Committing changes') try: self.session.commit() except Exception as e: log_exception(e, 'Couldn\'t commit database changes') try: logger.info(f'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(f'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 as e: 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(f'Committing changes') try: self.session.commit() except Exception as e: log_exception(e, 'Couldn\'t commit database changes') try: logger.info(f'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=f'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(f'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(f'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(f'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(f'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(f'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}') 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(f'same_position is True') position = last_lineup.position # pos_text = '' # view = None else: logger.info(f'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(f'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(f'go get position rating') try: pos_rating = await get_position(self.session, human_batter_card, position) except PositionNotFoundException as e: 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(f'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(f'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(f'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(f'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(f'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(f'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(f'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(f'init complete') async def callback(self, interaction: discord.Interaction): logger.info(f'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(f'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(f'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=f'Oh no, the evolution failed! Go ping the shit out of Cal so he can evolve it for you!')