import logging from typing import Literal import discord from discord import app_commands from discord.app_commands import Choice from discord.ext import commands, tasks import pygsheets from api_calls import db_get from command_logic.logic_gameplay import advance_runners, bunts, chaos, complete_game, doubles, flyballs, get_lineups_from_sheets, checks_log_interaction, complete_play, get_scorebug_embed, hit_by_pitch, homeruns, is_game_over, popouts, show_defense_cards, singles, strikeouts, triples, undo_play, update_game_settings, walks from dice import ab_roll from exceptions import GameNotFoundException, TeamNotFoundException, PlayNotFoundException, GameException, log_exception from helpers import DEFENSE_LITERAL, PD_PLAYERS_ROLE_NAME, get_channel, team_role, user_has_role, random_gif, random_from_list # from in_game import ai_manager from in_game.ai_manager import get_starting_pitcher, get_starting_lineup from in_game.game_helpers import PUBLIC_FIELDS_CATEGORY_NAME, legal_check from in_game.gameplay_models import Lineup, Play, Session, engine, player_description, select, Game from in_game.gameplay_queries import get_and_cache_position, get_channel_game_or_none, get_active_games_by_team, get_game_lineups, get_team_or_none, get_card_or_none from utilities.buttons import Confirm, ScorebugButtons, ask_confirm CLASSIC_EMBED = True logger = logging.getLogger('discord_app') class Gameplay(commands.Cog): def __init__(self, bot): self.bot = bot self.sheets = None self.get_sheets.start() @tasks.loop(count=1) async def get_sheets(self): logger.info(f'Getting sheets') self.sheets = pygsheets.authorize(service_file='storage/paper-dynasty-service-creds.json', retries=1) @get_sheets.before_loop async def before_get_sheets(self): logger.info(f'Waiting to get sheets') await self.bot.wait_until_ready() async def cog_command_error(self, ctx, error): logger.error(msg=error, stack_info=True) await ctx.send(f'{error}\n\nRun !help to see the command requirements') async def slash_error(self, ctx, error): logger.error(msg=error, stack_info=True) await ctx.send(f'{error[:1600]}') async def post_play(self, session: Session, interaction: discord.Interaction, this_play: Play, buffer_message: str = None): logger.info(f'post_play - Posting new play') if is_game_over(this_play): logger.info(f'Game {this_play.game.id} seems to be over') await interaction.edit_original_response(content=f'Looks like this one is over!') submit_game = await ask_confirm( interaction=interaction, question=f'Final score: {this_play.game.away_team.abbrev} {this_play.away_score} - {this_play.home_score} {this_play.game.home_team.abbrev}\n{this_play.scorebug_ascii}\nShould I go ahead and submit this game or roll it back a play?', custom_confirm_label='Submit', custom_cancel_label='Roll Back' ) if submit_game: logger.info(f'post_play - is_game_over - {interaction.user.display_name} rejected game completion') await complete_game(session, interaction, this_play) return else: logger.warning(f'post_play - is_game_over - {interaction.user.display_name} rejected game completion in Game {this_play.game.id}') cal_channel = get_channel(interaction, 'commissioners-office') await cal_channel.send(content=f'{interaction.user.display_name} just rejected game completion down in {interaction.channel.mention}') this_play = undo_play(session, this_play) await self.post_play(session, interaction, this_play) await interaction.channel.send(content=f'I let Cal know his bot is stupid') scorebug_buttons, this_ab_roll = None, None scorebug_embed = await get_scorebug_embed(session, this_play.game, full_length=False, classic=CLASSIC_EMBED) if this_play.game.roll_buttons and interaction.user.id in [this_play.game.away_team.gmid, this_play.game.home_team.gmid]: scorebug_buttons = ScorebugButtons(this_play, scorebug_embed) if this_play.on_base_code == 0 and this_play.game.auto_roll and not this_play.batter.team.is_ai: this_ab_roll = ab_roll(this_play.batter.team, this_play.game, allow_chaos=False) scorebug_buttons = None if this_ab_roll is not None and this_ab_roll.d_six_one > 3: scorebug_embed.set_image(url=this_play.pitcher.player.pitcher_card_url) if buffer_message is not None: await interaction.edit_original_response( content=buffer_message ) await interaction.channel.send( content=None, embed=scorebug_embed, view=scorebug_buttons ) else: await interaction.edit_original_response( content=None, embed=scorebug_embed, view=scorebug_buttons ) if this_ab_roll is not None: await interaction.channel.send( content=None, embeds=this_ab_roll.embeds ) async def complete_and_post_play(self, session: Session, interaction: discord.Interaction, this_play: Play, buffer_message: str = None): next_play = complete_play(session, this_play) logger.info(f'Completed play {this_play.id}') await self.post_play(session, interaction, next_play, buffer_message) group_new_game = app_commands.Group(name='new-game', description='Start a new baseball game') @group_new_game.command(name='mlb-campaign', description='Start a new MLB campaign game against an AI') @app_commands.describe( sp_card_id='Light gray number to the left of the pitcher\'s name on your depth chart' ) @app_commands.choices(league=[ Choice(value='minor-league', name='Minor League'), Choice(value='flashback', name='Flashback'), Choice(value='major-league', name='Major League'), Choice(value='hall-of-fame', name='Hall of Fame') ]) @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def new_game_mlb_campaign_command( self, interaction: discord.Interaction, league: Choice[str], away_team_abbrev: str, home_team_abbrev: str, sp_card_id: int ): await interaction.response.defer() with Session(engine) as session: conflict = get_channel_game_or_none(session, interaction.channel_id) if conflict is not None: await interaction.edit_original_response( content=f'Ope. There is already a game going on in this channel. Please wait for it to complete ' f'before starting a new one.' ) return if interaction.channel.category is None or interaction.channel.category.name != PUBLIC_FIELDS_CATEGORY_NAME: await interaction.edit_original_response( content=f'Why don\'t you head down to one of the Public Fields that way other humans can help if anything pops up?' ) return try: away_team = await get_team_or_none(session, team_abbrev=away_team_abbrev) except LookupError as e: await interaction.edit_original_response( content=f'Hm. I\'m not sure who **{away_team_abbrev}** is - check on that and try again!' ) return try: home_team = await get_team_or_none(session, team_abbrev=home_team_abbrev) except LookupError as e: await interaction.edit_original_response( content=f'Hm. I\'m not sure who **{home_team_abbrev}** is - check on that and try again!' ) return if not away_team.is_ai ^ home_team.is_ai: await interaction.edit_original_response( content=f'I don\'t see an AI team in this MLB Campaign game. Run `/new-game mlb-campaign` again with an AI for a campaign game or `/new-game ` for a PvP game.' ) return ai_team = away_team if away_team.is_ai else home_team human_team = away_team if home_team.is_ai else home_team conflict_games = get_active_games_by_team(session, team=human_team) if len(conflict_games) > 0: await interaction.edit_original_response( content=f'Ope. The {human_team.sname} are already playing over in {interaction.guild.get_channel(conflict_games[0].channel_id).mention}' ) return current = await db_get('current') week_num = current['week'] logger.info(f'gameplay - new_game_mlb_campaign - Season: {current["season"]} / Week: {week_num} / Away Team: {away_team.description} / Home Team: {home_team.description}') def role_error(required_role: str, league_name: str, lower_league: str): return f'Ope. Looks like you haven\'t received the **{required_role}** role, yet!\n\nTo play **{league_name}** games, you need to defeat all 30 MLB teams in the {lower_league} campaign. You can see your progress with `/record`.\n\nIf you have completed the {lower_league} campaign, go ping Cal to get your new role!' if league.value == 'flashback': if not user_has_role(interaction.user, 'PD - Major League'): await interaction.edit_original_response( content=role_error('PD - Major League', league_name='Flashback', lower_league='Minor League') ) return elif league.value == 'major-league': if not user_has_role(interaction.user, 'PD - Major League'): await interaction.edit_original_response( content=role_error('PD - Major League', league_name='Major League', lower_league='Minor League') ) return elif league.value == 'hall-of-fame': if not user_has_role(interaction.user, 'PD - Hall of Fame'): await interaction.edit_original_response( content=role_error('PD - Hall of Fame', league_name='Hall of Fame', lower_league='Major League') ) return this_game = Game( away_team_id=away_team.id, home_team_id=home_team.id, channel_id=interaction.channel_id, season=current['season'], week=week_num, first_message=None if interaction.message is None else interaction.message.channel.id, ai_team='away' if away_team.is_ai else 'home', game_type=league.value ) game_info_log = f'{league.name} game between {away_team.description} and {home_team.description} / first message: {this_game.first_message}' logger.info(game_info_log) # Get Human SP card human_sp_card = await get_card_or_none(session, card_id=sp_card_id) if human_sp_card is None: await interaction.channel.send( f'Uh oh. I can\'t find a card with ID {sp_card_id}. Will you double check that before we get started?' ) return if human_sp_card.team_id != human_team.id: logger.error(f'Card_id {sp_card_id} does not belong to {human_team.abbrev} in Game {this_game.id}') await interaction.channel.send( f'Uh oh. Card ID {sp_card_id} 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_and_cache_position(session, human_sp_card, 'P') legal_data = await legal_check([sp_card_id], difficulty_name=league.value) 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 {league.name} games. You can start a new game once you pick a new SP.' ) return human_sp_lineup = Lineup( team_id=human_team.id, player_id=human_sp_card.player.id, card_id=sp_card_id, position='P', batting_order=10, is_fatigued=False, game=this_game ) # session.add(human_sp_lineup) # Get AI SP await interaction.edit_original_response( content=f'{ai_team.gmname} is looking for a SP to counter {human_sp_card.player.name}...' ) ai_sp_lineup = await get_starting_pitcher( session, ai_team, this_game, True if home_team.is_ai else False, league.value ) await interaction.edit_original_response( content=f'The {ai_team.sname} are starting **{ai_sp_lineup.player.name_with_desc}**:\n\n{ai_sp_lineup.player.pitcher_card_url}' ) # Get AI Lineup final_message = await interaction.channel.send( content=f'{ai_team.gmname} is filling out the {ai_team.sname} lineup card...' ) batter_lineups = await get_starting_lineup( session, team=ai_team, game=this_game, league_name=league.value, sp_name=human_sp_card.player.name ) # Commit game and lineups session.add(this_game) session.commit() await final_message.edit(content=f'The {ai_team.sname} lineup is in, pulling in scouting data...') for batter in batter_lineups: if batter.position != 'DH': await get_and_cache_position(session, batter.card, batter.position) # session.refresh(this_game) away_role = await team_role(interaction, away_team) home_role = await team_role(interaction, home_team) embed = await get_scorebug_embed(session, this_game) embed.clear_fields() embed.add_field( name=f'{ai_team.abbrev} Lineup', value=this_game.team_lineup(session, ai_team) ) await final_message.edit( content=f'{away_role.mention} @ {home_role.mention} is set!\n\n' f'Go ahead and set lineups with the `/read-lineup` command!', embed=embed ) @commands.command(name='force-endgame', help='Mod: Force a game to end without stats') async def force_end_game_command(self, ctx: commands.Context): with Session(engine) as session: this_game = get_channel_game_or_none(session, ctx.channel.id) if this_game is None: await ctx.send(f'I do not see a game here - are you in the right place?') return try: await ctx.send( content=None, embed=await get_scorebug_embed(session, this_game, full_length=True) ) except Exception as e: logger.error(f'Unable to display scorebug while forcing game to end: {e}') await ctx.send(content='This game is so boned that I can\'t display the scorebug.') nuke_game = await ask_confirm( ctx, question=f'Is this the game I should nuke?', label_type='yes', timeout=15, ) # if view.value: if nuke_game: session.delete(this_game) session.commit() await ctx.channel.send(content=random_gif(random_from_list(['i killed it', 'deed is done', 'gone forever']))) else: await ctx.send(f'It stays. For now.') @app_commands.command(name='read-lineup', description='Import a saved lineup for this channel\'s PD game.') @app_commands.describe( roster='Which roster to pull from your sheet?', lineup='Which handedness lineup are you using?' ) @app_commands.choices( roster=[ Choice(value='1', name='Primary'), Choice(value='2', name='Secondary'), Choice(value='3', name='Ranked') ], lineup=[ Choice(value='1', name='v Right'), Choice(value='2', name='v Left') ] ) @app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME) async def read_lineup_command(self, interaction: discord.Interaction, roster: Choice[str], lineup: Choice[str]): await interaction.response.defer() with Session(engine) as session: this_game = get_channel_game_or_none(session, interaction.channel_id) if this_game is None: await interaction.edit_original_response( content=f'Hm. I don\'t see a game going on in this channel. Am I drunk?' ) return lineup_team = this_game.away_team if this_game.ai_team == 'home' else this_game.home_team if interaction.user.id != lineup_team.gmid: logger.info(f'{interaction.user.name} tried to run a command in Game {this_game.id} when they aren\'t a GM in the game.') await interaction.edit_original_response(content='Bruh. Only GMs of the active teams can pull lineups.') return existing_lineups = get_game_lineups( session=session, this_game=this_game, specific_team=lineup_team, is_active=True ) if len(existing_lineups) > 1: await interaction.edit_original_response( f'It looks like the {lineup_team.sname} already have a lineup. Run `/substitution` to make changes.' ) return await interaction.edit_original_response(content='Okay, let\'s put this lineup card together...') if this_game.away_team == lineup_team: this_game.away_roster_id = int(roster.value) else: this_game.home_roster_id = int(roster.value) session.add(this_game) human_lineups = await get_lineups_from_sheets(session, self.sheets, this_game, this_team=lineup_team, lineup_num=lineup.name, roster_num=int(roster.value)) await interaction.edit_original_response(content='Heard from sheets, pulling in scouting data...') for batter in human_lineups: session.add(batter) session.commit() for batter in human_lineups: if batter.position != 'DH': await get_and_cache_position(session, batter.card, batter.position) this_play = this_game.initialize_play(session) await self.post_play(session, interaction, this_play) @app_commands.command(name='gamestate', description='Post the current game state') async def gamestate_command(self, interaction: discord.Interaction, include_lineups: bool = False): await interaction.response.defer(ephemeral=True, thinking=True) with Session(engine) as session: this_game = get_channel_game_or_none(session, interaction.channel_id) if this_game is None: await interaction.edit_original_response( content=f'Hm. I don\'t see a game going on in this channel. Am I drunk?' ) return this_play = this_game.current_play_or_none(session) await self.post_play(session, interaction, this_play) @app_commands.command(name='settings-ingame', description='Change in-game settings') @app_commands.describe( roll_buttons='Display the "Roll AB" and "Check Jump" buttons along with the scorebug', auto_roll='When there are no baserunners, automatically roll the next AB' ) async def game_settings_command(self, interaction: discord.Interaction, roll_buttons: bool = None, auto_roll: bool = None): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='settings-ingame') await interaction.edit_original_response(content=None, embed=await update_game_settings( session, interaction, this_game, roll_buttons=roll_buttons, auto_roll=auto_roll )) group_log = app_commands.Group(name='log', description='Log a play in this channel\'s game') @group_log.command(name='flyball', description='Flyballs: a, b, ballpark, bq, c') async def log_flyball(self, interaction: discord.Interaction, flyball_type: Literal['a', 'b', 'ballpark', 'b?', 'c']): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log flyball') logger.info(f'log flyball {flyball_type} - this_play: {this_play}') this_play = await flyballs(session, interaction, this_play, flyball_type) await self.complete_and_post_play( session, interaction, this_play, buffer_message='Double logged' if this_play.starting_outs + this_play.outs < 3 and ((this_play.on_second and flyball_type == 'b') or (this_play.on_third and flyball_type == '?b')) else None ) @group_log.command(name='single', description='Singles: *, **, ballpark, uncapped') async def log_single( self, interaction: discord.Interaction, single_type: Literal['*', '**', 'ballpark', 'uncapped']): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log single') this_play = await singles(session, interaction, this_play, single_type) logger.info(f'log single {single_type} - this_play: {this_play}') await self.complete_and_post_play(session, interaction, this_play, buffer_message='Single logged' if ((this_play.on_first or this_play.on_second) and single_type == 'uncapped') else None) # complete_play(session, this_play) # if ((this_play.on_first or this_play.on_second) and single_type == 'uncapped'): # await interaction.edit_original_response(content='Single logged') # await interaction.channel.send( # content=None, # embed=await get_scorebug_embed(session, this_play.game, full_length=False, classic=CLASSIC_EMBED) # ) # else: # await interaction.edit_original_response( # content=None, # embed=await get_scorebug_embed(session, this_play.game, full_length=False, classic=CLASSIC_EMBED) # ) @group_log.command(name='double', description='Doubles: **, ***, uncapped') async def log_double(self, interaction: discord.Interaction, double_type: Literal['**', '***', 'uncapped']): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log double') logger.info(f'log double {double_type} - this_play: {this_play}') this_play = await doubles(session, interaction, this_play, double_type) await self.complete_and_post_play(session, interaction, this_play, buffer_message='Double logged' if (this_play.on_first and double_type == 'uncapped') else None) @group_log.command(name='triple', description='Triples: no sub-types') async def log_triple(self, interaction: discord.Interaction): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log triple') logger.info(f'log triple - this_play: {this_play}') this_play = await triples(session, interaction, this_play) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name='homerun', description='Home Runs: ballpark, no-doubt') async def log_homerun(self, interaction: discord.Interaction, homerun_type: Literal['ballpark', 'no-doubt']): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log homerun') logger.info(f'log homerun {homerun_type} - this_play: {this_play}') this_play = await homeruns(session, interaction, this_play, homerun_type) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name='walk', description='Walks: unintentional (default), intentional') async def log_walk(self, interaction: discord.Interaction, walk_type: Literal['unintentional', 'intentional'] = 'unintentional'): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log walk') logger.info(f'log walk {walk_type} - this_play: {this_play}') this_play = await walks(session, interaction, this_play, walk_type) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name='strikeout', description='Strikeout') async def log_strikeout(self, interaction: discord.Interaction): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log strikeout') logger.info(f'log strikeout - this_play: {this_play}') this_play = await strikeouts(session, interaction, this_play) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name='popout', description='Popout') async def log_popout(self, interaction: discord.Interaction): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log popout') logger.info(f'log popout - this_play: {this_play}') this_play = await popouts(session, interaction, this_play) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name='hit-by-pitch', description='Hit by pitch: batter to first; runners advance if forced') async def log_hit_by_pitch(self, interaction: discord.Interaction): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log hit-by-pitch') logger.info(f'log hit-by-pitch - this_play: {this_play}') this_play = await hit_by_pitch(session, interaction, this_play) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name='chaos', description='Chaos: wild-pitch, passed-ball, balk, pickoff') async def log_chaos(self, interaction: discord.Interaction, chaos_type: Literal['wild-pitch', 'passed-ball', 'balk', 'pickoff']): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log hit-by-pitch') if this_play.on_base_code == 0: await interaction.edit_original_response( content=f'There cannot be chaos when the bases are empty.' ) return logger.info(f'log chaos - this_play: {this_play}') this_play = await chaos(session, interaction, this_play, chaos_type) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name='bunt', description='Bunts: sacrifice, bad, popout, double-play, defense') async def log_sac_bunt(self, interaction: discord.Interaction, bunt_type: Literal['sacrifice', 'bad', 'popout', 'double-play', 'defense']): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log bunt') if this_play.on_base_code == 0: await interaction.edit_original_response( content=f'You cannot bunt when the bases are empty.' ) return elif this_play.starting_outs == 2: await interaction.edit_original_response( content=f'You cannot bunt with two outs.' ) return logger.info(f'log bunt - this_play: {this_play}') this_play = await bunts(session, interaction, this_play, bunt_type) await self.complete_and_post_play(session, interaction, this_play) @group_log.command(name='undo-play', description='Roll back most recent play from the log') async def log_undo_play_command(self, interaction: discord.Interaction): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='log undo-play') logger.info(f'log undo-play - this_play: {this_play}') this_play = undo_play(session, this_play) await self.post_play(session, interaction, this_play) group_show = app_commands.Group(name='show-card', description='Display the player card for an active player') @group_show.command(name='defense', description='Display a defender\'s player card') async def show_defense_command(self, interaction: discord.Interaction, position: DEFENSE_LITERAL): with Session(engine) as session: this_game, owner_team, this_play = await checks_log_interaction(session, interaction, command_name='show-card defense') logger.info(f'show-card defense - position: {position}') await show_defense_cards(session, interaction, this_play, position) async def setup(bot): await bot.add_cog(Gameplay(bot))