cal/paper-dynasty-discord
1
0
Fork 0
Code Issues 14 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Merge pull request 'perf: parallelize roll_for_cards and bump pack limit to 20' (#102) from performance/97-parallelize-roll-for-cards into next-release
All checks were successful
Build Docker Image / build (push) Successful in 1m25s
Details
Build Docker Image / build (pull_request) Successful in 54s
Details

Browse Source 
Reviewed-on: #102
This commit is contained in:
cal 2026-03-20 15:34:06 +00:00
commit 44d83b321f
3 changed files with 1108 additions and 433 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

620
discord_ui/selectors.py
View File

@ -3,126 +3,148 @@ Discord Select UI components.
Contains all Select classes for various team, cardset, and pack selections.
"""
import logging
import discord
from typing import Literal, Optional
from helpers.constants import ALL_MLB_TEAMS, IMAGES, normalize_franchise
logger = logging.getLogger('discord_app')
logger = logging.getLogger("discord_app")
# Team name to ID mappings
AL_TEAM_IDS = {
'Baltimore Orioles': 3,
'Boston Red Sox': 4,
'Chicago White Sox': 6,
'Cleveland Guardians': 8,
'Detroit Tigers': 10,
'Houston Astros': 11,
'Kansas City Royals': 12,
'Los Angeles Angels': 13,
'Minnesota Twins': 17,
'New York Yankees': 19,
'Oakland Athletics': 20,
'Athletics': 20, # Alias for post-Oakland move
'Seattle Mariners': 24,
'Tampa Bay Rays': 27,
'Texas Rangers': 28,
'Toronto Blue Jays': 29
"Baltimore Orioles": 3,
"Boston Red Sox": 4,
"Chicago White Sox": 6,
"Cleveland Guardians": 8,
"Detroit Tigers": 10,
"Houston Astros": 11,
"Kansas City Royals": 12,
"Los Angeles Angels": 13,
"Minnesota Twins": 17,
"New York Yankees": 19,
"Oakland Athletics": 20,
"Athletics": 20, # Alias for post-Oakland move
"Seattle Mariners": 24,
"Tampa Bay Rays": 27,
"Texas Rangers": 28,
"Toronto Blue Jays": 29,
}
NL_TEAM_IDS = {
'Arizona Diamondbacks': 1,
'Atlanta Braves': 2,
'Chicago Cubs': 5,
'Cincinnati Reds': 7,
'Colorado Rockies': 9,
'Los Angeles Dodgers': 14,
'Miami Marlins': 15,
'Milwaukee Brewers': 16,
'New York Mets': 18,
'Philadelphia Phillies': 21,
'Pittsburgh Pirates': 22,
'San Diego Padres': 23,
'San Francisco Giants': 25,
'St Louis Cardinals': 26, # Note: constants has 'St Louis Cardinals' not 'St. Louis Cardinals'
'Washington Nationals': 30
"Arizona Diamondbacks": 1,
"Atlanta Braves": 2,
"Chicago Cubs": 5,
"Cincinnati Reds": 7,
"Colorado Rockies": 9,
"Los Angeles Dodgers": 14,
"Miami Marlins": 15,
"Milwaukee Brewers": 16,
"New York Mets": 18,
"Philadelphia Phillies": 21,
"Pittsburgh Pirates": 22,
"San Diego Padres": 23,
"San Francisco Giants": 25,
"St Louis Cardinals": 26, # Note: constants has 'St Louis Cardinals' not 'St. Louis Cardinals'
"Washington Nationals": 30,
}
# Get AL teams from constants
AL_TEAMS = [team for team in ALL_MLB_TEAMS.keys() if team in AL_TEAM_IDS]
NL_TEAMS = [team for team in ALL_MLB_TEAMS.keys() if team in NL_TEAM_IDS or team == 'St Louis Cardinals']
NL_TEAMS = [
team
for team in ALL_MLB_TEAMS.keys()
if team in NL_TEAM_IDS or team == "St Louis Cardinals"
]
# Cardset mappings
CARDSET_LABELS_TO_IDS = {
'2022 Season': 3,
'2022 Promos': 4,
'2021 Season': 1,
'2019 Season': 5,
'2013 Season': 6,
'2012 Season': 7,
'Mario Super Sluggers': 8,
'2023 Season': 9,
'2016 Season': 11,
'2008 Season': 12,
'2018 Season': 13,
'2024 Season': 17,
'2024 Promos': 18,
'1998 Season': 20,
'2025 Season': 24,
'2005 Live': 27,
'Pokemon - Brilliant Stars': 23
"2022 Season": 3,
"2022 Promos": 4,
"2021 Season": 1,
"2019 Season": 5,
"2013 Season": 6,
"2012 Season": 7,
"Mario Super Sluggers": 8,
"2023 Season": 9,
"2016 Season": 11,
"2008 Season": 12,
"2018 Season": 13,
"2024 Season": 17,
"2024 Promos": 18,
"1998 Season": 20,
"2025 Season": 24,
"2005 Live": 27,
"Pokemon - Brilliant Stars": 23,
}
def _get_team_id(team_name: str, league: Literal['AL', 'NL']) -> int:
def _get_team_id(team_name: str, league: Literal["AL", "NL"]) -> int:
"""Get team ID from team name and league."""
if league == 'AL':
if league == "AL":
return AL_TEAM_IDS.get(team_name)
else:
# Handle the St. Louis Cardinals special case
if team_name == 'St. Louis Cardinals':
return NL_TEAM_IDS.get('St Louis Cardinals')
if team_name == "St. Louis Cardinals":
return NL_TEAM_IDS.get("St Louis Cardinals")
return NL_TEAM_IDS.get(team_name)
class SelectChoicePackTeam(discord.ui.Select):
def __init__(self, which: Literal['AL', 'NL'], team, cardset_id: Optional[int] = None):
def __init__(
self, which: Literal["AL", "NL"], team, cardset_id: Optional[int] = None
):
self.which = which
self.owner_team = team
self.cardset_id = cardset_id
if which == 'AL':
if which == "AL":
options = [discord.SelectOption(label=team) for team in AL_TEAMS]
else:
# Handle St. Louis Cardinals display name
options = [discord.SelectOption(label='St. Louis Cardinals' if team == 'St Louis Cardinals' else team)
for team in NL_TEAMS]
super().__init__(placeholder=f'Select an {which} team', options=options)
options = [
discord.SelectOption(
label=(
"St. Louis Cardinals" if team == "St Louis Cardinals" else team
)
)
for team in NL_TEAMS
]
super().__init__(placeholder=f"Select an {which} team", options=options)
async def callback(self, interaction: discord.Interaction):
# Import here to avoid circular imports
from api_calls import db_get, db_patch
from helpers import open_choice_pack
team_id = _get_team_id(self.values[0], self.which)
if team_id is None:
raise ValueError(f'Unknown team: {self.values[0]}')
raise ValueError(f"Unknown team: {self.values[0]}")
await interaction.response.edit_message(content=f'You selected the **{self.values[0]}**', view=None)
await interaction.response.edit_message(
content=f"You selected the **{self.values[0]}**", view=None
)
# Get the selected packs
params = [
('pack_type_id', 8), ('team_id', self.owner_team['id']), ('opened', False), ('limit', 1),
('exact_match', True)
("pack_type_id", 8),
("team_id", self.owner_team["id"]),
("opened", False),
("limit", 1),
("exact_match", True),
]
if self.cardset_id is not None:
params.append(('pack_cardset_id', self.cardset_id))
p_query = await db_get('packs', params=params)
if p_query['count'] == 0:
logger.error(f'open-packs - no packs found with params: {params}')
raise ValueError(f'Unable to open packs')
params.append(("pack_cardset_id", self.cardset_id))
p_query = await db_get("packs", params=params)
if p_query["count"] == 0:
logger.error(f"open-packs - no packs found with params: {params}")
raise ValueError("Unable to open packs")
this_pack = await db_patch('packs', object_id=p_query['packs'][0]['id'], params=[('pack_team_id', team_id)])
this_pack = await db_patch(
"packs",
object_id=p_query["packs"][0]["id"],
params=[("pack_team_id", team_id)],
)
await open_choice_pack(this_pack, self.owner_team, interaction, self.cardset_id)
@ -130,104 +152,116 @@ class SelectChoicePackTeam(discord.ui.Select):
class SelectOpenPack(discord.ui.Select):
def __init__(self, options: list, team: dict):
self.owner_team = team
super().__init__(placeholder='Select a Pack Type', options=options)
super().__init__(placeholder="Select a Pack Type", options=options)
async def callback(self, interaction: discord.Interaction):
# Import here to avoid circular imports
from api_calls import db_get
from helpers import open_st_pr_packs, open_choice_pack
logger.info(f'SelectPackChoice - selection: {self.values[0]}')
pack_vals = self.values[0].split('-')
logger.info(f'pack_vals: {pack_vals}')
logger.info(f"SelectPackChoice - selection: {self.values[0]}")
pack_vals = self.values[0].split("-")
logger.info(f"pack_vals: {pack_vals}")
# Get the selected packs
params = [('team_id', self.owner_team['id']), ('opened', False), ('limit', 5), ('exact_match', True)]
params = [
("team_id", self.owner_team["id"]),
("opened", False),
("limit", 20),
("exact_match", True),
]
open_type = 'standard'
if 'Standard' in pack_vals:
open_type = 'standard'
params.append(('pack_type_id', 1))
elif 'Premium' in pack_vals:
open_type = 'standard'
params.append(('pack_type_id', 3))
elif 'Daily' in pack_vals:
params.append(('pack_type_id', 4))
elif 'Promo Choice' in pack_vals:
open_type = 'choice'
params.append(('pack_type_id', 9))
elif 'MVP' in pack_vals:
open_type = 'choice'
params.append(('pack_type_id', 5))
elif 'All Star' in pack_vals:
open_type = 'choice'
params.append(('pack_type_id', 6))
elif 'Mario' in pack_vals:
open_type = 'choice'
params.append(('pack_type_id', 7))
elif 'Team Choice' in pack_vals:
open_type = 'choice'
params.append(('pack_type_id', 8))
open_type = "standard"
if "Standard" in pack_vals:
open_type = "standard"
params.append(("pack_type_id", 1))
elif "Premium" in pack_vals:
open_type = "standard"
params.append(("pack_type_id", 3))
elif "Daily" in pack_vals:
params.append(("pack_type_id", 4))
elif "Promo Choice" in pack_vals:
open_type = "choice"
params.append(("pack_type_id", 9))
elif "MVP" in pack_vals:
open_type = "choice"
params.append(("pack_type_id", 5))
elif "All Star" in pack_vals:
open_type = "choice"
params.append(("pack_type_id", 6))
elif "Mario" in pack_vals:
open_type = "choice"
params.append(("pack_type_id", 7))
elif "Team Choice" in pack_vals:
open_type = "choice"
params.append(("pack_type_id", 8))
else:
raise KeyError(f'Cannot identify pack details: {pack_vals}')
raise KeyError(f"Cannot identify pack details: {pack_vals}")
# If team isn't already set on team choice pack, make team pack selection now
await interaction.response.edit_message(view=None)
cardset_id = None
# Handle Team Choice packs with no team/cardset assigned
if 'Team Choice' in pack_vals and 'Team' not in pack_vals and 'Cardset' not in pack_vals:
if (
"Team Choice" in pack_vals
and "Team" not in pack_vals
and "Cardset" not in pack_vals
):
await interaction.followup.send(
content='This Team Choice pack needs to be assigned a team and cardset. '
'Please contact an admin to configure this pack.',
ephemeral=True
content="This Team Choice pack needs to be assigned a team and cardset. "
"Please contact an admin to configure this pack.",
ephemeral=True,
)
return
elif 'Team Choice' in pack_vals and 'Cardset' in pack_vals:
elif "Team Choice" in pack_vals and "Cardset" in pack_vals:
# cardset_id = pack_vals[2]
cardset_index = pack_vals.index('Cardset')
cardset_index = pack_vals.index("Cardset")
cardset_id = pack_vals[cardset_index + 1]
params.append(('pack_cardset_id', cardset_id))
if 'Team' not in pack_vals:
params.append(("pack_cardset_id", cardset_id))
if "Team" not in pack_vals:
view = SelectView(
[SelectChoicePackTeam('AL', self.owner_team, cardset_id),
SelectChoicePackTeam('NL', self.owner_team, cardset_id)],
timeout=30
[
SelectChoicePackTeam("AL", self.owner_team, cardset_id),
SelectChoicePackTeam("NL", self.owner_team, cardset_id),
],
timeout=30,
)
await interaction.followup.send(
content='Please select a team for your Team Choice pack:',
view=view
content="Please select a team for your Team Choice pack:", view=view
)
return
params.append(('pack_team_id', pack_vals[pack_vals.index('Team') + 1]))
else:
if 'Team' in pack_vals:
params.append(('pack_team_id', pack_vals[pack_vals.index('Team') + 1]))
if 'Cardset' in pack_vals:
cardset_id = pack_vals[pack_vals.index('Cardset') + 1]
params.append(('pack_cardset_id', cardset_id))
p_query = await db_get('packs', params=params)
if p_query['count'] == 0:
logger.error(f'open-packs - no packs found with params: {params}')
params.append(("pack_team_id", pack_vals[pack_vals.index("Team") + 1]))
else:
if "Team" in pack_vals:
params.append(("pack_team_id", pack_vals[pack_vals.index("Team") + 1]))
if "Cardset" in pack_vals:
cardset_id = pack_vals[pack_vals.index("Cardset") + 1]
params.append(("pack_cardset_id", cardset_id))
p_query = await db_get("packs", params=params)
if p_query["count"] == 0:
logger.error(f"open-packs - no packs found with params: {params}")
await interaction.followup.send(
content='Unable to find the selected pack. Please contact an admin.',
ephemeral=True
content="Unable to find the selected pack. Please contact an admin.",
ephemeral=True,
)
return
# Open the packs
try:
if open_type == 'standard':
await open_st_pr_packs(p_query['packs'], self.owner_team, interaction)
elif open_type == 'choice':
await open_choice_pack(p_query['packs'][0], self.owner_team, interaction, cardset_id)
if open_type == "standard":
await open_st_pr_packs(p_query["packs"], self.owner_team, interaction)
elif open_type == "choice":
await open_choice_pack(
p_query["packs"][0], self.owner_team, interaction, cardset_id
)
except Exception as e:
logger.error(f'Failed to open pack: {e}')
logger.error(f"Failed to open pack: {e}")
await interaction.followup.send(
content=f'Failed to open pack. Please contact an admin. Error: {str(e)}',
ephemeral=True
content=f"Failed to open pack. Please contact an admin. Error: {str(e)}",
ephemeral=True,
)
return
@ -235,275 +269,317 @@ class SelectOpenPack(discord.ui.Select):
class SelectPaperdexCardset(discord.ui.Select):
def __init__(self):
options = [
discord.SelectOption(label='2005 Live'),
discord.SelectOption(label='2025 Season'),
discord.SelectOption(label='1998 Season'),
discord.SelectOption(label='2024 Season'),
discord.SelectOption(label='2023 Season'),
discord.SelectOption(label='2022 Season'),
discord.SelectOption(label='2022 Promos'),
discord.SelectOption(label='2021 Season'),
discord.SelectOption(label='2019 Season'),
discord.SelectOption(label='2018 Season'),
discord.SelectOption(label='2016 Season'),
discord.SelectOption(label='2013 Season'),
discord.SelectOption(label='2012 Season'),
discord.SelectOption(label='2008 Season'),
discord.SelectOption(label='Mario Super Sluggers')
discord.SelectOption(label="2005 Live"),
discord.SelectOption(label="2025 Season"),
discord.SelectOption(label="1998 Season"),
discord.SelectOption(label="2024 Season"),
discord.SelectOption(label="2023 Season"),
discord.SelectOption(label="2022 Season"),
discord.SelectOption(label="2022 Promos"),
discord.SelectOption(label="2021 Season"),
discord.SelectOption(label="2019 Season"),
discord.SelectOption(label="2018 Season"),
discord.SelectOption(label="2016 Season"),
discord.SelectOption(label="2013 Season"),
discord.SelectOption(label="2012 Season"),
discord.SelectOption(label="2008 Season"),
discord.SelectOption(label="Mario Super Sluggers"),
]
super().__init__(placeholder='Select a Cardset', options=options)
super().__init__(placeholder="Select a Cardset", options=options)
async def callback(self, interaction: discord.Interaction):
# Import here to avoid circular imports
from api_calls import db_get
from helpers import get_team_by_owner, paperdex_cardset_embed, embed_pagination
logger.info(f'SelectPaperdexCardset - selection: {self.values[0]}')
logger.info(f"SelectPaperdexCardset - selection: {self.values[0]}")
cardset_id = CARDSET_LABELS_TO_IDS.get(self.values[0])
if cardset_id is None:
raise ValueError(f'Unknown cardset: {self.values[0]}')
raise ValueError(f"Unknown cardset: {self.values[0]}")
c_query = await db_get('cardsets', object_id=cardset_id, none_okay=False)
await interaction.response.edit_message(content=f'Okay, sifting through your cards...', view=None)
c_query = await db_get("cardsets", object_id=cardset_id, none_okay=False)
await interaction.response.edit_message(
content="Okay, sifting through your cards...", view=None
)
cardset_embeds = await paperdex_cardset_embed(
team=await get_team_by_owner(interaction.user.id),
this_cardset=c_query
team=await get_team_by_owner(interaction.user.id), this_cardset=c_query
)
await embed_pagination(cardset_embeds, interaction.channel, interaction.user)
class SelectPaperdexTeam(discord.ui.Select):
def __init__(self, which: Literal['AL', 'NL']):
def __init__(self, which: Literal["AL", "NL"]):
self.which = which
if which == 'AL':
if which == "AL":
options = [discord.SelectOption(label=team) for team in AL_TEAMS]
else:
# Handle St. Louis Cardinals display name
options = [discord.SelectOption(label='St. Louis Cardinals' if team == 'St Louis Cardinals' else team)
for team in NL_TEAMS]
super().__init__(placeholder=f'Select an {which} team', options=options)
options = [
discord.SelectOption(
label=(
"St. Louis Cardinals" if team == "St Louis Cardinals" else team
)
)
for team in NL_TEAMS
]
super().__init__(placeholder=f"Select an {which} team", options=options)
async def callback(self, interaction: discord.Interaction):
# Import here to avoid circular imports
from api_calls import db_get
from helpers import get_team_by_owner, paperdex_team_embed, embed_pagination
team_id = _get_team_id(self.values[0], self.which)
if team_id is None:
raise ValueError(f'Unknown team: {self.values[0]}')
raise ValueError(f"Unknown team: {self.values[0]}")
t_query = await db_get('teams', object_id=team_id, none_okay=False)
await interaction.response.edit_message(content=f'Okay, sifting through your cards...', view=None)
t_query = await db_get("teams", object_id=team_id, none_okay=False)
await interaction.response.edit_message(
content="Okay, sifting through your cards...", view=None
)
team_embeds = await paperdex_team_embed(team=await get_team_by_owner(interaction.user.id), mlb_team=t_query)
team_embeds = await paperdex_team_embed(
team=await get_team_by_owner(interaction.user.id), mlb_team=t_query
)
await embed_pagination(team_embeds, interaction.channel, interaction.user)
class SelectBuyPacksCardset(discord.ui.Select):
def __init__(self, team: dict, quantity: int, pack_type_id: int, pack_embed: discord.Embed, cost: int):
def __init__(
self,
team: dict,
quantity: int,
pack_type_id: int,
pack_embed: discord.Embed,
cost: int,
):
options = [
discord.SelectOption(label='2005 Live'),
discord.SelectOption(label='2025 Season'),
discord.SelectOption(label='1998 Season'),
discord.SelectOption(label='Pokemon - Brilliant Stars'),
discord.SelectOption(label='2024 Season'),
discord.SelectOption(label='2023 Season'),
discord.SelectOption(label='2022 Season'),
discord.SelectOption(label='2021 Season'),
discord.SelectOption(label='2019 Season'),
discord.SelectOption(label='2018 Season'),
discord.SelectOption(label='2016 Season'),
discord.SelectOption(label='2013 Season'),
discord.SelectOption(label='2012 Season'),
discord.SelectOption(label='2008 Season')
discord.SelectOption(label="2005 Live"),
discord.SelectOption(label="2025 Season"),
discord.SelectOption(label="1998 Season"),
discord.SelectOption(label="Pokemon - Brilliant Stars"),
discord.SelectOption(label="2024 Season"),
discord.SelectOption(label="2023 Season"),
discord.SelectOption(label="2022 Season"),
discord.SelectOption(label="2021 Season"),
discord.SelectOption(label="2019 Season"),
discord.SelectOption(label="2018 Season"),
discord.SelectOption(label="2016 Season"),
discord.SelectOption(label="2013 Season"),
discord.SelectOption(label="2012 Season"),
discord.SelectOption(label="2008 Season"),
]
self.team = team
self.quantity = quantity
self.pack_type_id = pack_type_id
self.pack_embed = pack_embed
self.cost = cost
super().__init__(placeholder='Select a Cardset', options=options)
super().__init__(placeholder="Select a Cardset", options=options)
async def callback(self, interaction: discord.Interaction):
# Import here to avoid circular imports
from api_calls import db_post
from discord_ui.confirmations import Confirm
logger.info(f'SelectBuyPacksCardset - selection: {self.values[0]}')
logger.info(f"SelectBuyPacksCardset - selection: {self.values[0]}")
cardset_id = CARDSET_LABELS_TO_IDS.get(self.values[0])
if cardset_id is None:
raise ValueError(f'Unknown cardset: {self.values[0]}')
if self.values[0] == 'Pokemon - Brilliant Stars':
self.pack_embed.set_image(url=IMAGES['pack-pkmnbs'])
raise ValueError(f"Unknown cardset: {self.values[0]}")
self.pack_embed.description = f'{self.pack_embed.description} - {self.values[0]}'
if self.values[0] == "Pokemon - Brilliant Stars":
self.pack_embed.set_image(url=IMAGES["pack-pkmnbs"])
self.pack_embed.description = (
f"{self.pack_embed.description} - {self.values[0]}"
)
view = Confirm(responders=[interaction.user], timeout=30)
await interaction.response.edit_message(
content=None,
embed=self.pack_embed,
view=None
content=None, embed=self.pack_embed, view=None
)
question = await interaction.channel.send(
content=f'Your Wallet: {self.team["wallet"]}₼\n'
f'Pack{"s" if self.quantity > 1 else ""} Price: {self.cost}₼\n'
f'After Purchase: {self.team["wallet"] - self.cost}₼\n\n'
f'Would you like to make this purchase?',
view=view
content=f"Your Wallet: {self.team['wallet']}₼\n"
f"Pack{'s' if self.quantity > 1 else ''} Price: {self.cost}₼\n"
f"After Purchase: {self.team['wallet'] - self.cost}₼\n\n"
f"Would you like to make this purchase?",
view=view,
)
await view.wait()
if not view.value:
await question.edit(
content='Saving that money. Smart.',
view=None
)
await question.edit(content="Saving that money. Smart.", view=None)
return
p_model = {
'team_id': self.team['id'],
'pack_type_id': self.pack_type_id,
'pack_cardset_id': cardset_id
"team_id": self.team["id"],
"pack_type_id": self.pack_type_id,
"pack_cardset_id": cardset_id,
}
await db_post('packs', payload={'packs': [p_model for x in range(self.quantity)]})
await db_post(f'teams/{self.team["id"]}/money/-{self.cost}')
await db_post(
"packs", payload={"packs": [p_model for x in range(self.quantity)]}
)
await db_post(f"teams/{self.team['id']}/money/-{self.cost}")
await question.edit(
content=f'{"They are" if self.quantity > 1 else "It is"} all yours! Go rip \'em with `/open-packs`',
view=None
content=f"{'They are' if self.quantity > 1 else 'It is'} all yours! Go rip 'em with `/open-packs`",
view=None,
)
class SelectBuyPacksTeam(discord.ui.Select):
def __init__(
self, which: Literal['AL', 'NL'], team: dict, quantity: int, pack_type_id: int, pack_embed: discord.Embed,
cost: int):
self,
which: Literal["AL", "NL"],
team: dict,
quantity: int,
pack_type_id: int,
pack_embed: discord.Embed,
cost: int,
):
self.which = which
self.team = team
self.quantity = quantity
self.pack_type_id = pack_type_id
self.pack_embed = pack_embed
self.cost = cost
if which == 'AL':
if which == "AL":
options = [discord.SelectOption(label=team) for team in AL_TEAMS]
else:
# Handle St. Louis Cardinals display name
options = [discord.SelectOption(label='St. Louis Cardinals' if team == 'St Louis Cardinals' else team)
for team in NL_TEAMS]
super().__init__(placeholder=f'Select an {which} team', options=options)
options = [
discord.SelectOption(
label=(
"St. Louis Cardinals" if team == "St Louis Cardinals" else team
)
)
for team in NL_TEAMS
]
super().__init__(placeholder=f"Select an {which} team", options=options)
async def callback(self, interaction: discord.Interaction):
# Import here to avoid circular imports
from api_calls import db_post
from discord_ui.confirmations import Confirm
team_id = _get_team_id(self.values[0], self.which)
if team_id is None:
raise ValueError(f'Unknown team: {self.values[0]}')
raise ValueError(f"Unknown team: {self.values[0]}")
self.pack_embed.description = f'{self.pack_embed.description} - {self.values[0]}'
self.pack_embed.description = (
f"{self.pack_embed.description} - {self.values[0]}"
)
view = Confirm(responders=[interaction.user], timeout=30)
await interaction.response.edit_message(
content=None,
embed=self.pack_embed,
view=None
content=None, embed=self.pack_embed, view=None
)
question = await interaction.channel.send(
content=f'Your Wallet: {self.team["wallet"]}₼\n'
f'Pack{"s" if self.quantity > 1 else ""} Price: {self.cost}₼\n'
f'After Purchase: {self.team["wallet"] - self.cost}₼\n\n'
f'Would you like to make this purchase?',
view=view
content=f"Your Wallet: {self.team['wallet']}₼\n"
f"Pack{'s' if self.quantity > 1 else ''} Price: {self.cost}₼\n"
f"After Purchase: {self.team['wallet'] - self.cost}₼\n\n"
f"Would you like to make this purchase?",
view=view,
)
await view.wait()
if not view.value:
await question.edit(
content='Saving that money. Smart.',
view=None
)
await question.edit(content="Saving that money. Smart.", view=None)
return
p_model = {
'team_id': self.team['id'],
'pack_type_id': self.pack_type_id,
'pack_team_id': team_id
"team_id": self.team["id"],
"pack_type_id": self.pack_type_id,
"pack_team_id": team_id,
}
await db_post('packs', payload={'packs': [p_model for x in range(self.quantity)]})
await db_post(f'teams/{self.team["id"]}/money/-{self.cost}')
await db_post(
"packs", payload={"packs": [p_model for x in range(self.quantity)]}
)
await db_post(f"teams/{self.team['id']}/money/-{self.cost}")
await question.edit(
content=f'{"They are" if self.quantity > 1 else "It is"} all yours! Go rip \'em with `/open-packs`',
view=None
content=f"{'They are' if self.quantity > 1 else 'It is'} all yours! Go rip 'em with `/open-packs`",
view=None,
)
class SelectUpdatePlayerTeam(discord.ui.Select):
def __init__(self, which: Literal['AL', 'NL'], player: dict, reporting_team: dict, bot):
def __init__(
self, which: Literal["AL", "NL"], player: dict, reporting_team: dict, bot
):
self.bot = bot
self.which = which
self.player = player
self.reporting_team = reporting_team
if which == 'AL':
if which == "AL":
options = [discord.SelectOption(label=team) for team in AL_TEAMS]
else:
# Handle St. Louis Cardinals display name
options = [discord.SelectOption(label='St. Louis Cardinals' if team == 'St Louis Cardinals' else team)
for team in NL_TEAMS]
super().__init__(placeholder=f'Select an {which} team', options=options)
options = [
discord.SelectOption(
label=(
"St. Louis Cardinals" if team == "St Louis Cardinals" else team
)
)
for team in NL_TEAMS
]
super().__init__(placeholder=f"Select an {which} team", options=options)
async def callback(self, interaction: discord.Interaction):
# Import here to avoid circular imports
from api_calls import db_patch, db_post
from discord_ui.confirmations import Confirm
from helpers import player_desc, send_to_channel
# Check if already assigned - compare against both normalized franchise and full mlbclub
normalized_selection = normalize_franchise(self.values[0])
if normalized_selection == self.player['franchise'] or self.values[0] == self.player['mlbclub']:
if (
normalized_selection == self.player["franchise"]
or self.values[0] == self.player["mlbclub"]
):
await interaction.response.send_message(
content=f'Thank you for the help, but it looks like somebody beat you to it! '
f'**{player_desc(self.player)}** is already assigned to the **{self.player["mlbclub"]}**.'
content=f"Thank you for the help, but it looks like somebody beat you to it! "
f"**{player_desc(self.player)}** is already assigned to the **{self.player['mlbclub']}**."
)
return
view = Confirm(responders=[interaction.user], timeout=15)
await interaction.response.edit_message(
content=f'Should I update **{player_desc(self.player)}**\'s team to the **{self.values[0]}**?',
view=None
)
question = await interaction.channel.send(
content=None,
view=view
content=f"Should I update **{player_desc(self.player)}**'s team to the **{self.values[0]}**?",
view=None,
)
question = await interaction.channel.send(content=None, view=view)
await view.wait()
if not view.value:
await question.edit(
content='That didnt\'t sound right to me, either. Let\'s not touch that.',
view=None
content="That didnt't sound right to me, either. Let's not touch that.",
view=None,
)
return
else:
await question.delete()
await db_patch('players', object_id=self.player['player_id'], params=[
('mlbclub', self.values[0]), ('franchise', normalize_franchise(self.values[0]))
])
await db_post(f'teams/{self.reporting_team["id"]}/money/25')
await send_to_channel(
self.bot, 'pd-news-ticker',
content=f'{interaction.user.name} just updated **{player_desc(self.player)}**\'s team to the '
f'**{self.values[0]}**'
await db_patch(
"players",
object_id=self.player["player_id"],
params=[
("mlbclub", self.values[0]),
("franchise", normalize_franchise(self.values[0])),
],
)
await interaction.channel.send(f'All done!')
await db_post(f"teams/{self.reporting_team['id']}/money/25")
await send_to_channel(
self.bot,
"pd-news-ticker",
content=f"{interaction.user.name} just updated **{player_desc(self.player)}**'s team to the "
f"**{self.values[0]}**",
)
await interaction.channel.send("All done!")
class SelectView(discord.ui.View):
@ -511,4 +587,4 @@ class SelectView(discord.ui.View):
super().__init__(timeout=timeout)
for x in select_objects:
self.add_item(x)
self.add_item(x)

391
helpers/main.py
View File

@ -2,35 +2,23 @@ import asyncio
import datetime
import logging
import math
import os
import random
import traceback
import discord
import pygsheets
import aiohttp
from discord.ext import commands
from api_calls import *
from bs4 import BeautifulSoup
from difflib import get_close_matches
from dataclasses import dataclass
from typing import Optional, Literal, Union, List
from typing import Optional, Union, List
from exceptions import log_exception
from in_game.gameplay_models import Team
from constants import *
from discord_ui import *
from random_content import *
from utils import (
position_name_to_abbrev,
user_has_role,
get_roster_sheet_legacy,
get_roster_sheet,
get_player_url,
owner_only,
get_cal_user,
get_context_user,
)
from search_utils import *
from discord_utils import *
@ -182,7 +170,7 @@ async def get_card_embeds(card, include_stats=False) -> list:
]
if any(bool_list):
if count == 1:
coll_string = f"Only you"
coll_string = "Only you"
else:
coll_string = (
f"You and {count - 1} other{'s' if count - 1 != 1 else ''}"
@ -190,7 +178,7 @@ async def get_card_embeds(card, include_stats=False) -> list:
elif count:
coll_string = f"{count} other team{'s' if count != 1 else ''}"
else:
coll_string = f"0 teams"
coll_string = "0 teams"
embed.add_field(name="Collected By", value=coll_string)
else:
embed.add_field(
@ -229,7 +217,7 @@ async def get_card_embeds(card, include_stats=False) -> list:
)
if evo_mon is not None:
embed.add_field(name="Evolves Into", value=f"{evo_mon['p_name']}")
except Exception as e:
except Exception:
logging.error(
"could not pull evolution: {e}", exc_info=True, stack_info=True
)
@ -240,7 +228,7 @@ async def get_card_embeds(card, include_stats=False) -> list:
)
if evo_mon is not None:
embed.add_field(name="Evolves From", value=f"{evo_mon['p_name']}")
except Exception as e:
except Exception:
logging.error(
"could not pull evolution: {e}", exc_info=True, stack_info=True
)
@ -342,7 +330,7 @@ async def display_cards(
)
try:
cards.sort(key=lambda x: x["player"]["rarity"]["value"])
logger.debug(f"Cards sorted successfully")
logger.debug("Cards sorted successfully")
card_embeds = [await get_card_embeds(x) for x in cards]
logger.debug(f"Created {len(card_embeds)} card embeds")
@ -363,15 +351,15 @@ async def display_cards(
r_emoji = "→"
view.left_button.disabled = True
view.left_button.label = f"{l_emoji}Prev: -/{len(card_embeds)}"
view.cancel_button.label = f"Close Pack"
view.cancel_button.label = "Close Pack"
view.right_button.label = f"Next: {page_num + 2}/{len(card_embeds)}{r_emoji}"
if len(cards) == 1:
view.right_button.disabled = True
logger.debug(f"Pagination view created successfully")
logger.debug("Pagination view created successfully")
if pack_cover:
logger.debug(f"Sending pack cover message")
logger.debug("Sending pack cover message")
msg = await channel.send(
content=None,
embed=image_embed(pack_cover, title=f"{team['lname']}", desc=pack_name),
@ -383,7 +371,7 @@ async def display_cards(
content=None, embeds=card_embeds[page_num], view=view
)
logger.debug(f"Initial message sent successfully")
logger.debug("Initial message sent successfully")
except Exception as e:
logger.error(
f"Error creating view or sending initial message: {e}", exc_info=True
@ -400,12 +388,12 @@ async def display_cards(
f"{user.mention} you've got {len(cards)} cards here"
)
logger.debug(f"Follow-up message sent successfully")
logger.debug("Follow-up message sent successfully")
except Exception as e:
logger.error(f"Error sending follow-up message: {e}", exc_info=True)
return False
logger.debug(f"Starting main interaction loop")
logger.debug("Starting main interaction loop")
while True:
try:
logger.debug(f"Waiting for user interaction on page {page_num}")
@ -471,7 +459,7 @@ async def display_cards(
),
view=view,
)
logger.debug(f"MVP display updated successfully")
logger.debug("MVP display updated successfully")
except Exception as e:
logger.error(
f"Error processing shiny card on page {page_num}: {e}", exc_info=True
@ -479,19 +467,19 @@ async def display_cards(
# Continue with regular flow instead of crashing
try:
tmp_msg = await channel.send(
content=f"<@&1163537676885033010> we've got an MVP!"
content="<@&1163537676885033010> we've got an MVP!"
)
await follow_up.edit(
content=f"<@&1163537676885033010> we've got an MVP!"
content="<@&1163537676885033010> we've got an MVP!"
)
await tmp_msg.delete()
except discord.errors.NotFound:
# Role might not exist or message was already deleted
await follow_up.edit(content=f"We've got an MVP!")
await follow_up.edit(content="We've got an MVP!")
except Exception as e:
# Log error but don't crash the function
logger.error(f"Error handling MVP notification: {e}")
await follow_up.edit(content=f"We've got an MVP!")
await follow_up.edit(content="We've got an MVP!")
await view.wait()
view = Pagination([user], timeout=10)
@ -499,7 +487,7 @@ async def display_cards(
view.right_button.label = (
f"Next: {page_num + 2}/{len(card_embeds)}{r_emoji}"
)
view.cancel_button.label = f"Close Pack"
view.cancel_button.label = "Close Pack"
view.left_button.label = f"{l_emoji}Prev: {page_num}/{len(card_embeds)}"
if page_num == 0:
view.left_button.label = f"{l_emoji}Prev: -/{len(card_embeds)}"
@ -547,7 +535,7 @@ async def embed_pagination(
l_emoji = ""
r_emoji = ""
view.right_button.label = f"Next: {page_num + 2}/{len(all_embeds)}{r_emoji}"
view.cancel_button.label = f"Cancel"
view.cancel_button.label = "Cancel"
view.left_button.label = f"{l_emoji}Prev: {page_num}/{len(all_embeds)}"
if page_num == 0:
view.left_button.label = f"{l_emoji}Prev: -/{len(all_embeds)}"
@ -582,7 +570,7 @@ async def embed_pagination(
view = Pagination([user], timeout=timeout)
view.right_button.label = f"Next: {page_num + 2}/{len(all_embeds)}{r_emoji}"
view.cancel_button.label = f"Cancel"
view.cancel_button.label = "Cancel"
view.left_button.label = f"{l_emoji}Prev: {page_num}/{len(all_embeds)}"
if page_num == 0:
view.left_button.label = f"{l_emoji}Prev: -/{len(all_embeds)}"
@ -661,21 +649,15 @@ async def get_test_pack(ctx, team):
async def roll_for_cards(all_packs: list, extra_val=None) -> list:
"""Open packs by rolling dice, fetching random players, and creating cards.
Parallelizes DB calls: one fetch per rarity tier across all packs,
then gathers all card creates and pack patches concurrently.
"""
Pack odds are calculated based on the pack type
Parameters
----------
extra_val
all_packs
Returns
-------
"""
all_players = []
team = all_packs[0]["team"]
pack_ids = []
# --- Phase A: Roll dice for every pack (CPU-only, no I/O) ---
pack_counts = []
for pack in all_packs:
counts = {
"Rep": {"count": 0, "rarity": 0},
@ -685,10 +667,9 @@ async def roll_for_cards(all_packs: list, extra_val=None) -> list:
"MVP": {"count": 0, "rarity": 5},
"HoF": {"count": 0, "rarity": 8},
}
this_pack_players = []
if pack["pack_type"]["name"] == "Standard":
# Cards 1 - 2
for x in range(2):
for _ in range(2):
d_1000 = random.randint(1, 1000)
if d_1000 <= 450:
counts["Rep"]["count"] += 1
@ -808,7 +789,6 @@ async def roll_for_cards(all_packs: list, extra_val=None) -> list:
logger.info(
f"Building Check-In Pack // extra_val (type): {extra_val} {type(extra_val)}"
)
# Single Card
mod = 0
if isinstance(extra_val, int):
mod = extra_val
@ -826,106 +806,195 @@ async def roll_for_cards(all_packs: list, extra_val=None) -> list:
else:
raise TypeError(f"Pack type not recognized: {pack['pack_type']['name']}")
pull_notifs = []
for key in counts:
mvp_flag = None
pack_counts.append(counts)
if counts[key]["count"] > 0:
params = [
("min_rarity", counts[key]["rarity"]),
("max_rarity", counts[key]["rarity"]),
("limit", counts[key]["count"]),
]
if all_packs[0]["pack_team"] is not None:
params.extend(
[
("franchise", all_packs[0]["pack_team"]["sname"]),
("in_packs", True),
]
)
elif all_packs[0]["pack_cardset"] is not None:
params.append(("cardset_id", all_packs[0]["pack_cardset"]["id"]))
else:
params.append(("in_packs", True))
# --- Phase B: Fetch players — one call per rarity tier, all gathered ---
# Sum counts across all packs per rarity tier
rarity_keys = ["Rep", "Res", "Sta", "All", "MVP", "HoF"]
summed = {key: 0 for key in rarity_keys}
for counts in pack_counts:
for key in rarity_keys:
summed[key] += counts[key]["count"]
pl = await db_get("players/random", params=params)
if pl["count"] != counts[key]["count"]:
mvp_flag = counts[key]["count"] - pl["count"]
logging.info(
f"Set mvp flag to {mvp_flag} / cardset_id: {all_packs[0]['pack_cardset']['id']}"
)
for x in pl["players"]:
this_pack_players.append(x)
all_players.append(x)
if x["rarity"]["value"] >= 3:
pull_notifs.append(x)
if mvp_flag and all_packs[0]["pack_cardset"]["id"] not in [23]:
logging.info(f"Adding {mvp_flag} MVPs for missing cards")
pl = await db_get(
"players/random", params=[("min_rarity", 5), ("limit", mvp_flag)]
)
for x in pl["players"]:
this_pack_players.append(x)
all_players.append(x)
# Add dupes of Replacement/Reserve cards
elif mvp_flag:
logging.info(f"Adding {mvp_flag} duplicate pokemon cards")
for count in range(mvp_flag):
logging.info(f"Adding {pl['players'][0]['p_name']} to the pack")
this_pack_players.append(x)
all_players.append(pl["players"][0])
success = await db_post(
"cards",
payload={
"cards": [
{
"player_id": x["player_id"],
"team_id": pack["team"]["id"],
"pack_id": pack["id"],
}
for x in this_pack_players
]
},
timeout=10,
# Build shared filter params
base_params = []
if all_packs[0]["pack_team"] is not None:
base_params.extend(
[
("franchise", all_packs[0]["pack_team"]["sname"]),
("in_packs", True),
]
)
if not success:
raise ConnectionError(f"Failed to create this pack of cards.")
elif all_packs[0]["pack_cardset"] is not None:
base_params.append(("cardset_id", all_packs[0]["pack_cardset"]["id"]))
else:
base_params.append(("in_packs", True))
await db_patch(
"packs",
object_id=pack["id"],
params=[
(
"open_time",
int(datetime.datetime.timestamp(datetime.datetime.now()) * 1000),
)
],
)
pack_ids.append(pack["id"])
# Fire one request per non-zero rarity tier concurrently
rarity_values = {
"Rep": 0,
"Res": 1,
"Sta": 2,
"All": 3,
"MVP": 5,
"HoF": 8,
}
fetch_keys = [key for key in rarity_keys if summed[key] > 0]
fetch_coros = []
for key in fetch_keys:
params = [
("min_rarity", rarity_values[key]),
("max_rarity", rarity_values[key]),
("limit", summed[key]),
] + base_params
fetch_coros.append(db_get("players/random", params=params))
for pull in pull_notifs:
logger.info(f"good pull: {pull}")
await db_post(
"notifs",
payload={
"created": int(
datetime.datetime.timestamp(datetime.datetime.now()) * 1000
),
"title": "Rare Pull",
"field_name": f"{player_desc(pull)} ({pull['rarity']['name']})",
"message": f"Pulled by {team['abbrev']}",
"about": f"Player-{pull['player_id']}",
},
fetch_results = await asyncio.gather(*fetch_coros)
# Map results back: rarity key -> list of players
fetched_players = {}
for key, result in zip(fetch_keys, fetch_results):
fetched_players[key] = result.get("players", [])
# Handle shortfalls — collect total MVP backfill needed
total_mvp_shortfall = 0
# Track per-tier shortfall for dupe-branch (cardset 23 exclusion)
tier_shortfalls = {}
for key in fetch_keys:
returned = len(fetched_players[key])
requested = summed[key]
if returned < requested:
shortfall = requested - returned
tier_shortfalls[key] = shortfall
total_mvp_shortfall += shortfall
logging.info(
f"Shortfall in {key}: requested {requested}, got {returned} "
f"(cardset_id: {all_packs[0]['pack_cardset']['id'] if all_packs[0]['pack_cardset'] else 'N/A'})"
)
return pack_ids
# Fetch MVP backfill or duplicate existing players
backfill_players = []
is_dupe_cardset = all_packs[0]["pack_cardset"] is not None and all_packs[0][
"pack_cardset"
]["id"] in [23]
if total_mvp_shortfall > 0 and not is_dupe_cardset:
logging.info(f"Adding {total_mvp_shortfall} MVPs for missing cards")
mvp_result = await db_get(
"players/random",
params=[("min_rarity", 5), ("limit", total_mvp_shortfall)],
)
backfill_players = mvp_result.get("players", [])
elif total_mvp_shortfall > 0 and is_dupe_cardset:
logging.info(
f"Adding {total_mvp_shortfall} duplicate cards for excluded cardset"
)
# Duplicate from first available player in the fetched results
for key in fetch_keys:
if fetched_players[key]:
for _ in range(total_mvp_shortfall):
backfill_players.append(fetched_players[key][0])
break
# Slice fetched players back into per-pack groups
# Track consumption offset per rarity tier
tier_offsets = {key: 0 for key in rarity_keys}
backfill_offset = 0
per_pack_players = []
all_pull_notifs = []
for pack_idx, counts in enumerate(pack_counts):
this_pack_players = []
pack_shortfall = 0
for key in rarity_keys:
needed = counts[key]["count"]
if needed == 0:
continue
available = fetched_players.get(key, [])
start = tier_offsets[key]
end = start + needed
got = available[start:end]
this_pack_players.extend(got)
tier_offsets[key] = end
# Track shortfall for this pack
if len(got) < needed:
pack_shortfall += needed - len(got)
# Distribute backfill players to this pack
if pack_shortfall > 0 and backfill_offset < len(backfill_players):
bf_slice = backfill_players[
backfill_offset : backfill_offset + pack_shortfall
]
this_pack_players.extend(bf_slice)
backfill_offset += len(bf_slice)
# Collect rare pull notifications
for player in this_pack_players:
if player["rarity"]["value"] >= 3:
all_pull_notifs.append(player)
per_pack_players.append(this_pack_players)
# --- Phase C: Write cards + mark packs opened, all gathered ---
open_time = int(datetime.datetime.timestamp(datetime.datetime.now()) * 1000)
write_coros = []
for pack, this_pack_players in zip(all_packs, per_pack_players):
write_coros.append(
db_post(
"cards",
payload={
"cards": [
{
"player_id": p["player_id"],
"team_id": pack["team"]["id"],
"pack_id": pack["id"],
}
for p in this_pack_players
]
},
timeout=10,
)
)
write_coros.append(
db_patch(
"packs",
object_id=pack["id"],
params=[("open_time", open_time)],
)
)
write_results = await asyncio.gather(*write_coros)
# Check card creation results (every other result starting at index 0)
for i in range(0, len(write_results), 2):
if not write_results[i]:
raise ConnectionError("Failed to create this pack of cards.")
# --- Gather notification posts ---
if all_pull_notifs:
notif_coros = []
for pull in all_pull_notifs:
logger.info(f"good pull: {pull}")
notif_coros.append(
db_post(
"notifs",
payload={
"created": int(
datetime.datetime.timestamp(datetime.datetime.now()) * 1000
),
"title": "Rare Pull",
"field_name": f"{player_desc(pull)} ({pull['rarity']['name']})",
"message": f"Pulled by {team['abbrev']}",
"about": f"Player-{pull['player_id']}",
},
)
)
await asyncio.gather(*notif_coros)
return [pack["id"] for pack in all_packs]
async def give_packs(team: dict, num_packs: int, pack_type: dict = None) -> dict:
@ -962,7 +1031,7 @@ def get_sheets(bot):
except Exception as e:
logger.error(f"Could not grab sheets auth: {e}")
raise ConnectionError(
f"Bot has not authenticated with discord; please try again in 1 minute."
"Bot has not authenticated with discord; please try again in 1 minute."
)
@ -1072,7 +1141,7 @@ def get_blank_team_card(player):
def get_rosters(team, bot, roster_num: Optional[int] = None) -> list:
sheets = get_sheets(bot)
this_sheet = sheets.open_by_key(team["gsheet"])
r_sheet = this_sheet.worksheet_by_title(f"My Rosters")
r_sheet = this_sheet.worksheet_by_title("My Rosters")
logger.debug(f"this_sheet: {this_sheet} / r_sheet = {r_sheet}")
all_rosters = [None, None, None]
@ -1153,11 +1222,11 @@ def get_roster_lineups(team, bot, roster_num, lineup_num) -> list:
try:
lineup_cells = [(row[0].value, int(row[1].value)) for row in raw_cells]
except ValueError as e:
except ValueError:
logger.error(f"Could not pull roster for {team['abbrev']} due to a ValueError")
raise ValueError(
f"Uh oh. Looks like your roster might not be saved. I am reading blanks when I try to "
f"get the card IDs"
"Uh oh. Looks like your roster might not be saved. I am reading blanks when I try to "
"get the card IDs"
)
logger.debug(f"lineup_cells: {lineup_cells}")
@ -1552,7 +1621,7 @@ def get_ratings_guide(sheets):
}
for x in p_data
]
except Exception as e:
except Exception:
return {"valid": False}
return {"valid": True, "batter_ratings": batters, "pitcher_ratings": pitchers}
@ -1764,7 +1833,7 @@ async def open_st_pr_packs(all_packs: list, team: dict, context):
pack_ids = await roll_for_cards(all_packs)
if not pack_ids:
logger.error(f"open_packs - unable to roll_for_cards for packs: {all_packs}")
raise ValueError(f"I was not able to unpack these cards")
raise ValueError("I was not able to unpack these cards")
all_cards = []
for p_id in pack_ids:
@ -1775,7 +1844,7 @@ async def open_st_pr_packs(all_packs: list, team: dict, context):
if not all_cards:
logger.error(f"open_packs - unable to get cards for packs: {pack_ids}")
raise ValueError(f"I was not able to display these cards")
raise ValueError("I was not able to display these cards")
# Present cards to opening channel
if type(context) == commands.Context:
@ -1834,7 +1903,7 @@ async def get_choice_from_cards(
view = Pagination([interaction.user], timeout=30)
view.left_button.disabled = True
view.left_button.label = f"Prev: -/{len(card_embeds)}"
view.cancel_button.label = f"Take This Card"
view.cancel_button.label = "Take This Card"
view.cancel_button.style = discord.ButtonStyle.success
view.cancel_button.disabled = True
view.right_button.label = f"Next: 1/{len(card_embeds)}"
@ -1852,7 +1921,7 @@ async def get_choice_from_cards(
view = Pagination([interaction.user], timeout=30)
view.left_button.label = f"Prev: -/{len(card_embeds)}"
view.left_button.disabled = True
view.cancel_button.label = f"Take This Card"
view.cancel_button.label = "Take This Card"
view.cancel_button.style = discord.ButtonStyle.success
view.right_button.label = f"Next: {page_num + 1}/{len(card_embeds)}"
@ -1895,7 +1964,7 @@ async def get_choice_from_cards(
view = Pagination([interaction.user], timeout=30)
view.left_button.label = f"Prev: {page_num - 1}/{len(card_embeds)}"
view.cancel_button.label = f"Take This Card"
view.cancel_button.label = "Take This Card"
view.cancel_button.style = discord.ButtonStyle.success
view.right_button.label = f"Next: {page_num + 1}/{len(card_embeds)}"
if page_num == 1:
@ -1941,7 +2010,7 @@ async def open_choice_pack(
players = pl["players"]
elif pack_type == "Team Choice":
if this_pack["pack_team"] is None:
raise KeyError(f"Team not listed for Team Choice pack")
raise KeyError("Team not listed for Team Choice pack")
d1000 = random.randint(1, 1000)
pack_cover = this_pack["pack_team"]["logo"]
@ -1980,7 +2049,7 @@ async def open_choice_pack(
rarity_id += 1
elif pack_type == "Promo Choice":
if this_pack["pack_cardset"] is None:
raise KeyError(f"Cardset not listed for Promo Choice pack")
raise KeyError("Cardset not listed for Promo Choice pack")
d1000 = random.randint(1, 1000)
pack_cover = IMAGES["mvp-hype"]
@ -2037,8 +2106,8 @@ async def open_choice_pack(
rarity_id += 3
if len(players) == 0:
logger.error(f"Could not create choice pack")
raise ConnectionError(f"Could not create choice pack")
logger.error("Could not create choice pack")
raise ConnectionError("Could not create choice pack")
if type(context) == commands.Context:
author = context.author
@ -2061,7 +2130,7 @@ async def open_choice_pack(
view = Pagination([author], timeout=30)
view.left_button.disabled = True
view.left_button.label = f"Prev: -/{len(card_embeds)}"
view.cancel_button.label = f"Take This Card"
view.cancel_button.label = "Take This Card"
view.cancel_button.style = discord.ButtonStyle.success
view.cancel_button.disabled = True
view.right_button.label = f"Next: 1/{len(card_embeds)}"
@ -2079,10 +2148,10 @@ async def open_choice_pack(
)
if rarity_id >= 5:
tmp_msg = await pack_channel.send(
content=f"<@&1163537676885033010> we've got an MVP!"
content="<@&1163537676885033010> we've got an MVP!"
)
else:
tmp_msg = await pack_channel.send(content=f"We've got a choice pack here!")
tmp_msg = await pack_channel.send(content="We've got a choice pack here!")
while True:
await view.wait()
@ -2097,7 +2166,7 @@ async def open_choice_pack(
)
except Exception as e:
logger.error(f"failed to create cards: {e}")
raise ConnectionError(f"Failed to distribute these cards.")
raise ConnectionError("Failed to distribute these cards.")
await db_patch(
"packs",
@ -2131,7 +2200,7 @@ async def open_choice_pack(
view = Pagination([author], timeout=30)
view.left_button.label = f"Prev: {page_num - 1}/{len(card_embeds)}"
view.cancel_button.label = f"Take This Card"
view.cancel_button.label = "Take This Card"
view.cancel_button.style = discord.ButtonStyle.success
view.right_button.label = f"Next: {page_num + 1}/{len(card_embeds)}"
if page_num == 1:

530
tests/test_roll_for_cards.py Normal file
View File

@ -0,0 +1,530 @@
"""Tests for roll_for_cards parallelized implementation.
Validates dice rolling, batched player fetches, card creation,
pack marking, MVP backfill, cardset-23 dupe fallback, and notifications.
"""
import os
from unittest.mock import AsyncMock, patch
import pytest
def _make_player(
player_id, rarity_value=0, rarity_name="Replacement", p_name="Test Player"
):
"""Factory for player dicts matching API shape."""
return {
"player_id": player_id,
"rarity": {"value": rarity_value, "name": rarity_name},
"p_name": p_name,
"description": f"2024 {p_name}",
}
_UNSET = object()
def _make_pack(
pack_id, team_id=1, pack_type="Standard", pack_team=None, pack_cardset=_UNSET
):
"""Factory for pack dicts matching API shape."""
return {
"id": pack_id,
"team": {"id": team_id, "abbrev": "TST"},
"pack_type": {"name": pack_type},
"pack_team": pack_team,
"pack_cardset": {"id": 10} if pack_cardset is _UNSET else pack_cardset,
}
def _random_response(players):
"""Wrap a list of player dicts in the API response shape."""
return {"count": len(players), "players": players}
@pytest.fixture
def mock_db():
"""Patch db_get, db_post, db_patch in helpers.main."""
with (
patch("helpers.main.db_get", new_callable=AsyncMock) as mock_get,
patch("helpers.main.db_post", new_callable=AsyncMock) as mock_post,
patch("helpers.main.db_patch", new_callable=AsyncMock) as mock_patch,
):
mock_post.return_value = True
yield mock_get, mock_post, mock_patch
class TestSinglePack:
"""Single pack opening — verifies basic flow."""
async def test_single_standard_pack_creates_cards_and_marks_opened(self, mock_db):
"""A single standard pack should fetch players, create cards, and patch open_time.
Why: Validates the core happy path — dice roll → fetch → create → mark opened.
"""
mock_get, mock_post, mock_patch = mock_db
# Return enough players for any rarity tier requested
mock_get.return_value = _random_response([_make_player(i) for i in range(10)])
pack = _make_pack(100)
from helpers.main import roll_for_cards
result = await roll_for_cards([pack])
assert result == [100]
# At least one db_get for player fetches
assert mock_get.call_count >= 1
# Exactly one db_post for cards (may have notif posts too)
card_posts = [c for c in mock_post.call_args_list if c.args[0] == "cards"]
assert len(card_posts) == 1
# Exactly one db_patch for marking pack opened
assert mock_patch.call_count == 1
assert mock_patch.call_args.kwargs["object_id"] == 100
async def test_checkin_pack_uses_extra_val(self, mock_db):
"""Check-In Player packs should apply extra_val modifier to dice range.
Why: extra_val shifts the d1000 ceiling, affecting rarity odds for check-in rewards.
"""
mock_get, mock_post, mock_patch = mock_db
mock_get.return_value = _random_response([_make_player(1)])
pack = _make_pack(200, pack_type="Check-In Player")
from helpers.main import roll_for_cards
result = await roll_for_cards([pack], extra_val=500)
assert result == [200]
assert mock_get.call_count >= 1
async def test_unknown_pack_type_raises(self, mock_db):
"""Unrecognized pack types must raise TypeError.
Why: Guards against silent failures if a new pack type is added without dice logic.
"""
mock_get, mock_post, mock_patch = mock_db
pack = _make_pack(300, pack_type="Unknown")
from helpers.main import roll_for_cards
with pytest.raises(TypeError, match="Pack type not recognized"):
await roll_for_cards([pack])
class TestMultiplePacks:
"""Multiple packs — verifies batching and distribution."""
async def test_multiple_packs_return_all_ids(self, mock_db):
"""Opening multiple packs should return all pack IDs.
Why: Callers use the returned IDs to know which packs were successfully opened.
"""
mock_get, mock_post, mock_patch = mock_db
mock_get.return_value = _random_response([_make_player(i) for i in range(50)])
packs = [_make_pack(i) for i in range(5)]
from helpers.main import roll_for_cards
result = await roll_for_cards(packs)
assert result == [0, 1, 2, 3, 4]
async def test_multiple_packs_batch_fetches(self, mock_db):
"""Multiple packs should batch fetches — one db_get per rarity tier, not per pack.
Why: This is the core performance optimization. 5 packs should NOT make 20-30 calls.
"""
mock_get, mock_post, mock_patch = mock_db
mock_get.return_value = _random_response([_make_player(i) for i in range(50)])
packs = [_make_pack(i) for i in range(5)]
from helpers.main import roll_for_cards
await roll_for_cards(packs)
# Standard packs have up to 6 rarity tiers, but typically fewer are non-zero.
# The key assertion: far fewer fetches than 5 packs * ~4 tiers = 20.
player_fetches = [
c for c in mock_get.call_args_list if c.args[0] == "players/random"
]
# At most 6 tier fetches + possible 1 MVP backfill = 7
assert len(player_fetches) <= 7
async def test_multiple_packs_create_cards_per_pack(self, mock_db):
"""Each pack should get its own db_post('cards') call with correct pack_id.
Why: Cards must be associated with the correct pack for display and tracking.
"""
mock_get, mock_post, mock_patch = mock_db
mock_get.return_value = _random_response([_make_player(i) for i in range(50)])
packs = [_make_pack(i) for i in range(3)]
from helpers.main import roll_for_cards
await roll_for_cards(packs)
card_posts = [c for c in mock_post.call_args_list if c.args[0] == "cards"]
assert len(card_posts) == 3
# Each card post should reference the correct pack_id
for i, post_call in enumerate(card_posts):
payload = post_call.kwargs["payload"]
pack_ids_in_cards = {card["pack_id"] for card in payload["cards"]}
assert pack_ids_in_cards == {i}
class TestMVPBackfill:
"""MVP fallback when a rarity tier returns fewer players than requested."""
async def test_shortfall_triggers_mvp_backfill(self, mock_db):
"""When a tier returns fewer players than needed, MVP backfill should fire.
Why: Packs must always contain the expected number of cards. Shortfalls are
filled with MVP-tier players as a fallback.
"""
mock_get, mock_post, mock_patch = mock_db
call_count = 0
async def side_effect(endpoint, params=None):
nonlocal call_count
call_count += 1
if params and any(
p[0] == "min_rarity"
and p[1] == 5
and any(q[0] == "max_rarity" for q in params) is False
for p in params
):
# MVP backfill call (no max_rarity)
return _random_response([_make_player(900, 5, "MVP")])
# For tier-specific calls, check if this is the MVP backfill
if params:
param_dict = dict(params)
if "max_rarity" not in param_dict:
return _random_response([_make_player(900, 5, "MVP")])
# Return fewer than requested to trigger shortfall
requested = 5
if params:
for key, val in params:
if key == "limit":
requested = val
break
return _random_response(
[_make_player(i) for i in range(max(0, requested - 1))]
)
mock_get.side_effect = side_effect
pack = _make_pack(100)
from helpers.main import roll_for_cards
result = await roll_for_cards([pack])
assert result == [100]
# Should have at least the tier fetch + backfill call
assert mock_get.call_count >= 2
class TestCardsetExclusion:
"""Cardset 23 should duplicate existing players instead of MVP backfill."""
async def test_cardset_23_duplicates_instead_of_mvp(self, mock_db):
"""For cardset 23, shortfalls should duplicate existing players, not fetch MVPs.
Why: Cardset 23 (special/limited cardset) shouldn't pull from the MVP pool —
it should fill gaps by duplicating from what's already available.
"""
mock_get, mock_post, mock_patch = mock_db
async def side_effect(endpoint, params=None):
if params:
param_dict = dict(params)
# If this is a backfill call (no max_rarity), it shouldn't happen
if "max_rarity" not in param_dict:
pytest.fail("Should not make MVP backfill call for cardset 23")
# Return fewer than requested
return _random_response([_make_player(1)])
mock_get.side_effect = side_effect
pack = _make_pack(100, pack_cardset={"id": 23})
from helpers.main import roll_for_cards
# Force specific dice rolls to ensure a shortfall
with patch("helpers.main.random.randint", return_value=1):
# d1000=1 for Standard: Rep, Rep, Rep, Rep, Rep → 5 Reps needed
result = await roll_for_cards([pack])
assert result == [100]
class TestNotifications:
"""Rare pull notifications should be gathered and sent."""
async def test_rare_pulls_generate_notifications(self, mock_db):
"""Players with rarity >= 3 should trigger notification posts.
Why: Rare pulls are announced to the community — all notifs should be sent.
"""
mock_get, mock_post, mock_patch = mock_db
rare_player = _make_player(
42, rarity_value=3, rarity_name="All-Star", p_name="Mike Trout"
)
mock_get.return_value = _random_response([rare_player])
pack = _make_pack(100)
# Force all dice to land on All-Star tier (d1000=951 for card 3)
from helpers.main import roll_for_cards
with patch("helpers.main.random.randint", return_value=960):
await roll_for_cards([pack])
notif_posts = [c for c in mock_post.call_args_list if c.args[0] == "notifs"]
assert len(notif_posts) >= 1
payload = notif_posts[0].kwargs["payload"]
assert payload["title"] == "Rare Pull"
assert "Mike Trout" in payload["field_name"]
async def test_no_notifications_for_common_pulls(self, mock_db):
"""Players with rarity < 3 should NOT trigger notifications.
Why: Only rare pulls are noteworthy — common cards would spam the notif feed.
"""
mock_get, mock_post, mock_patch = mock_db
common_player = _make_player(1, rarity_value=0, rarity_name="Replacement")
mock_get.return_value = _random_response([common_player])
pack = _make_pack(100)
from helpers.main import roll_for_cards
# Force low dice rolls (all Replacement)
with patch("helpers.main.random.randint", return_value=1):
await roll_for_cards([pack])
notif_posts = [c for c in mock_post.call_args_list if c.args[0] == "notifs"]
assert len(notif_posts) == 0
class TestErrorHandling:
"""Error propagation from gathered writes."""
async def test_card_creation_failure_raises(self, mock_db):
"""If db_post('cards') returns falsy, ConnectionError must propagate.
Why: Card creation failure means the pack wasn't properly opened — caller
needs to know so it can report the error to the user.
"""
mock_get, mock_post, mock_patch = mock_db
mock_get.return_value = _random_response([_make_player(1)])
mock_post.return_value = False # Simulate failure
pack = _make_pack(100)
from helpers.main import roll_for_cards
with pytest.raises(ConnectionError, match="Failed to create"):
await roll_for_cards([pack])
class TestPackTeamFiltering:
"""Verify correct filter params are passed to player fetch."""
async def test_pack_team_adds_franchise_filter(self, mock_db):
"""When pack has a pack_team, franchise filter should be applied.
Why: Team-specific packs should only contain players from that franchise.
"""
mock_get, mock_post, mock_patch = mock_db
mock_get.return_value = _random_response([_make_player(1)])
pack = _make_pack(
100,
pack_team={"sname": "NYY"},
pack_cardset=None,
)
from helpers.main import roll_for_cards
with patch("helpers.main.random.randint", return_value=1):
await roll_for_cards([pack])
# Check that tier-fetch calls (those with max_rarity) include franchise filter
tier_calls = [
c
for c in mock_get.call_args_list
if any(p[0] == "max_rarity" for p in (c.kwargs.get("params") or []))
]
assert len(tier_calls) >= 1
for c in tier_calls:
param_dict = dict(c.kwargs.get("params") or [])
assert param_dict.get("franchise") == "NYY"
assert param_dict.get("in_packs") is True
async def test_no_team_no_cardset_adds_in_packs(self, mock_db):
"""When pack has no team or cardset, in_packs filter should be applied.
Why: Generic packs still need the in_packs filter to exclude non-packable players.
"""
mock_get, mock_post, mock_patch = mock_db
mock_get.return_value = _random_response([_make_player(1)])
pack = _make_pack(100, pack_team=None, pack_cardset=None)
from helpers.main import roll_for_cards
with patch("helpers.main.random.randint", return_value=1):
await roll_for_cards([pack])
# Check that tier-fetch calls (those with max_rarity) include in_packs filter
tier_calls = [
c
for c in mock_get.call_args_list
if any(p[0] == "max_rarity" for p in (c.kwargs.get("params") or []))
]
assert len(tier_calls) >= 1
for c in tier_calls:
param_dict = dict(c.kwargs.get("params") or [])
assert param_dict.get("in_packs") is True
# ---------------------------------------------------------------------------
# Integration tests — hit real dev API for reads, mock all writes
# ---------------------------------------------------------------------------
requires_api = pytest.mark.skipif(
not os.environ.get("API_TOKEN"),
reason="API_TOKEN not set — skipping integration tests",
)
@requires_api
class TestIntegrationRealFetches:
"""Integration tests that hit the real dev API for player fetches.
Only db_get is real — db_post and db_patch are mocked to prevent writes.
Run with: API_TOKEN=<token> python -m pytest tests/test_roll_for_cards.py -k integration -v
"""
@pytest.fixture
def mock_writes(self):
"""Mock only write operations, let reads hit the real API."""
with (
patch("helpers.main.db_post", new_callable=AsyncMock) as mock_post,
patch("helpers.main.db_patch", new_callable=AsyncMock) as mock_patch,
):
mock_post.return_value = True
yield mock_post, mock_patch
async def test_integration_single_pack_fetches_real_players(self, mock_writes):
"""A single standard pack should fetch real players from the dev API.
Why: Validates that the batched fetch params (min_rarity, max_rarity, limit,
in_packs) produce valid responses from the real API and that the returned
players have the expected structure.
"""
mock_post, mock_patch = mock_writes
pack = _make_pack(9999)
from helpers.main import roll_for_cards
result = await roll_for_cards([pack])
assert result == [9999]
# Cards were "created" (mocked)
card_posts = [c for c in mock_post.call_args_list if c.args[0] == "cards"]
assert len(card_posts) == 1
payload = card_posts[0].kwargs["payload"]
# Standard pack produces 5 cards
assert len(payload["cards"]) == 5
# Each card has the expected structure
for card in payload["cards"]:
assert "player_id" in card
assert card["team_id"] == 1
assert card["pack_id"] == 9999
async def test_integration_multiple_packs_batch_correctly(self, mock_writes):
"""Multiple packs should batch fetches and distribute players correctly.
Why: Validates the core optimization — summing counts across packs, making
fewer API calls, and slicing players back into per-pack groups with real data.
"""
mock_post, mock_patch = mock_writes
packs = [_make_pack(i + 9000) for i in range(3)]
from helpers.main import roll_for_cards
result = await roll_for_cards(packs)
assert result == [9000, 9001, 9002]
card_posts = [c for c in mock_post.call_args_list if c.args[0] == "cards"]
assert len(card_posts) == 3
# Each pack should have exactly 5 cards (Standard packs)
total_cards = 0
for post_call in card_posts:
cards = post_call.kwargs["payload"]["cards"]
assert len(cards) == 5
total_cards += len(cards)
assert total_cards == 15
async def test_integration_players_have_valid_rarity(self, mock_writes):
"""Fetched players should have rarity values matching their requested tier.
Why: Confirms the API respects min_rarity/max_rarity filters and that
the player distribution logic assigns correct-tier players to each pack.
"""
mock_post, mock_patch = mock_writes
pack = _make_pack(9999)
from helpers.main import roll_for_cards
# Use fixed dice to get known rarity distribution
# d1000=500 for Standard: Rep, Res, Sta, Res, Sta (mix of low tiers)
with patch("helpers.main.random.randint", return_value=500):
await roll_for_cards([pack])
card_posts = [c for c in mock_post.call_args_list if c.args[0] == "cards"]
assert len(card_posts) == 1
cards = card_posts[0].kwargs["payload"]["cards"]
# All cards should have valid player_ids (positive integers from real API)
for card in cards:
assert isinstance(card["player_id"], int)
assert card["player_id"] > 0
async def test_integration_cardset_filter(self, mock_writes):
"""Packs with a specific cardset should only fetch players from that cardset.
Why: Validates that the cardset_id parameter is correctly passed through
the batched fetch and the API filters accordingly.
"""
mock_post, mock_patch = mock_writes
pack = _make_pack(9999, pack_cardset={"id": 24})
from helpers.main import roll_for_cards
with patch("helpers.main.random.randint", return_value=500):
result = await roll_for_cards([pack])
assert result == [9999]
card_posts = [c for c in mock_post.call_args_list if c.args[0] == "cards"]
assert len(card_posts) == 1
assert len(card_posts[0].kwargs["payload"]["cards"]) == 5
async def test_integration_checkin_pack(self, mock_writes):
"""Check-In Player pack should fetch exactly 1 player from the real API.
Why: Check-in packs produce a single card — validates the simplest
path through the batched fetch logic with real data.
"""
mock_post, mock_patch = mock_writes
pack = _make_pack(9999, pack_type="Check-In Player")
from helpers.main import roll_for_cards
result = await roll_for_cards([pack])
assert result == [9999]
card_posts = [c for c in mock_post.call_args_list if c.args[0] == "cards"]
assert len(card_posts) == 1
# Check-in packs produce exactly 1 card
assert len(card_posts[0].kwargs["payload"]["cards"]) == 1