import discord from discord.ext import commands from discord import ui import slot_db as sdb from datetime import datetime, timedelta import asyncio SLOT_ADMIN_ROLES = ["Admin", "Slot Admin", "Owner"] SLOT_MOD_ROLES = ["Admin", "Slot Admin", "Owner", "Moderator", "Staff"] SLOT_COLOR = 0x5865F2 SUCCESS_COLOR = 0x2ECC71 ERROR_COLOR = 0xE74C3C WARNING_COLOR = 0xF39C12 def has_slot_admin(member: discord.Member) -> bool: role_names = [r.name for r in member.roles] return any(r in SLOT_ADMIN_ROLES for r in role_names) or member.guild_permissions.administrator def has_slot_mod(member: discord.Member) -> bool: role_names = [r.name for r in member.roles] return any(r in SLOT_MOD_ROLES for r in role_names) or member.guild_permissions.administrator def s_embed(title: str, desc: str = "", color: int = SLOT_COLOR) -> discord.Embed: e = discord.Embed(title=title, description=desc, color=color) e.set_footer(text="SlotManager Bot") return e def ok(title: str, desc: str = "") -> discord.Embed: return s_embed(title, desc, SUCCESS_COLOR) def err(title: str, desc: str = "") -> discord.Embed: return s_embed(title, desc, ERROR_COLOR) def warn(title: str, desc: str = "") -> discord.Embed: return s_embed(title, desc, WARNING_COLOR) def slot_status_icon(status: str) -> str: return {"active": "🟢", "hold": "🟡", "revoked": "🔴", "expired": "⚫"}.get(status, "⚪") def format_slot_row(row) -> dict: return { "slot_id": row[0], "user_id": row[1], "channel_id": row[2], "category_id": row[3], "key_code": row[4], "status": row[5], "ping_count": row[6], "ping_limit": row[7], "revoke_reason": row[8], "revoked_by": row[9], "revoked_at": row[10], "expires_at": row[11], "created_at": row[12] } class RevokeModal(ui.Modal, title="Revoke Slot"): reason = ui.TextInput(label="Reason", placeholder="Enter revocation reason...", max_length=300, required=False) def __init__(self, target: discord.Member, cog): super().__init__() self.target = target self.cog = cog async def on_submit(self, interaction: discord.Interaction): await self.cog._do_revoke(interaction, self.target, str(self.reason) or "No reason provided.") class RevokeSelectView(ui.View): def __init__(self, members: list, cog): super().__init__(timeout=60) options = [ discord.SelectOption(label=m.display_name[:25], value=str(m.id), description=f"ID: {m.id}") for m in members[:25] ] select = ui.Select(placeholder="Select a user to revoke...", options=options) select.callback = self.on_select self.add_item(select) self._members = {str(m.id): m for m in members} self.cog = cog async def on_select(self, interaction: discord.Interaction): member = self._members.get(interaction.data["values"][0]) if not member: await interaction.response.send_message(embed=err("Not Found", "Member not found."), ephemeral=True) return await interaction.response.send_modal(RevokeModal(member, self.cog)) class SlotManager(commands.Cog): def __init__(self, bot): self.bot = bot async def _do_revoke(self, ctx_or_interaction, target: discord.Member, reason: str): slot = await sdb.get_slot(target.id) if not slot: msg = embed=err("No Slot", f"{target.mention} does not have a slot.") elif slot[5] == "revoked": msg = warn("Already Revoked", f"{target.mention}'s slot is already revoked.") else: s = format_slot_row(slot) who = ctx_or_interaction.user.id if hasattr(ctx_or_interaction, "user") else ctx_or_interaction.author.id await sdb.update_slot(s["slot_id"], status="revoked", revoke_reason=reason, revoked_by=who, revoked_at=datetime.utcnow().isoformat()) msg = ok("Slot Revoked", f"{target.mention}'s slot has been revoked.\n**Reason:** {reason}") try: dm_embed = s_embed("⛔ Slot Revoked", f"Your slot in **{ctx_or_interaction.guild.name}** has been revoked.\n**Reason:** {reason}", ERROR_COLOR) await target.send(embed=dm_embed) except discord.Forbidden: pass if hasattr(ctx_or_interaction, "response"): await ctx_or_interaction.response.send_message(embed=msg) else: await ctx_or_interaction.send(embed=msg) @commands.command(name="myslot") async def myslot(self, ctx): slot = await sdb.get_slot(ctx.author.id) if not slot: await ctx.send(embed=err("No Slot", "You do not have a slot. Redeem a key with `+redeem `.")) return s = format_slot_row(slot) icon = slot_status_icon(s["status"]) embed = discord.Embed(title=f"{icon} Your Slot — #{s['slot_id']}", color=SLOT_COLOR) embed.set_thumbnail(url=ctx.author.display_avatar.url) embed.add_field(name="Status", value=f"`{s['status'].capitalize()}`", inline=True) embed.add_field(name="Pings", value=f"`{s['ping_count']} / {s['ping_limit']}`", inline=True) channel = ctx.guild.get_channel(s["channel_id"]) if s["channel_id"] else None embed.add_field(name="Channel", value=channel.mention if channel else "`Not assigned`", inline=True) embed.add_field(name="Key Used", value=f"`{s['key_code'] or 'Manual'}`", inline=True) embed.add_field(name="Expires", value=f"" if s["expires_at"] else "`Never`", inline=True) embed.add_field(name="Created", value=f"", inline=True) if s["status"] == "revoked" and s["revoke_reason"]: embed.add_field(name="Revoke Reason", value=s["revoke_reason"], inline=False) embed.set_footer(text="SlotManager Bot") await ctx.send(embed=embed) @commands.command(name="slotinfo") async def slotinfo(self, ctx, member: discord.Member): if not has_slot_mod(ctx.author): await ctx.send(embed=err("No Permission", "Staff only.")) return slot = await sdb.get_slot(member.id) if not slot: await ctx.send(embed=err("No Slot", f"{member.mention} does not have a slot.")) return s = format_slot_row(slot) icon = slot_status_icon(s["status"]) embed = discord.Embed(title=f"{icon} Slot Info — {member.display_name}", color=SLOT_COLOR) embed.set_thumbnail(url=member.display_avatar.url) embed.add_field(name="User", value=f"{member.mention} (`{member.id}`)", inline=False) embed.add_field(name="Slot ID", value=f"`#{s['slot_id']}`", inline=True) embed.add_field(name="Status", value=f"`{s['status'].capitalize()}`", inline=True) embed.add_field(name="Pings", value=f"`{s['ping_count']} / {s['ping_limit']}`", inline=True) channel = ctx.guild.get_channel(s["channel_id"]) if s["channel_id"] else None embed.add_field(name="Channel", value=channel.mention if channel else "`Not assigned`", inline=True) embed.add_field(name="Key Used", value=f"`{s['key_code'] or 'Manual'}`", inline=True) embed.add_field(name="Expires", value=f"" if s["expires_at"] else "`Never`", inline=True) embed.add_field(name="Created", value=f"`{s['created_at'][:16]}`", inline=True) if s["revoke_reason"]: embed.add_field(name="Revoke Reason", value=s["revoke_reason"], inline=False) if s["revoked_by"]: embed.add_field(name="Revoked By", value=f"<@{s['revoked_by']}>", inline=True) embed.set_footer(text="SlotManager Bot") await ctx.send(embed=embed) @commands.command(name="create") async def create_slot_cmd(self, ctx, member: discord.Member, duration: str = None, ping_limit: int = 10): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return existing = await sdb.get_slot(member.id) if existing and existing[5] == "active": await ctx.send(embed=warn("Already Has Slot", f"{member.mention} already has an active slot.")) return expires_at = None if duration: try: days = int(duration.rstrip("d").rstrip("D")) expires_at = (datetime.utcnow() + timedelta(days=days)).isoformat() except ValueError: await ctx.send(embed=err("Invalid Duration", "Use format like `30d` for 30 days.")) return slot_id = await sdb.create_slot(member.id, expires_at=expires_at, ping_limit=ping_limit) embed = ok("Slot Created", f"Slot `#{slot_id}` created for {member.mention}\n**Duration:** `{'Permanent' if not expires_at else duration}`\n**Ping Limit:** `{ping_limit}`") await ctx.send(embed=embed) try: dm_embed = s_embed("🎉 Slot Assigned", f"You have been assigned a slot in **{ctx.guild.name}**!\n**Slot ID:** `#{slot_id}`\n**Ping Limit:** `{ping_limit}`\n**Expires:** {'Never' if not expires_at else f''}", SUCCESS_COLOR) await member.send(embed=dm_embed) except discord.Forbidden: pass @commands.command(name="deleteslots") async def deleteslots(self, ctx, member: discord.Member): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return deleted = await sdb.delete_slot(member.id) if deleted: await ctx.send(embed=ok("Slot Deleted", f"Slot for {member.mention} has been deleted.")) else: await ctx.send(embed=err("No Slot", f"{member.mention} does not have a slot.")) @commands.command(name="hold") async def hold_slot(self, ctx, member: discord.Member, *, reason: str = "No reason provided"): if not has_slot_mod(ctx.author): await ctx.send(embed=err("No Permission", "Staff only.")) return slot = await sdb.get_slot(member.id) if not slot: await ctx.send(embed=err("No Slot", f"{member.mention} has no slot.")) return if slot[5] == "hold": await ctx.send(embed=warn("Already On Hold", f"{member.mention}'s slot is already on hold.")) return await sdb.update_slot(slot[0], status="hold", revoke_reason=reason) await ctx.send(embed=ok("Slot On Hold", f"{member.mention}'s slot is now on hold.\n**Reason:** {reason}")) try: await member.send(embed=warn("⏸️ Slot On Hold", f"Your slot in **{ctx.guild.name}** has been placed on hold.\n**Reason:** {reason}")) except discord.Forbidden: pass @commands.command(name="unhold") async def unhold_slot(self, ctx, member: discord.Member): if not has_slot_mod(ctx.author): await ctx.send(embed=err("No Permission", "Staff only.")) return slot = await sdb.get_slot(member.id) if not slot: await ctx.send(embed=err("No Slot", f"{member.mention} has no slot.")) return if slot[5] != "hold": await ctx.send(embed=warn("Not On Hold", f"{member.mention}'s slot is not currently on hold.")) return await sdb.update_slot(slot[0], status="active", revoke_reason=None) await ctx.send(embed=ok("Slot Resumed", f"{member.mention}'s slot has been taken off hold and is now active.")) try: await member.send(embed=ok("▶️ Slot Resumed", f"Your slot in **{ctx.guild.name}** has been resumed and is now active.")) except discord.Forbidden: pass @commands.command(name="revoke") async def revoke_slot(self, ctx, member: discord.Member, *, reason: str = "No reason provided"): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return await self._do_revoke(ctx, member, reason) @commands.command(name="unrevoke") async def unrevoke_slot(self, ctx, member: discord.Member): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return slot = await sdb.get_slot(member.id) if not slot: await ctx.send(embed=err("No Slot", f"{member.mention} has no slot.")) return if slot[5] != "revoked": await ctx.send(embed=warn("Not Revoked", f"{member.mention}'s slot is not currently revoked.")) return await sdb.update_slot(slot[0], status="active", revoke_reason=None, revoked_by=None, revoked_at=None) await ctx.send(embed=ok("Slot Restored", f"{member.mention}'s slot has been restored to active.")) try: await member.send(embed=ok("✅ Slot Restored", f"Your slot in **{ctx.guild.name}** has been restored and is now active.")) except discord.Forbidden: pass @commands.command(name="rpanel") async def rpanel(self, ctx): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return active_slots = await sdb.get_all_slots("active") if not active_slots: await ctx.send(embed=warn("No Active Slots", "There are no active slots to revoke.")) return members = [] for row in active_slots: m = ctx.guild.get_member(row[1]) if m: members.append(m) if not members: await ctx.send(embed=warn("No Members", "Could not resolve any members for active slots.")) return embed = s_embed("⛔ Revoke Panel", "Select a user from the dropdown to revoke their slot.") await ctx.send(embed=embed, view=RevokeSelectView(members[:25], self)) @commands.command(name="redeem") async def redeem_key(self, ctx, key_code: str): result = await sdb.redeem_key(key_code.upper(), ctx.author.id) if not result["success"]: await ctx.send(embed=err("Redemption Failed", result["reason"])) return expires_text = "Never" if result["expires_at"]: ts = int(datetime.fromisoformat(result["expires_at"]).timestamp()) expires_text = f"" embed = ok("🎉 Key Redeemed!", f"Your slot has been activated!\n\n**Slot ID:** `#{result['slot_id']}`\n**Ping Limit:** `{result['ping_limit']}`\n**Expires:** {expires_text}") await ctx.send(embed=embed) @commands.command(name="renew") async def renew_slot(self, ctx, member: discord.Member, duration: str = "30d"): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return slot = await sdb.get_slot(member.id) if not slot: await ctx.send(embed=err("No Slot", f"{member.mention} does not have a slot.")) return try: days = int(duration.rstrip("dD")) except ValueError: await ctx.send(embed=err("Invalid Duration", "Use format like `30d`.")) return if slot[11]: current_expiry = datetime.fromisoformat(slot[11]) base = max(current_expiry, datetime.utcnow()) else: base = datetime.utcnow() new_expiry = (base + timedelta(days=days)).isoformat() await sdb.update_slot(slot[0], expires_at=new_expiry, status="active") ts = int(datetime.fromisoformat(new_expiry).timestamp()) await ctx.send(embed=ok("Slot Renewed", f"{member.mention}'s slot renewed by `{days}` days.\n**New Expiry:** ")) try: await member.send(embed=ok("🔄 Slot Renewed", f"Your slot in **{ctx.guild.name}** has been renewed!\n**New Expiry:** ")) except discord.Forbidden: pass @commands.command(name="create_key") async def create_key_cmd(self, ctx, duration: str = None, ping_limit: int = 10, count: int = 1): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return duration_days = None if duration and duration.lower() != "perm": try: duration_days = int(duration.rstrip("dD")) except ValueError: await ctx.send(embed=err("Invalid Duration", "Use `perm` or `30d` format.")) return count = min(count, 20) keys = await sdb.create_key(ctx.author.id, duration_days, ping_limit, count) duration_text = "Permanent" if not duration_days else f"{duration_days} days" keys_text = "\n".join(f"`{k}`" for k in keys) embed = ok( f"{'Key' if count == 1 else f'{count} Keys'} Generated", f"**Duration:** `{duration_text}`\n**Ping Limit:** `{ping_limit}`\n\n**{'Key' if count == 1 else 'Keys'}:**\n{keys_text}" ) await ctx.send(embed=embed) @commands.command(name="delete_key") async def delete_key_cmd(self, ctx, key_code: str): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return deleted = await sdb.delete_key(key_code.upper()) if deleted: await ctx.send(embed=ok("Key Deleted", f"Key `{key_code.upper()}` has been deleted.")) else: await ctx.send(embed=err("Not Found", f"Key `{key_code.upper()}` not found.")) @commands.command(name="keys") async def list_keys(self, ctx): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return keys = await sdb.get_all_keys(active_only=True) if not keys: await ctx.send(embed=warn("No Keys", "There are no active (unredeemed) keys.")) return lines = [] for k in keys[:20]: dur = f"{k[3]}d" if k[3] else "Perm" lines.append(f"`{k[1]}` — `{dur}` | Pings: `{k[5]}` | By: <@{k[2]}>") embed = s_embed(f"🔑 Active Keys ({len(keys)})", "\n".join(lines)) if len(keys) > 20: embed.set_footer(text=f"Showing 20 of {len(keys)} keys • SlotManager Bot") await ctx.send(embed=embed) @commands.command(name="ausers") async def ausers(self, ctx): if not has_slot_mod(ctx.author): await ctx.send(embed=err("No Permission", "Staff only.")) return slots = await sdb.get_all_slots("active") if not slots: await ctx.send(embed=warn("No Active Users", "No users currently have active slots.")) return lines = [] for row in slots[:30]: s = format_slot_row(row) member = ctx.guild.get_member(s["user_id"]) name = member.display_name if member else f"User {s['user_id']}" expires = f"" if s["expires_at"] else "`Never`" lines.append(f"🟢 **{name}** — Slot `#{s['slot_id']}` | Expires: {expires}") embed = s_embed(f"✅ Active Slot Users ({len(slots)})", "\n".join(lines)) await ctx.send(embed=embed) @commands.command(name="pusers") async def pusers(self, ctx): if not has_slot_mod(ctx.author): await ctx.send(embed=err("No Permission", "Staff only.")) return slots = await sdb.get_all_slots("active") permanent = [s for s in slots if s[11] is None] if not permanent: await ctx.send(embed=warn("No Permanent Users", "No users have permanent slots.")) return lines = [] for row in permanent[:30]: s = format_slot_row(row) member = ctx.guild.get_member(s["user_id"]) name = member.display_name if member else f"User {s['user_id']}" lines.append(f"♾️ **{name}** — Slot `#{s['slot_id']}`") embed = s_embed(f"♾️ Permanent Slot Users ({len(permanent)})", "\n".join(lines)) await ctx.send(embed=embed) @commands.command(name="rusers") async def rusers(self, ctx): if not has_slot_mod(ctx.author): await ctx.send(embed=err("No Permission", "Staff only.")) return slots = await sdb.get_all_slots("revoked") if not slots: await ctx.send(embed=warn("No Revoked Users", "No users have revoked slots.")) return lines = [] for row in slots[:30]: s = format_slot_row(row) member = ctx.guild.get_member(s["user_id"]) name = member.display_name if member else f"User {s['user_id']}" reason = s["revoke_reason"] or "No reason" lines.append(f"🔴 **{name}** — `{reason[:40]}`") embed = s_embed(f"⛔ Revoked Slot Users ({len(slots)})", "\n".join(lines)) await ctx.send(embed=embed) @commands.command(name="pings") async def pings_cmd(self, ctx, member: discord.Member = None): target = member or ctx.author if target != ctx.author and not has_slot_mod(ctx.author): await ctx.send(embed=err("No Permission", "Staff only to view others' pings.")) return slot = await sdb.get_slot(target.id) if not slot: await ctx.send(embed=err("No Slot", f"{target.mention} does not have a slot.")) return s = format_slot_row(slot) remaining = s["ping_limit"] - s["ping_count"] bar_filled = int((s["ping_count"] / s["ping_limit"]) * 10) if s["ping_limit"] > 0 else 0 bar = "█" * bar_filled + "░" * (10 - bar_filled) embed = discord.Embed( title=f"🔔 Ping Usage — {target.display_name}", description=f"`{bar}` **{s['ping_count']}/{s['ping_limit']}**\n\n✅ Used: `{s['ping_count']}` | 🔲 Remaining: `{remaining}`", color=SLOT_COLOR ) embed.set_footer(text="SlotManager Bot") await ctx.send(embed=embed) @commands.command(name="setpings") async def setpings(self, ctx, member: discord.Member, limit: int): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return slot = await sdb.get_slot(member.id) if not slot: await ctx.send(embed=err("No Slot", f"{member.mention} does not have a slot.")) return await sdb.update_slot(slot[0], ping_limit=limit) await ctx.send(embed=ok("Ping Limit Set", f"{member.mention}'s ping limit updated to `{limit}`.")) @commands.command(name="resetping") async def resetping(self, ctx, member: discord.Member): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return slot = await sdb.get_slot(member.id) if not slot: await ctx.send(embed=err("No Slot", f"{member.mention} does not have a slot.")) return await sdb.reset_ping(member.id) await ctx.send(embed=ok("Ping Reset", f"{member.mention}'s ping count has been reset to `0`.")) @commands.command(name="resetpings") async def resetpings(self, ctx): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return await sdb.reset_all_pings() await ctx.send(embed=ok("All Pings Reset", "All users' ping counts have been reset to `0`.")) @commands.command(name="search") async def search_slots(self, ctx, *, query: str): if not has_slot_mod(ctx.author): await ctx.send(embed=err("No Permission", "Staff only.")) return results = await sdb.search_slots(query) if not results: await ctx.send(embed=warn("No Results", f"No slots found matching `{query}`.")) return lines = [] for row in results[:15]: s = format_slot_row(row) member = ctx.guild.get_member(s["user_id"]) name = member.display_name if member else f"User {s['user_id']}" icon = slot_status_icon(s["status"]) lines.append(f"{icon} **{name}** — Slot `#{s['slot_id']}` | `{s['status']}`") embed = s_embed(f"🔍 Search: `{query}` ({len(results)} result(s))", "\n".join(lines)) await ctx.send(embed=embed) @commands.command(name="nuke") async def nuke_channel(self, ctx, channel: discord.TextChannel = None, confirm: str = None): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return target = channel or ctx.channel if confirm != "CONFIRM": await ctx.send(embed=warn( "⚠️ Confirm Nuke", f"This will delete and recreate {target.mention}, clearing all messages.\n\n" f"Run `+nuke #{target.name} CONFIRM` to proceed." )) return position = target.position category = target.category overwrites = target.overwrites topic = target.topic name = target.name await target.delete(reason=f"Nuke by {ctx.author}") new_channel = await ctx.guild.create_text_channel( name, category=category, position=position, overwrites=overwrites, topic=topic ) await new_channel.send(embed=ok("💥 Channel Nuked", f"Channel has been nuked by {ctx.author.mention}.")) @commands.command(name="restoreslots") async def restoreslots(self, ctx): if not has_slot_admin(ctx.author): await ctx.send(embed=err("No Permission", "Admins only.")) return async with __import__("aiosqlite").connect(sdb.SLOT_DB_PATH) as _db: await _db.execute("UPDATE slots SET status='active', revoke_reason=NULL, revoked_by=NULL, revoked_at=NULL WHERE status='revoked'") changes = _db.total_changes await _db.commit() await ctx.send(embed=ok("Slots Restored", f"All revoked slots have been restored to active.")) @commands.command(name="help") async def help_cmd(self, ctx): embed = discord.Embed( title="📋 SlotManager Commands", description=f"Prefix: `+` | All available commands listed below", color=SLOT_COLOR ) embed.add_field(name="👤 User Commands", value=( "`myslot` — View your slot info\n" "`pings` — View your ping usage\n" "`redeem ` — Redeem a key for a slot" ), inline=False) embed.add_field(name="🛡️ Staff Commands", value=( "`slotinfo @user` — Detailed slot info\n" "`ausers` — List active slot users\n" "`pusers` — List permanent slot users\n" "`rusers` — List revoked users\n" "`pings @user` — View a user's ping usage\n" "`hold @user [reason]` — Put slot on hold\n" "`unhold @user` — Release hold\n" "`search ` — Search slots" ), inline=False) embed.add_field(name="⚙️ Admin Commands", value=( "`create @user [duration] [ping_limit]` — Create slot\n" "`deleteslots @user` — Delete a slot\n" "`revoke @user [reason]` — Revoke slot\n" "`unrevoke @user` — Restore slot\n" "`rpanel` — Interactive revoke panel\n" "`renew @user [duration]` — Renew slot\n" "`create_key [duration] [ping_limit] [count]` — Generate keys\n" "`delete_key

` — Delete a key\n"
            "`keys` — List all active keys\n"
            "`setpings @user ` — Set ping limit\n"
            "`resetping @user` — Reset user ping count\n"
            "`resetpings` — Reset ALL ping counts\n"
            "`restoreslots` — Restore all revoked slots\n"
            "`nuke [#channel] CONFIRM` — Nuke a channel"
        ), inline=False)
        embed.set_footer(text="SlotManager Bot • Duration format: 30d | perm = permanent")
        await ctx.send(embed=embed)


async def setup(bot):
    await bot.add_cog(SlotManager(bot))