347 lines
16 KiB
Python
347 lines
16 KiB
Python
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 get_lineups_from_sheets, checks_log_interaction
|
|
from exceptions import GameNotFoundException, TeamNotFoundException, PlayNotFoundException, GameException
|
|
from helpers import PD_PLAYERS_ROLE_NAME, 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, Session, engine, player_description, select, Game
|
|
from in_game.gameplay_queries import 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
|
|
|
|
|
|
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):
|
|
logging.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):
|
|
logging.info(f'Waiting to get sheets')
|
|
await self.bot.wait_until_ready()
|
|
|
|
async def cog_command_error(self, ctx, error):
|
|
await ctx.send(f'{error}\n\nRun !help <command_name> to see the command requirements')
|
|
|
|
async def slash_error(self, ctx, error):
|
|
await ctx.send(f'{error[:1600]}')
|
|
|
|
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 <ranked / unlimited>` 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 away_team
|
|
|
|
conflict_games = get_active_games_by_team(session, team_id=human_team.id)
|
|
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']
|
|
logging.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_num=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}'
|
|
logging.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:
|
|
logging.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
|
|
|
|
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.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.with_desc}**:\n\n{ai_sp_lineup.player.p_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()
|
|
# session.refresh(this_game)
|
|
|
|
away_role = await team_role(interaction, away_team)
|
|
home_role = await team_role(interaction, home_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=this_game.get_scorebug_embed(session)
|
|
)
|
|
|
|
@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
|
|
|
|
await ctx.send(
|
|
content=None,
|
|
embed=this_game.get_scorebug_embed(session, include_lineups=True)
|
|
)
|
|
|
|
view = Confirm(responders=[ctx.author], timeout=60, label_type='confirm')
|
|
question = await ctx.send(f'Is this the game I should nuke?', view=view)
|
|
await view.wait()
|
|
|
|
if view.value:
|
|
session.delete(this_game)
|
|
session.commit()
|
|
await question.edit(
|
|
content=random_gif(random_from_list([
|
|
'i killed it', 'deed is done', 'gone forever'
|
|
])),
|
|
view=None
|
|
)
|
|
else:
|
|
await question.edit(
|
|
content=f'~~Is this the game I should nuke?~~\n\nIt stays.',
|
|
view=None
|
|
)
|
|
|
|
@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:
|
|
logging.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))
|
|
|
|
for batter in human_lineups:
|
|
session.add(batter)
|
|
|
|
session.commit()
|
|
|
|
await interaction.edit_original_response(content=None, embed=this_game.get_scorebug_embed(session))
|
|
|
|
@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
|
|
|
|
await interaction.edit_original_response(
|
|
content=None,
|
|
embed=this_game.get_scorebug_embed(session, full_length=include_lineups)
|
|
)
|
|
|
|
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='flyball')
|
|
|
|
|
|
async def setup(bot):
|
|
await bot.add_cog(Gameplay(bot)) |