8b2a442385
All checks were successful
Build Docker Image / build (pull_request) Successful in 3m0s
Add logging, user feedback, and wipe_team cleanup to the previously silent ZeroDivisionError handlers in the gauntlet draft flow. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
303 lines
12 KiB
Python
303 lines
12 KiB
Python
# Gauntlet Module
|
|
# Contains gauntlet game mode functionality from the original players.py
|
|
|
|
from discord.ext import commands
|
|
from discord import app_commands
|
|
import discord
|
|
from typing import Optional
|
|
|
|
# Import specific utilities needed by this module
|
|
import logging
|
|
import datetime
|
|
from sqlmodel import Session
|
|
from api_calls import db_get, db_post, db_patch, db_delete, get_team_by_abbrev
|
|
from helpers import (
|
|
ACTIVE_EVENT_LITERAL,
|
|
PD_PLAYERS_ROLE_NAME,
|
|
get_team_embed,
|
|
get_team_by_owner,
|
|
legal_channel,
|
|
Confirm,
|
|
send_to_channel,
|
|
)
|
|
from helpers.utils import get_roster_sheet, get_cal_user
|
|
from utilities.buttons import ask_with_buttons
|
|
from in_game.gameplay_models import engine
|
|
from in_game.gameplay_queries import get_team_or_none
|
|
|
|
logger = logging.getLogger("discord_app")
|
|
|
|
# Try to import gauntlets module, provide fallback if not available
|
|
try:
|
|
import gauntlets
|
|
|
|
GAUNTLETS_AVAILABLE = True
|
|
except ImportError:
|
|
logger.warning(
|
|
"Gauntlets module not available - gauntlet commands will have limited functionality"
|
|
)
|
|
GAUNTLETS_AVAILABLE = False
|
|
gauntlets = None
|
|
|
|
|
|
class Gauntlet(commands.Cog):
|
|
"""Gauntlet game mode functionality for Paper Dynasty."""
|
|
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
|
|
group_gauntlet = app_commands.Group(
|
|
name="gauntlets", description="Check your progress or start a new Gauntlet"
|
|
)
|
|
|
|
@group_gauntlet.command(
|
|
name="status", description="View status of current Gauntlet run"
|
|
)
|
|
@app_commands.describe(
|
|
team_abbrev="To check the status of a team's active run, enter their abbreviation"
|
|
)
|
|
@app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME)
|
|
async def gauntlet_run_command(
|
|
self,
|
|
interaction: discord.Interaction,
|
|
event_name: ACTIVE_EVENT_LITERAL, # type: ignore
|
|
team_abbrev: Optional[str] = None,
|
|
):
|
|
"""View status of current gauntlet run - corrected to match original business logic."""
|
|
await interaction.response.defer()
|
|
|
|
e_query = await db_get(
|
|
"events", params=[("name", event_name), ("active", True)]
|
|
)
|
|
if not e_query or e_query.get("count", 0) == 0:
|
|
await interaction.edit_original_response(
|
|
content=f"Hmm...looks like that event is inactive."
|
|
)
|
|
return
|
|
else:
|
|
this_event = e_query["events"][0]
|
|
|
|
this_run, this_team = None, None
|
|
if team_abbrev:
|
|
if "Gauntlet-" not in team_abbrev:
|
|
team_abbrev = f"Gauntlet-{team_abbrev}"
|
|
t_query = await db_get("teams", params=[("abbrev", team_abbrev)])
|
|
if t_query and t_query.get("count", 0) != 0:
|
|
this_team = t_query["teams"][0]
|
|
r_query = await db_get(
|
|
"gauntletruns",
|
|
params=[
|
|
("team_id", this_team["id"]),
|
|
("is_active", True),
|
|
("gauntlet_id", this_event["id"]),
|
|
],
|
|
)
|
|
|
|
if r_query and r_query.get("count", 0) != 0:
|
|
this_run = r_query["runs"][0]
|
|
else:
|
|
await interaction.edit_original_response(
|
|
content=f'I do not see an active run for the {this_team["lname"]}.'
|
|
)
|
|
return
|
|
else:
|
|
await interaction.edit_original_response(
|
|
content=f"I do not see an active run for {team_abbrev.upper()}."
|
|
)
|
|
return
|
|
|
|
# Use gauntlets module if available, otherwise show error
|
|
if GAUNTLETS_AVAILABLE and gauntlets:
|
|
await interaction.edit_original_response(
|
|
content=None,
|
|
embed=await gauntlets.get_embed(this_run, this_event, this_team), # type: ignore
|
|
)
|
|
else:
|
|
await interaction.edit_original_response(
|
|
content="Gauntlet status unavailable - gauntlets module not loaded."
|
|
)
|
|
|
|
@group_gauntlet.command(name="start", description="Start a new Gauntlet run")
|
|
@app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME)
|
|
async def gauntlet_start_command(self, interaction: discord.Interaction):
|
|
"""Start a new gauntlet run."""
|
|
|
|
# Channel restriction - must be in a 'hello' channel (private channel)
|
|
if (
|
|
interaction.channel
|
|
and hasattr(interaction.channel, "name")
|
|
and "hello" not in str(interaction.channel.name)
|
|
):
|
|
await interaction.response.send_message(
|
|
content="The draft will probably take you about 15 minutes. Why don't you head to your private "
|
|
"channel to run the draft?",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
logger.info(f"Starting a gauntlet run for user {interaction.user.name}")
|
|
await interaction.response.defer()
|
|
|
|
with Session(engine) as session:
|
|
main_team = await get_team_or_none(
|
|
session, gm_id=interaction.user.id, main_team=True
|
|
)
|
|
draft_team = await get_team_or_none(
|
|
session, gm_id=interaction.user.id, gauntlet_team=True
|
|
)
|
|
|
|
# Get active events
|
|
e_query = await db_get("events", params=[("active", True)])
|
|
if not e_query or e_query.get("count", 0) == 0:
|
|
await interaction.edit_original_response(
|
|
content="Hmm...I don't see any active events."
|
|
)
|
|
return
|
|
elif e_query.get("count", 0) == 1:
|
|
this_event = e_query["events"][0]
|
|
else:
|
|
event_choice = await ask_with_buttons(
|
|
interaction,
|
|
button_options=[x["name"] for x in e_query["events"]],
|
|
question="Which event would you like to take on?",
|
|
timeout=3,
|
|
delete_question=False,
|
|
)
|
|
this_event = [
|
|
event
|
|
for event in e_query["events"]
|
|
if event["name"] == event_choice
|
|
][0]
|
|
|
|
logger.info(f"this_event: {this_event}")
|
|
|
|
first_flag = draft_team is None
|
|
if draft_team is not None:
|
|
r_query = await db_get(
|
|
"gauntletruns",
|
|
params=[
|
|
("team_id", draft_team.id),
|
|
("gauntlet_id", this_event["id"]),
|
|
("is_active", True),
|
|
],
|
|
)
|
|
|
|
if r_query and r_query.get("count", 0) != 0:
|
|
await interaction.edit_original_response(
|
|
content=f'Looks like you already have a {r_query["runs"][0]["gauntlet"]["name"]} run active! '
|
|
f"You can check it out with the `/gauntlets status` command."
|
|
)
|
|
return
|
|
|
|
try:
|
|
draft_embed = await gauntlets.run_draft(interaction, main_team, this_event, draft_team) # type: ignore
|
|
except ZeroDivisionError as e:
|
|
logger.error(
|
|
f'ZeroDivisionError in {this_event["name"]} draft for the {main_team.sname if main_team else "unknown"}: {e}'
|
|
)
|
|
await gauntlets.wipe_team(draft_team, interaction) # type: ignore
|
|
await interaction.followup.send(
|
|
content=f"Shoot - it looks like we ran into an issue running the draft. I had to clear it all out "
|
|
f"for now. I let {get_cal_user(interaction).mention} know what happened so he better "
|
|
f"fix it quick."
|
|
)
|
|
return
|
|
except Exception as e:
|
|
logger.error(
|
|
f'Failed to run {this_event["name"]} draft for the {main_team.sname if main_team else "unknown"}: {e}'
|
|
)
|
|
await gauntlets.wipe_team(draft_team, interaction) # type: ignore
|
|
await interaction.followup.send(
|
|
content=f"Shoot - it looks like we ran into an issue running the draft. I had to clear it all out "
|
|
f"for now. I let {get_cal_user(interaction).mention} know what happened so he better "
|
|
f"fix it quick."
|
|
)
|
|
return
|
|
|
|
if first_flag:
|
|
await interaction.followup.send(
|
|
f"Good luck, champ in the making! To start playing, follow these steps:\n\n"
|
|
f"1) Make a copy of the Team Sheet Template found in `/help-pd links`\n"
|
|
f"2) Run `/newsheet` to link it to your Gauntlet team\n"
|
|
f'3) Go play your first game with `/new-game gauntlet {this_event["name"]}`'
|
|
)
|
|
else:
|
|
await interaction.followup.send(
|
|
f"Good luck, champ in the making! In your team sheet, sync your cards with **Paper Dynasty** -> "
|
|
f"**Data Imports** -> **My Cards** then you can set your lineup here and you'll be ready to go!\n\n"
|
|
f"{get_roster_sheet(draft_team)}"
|
|
)
|
|
|
|
await send_to_channel(
|
|
bot=self.bot,
|
|
channel_name="pd-news-ticker",
|
|
content=f'The {main_team.lname if main_team else "Unknown Team"} have entered the {this_event["name"]} Gauntlet!',
|
|
embed=draft_embed,
|
|
)
|
|
|
|
@group_gauntlet.command(
|
|
name="reset", description="Wipe your current team so you can re-draft"
|
|
)
|
|
@app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME)
|
|
async def gauntlet_reset_command(self, interaction: discord.Interaction, event_name: ACTIVE_EVENT_LITERAL): # type: ignore
|
|
"""Reset current gauntlet run."""
|
|
await interaction.response.defer()
|
|
main_team = await get_team_by_owner(interaction.user.id)
|
|
draft_team = await get_team_by_abbrev(f'Gauntlet-{main_team["abbrev"]}')
|
|
if draft_team is None:
|
|
await interaction.edit_original_response(
|
|
content="Hmm, I can't find a gauntlet team for you. Have you signed up already?"
|
|
)
|
|
return
|
|
|
|
e_query = await db_get(
|
|
"events", params=[("name", event_name), ("active", True)]
|
|
)
|
|
if e_query["count"] == 0:
|
|
await interaction.edit_original_response(
|
|
content="Hmm...looks like that event is inactive."
|
|
)
|
|
return
|
|
else:
|
|
this_event = e_query["events"][0]
|
|
|
|
r_query = await db_get(
|
|
"gauntletruns",
|
|
params=[
|
|
("team_id", draft_team["id"]),
|
|
("is_active", True),
|
|
("gauntlet_id", this_event["id"]),
|
|
],
|
|
)
|
|
|
|
if r_query and r_query.get("count", 0) != 0:
|
|
this_run = r_query["runs"][0]
|
|
else:
|
|
await interaction.edit_original_response(
|
|
content=f'I do not see an active run for the {draft_team["lname"]}.'
|
|
)
|
|
return
|
|
|
|
view = Confirm(responders=[interaction.user], timeout=60)
|
|
conf_string = f"Are you sure you want to wipe your active run?"
|
|
await interaction.edit_original_response(content=conf_string, view=view)
|
|
await view.wait()
|
|
|
|
if view.value:
|
|
await gauntlets.end_run(this_run, this_event, draft_team, force_end=True) # type: ignore
|
|
await interaction.edit_original_response(
|
|
content=f"Your {event_name} run has been reset. Run `/gauntlets start` to redraft!",
|
|
view=None,
|
|
)
|
|
|
|
else:
|
|
await interaction.edit_original_response(
|
|
content=f"~~{conf_string}~~\n\nNo worries, I will leave it active.",
|
|
view=None,
|
|
)
|
|
|
|
|
|
async def setup(bot):
|
|
"""Setup function for the Gauntlet cog."""
|
|
await bot.add_cog(Gauntlet(bot))
|