Files
SREBOT/BOT/stack_manager.py
T
2026-05-16 16:04:37 -07:00

1248 lines
50 KiB
Python

"""
stack_manager.py
Stack management feature: leaders create a persistent embed, players apply to join,
leader accepts/declines applicants.
"""
# Standard Library Imports
import json
import time
from pathlib import Path
# Third-Party Library Imports
import discord
from discord import app_commands
from discord.ext import commands
# Local Module Imports
from .utils import (
DEFAULT_FOOTER_CAT,
STACKS_DIR,
load_json,
write_json,
is_blacklisted,
gate_entitle,
permission_fail,
esc,
collect_command_stats,
command_locale,
)
from .utils import t, guild_lang
MAX_MEMBERS = 8
MAX_REQUESTERS = 20
# ============================================================================
# HELPERS
# ============================================================================
def get_stack_path(guild_id: str, leader_discord_id: str) -> Path:
"""Return the filesystem Path for a stack's JSON file.
Args:
guild_id: Discord guild ID string.
leader_discord_id: Discord user ID string of the stack leader.
Returns:
Path object pointing to the stack JSON file.
"""
return STACKS_DIR / f"{guild_id}-{leader_discord_id}.json"
async def load_stack(guild_id: str, leader_discord_id: str) -> dict | None:
"""Load a stack's data from its JSON file asynchronously.
Args:
guild_id: Discord guild ID string.
leader_discord_id: Discord user ID string of the stack leader.
Returns:
The parsed stack dict, or None if the file does not exist or is empty.
"""
path = get_stack_path(guild_id, leader_discord_id)
if not path.exists():
return None
data = await load_json(path, None)
return data if data else None
def load_stack_sync(guild_id: str, leader_discord_id: str) -> dict | None:
"""Sync version for use in __init__ methods that can't be async."""
path = get_stack_path(guild_id, leader_discord_id)
if not path.exists():
return None
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
return data if data else None
except Exception:
return None
async def save_stack(data: dict) -> bool:
"""Persist a stack's data dict to its JSON file.
Args:
data: Stack data dict containing at least guild_id and leader_discord_id.
Returns:
True on success, False on failure.
"""
path = get_stack_path(data["guild_id"], data["leader_discord_id"])
STACKS_DIR.mkdir(parents=True, exist_ok=True)
return await write_json(path, data)
def build_stack_embed(data: dict, lang: str = "en") -> discord.Embed:
"""Build the public-facing Discord embed for a stack.
Shows the stack name, member list with vehicles, leader star indicator,
and any pending applicants in the queue.
Args:
data: Stack data dict with members, requesters, leader info, and name.
lang: Language code for i18n.
Returns:
A discord.Embed ready to send or edit into a message.
"""
leader_nick = esc(data["leader_nick"])
members = data.get("members", [])
requesters = data.get("requesters", [])
embed = discord.Embed(
title=data.get("name") or t(lang, "stacks.stack_title", leader=data['leader_nick']),
color=discord.Color.blurple(),
)
if members:
lines = []
for m in members:
nick = esc(m['nick'])
vehicle = esc(m['vehicle'])
if m["discord_id"] == data["leader_discord_id"]:
lines.append(f"{nick}{vehicle} (⭐)")
else:
lines.append(f"{nick}{vehicle}")
members_text = "\n".join(lines)
else:
members_text = t(lang, "stacks.no_members")
embed.add_field(
name=t(lang, "stacks.members_field", count=len(members), max=MAX_MEMBERS),
value=members_text,
inline=False,
)
if requesters:
req_lines = [
f"{esc(r['nick'])}{esc(r['vehicle'])}"
for r in requesters
]
embed.add_field(
name=t(lang, "stacks.queue_field", count=len(requesters), max=MAX_REQUESTERS),
value="\n".join(req_lines),
inline=False,
)
embed.set_footer(text=DEFAULT_FOOTER_CAT)
return embed
def _build_manage_embed(data: dict, lang: str = "en") -> discord.Embed:
"""Build the ephemeral management embed showing members and queue.
Args:
data: Stack data dict with members and requesters lists.
lang: Language code for i18n.
Returns:
A discord.Embed for the leader's management panel.
"""
requesters = data.get("requesters", [])
members = data.get("members", [])
embed = discord.Embed(title=t(lang, "stacks.manage_title"), color=discord.Color.blurple())
embed.add_field(
name=t(lang, "stacks.members_field", count=len(members), max=MAX_MEMBERS),
value="\n".join(
f"{esc(m['nick'])}{esc(m['vehicle'])}"
for m in members
) or t(lang, "common.none_option"),
inline=False,
)
embed.add_field(
name=t(lang, "stacks.queue_field", count=len(requesters), max=MAX_REQUESTERS),
value="\n".join(
f"{esc(r['nick'])}{esc(r['vehicle'])}"
for r in requesters
) or t(lang, "stacks.no_pending_requests"),
inline=False,
)
embed.set_footer(text=DEFAULT_FOOTER_CAT)
return embed
async def _do_transfer(
data: dict,
new_leader_id: str,
public_stack_view: "StackView",
lang: str = "en",
) -> str | None:
"""
Transfer stack ownership to new_leader_id.
Renames the JSON file, updates public_stack_view in place.
Returns an error string on failure, or None on success.
"""
guild_id = data["guild_id"]
old_leader_id = data["leader_discord_id"]
new_leader = next((m for m in data["members"] if m["discord_id"] == new_leader_id), None)
if new_leader is None:
return t(lang, "stacks.member_not_exists")
new_path = get_stack_path(guild_id, new_leader_id)
if new_path.exists():
return t(lang, "stacks.already_has_stack")
old_path = get_stack_path(guild_id, old_leader_id)
data["leader_discord_id"] = new_leader["discord_id"]
data["leader_nick"] = new_leader["nick"]
data["leader_vehicle"] = new_leader["vehicle"]
data["members"] = [m for m in data["members"] if m["discord_id"] != old_leader_id]
await save_stack(data) # writes to new path
old_path.unlink(missing_ok=True)
public_stack_view.leader_discord_id = new_leader_id
return None
async def refresh_stack_message(view: "StackView") -> None:
"""Re-render and edit the public stack message with current data.
Args:
view: The StackView whose associated message should be refreshed.
"""
if view.message is None:
return
data = await load_stack(view.guild_id, view.leader_discord_id)
if data is None:
return
embed = build_stack_embed(data, lang=view.lang)
await view.message.edit(embed=embed, view=view)
# ============================================================================
# MODAL
# ============================================================================
class JoinStackModal(discord.ui.Modal, title="Apply to Join Stack"):
"""Modal presented to users requesting to join a stack.
Collects the vehicle the player intends to use and adds them
to the stack's requester queue upon submission.
"""
vehicle_input = discord.ui.TextInput(
label="What will you play?",
placeholder="e.g. F-16C, WZ305...",
max_length=100,
required=True,
)
def __init__(self, guild_id: str, leader_discord_id: str, stack_view: "StackView", lang: str = "en"):
super().__init__()
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.stack_view = stack_view
self.lang = lang
self.title = t(lang, "stacks.join_modal_title")
self.vehicle_input.label = t(lang, "stacks.join_vehicle_label")
self.vehicle_input.placeholder = t(lang, "stacks.join_vehicle_placeholder")
async def on_submit(self, interaction: discord.Interaction) -> None:
"""Validate the application and append the user to the stack's request queue."""
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(
t(self.lang, "stacks.no_longer_exists"), ephemeral=True
)
return
user_id = str(interaction.user.id)
nick = interaction.user.display_name
vehicle = self.vehicle_input.value.strip()
if any(m["discord_id"] == user_id for m in data["members"]):
await interaction.response.send_message(
t(self.lang, "stacks.already_member"), ephemeral=True
)
return
if any(r["discord_id"] == user_id for r in data["requesters"]):
await interaction.response.send_message(
t(self.lang, "stacks.already_applied"), ephemeral=True
)
return
if len(data["requesters"]) >= MAX_REQUESTERS:
await interaction.response.send_message(
t(self.lang, "stacks.queue_full", max=MAX_REQUESTERS),
ephemeral=True,
)
return
data["requesters"].append({"discord_id": user_id, "nick": nick, "vehicle": vehicle})
await save_stack(data)
await refresh_stack_message(self.stack_view)
await interaction.response.send_message(
t(self.lang, "stacks.application_sent"), ephemeral=True
)
# ============================================================================
# DISBAND CONFIRMATION VIEW
# ============================================================================
class DisbandConfirmView(discord.ui.View):
"""Ephemeral confirmation view with Yes/Cancel buttons for disbanding a stack."""
def __init__(self, stack_view: "StackView", lang: str = "en"):
super().__init__(timeout=30)
self.stack_view = stack_view
self.lang = lang
self.confirm.label = t(lang, "buttons.yes_disband")
self.cancel.label = t(lang, "buttons.cancel")
@discord.ui.button(label="Yes, Disband", style=discord.ButtonStyle.red)
async def confirm(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
await interaction.response.edit_message(
content=t(self.lang, "stacks.stack_disbanded"), embed=None, view=None
)
path = get_stack_path(self.stack_view.guild_id, self.stack_view.leader_discord_id)
path.unlink(missing_ok=True)
embed = discord.Embed(
title=t(self.lang, "stacks.disbanded_title"),
description=t(self.lang, "stacks.disbanded_desc"),
color=discord.Color.dark_red(),
)
embed.set_footer(text=DEFAULT_FOOTER_CAT)
if self.stack_view.message:
await self.stack_view.message.edit(embed=embed, view=None)
self.stack_view.stop()
@discord.ui.button(label="Cancel", style=discord.ButtonStyle.grey)
async def cancel(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
await interaction.response.edit_message(content=t(self.lang, "stacks.cancelled"), embed=None, view=None)
# ============================================================================
# TRANSFER LEAVE VIEW (ephemeral, leader-only)
# ============================================================================
class TransferLeaveView(discord.ui.View):
"""Shown when the leader clicks Leave / Withdraw and other members exist."""
def __init__(self, guild_id: str, leader_discord_id: str, public_stack_view: "StackView", lang: str = "en"):
super().__init__(timeout=60)
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.public_stack_view = public_stack_view
self.lang = lang
self.selected_uid: str | None = None
data = load_stack_sync(guild_id, leader_discord_id)
non_leader = [
m for m in (data.get("members", []) if data else [])
if m["discord_id"] != leader_discord_id
]
options = [
discord.SelectOption(
label=f"{m['nick']}{m['vehicle']}"[:100],
value=m["discord_id"],
)
for m in non_leader
]
select = discord.ui.Select(placeholder=t(lang, "stacks.select_new_leader"), options=options)
select.callback = self._on_select
self.add_item(select)
transfer_btn = discord.ui.Button(label=t(lang, "buttons.transfer_leave"), style=discord.ButtonStyle.blurple)
transfer_btn.callback = self._on_transfer
self.add_item(transfer_btn)
async def _on_select(self, interaction: discord.Interaction) -> None:
if interaction.data:
self.selected_uid = interaction.data["values"][0] # type: ignore[typeddict-item]
await interaction.response.defer()
async def _on_transfer(self, interaction: discord.Interaction) -> None:
if not self.selected_uid:
await interaction.response.send_message(
t(self.lang, "stacks.select_member_transfer"), ephemeral=True
)
return
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
new_nick = next(
(m["nick"] for m in data["members"] if m["discord_id"] == self.selected_uid), "?"
)
error = await _do_transfer(data, self.selected_uid, self.public_stack_view, lang=self.lang)
if error:
await interaction.response.send_message(error, ephemeral=True)
return
await refresh_stack_message(self.public_stack_view)
await interaction.response.edit_message(
content=t(self.lang, "stacks.ownership_transferred", nick=esc(new_nick)),
view=None,
)
# ============================================================================
# MANAGE SUB-VIEWS (ephemeral, leader-only)
# ============================================================================
class AcceptMembersView(discord.ui.View):
"""Sub-view: accept or decline applicants from the queue."""
def __init__(self, guild_id: str, leader_discord_id: str, public_stack_view: "StackView", lang: str = "en"):
super().__init__(timeout=300)
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.public_stack_view = public_stack_view
self.lang = lang
self.selected_requester_uids: list[str] = []
data = load_stack_sync(guild_id, leader_discord_id)
requesters = data.get("requesters", []) if data else []
if requesters:
req_options = [
discord.SelectOption(
label=f"{r['nick']}{r['vehicle']}"[:100],
value=r["discord_id"],
)
for r in requesters
]
req_select = discord.ui.Select(
placeholder=t(lang, "stacks.select_applicants"),
options=req_options,
min_values=1,
max_values=len(req_options),
)
req_select.callback = self._on_requester_select
self.add_item(req_select)
accept_btn = discord.ui.Button(label=t(lang, "buttons.accept_selected"), style=discord.ButtonStyle.green)
accept_btn.callback = self._on_accept
self.add_item(accept_btn)
accept_all_btn = discord.ui.Button(label=t(lang, "buttons.accept_all"), style=discord.ButtonStyle.green)
accept_all_btn.callback = self._on_accept_all
self.add_item(accept_all_btn)
decline_btn = discord.ui.Button(label=t(lang, "buttons.decline_selected"), style=discord.ButtonStyle.red)
decline_btn.callback = self._on_decline
self.add_item(decline_btn)
else:
disabled_req = discord.ui.Select(
placeholder=t(lang, "stacks.no_pending_applications"),
options=[discord.SelectOption(label="—", value="none")],
disabled=True,
)
self.add_item(disabled_req)
back_btn = discord.ui.Button(label=t(lang, "buttons.back"), style=discord.ButtonStyle.grey)
back_btn.callback = self._on_back
self.add_item(back_btn)
async def _on_requester_select(self, interaction: discord.Interaction) -> None:
if interaction.data:
self.selected_requester_uids = interaction.data["values"] # type: ignore[typeddict-item]
await interaction.response.defer()
async def _on_accept(self, interaction: discord.Interaction) -> None:
if not self.selected_requester_uids:
await interaction.response.send_message(
t(self.lang, "stacks.select_applicant_first"), ephemeral=True
)
return
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
available_slots = MAX_MEMBERS - len(data["members"])
to_accept_ids = self.selected_requester_uids[:available_slots]
if not to_accept_ids:
await interaction.response.send_message(
t(self.lang, "stacks.stack_full", max=MAX_MEMBERS), ephemeral=True
)
return
accepted = [r for r in data["requesters"] if r["discord_id"] in to_accept_ids]
data["members"].extend(accepted)
accepted_ids = {r["discord_id"] for r in accepted}
data["requesters"] = [r for r in data["requesters"] if r["discord_id"] not in accepted_ids]
await save_stack(data)
self.selected_requester_uids = []
await refresh_stack_message(self.public_stack_view)
new_view = AcceptMembersView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(embed=_build_manage_embed(data, lang=self.lang), view=new_view)
async def _on_accept_all(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
available_slots = MAX_MEMBERS - len(data["members"])
if available_slots <= 0:
await interaction.response.send_message(
t(self.lang, "stacks.stack_full", max=MAX_MEMBERS), ephemeral=True
)
return
to_accept = data["requesters"][:available_slots]
accepted_ids = {r["discord_id"] for r in to_accept}
data["members"].extend(to_accept)
data["requesters"] = [r for r in data["requesters"] if r["discord_id"] not in accepted_ids]
await save_stack(data)
self.selected_requester_uids = []
await refresh_stack_message(self.public_stack_view)
new_view = AcceptMembersView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(embed=_build_manage_embed(data, lang=self.lang), view=new_view)
async def _on_decline(self, interaction: discord.Interaction) -> None:
if not self.selected_requester_uids:
await interaction.response.send_message(
t(self.lang, "stacks.select_applicant_first"), ephemeral=True
)
return
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
declined_ids = set(self.selected_requester_uids)
data["requesters"] = [r for r in data["requesters"] if r["discord_id"] not in declined_ids]
await save_stack(data)
self.selected_requester_uids = []
await refresh_stack_message(self.public_stack_view)
new_view = AcceptMembersView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(embed=_build_manage_embed(data, lang=self.lang), view=new_view)
async def _on_back(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
new_view = ManageStackView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(
content=None, embed=_build_manage_embed(data, lang=self.lang) if data else None, view=new_view
)
class RemoveMembersView(discord.ui.View):
"""Sub-view: remove members and/or requesters by various bulk or selective actions."""
def __init__(self, guild_id: str, leader_discord_id: str, public_stack_view: "StackView", lang: str = "en"):
super().__init__(timeout=300)
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.public_stack_view = public_stack_view
self.lang = lang
# values are prefixed: "m:{id}" for members, "r:{id}" for requesters
self.selected_values: list[str] = []
data = load_stack_sync(guild_id, leader_discord_id)
non_leader_members = [
m for m in (data.get("members", []) if data else [])
if m["discord_id"] != leader_discord_id
]
requesters = data.get("requesters", []) if data else []
all_options = [
discord.SelectOption(
label=f"[Active] {m['nick']}{m['vehicle']}"[:100],
value=f"m:{m['discord_id']}",
)
for m in non_leader_members
] + [
discord.SelectOption(
label=f"[Queue] {r['nick']}{r['vehicle']}"[:100],
value=f"r:{r['discord_id']}",
)
for r in requesters
]
if all_options:
select = discord.ui.Select(
placeholder=t(lang, "stacks.select_to_remove"),
options=all_options,
min_values=1,
max_values=len(all_options),
)
select.callback = self._on_select
self.add_item(select)
else:
self.add_item(discord.ui.Select(
placeholder=t(lang, "stacks.no_members_or_applicants"),
options=[discord.SelectOption(label="—", value="none")],
disabled=True,
))
remove_all_btn = discord.ui.Button(label=t(lang, "buttons.remove_all"), style=discord.ButtonStyle.red)
remove_all_btn.callback = self._on_remove_all
self.add_item(remove_all_btn)
remove_active_btn = discord.ui.Button(label=t(lang, "buttons.remove_active"), style=discord.ButtonStyle.red)
remove_active_btn.callback = self._on_remove_active
self.add_item(remove_active_btn)
remove_queued_btn = discord.ui.Button(label=t(lang, "buttons.remove_queued"), style=discord.ButtonStyle.red)
remove_queued_btn.callback = self._on_remove_queued
self.add_item(remove_queued_btn)
remove_selected_btn = discord.ui.Button(label=t(lang, "buttons.remove_selected"), style=discord.ButtonStyle.red)
remove_selected_btn.callback = self._on_remove_selected
self.add_item(remove_selected_btn)
back_btn = discord.ui.Button(label=t(lang, "buttons.back"), style=discord.ButtonStyle.grey)
back_btn.callback = self._on_back
self.add_item(back_btn)
async def _on_select(self, interaction: discord.Interaction) -> None:
if interaction.data:
self.selected_values = interaction.data["values"] # type: ignore[typeddict-item]
await interaction.response.defer()
async def _apply_removal(
self,
interaction: discord.Interaction,
remove_member_ids: set[str],
remove_requester_ids: set[str],
) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
remove_member_ids -= {self.leader_discord_id}
data["members"] = [m for m in data["members"] if m["discord_id"] not in remove_member_ids]
data["requesters"] = [r for r in data["requesters"] if r["discord_id"] not in remove_requester_ids]
await save_stack(data)
self.selected_values = []
await refresh_stack_message(self.public_stack_view)
new_view = RemoveMembersView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(embed=_build_manage_embed(data, lang=self.lang), view=new_view)
async def _on_remove_all(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
member_ids = {m["discord_id"] for m in data.get("members", [])}
requester_ids = {r["discord_id"] for r in data.get("requesters", [])}
await self._apply_removal(interaction, member_ids, requester_ids)
async def _on_remove_active(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
member_ids = {m["discord_id"] for m in data.get("members", [])}
await self._apply_removal(interaction, member_ids, set())
async def _on_remove_queued(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
requester_ids = {r["discord_id"] for r in data.get("requesters", [])}
await self._apply_removal(interaction, set(), requester_ids)
async def _on_remove_selected(self, interaction: discord.Interaction) -> None:
if not self.selected_values:
await interaction.response.send_message(
t(self.lang, "stacks.select_person_first"), ephemeral=True
)
return
member_ids = {v[2:] for v in self.selected_values if v.startswith("m:")}
requester_ids = {v[2:] for v in self.selected_values if v.startswith("r:")}
await self._apply_removal(interaction, member_ids, requester_ids)
async def _on_back(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
new_view = ManageStackView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(
content=None, embed=_build_manage_embed(data, lang=self.lang) if data else None, view=new_view
)
class PingMessageModal(discord.ui.Modal, title="Ping Message"):
"""Modal for composing an optional custom ping message."""
message = discord.ui.TextInput(
label="Custom message (optional)",
placeholder='e.g. Come now! Stack starting!',
required=False,
max_length=200,
)
def __init__(
self,
guild_id: str,
leader_discord_id: str,
ping_type: str,
selected_uids: list[str] | None = None,
lang: str = "en",
):
super().__init__()
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.ping_type = ping_type # "all" | "active" | "queued" | "select"
self.selected_uids = selected_uids or []
self.lang = lang
self.title = t(lang, "stacks.ping_modal_title")
self.message.label = t(lang, "stacks.ping_message_label")
self.message.placeholder = t(lang, "stacks.ping_message_placeholder")
async def on_submit(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
if self.ping_type == "all":
targets = (
[m["discord_id"] for m in data.get("members", []) if m["discord_id"] != self.leader_discord_id]
+ [r["discord_id"] for r in data.get("requesters", [])]
)
elif self.ping_type == "active":
targets = [m["discord_id"] for m in data.get("members", [])]
elif self.ping_type == "queued":
targets = [r["discord_id"] for r in data.get("requesters", [])]
else:
targets = self.selected_uids
if not targets:
await interaction.response.send_message(t(self.lang, "stacks.no_one_to_ping"), ephemeral=True)
return
mentions = " ".join(f"<@{uid}>" for uid in targets)
custom = self.message.value.strip()
if custom:
content = f"{mentions} {custom}"
else:
stack_name = data.get("name") or "the stack"
content = f"{mentions}\n-# {t(self.lang, 'stacks.ping_footer', leader=esc(data['leader_nick']), stack=stack_name)}"
await interaction.response.send_message(t(self.lang, "stacks.pinged"), ephemeral=True)
await interaction.channel.send(content) # type: ignore[union-attr]
class PingMembersView(discord.ui.View):
"""Sub-view: ping all, active members, queued applicants, or a custom selection."""
def __init__(self, guild_id: str, leader_discord_id: str, public_stack_view: "StackView", lang: str = "en"):
super().__init__(timeout=300)
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.public_stack_view = public_stack_view
self.lang = lang
self.selected_uids: list[str] = []
data = load_stack_sync(guild_id, leader_discord_id)
members = data.get("members", []) if data else []
requesters = data.get("requesters", []) if data else []
all_options = [
discord.SelectOption(
label=f"[Active] {m['nick']}{m['vehicle']}"[:100],
value=m["discord_id"],
)
for m in members
] + [
discord.SelectOption(
label=f"[Queue] {r['nick']}{r['vehicle']}"[:100],
value=r["discord_id"],
)
for r in requesters
]
if all_options:
select = discord.ui.Select(
placeholder=t(lang, "stacks.select_to_ping"),
options=all_options,
min_values=1,
max_values=len(all_options),
)
select.callback = self._on_select
self.add_item(select)
else:
self.add_item(discord.ui.Select(
placeholder=t(lang, "stacks.no_members_or_applicants"),
options=[discord.SelectOption(label="—", value="none")],
disabled=True,
))
ping_all_btn = discord.ui.Button(label=t(lang, "buttons.ping_all"), style=discord.ButtonStyle.green)
ping_all_btn.callback = self._on_ping_all
self.add_item(ping_all_btn)
ping_active_btn = discord.ui.Button(label=t(lang, "buttons.ping_active"), style=discord.ButtonStyle.blurple)
ping_active_btn.callback = self._on_ping_active
self.add_item(ping_active_btn)
ping_queued_btn = discord.ui.Button(label=t(lang, "buttons.ping_queued"), style=discord.ButtonStyle.blurple)
ping_queued_btn.callback = self._on_ping_queued
self.add_item(ping_queued_btn)
ping_select_btn = discord.ui.Button(label=t(lang, "buttons.ping_selected"), style=discord.ButtonStyle.blurple)
ping_select_btn.callback = self._on_ping_selected
self.add_item(ping_select_btn)
back_btn = discord.ui.Button(label=t(lang, "buttons.back"), style=discord.ButtonStyle.grey)
back_btn.callback = self._on_back
self.add_item(back_btn)
async def _on_select(self, interaction: discord.Interaction) -> None:
if interaction.data:
self.selected_uids = interaction.data["values"] # type: ignore[typeddict-item]
await interaction.response.defer()
async def _on_ping_all(self, interaction: discord.Interaction) -> None:
await interaction.response.send_modal(
PingMessageModal(self.guild_id, self.leader_discord_id, "all", lang=self.lang)
)
async def _on_ping_active(self, interaction: discord.Interaction) -> None:
await interaction.response.send_modal(
PingMessageModal(self.guild_id, self.leader_discord_id, "active", lang=self.lang)
)
async def _on_ping_queued(self, interaction: discord.Interaction) -> None:
await interaction.response.send_modal(
PingMessageModal(self.guild_id, self.leader_discord_id, "queued", lang=self.lang)
)
async def _on_ping_selected(self, interaction: discord.Interaction) -> None:
if not self.selected_uids:
await interaction.response.send_message(
t(self.lang, "stacks.select_from_dropdown"), ephemeral=True
)
return
await interaction.response.send_modal(
PingMessageModal(self.guild_id, self.leader_discord_id, "select", self.selected_uids, lang=self.lang)
)
async def _on_back(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
new_view = ManageStackView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(
content=None, embed=_build_manage_embed(data, lang=self.lang) if data else None, view=new_view
)
class RenameStackModal(discord.ui.Modal, title="Rename Stack"):
"""Modal for the stack leader to rename their stack."""
name = discord.ui.TextInput(
label="Stack name",
placeholder="e.g. Night Owls, Alpha Squad...",
max_length=50,
required=True,
)
def __init__(self, guild_id: str, leader_discord_id: str, public_stack_view: "StackView", lang: str = "en"):
super().__init__()
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.public_stack_view = public_stack_view
self.lang = lang
self.title = t(lang, "stacks.rename_modal_title")
self.name.label = t(lang, "stacks.rename_label")
self.name.placeholder = t(lang, "stacks.rename_placeholder")
async def on_submit(self, interaction: discord.Interaction) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
data["name"] = self.name.value.strip()
await save_stack(data)
await refresh_stack_message(self.public_stack_view)
await interaction.response.send_message(
t(self.lang, "stacks.stack_renamed", name=data['name']), ephemeral=True
)
# ============================================================================
# MANAGE VIEW (ephemeral, leader-only)
# ============================================================================
class ManageStackView(discord.ui.View):
"""Entry view with 4 navigation buttons for stack management.
Buttons: Accept Members, Remove Members, Ping Members, Rename Stack.
"""
def __init__(self, guild_id: str, leader_discord_id: str, public_stack_view: "StackView", lang: str = "en"):
super().__init__(timeout=300)
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.public_stack_view = public_stack_view
self.lang = lang
self.accept_btn.label = t(lang, "buttons.accept_members")
self.remove_btn.label = t(lang, "buttons.remove_members")
self.ping_btn.label = t(lang, "buttons.ping_members")
self.rename_btn.label = t(lang, "buttons.rename_stack")
@discord.ui.button(label="Accept Members", style=discord.ButtonStyle.green)
async def accept_btn(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
new_view = AcceptMembersView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(embed=_build_manage_embed(data, lang=self.lang), view=new_view)
@discord.ui.button(label="Remove Members", style=discord.ButtonStyle.red)
async def remove_btn(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
new_view = RemoveMembersView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(embed=_build_manage_embed(data, lang=self.lang), view=new_view)
@discord.ui.button(label="Ping Members", style=discord.ButtonStyle.blurple)
async def ping_btn(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
new_view = PingMembersView(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
await interaction.response.edit_message(embed=_build_manage_embed(data, lang=self.lang), view=new_view)
@discord.ui.button(label="Rename Stack", style=discord.ButtonStyle.grey)
async def rename_btn(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
await interaction.response.send_modal(
RenameStackModal(self.guild_id, self.leader_discord_id, self.public_stack_view, lang=self.lang)
)
# ============================================================================
# PUBLIC STACK VIEW
# ============================================================================
class StackView(discord.ui.View):
"""Public persistent view attached to the stack embed message.
Provides Join, Leave, Manage, and Disband buttons. Times out after
8 hours, at which point the stack is automatically cleaned up.
"""
def __init__(self, guild_id: str, leader_discord_id: str, lang: str = "en"):
super().__init__(timeout=28800) # 8 hours
self.guild_id = guild_id
self.leader_discord_id = leader_discord_id
self.lang = lang
self.message: discord.Message | None = None
self.join_btn.label = t(lang, "buttons.request_to_join")
self.leave_btn.label = t(lang, "buttons.leave_withdraw")
self.manage_btn.label = t(lang, "buttons.manage_stack")
self.disband_btn.label = t(lang, "buttons.disband_stack")
@discord.ui.button(label="Request to Join", style=discord.ButtonStyle.green)
async def join_btn(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
"""Open the JoinStackModal for the user to submit a join application."""
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(
t(self.lang, "stacks.no_longer_exists"), ephemeral=True
)
return
await interaction.response.send_modal(
JoinStackModal(self.guild_id, self.leader_discord_id, self, lang=self.lang)
)
@discord.ui.button(label="Leave / Withdraw", style=discord.ButtonStyle.grey)
async def leave_btn(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
"""Remove the user from members or requesters, or prompt ownership transfer if leader."""
user_id = str(interaction.user.id)
if user_id == self.leader_discord_id:
data = await load_stack(self.guild_id, self.leader_discord_id)
non_leader = [
m for m in (data.get("members", []) if data else [])
if m["discord_id"] != self.leader_discord_id
]
if not non_leader:
await interaction.response.send_message(
t(self.lang, "stacks.only_member_use_disband"),
ephemeral=True,
)
return
await interaction.response.send_message(
t(self.lang, "stacks.select_transfer_prompt"),
view=TransferLeaveView(self.guild_id, self.leader_discord_id, self, lang=self.lang),
ephemeral=True,
)
return
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(
t(self.lang, "stacks.no_longer_exists"), ephemeral=True
)
return
if any(m["discord_id"] == user_id for m in data["members"]):
data["members"] = [m for m in data["members"] if m["discord_id"] != user_id]
await save_stack(data)
await refresh_stack_message(self)
await interaction.response.send_message(t(self.lang, "stacks.left_stack"), ephemeral=True)
return
if any(r["discord_id"] == user_id for r in data["requesters"]):
data["requesters"] = [r for r in data["requesters"] if r["discord_id"] != user_id]
await save_stack(data)
await refresh_stack_message(self)
await interaction.response.send_message(
t(self.lang, "stacks.application_withdrawn"), ephemeral=True
)
return
await interaction.response.send_message(
t(self.lang, "stacks.not_member_or_applicant"), ephemeral=True
)
@discord.ui.button(label="Manage Stack ⚙️", style=discord.ButtonStyle.blurple)
async def manage_btn(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
"""Open the ephemeral management panel (leader-only)."""
if interaction.user.id != int(self.leader_discord_id):
await interaction.response.send_message(
t(self.lang, "stacks.leader_only_manage"), ephemeral=True
)
return
data = await load_stack(self.guild_id, self.leader_discord_id)
if data is None:
await interaction.response.send_message(t(self.lang, "stacks.stack_not_found"), ephemeral=True)
return
manage_view = ManageStackView(self.guild_id, self.leader_discord_id, self, lang=self.lang)
await interaction.response.send_message(
embed=_build_manage_embed(data, lang=self.lang), view=manage_view, ephemeral=True
)
@discord.ui.button(label="Disband Stack", style=discord.ButtonStyle.red)
async def disband_btn(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
"""Show a disband confirmation prompt (leader-only)."""
if interaction.user.id != int(self.leader_discord_id):
await interaction.response.send_message(
t(self.lang, "stacks.leader_only_disband"), ephemeral=True
)
return
await interaction.response.send_message(
t(self.lang, "stacks.confirm_disband"),
view=DisbandConfirmView(self, lang=self.lang),
ephemeral=True,
)
async def on_timeout(self) -> None:
path = get_stack_path(self.guild_id, self.leader_discord_id)
try:
path.unlink(missing_ok=True)
except Exception:
pass
if self.message is None:
return
try:
embed = discord.Embed(
title=t(self.lang, "stacks.expired_title"),
description=t(self.lang, "stacks.expired_desc"),
color=discord.Color.dark_grey(),
)
embed.set_footer(text=DEFAULT_FOOTER_CAT)
await self.message.edit(embed=embed, view=None)
except Exception:
pass
# ============================================================================
# FORCE CREATE VIEW (ephemeral, used when duplicate stack exists)
# ============================================================================
class ForceCreateView(discord.ui.View):
"""Shown when the user already has a stack JSON — lets them nuke it and start fresh."""
def __init__(self, guild_id: str, leader_id: str, nick: str, vehicle: str, channel_id: str, lang: str = "en"):
super().__init__(timeout=30)
self.guild_id = guild_id
self.leader_id = leader_id
self.nick = nick
self.vehicle = vehicle
self.channel_id = channel_id
self.lang = lang
self.confirm.label = t(lang, "buttons.force_disband_create")
self.cancel.label = t(lang, "buttons.cancel")
@discord.ui.button(label="Force Disband & Create New", style=discord.ButtonStyle.red)
async def confirm(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
get_stack_path(self.guild_id, self.leader_id).unlink(missing_ok=True)
data = {
"guild_id": self.guild_id,
"leader_discord_id": self.leader_id,
"leader_nick": self.nick,
"leader_vehicle": self.vehicle,
"channel_id": self.channel_id,
"message_id": None,
"name": None,
"created_at": int(time.time()),
"members": [
{"discord_id": self.leader_id, "nick": self.nick, "vehicle": self.vehicle}
],
"requesters": [],
}
STACKS_DIR.mkdir(parents=True, exist_ok=True)
await save_stack(data)
view = StackView(self.guild_id, self.leader_id, lang=self.lang)
embed = build_stack_embed(data, lang=self.lang)
msg = await interaction.channel.send(embed=embed, view=view) # type: ignore[union-attr]
view.message = msg
data["message_id"] = str(msg.id)
await save_stack(data)
await interaction.response.edit_message(
content=t(self.lang, "stacks.force_created"), view=None
)
@discord.ui.button(label="Cancel", style=discord.ButtonStyle.grey)
async def cancel(self, interaction: discord.Interaction, button: discord.ui.Button) -> None:
await interaction.response.edit_message(content=t(self.lang, "stacks.cancelled"), view=None)
# ============================================================================
# COMMAND REGISTRATION
# ============================================================================
def register_commands(bot: commands.Bot) -> None:
"""Register the /stack-create and /stack-manage slash commands on the bot.
Args:
bot: The Discord bot instance to register commands on.
"""
@is_blacklisted()
@gate_entitle("standard")
@bot.tree.command(name="stack-create", description=command_locale("Create a new player stack", "commands.stack_create.description"))
@app_commands.describe(vehicle=command_locale("What vehicle will you start with?", "commands.stack_create.vehicle"))
async def stack_create(interaction: discord.Interaction, vehicle: str) -> None:
await collect_command_stats(interaction)
await interaction.response.defer(ephemeral=False)
guild_id = str(interaction.guild_id)
lang = await guild_lang(interaction.guild_id) if interaction.guild_id else "en"
leader_id = str(interaction.user.id)
nick = interaction.user.display_name
stack_path = get_stack_path(guild_id, leader_id)
if stack_path.exists():
await interaction.followup.send(
t(lang, "stacks.already_active_stack"),
view=ForceCreateView(guild_id, leader_id, nick, vehicle.strip(), str(interaction.channel_id), lang=lang),
ephemeral=True,
)
return
data = {
"guild_id": guild_id,
"leader_discord_id": leader_id,
"leader_nick": nick,
"leader_vehicle": vehicle.strip(),
"channel_id": str(interaction.channel_id),
"message_id": None,
"name": None,
"created_at": int(time.time()),
"members": [
{"discord_id": leader_id, "nick": nick, "vehicle": vehicle.strip()}
],
"requesters": [],
}
STACKS_DIR.mkdir(parents=True, exist_ok=True)
await save_stack(data)
view = StackView(guild_id, leader_id, lang=lang)
embed = build_stack_embed(data, lang=lang)
msg = await interaction.followup.send(embed=embed, view=view, wait=True)
view.message = msg
data["message_id"] = str(msg.id)
await save_stack(data)
@stack_create.error
async def stack_create_error(interaction: discord.Interaction, error: Exception) -> None:
await permission_fail(interaction, error)
@is_blacklisted()
@gate_entitle("standard")
@bot.tree.command(name="stack-manage", description=command_locale("Re-post your active stack embed to this channel", "commands.stack_manage.description"))
async def stack_manage(interaction: discord.Interaction) -> None:
await collect_command_stats(interaction)
await interaction.response.defer(ephemeral=False)
guild_id = str(interaction.guild_id)
lang = await guild_lang(interaction.guild_id) if interaction.guild_id else "en"
leader_id = str(interaction.user.id)
data = await load_stack(guild_id, leader_id)
if data is None:
await interaction.followup.send(
t(lang, "stacks.no_active_stack"),
ephemeral=True,
)
return
view = StackView(guild_id, leader_id, lang=lang)
embed = build_stack_embed(data, lang=lang)
msg = await interaction.followup.send(embed=embed, view=view, wait=True)
view.message = msg
data["message_id"] = str(msg.id)
await save_stack(data)
@stack_manage.error
async def stack_manage_error(interaction: discord.Interaction, error: Exception) -> None:
await permission_fail(interaction, error)