""" Admin cog: Moderation and server management commands. Enhanced with rich emoji decorations and beautiful formatting. """ import datetime as dt import hashlib import json import random import re from typing import Optional import aiohttp import discord from discord.ext import commands from bot.theme import ( fancy_header, NEON_CYAN, NEON_PINK, NEON_PURPLE, NEON_LIME, NEON_ORANGE, NEON_RED, NEON_BLUE, NEON_YELLOW, panel_divider, success_embed, error_embed, warning_embed, info_embed, double_line, triple_line, shimmer, pick_neon_color, add_banner_to_embed ) from bot.i18n import get_cmd_desc from bot.emojis import ( ui, E_SHIELD, E_CROWN, E_TROPHY, E_FIRE, E_SPARKLE, E_LOCK, E_KEY, E_ARROW_BLUE, E_ARROW_GREEN, E_ARROW_PINK, E_ARROW_PURPLE, E_GEM, E_STAR ) # ═══════════════════════════════════════════════════════════════════════════════ # AWESOME ROLES DEFINITIONS # ═══════════════════════════════════════════════════════════════════════════════ AWESOME_ROLES = [ { "name": "✨ Cyan Legend", "color": NEON_CYAN, "permissions": ["view_channel", "send_messages", "embed_links", "attach_files", "read_message_history"], "hoist": True, "description": "Elite member with cyan glow" }, { "name": "<:animatedarrowpink:1477261266690113651> Pink Master", "color": NEON_PINK, "permissions": ["view_channel", "send_messages", "embed_links", "attach_files", "read_message_history", "add_reactions"], "hoist": True, "description": "Creative member with pink style" }, { "name": "<:animatedarrowgreen:1477261279428087979> Purple Elite", "color": NEON_PURPLE, "permissions": ["view_channel", "send_messages", "embed_links", "attach_files", "read_message_history", "use_external_emojis"], "hoist": True, "description": "Premium member with purple flair" }, { "name": "<:animatedarrowgreen:1477261279428087979> Green Champion", "color": NEON_LIME, "permissions": ["view_channel", "send_messages", "embed_links", "attach_files", "read_message_history", "stream"], "hoist": True, "description": "Active member with green energy" }, { "name": "🧡 Orange Hero", "color": NEON_ORANGE, "permissions": ["view_channel", "send_messages", "embed_links", "attach_files", "read_message_history", "connect", "speak"], "hoist": True, "description": "Helpful member with orange vibe" }, { "name": "💛 Golden Star", "color": NEON_YELLOW, "permissions": ["view_channel", "send_messages", "embed_links", "attach_files", "read_message_history", "change_nickname"], "hoist": True, "description": "Valued member with golden touch" }, { "name": "💙 Azure Guardian", "color": NEON_BLUE, "permissions": ["view_channel", "send_messages", "embed_links", "attach_files", "read_message_history", "manage_nicknames"], "hoist": True, "description": "Trusted member with azure power" }, ] PRESENCE_STATUS_MAP: dict[str, discord.Status] = { "online": discord.Status.online, "idle": discord.Status.idle, "dnd": discord.Status.dnd, "invisible": discord.Status.invisible, "offline": discord.Status.invisible, } PRESENCE_ACTIVITY_MAP: dict[str, discord.ActivityType] = { "playing": discord.ActivityType.playing, "watching": discord.ActivityType.watching, "listening": discord.ActivityType.listening, "competing": discord.ActivityType.competing, } class AdminMasterPanel(discord.ui.View): def __init__(self, cog: "Admin") -> None: super().__init__(timeout=None) self.cog = cog @discord.ui.select( placeholder="Admin Panel | لوحة الإدارة", min_values=1, max_values=1, custom_id="admin_master_panel_select", options=[ discord.SelectOption(label="💰 Economy Admin | إدارة الاقتصاد", value="economy"), discord.SelectOption(label="🛡️ Shield Control | تحكم الدرع", value="shield"), discord.SelectOption(label="📊 System Status | حالة النظام", value="status"), discord.SelectOption(label="🎁 Giveaways | القيافاوي", value="giveaways"), discord.SelectOption(label="🏆 Tournaments | البطولات", value="tournaments"), discord.SelectOption(label="📊 Polls | التصويتات", value="polls"), discord.SelectOption(label="🎫 Tickets | التذاكر", value="tickets"), discord.SelectOption(label="🤖 AI Admin | الذكاء الاصطناعي", value="ai_admin"), discord.SelectOption(label="🔧 Server Config | الإعدادات", value="config"), ], ) async def select_action(self, interaction: discord.Interaction, select: discord.ui.Select) -> None: choice = select.values[0] if choice == "economy": await interaction.response.edit_message( embed=await self.cog.build_economy_admin_embed(interaction.guild), view=EconomyAdminPanel(self.cog), ) return if choice == "shield": events_cog = self.cog.bot.get_cog("Events") if events_cog is None: await interaction.response.send_message("❌ Events cog not loaded.", ephemeral=True) return from bot.cogs.events import ShieldControlPanel panel = ShieldControlPanel(events_cog, interaction.guild.id) embed = await panel._build_embed() await interaction.response.edit_message(embed=embed, view=panel) return if choice == "status": synced = "Enabled" if self.cog.bot.db._hf_sync_enabled else "Disabled" activity_obj = getattr(self.cog.bot, "activity", None) activity_name = getattr(activity_obj, "name", None) or "CYBER // GRID" activity_type = str(getattr(activity_obj, "type", "playing")).replace("ActivityType.", "") status_name = str(getattr(self.cog.bot, "status", "online")).replace("Status.", "") await interaction.response.send_message( ( f"HF DB Sync: **{synced}**\n" f"Bot Status: **{status_name}**\n" f"Bot Activity: **{activity_type}** • **{activity_name}**\n\n" "Owner commands:\n" "`/set_bot_status `\n" "`/reset_bot_status`\n" "`/bot_status`" ), ephemeral=True, ) return if choice == "giveaways": await interaction.response.send_message( "**Giveaway Commands:**\n" "`/giveaway start` - Start a new giveaway\n" "`/giveaway end ` - End a giveaway\n" "`/giveaway reroll ` - Reroll a giveaway", ephemeral=True ) return if choice == "tournaments": await interaction.response.send_message( "**Tournament Commands:**\n" "`/tournament create` - Create a tournament\n" "`/tournament join` - Join a tournament\n" "`/tournament start` - Start a tournament\n" "`/tournament end` - End a tournament", ephemeral=True ) return if choice == "polls": await interaction.response.send_message( "**Poll Commands:**\n" "`/poll create` - Create a poll\n" "`/poll end ` - End a poll\n" "`/poll results ` - View poll results", ephemeral=True ) return if choice == "tickets": await interaction.response.send_message( "**Ticket Commands:**\n" "`/ticket_panel setup` - Setup ticket panel\n" "`/ticket close` - Close current ticket\n" "`/ticket delete` - Delete current ticket", ephemeral=True ) return if choice == "ai_admin": await interaction.response.send_message( "**AI Admin Commands:**\n" "`/ai_admin ` - Let AI manage the server\n" "`/ai_help` - Show AI capabilities\n\n" "Examples:\n" "- Create a giveaway for Nitro\n" "- Setup a Valorant tournament\n" "- Create a moderator role", ephemeral=True ) return if choice == "config": await interaction.response.send_message( "**Config Commands:**\n" "`/config panel` - Open config panel\n" "`/language set` - Set server language\n" "`/prefix set` - Set bot prefix", ephemeral=True ) return class EconomyAdminModalBase(discord.ui.Modal): member_id = discord.ui.TextInput( label="User ID / Mention", placeholder="123456789012345678 or @user", required=True, max_length=32, ) def __init__(self, cog: "Admin", title: str) -> None: super().__init__(title=title) self.cog = cog async def _resolve_member_or_reply(self, interaction: discord.Interaction) -> discord.Member | None: guild = interaction.guild if guild is None: await interaction.response.send_message("This action must be used in a server.", ephemeral=True) return None member = await self.cog.resolve_member(guild, str(self.member_id)) if member is None: await interaction.response.send_message("Could not find that member.", ephemeral=True) return None return member class AddCoinsModal(EconomyAdminModalBase): amount = discord.ui.TextInput(label="Coins To Add", placeholder="250", required=True, max_length=12) def __init__(self, cog: "Admin") -> None: super().__init__(cog, "➕ Add Coins") async def on_submit(self, interaction: discord.Interaction) -> None: member = await self._resolve_member_or_reply(interaction) if member is None: return amount = self.cog.parse_positive_int(str(self.amount)) if amount is None: await interaction.response.send_message("Amount must be a positive number.", ephemeral=True) return before, after = await self.cog.adjust_wallet(interaction.guild.id, member.id, amount) await interaction.response.send_message( f"✅ Added `{amount}` coins for {member.mention}. Wallet `{before}` → `{after}`.", ephemeral=True, ) class RemoveCoinsModal(EconomyAdminModalBase): amount = discord.ui.TextInput(label="Coins To Remove", placeholder="150", required=True, max_length=12) def __init__(self, cog: "Admin") -> None: super().__init__(cog, "➖ Remove Coins") async def on_submit(self, interaction: discord.Interaction) -> None: member = await self._resolve_member_or_reply(interaction) if member is None: return amount = self.cog.parse_positive_int(str(self.amount)) if amount is None: await interaction.response.send_message("Amount must be a positive number.", ephemeral=True) return before, after = await self.cog.adjust_wallet(interaction.guild.id, member.id, -amount) await interaction.response.send_message( f"✅ Removed `{amount}` coins for {member.mention}. Wallet `{before}` → `{after}`.", ephemeral=True, ) class SetBalanceModal(EconomyAdminModalBase): amount = discord.ui.TextInput(label="New Wallet Balance", placeholder="5000", required=True, max_length=12) def __init__(self, cog: "Admin") -> None: super().__init__(cog, "⚖️ Set Balance") async def on_submit(self, interaction: discord.Interaction) -> None: member = await self._resolve_member_or_reply(interaction) if member is None: return target = self.cog.parse_non_negative_int(str(self.amount)) if target is None: await interaction.response.send_message("Balance must be zero or more.", ephemeral=True) return before, after = await self.cog.set_wallet(interaction.guild.id, member.id, target) await interaction.response.send_message( f"✅ Set wallet for {member.mention}. `{before}` → `{after}`.", ephemeral=True, ) class SetSalaryModal(discord.ui.Modal): min_salary = discord.ui.TextInput(label="Minimum Work Salary", placeholder="50", required=True, max_length=12) max_salary = discord.ui.TextInput(label="Maximum Work Salary", placeholder="150", required=True, max_length=12) def __init__(self, cog: "Admin") -> None: super().__init__(title="💸 Set Salary") self.cog = cog async def on_submit(self, interaction: discord.Interaction) -> None: if interaction.guild is None: await interaction.response.send_message("This action must be used in a server.", ephemeral=True) return min_val = self.cog.parse_non_negative_int(str(self.min_salary)) max_val = self.cog.parse_non_negative_int(str(self.max_salary)) if min_val is None or max_val is None or min_val > max_val: await interaction.response.send_message("Use valid integers with min <= max.", ephemeral=True) return await self.cog.bot.db.execute( "INSERT INTO economy_salaries(guild_id, min_salary, max_salary) VALUES (?, ?, ?) " "ON CONFLICT(guild_id) DO UPDATE SET min_salary = excluded.min_salary, max_salary = excluded.max_salary", interaction.guild.id, min_val, max_val, ) await interaction.response.send_message( f"✅ Work salary updated to `{min_val}` - `{max_val}`.", ephemeral=True, ) class SetDailyModal(discord.ui.Modal): daily_min = discord.ui.TextInput(label="Minimum Daily Reward", placeholder="100", required=True, max_length=12) daily_max = discord.ui.TextInput(label="Maximum Daily Reward", placeholder="250", required=True, max_length=12) def __init__(self, cog: "Admin") -> None: super().__init__(title="📆 Set Daily") self.cog = cog async def on_submit(self, interaction: discord.Interaction) -> None: if interaction.guild is None: await interaction.response.send_message("This action must be used in a server.", ephemeral=True) return min_val = self.cog.parse_non_negative_int(str(self.daily_min)) max_val = self.cog.parse_non_negative_int(str(self.daily_max)) if min_val is None or max_val is None or min_val > max_val: await interaction.response.send_message("Use valid integers with min <= max.", ephemeral=True) return await self.cog.bot.db.execute( "INSERT INTO economy_salaries(guild_id, daily_min, daily_max) VALUES (?, ?, ?) " "ON CONFLICT(guild_id) DO UPDATE SET daily_min = excluded.daily_min, daily_max = excluded.daily_max", interaction.guild.id, min_val, max_val, ) await interaction.response.send_message( f"✅ Daily reward updated to `{min_val}` - `{max_val}`.", ephemeral=True, ) class EconomyAdminPanel(discord.ui.View): def __init__(self, cog: "Admin") -> None: super().__init__(timeout=300) self.cog = cog @discord.ui.button(label="➕ Add Coins", style=discord.ButtonStyle.success, row=0) async def add_coins(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: await interaction.response.send_modal(AddCoinsModal(self.cog)) @discord.ui.button(label="💸 Set Salary", style=discord.ButtonStyle.primary, row=0) async def set_salary(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: await interaction.response.send_modal(SetSalaryModal(self.cog)) @discord.ui.button(label="➖ Remove", style=discord.ButtonStyle.danger, row=1) async def remove_coins(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: await interaction.response.send_modal(RemoveCoinsModal(self.cog)) @discord.ui.button(label="📆 Set Daily", style=discord.ButtonStyle.secondary, row=1) async def set_daily(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: await interaction.response.send_modal(SetDailyModal(self.cog)) @discord.ui.button(label="⚖️ Set Balance", style=discord.ButtonStyle.primary, row=2) async def set_balance(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: await interaction.response.send_modal(SetBalanceModal(self.cog)) @discord.ui.button(label="⬅️ Back", style=discord.ButtonStyle.secondary, row=2) async def back(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: await interaction.response.edit_message(embed=await self.cog.build_admin_home_embed(interaction.guild), view=AdminMasterPanel(self.cog)) class EmojiCloneModal(discord.ui.Modal): source = discord.ui.TextInput( label="Source Emoji (tag/id/url/name)", placeholder="<:emoji:123...> or 123... or https://... or emoji_name", required=True, max_length=500, ) target_name = discord.ui.TextInput( label="New Name (optional)", placeholder="leave empty to auto-detect", required=False, max_length=32, ) def __init__(self, cog: "Admin") -> None: super().__init__(title="Clone Emoji") self.cog = cog async def on_submit(self, interaction: discord.Interaction) -> None: if not interaction.guild: await interaction.response.send_message("Server only.", ephemeral=True) return if not interaction.user.guild_permissions.manage_emojis: await interaction.response.send_message("Manage Emojis permission required.", ephemeral=True) return source = str(self.source.value).strip() name = str(self.target_name.value).strip() or None created, error_text = await self.cog._clone_emoji_to_guild( interaction.guild, interaction.user, source=source, name=name, ) if error_text: await interaction.response.send_message(error_text, ephemeral=True) return embed = success_embed( "Emoji Cloned", f"Created {created} as `:{created.name}:`", ) await interaction.response.send_message(embed=embed, ephemeral=True) class EmojiClonePanelView(discord.ui.View): def __init__(self, cog: "Admin") -> None: super().__init__(timeout=300) self.cog = cog async def _ensure_perm(self, interaction: discord.Interaction) -> bool: if not interaction.guild: await interaction.response.send_message("Server only.", ephemeral=True) return False if not interaction.user.guild_permissions.manage_emojis: await interaction.response.send_message("Manage Emojis permission required.", ephemeral=True) return False return True @discord.ui.button(label="Clone Emoji", style=discord.ButtonStyle.success, emoji="🧬", row=0) async def clone_emoji(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: if not await self._ensure_perm(interaction): return await interaction.response.send_modal(EmojiCloneModal(self.cog)) @discord.ui.button(label="Server Emojis", style=discord.ButtonStyle.primary, emoji="📋", row=0) async def list_server_emojis(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: if not await self._ensure_perm(interaction): return emojis = interaction.guild.emojis if interaction.guild else [] if not emojis: await interaction.response.send_message("No custom emojis in this server.", ephemeral=True) return lines = [f"{e} `:{e.name}:` `{str(e)}`" for e in emojis[:60]] embed = info_embed("Server Emoji List", "\n".join(lines)) await interaction.response.send_message(embed=embed, ephemeral=True) @discord.ui.button(label="Bot Emoji Picker", style=discord.ButtonStyle.secondary, emoji="🧩", row=1) async def picker(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: if not await self._ensure_perm(interaction): return custom_names = sorted({emoji.name for emoji in getattr(self.cog.bot, "emojis", []) if getattr(emoji, "name", None)}) if not custom_names: await interaction.response.send_message("No bot emoji catalog available.", ephemeral=True) return lines = [] for name in custom_names[:60]: emoji_obj = discord.utils.get(self.cog.bot.emojis, name=name) rendered = str(emoji_obj) if emoji_obj else name lines.append(f"{rendered} `{name}`") embed = info_embed( "Emoji Picker", "\n".join(lines) + "\n\nUse Clone Emoji button and paste any name/tag/URL/ID.", ) await interaction.response.send_message(embed=embed, ephemeral=True) @discord.ui.button(label="Close", style=discord.ButtonStyle.danger, emoji="✖", row=1) async def close_panel(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: if not await self._ensure_perm(interaction): return for item in self.children: item.disabled = True await interaction.response.edit_message(view=self) class Admin(commands.Cog): """Moderation commands with beautiful panels.""" def __init__(self, bot: commands.Bot) -> None: self.bot = bot async def cog_load(self) -> None: self.bot.add_view(AdminMasterPanel(self)) async def build_admin_home_embed(self, guild: discord.Guild | None = None) -> discord.Embed: embed = success_embed( "꧁⫷ 𝕄𝕠تآز 𝕊𝕪𝕤𝕥𝕖𝕞 ⫸꧂", "╔════╗\n💰 Economy Admin | إدارة الاقتصاد\n🛡️ Shield Control | تحكم الدرع\n📊 System Status | حالة النظام\n╚════╝", ) embed.set_footer(text="🏮 Powered by BOT- AI Suite 🏮") if guild: await add_banner_to_embed(embed, guild) return embed async def build_economy_admin_embed(self, guild: discord.Guild | None = None) -> discord.Embed: embed = info_embed( f"{ui('economy')} Economy Admin", "Choose one action below:\n" "➕ Add Coins\n" "💸 Set Salary\n" "➖ Remove Coins\n" "📆 Set Daily\n" "⚖️ Set Balance", ) if guild: await add_banner_to_embed(embed, guild) return embed @staticmethod def parse_positive_int(value: str) -> int | None: try: parsed = int(value.strip()) except (TypeError, ValueError): return None return parsed if parsed > 0 else None @staticmethod def parse_non_negative_int(value: str) -> int | None: try: parsed = int(value.strip()) except (TypeError, ValueError): return None return parsed if parsed >= 0 else None async def resolve_member(self, guild: discord.Guild, raw: str) -> discord.Member | None: match = re.search(r"\d{15,22}", raw or "") if not match: return None member_id = int(match.group(0)) member = guild.get_member(member_id) if member is not None: return member try: return await guild.fetch_member(member_id) except discord.HTTPException: return None async def get_wallet(self, guild_id: int, user_id: int) -> int: row = await self.bot.db.fetchone( "SELECT wallet FROM user_balance WHERE guild_id = ? AND user_id = ?", guild_id, user_id, ) return int(row[0]) if row else 0 async def adjust_wallet(self, guild_id: int, user_id: int, delta: int) -> tuple[int, int]: before = await self.get_wallet(guild_id, user_id) after = max(0, before + delta) await self.bot.db.execute( "INSERT INTO user_balance(guild_id, user_id, wallet, bank) VALUES (?, ?, ?, 0) " "ON CONFLICT(guild_id, user_id) DO UPDATE SET wallet = excluded.wallet", guild_id, user_id, after, ) return before, after async def set_wallet(self, guild_id: int, user_id: int, target: int) -> tuple[int, int]: before = await self.get_wallet(guild_id, user_id) after = max(0, target) await self.bot.db.execute( "INSERT INTO user_balance(guild_id, user_id, wallet, bank) VALUES (?, ?, ?, 0) " "ON CONFLICT(guild_id, user_id) DO UPDATE SET wallet = excluded.wallet", guild_id, user_id, after, ) return before, after async def _safe_ctx_send( self, ctx: commands.Context, *, content: str | None = None, embed: discord.Embed | None = None, view: discord.ui.View | None = None, ephemeral: bool = False, delete_after: float | None = None, ) -> None: kwargs: dict[str, object] = {} if content is not None: kwargs["content"] = content if embed is not None: kwargs["embed"] = embed if view is not None: kwargs["view"] = view kwargs["ephemeral"] = ephemeral if ctx.interaction: try: if not ctx.interaction.response.is_done(): await ctx.interaction.response.send_message(**kwargs) return await ctx.interaction.followup.send(**kwargs) return except (discord.NotFound, discord.InteractionResponded): pass except discord.HTTPException as exc: if exc.code not in {10062, 40060, 10008}: raise if ctx.channel: channel_kwargs = {k: v for k, v in kwargs.items() if k != "ephemeral"} msg = await ctx.channel.send(**channel_kwargs) if delete_after: await msg.delete(delay=delete_after) def _presence_status_label(self, status: discord.Status | str | None) -> str: value = str(status or "online").lower() return { "online": "online", "idle": "idle", "dnd": "dnd", "invisible": "invisible", "offline": "invisible", }.get(value, "online") def _presence_activity_label(self, activity_type: discord.ActivityType | str | None) -> str: value = str(activity_type or "playing").lower() return { "playing": "playing", "watching": "watching", "listening": "listening", "competing": "competing", }.get(value, "playing") async def _set_bot_presence( self, actor_id: int, *, status_name: str, activity_type_name: str, activity_text: str, ) -> None: resolved_status = PRESENCE_STATUS_MAP.get(status_name, discord.Status.online) resolved_activity_type = PRESENCE_ACTIVITY_MAP.get(activity_type_name, discord.ActivityType.playing) await self.bot.change_presence( status=resolved_status, activity=discord.Activity(type=resolved_activity_type, name=activity_text), ) await self.bot.db.execute( "INSERT INTO bot_presence_config(id, status, activity_type, activity_text, updated_by, updated_at) " "VALUES (1, ?, ?, ?, ?, ?) " "ON CONFLICT(id) DO UPDATE SET " "status = excluded.status, " "activity_type = excluded.activity_type, " "activity_text = excluded.activity_text, " "updated_by = excluded.updated_by, " "updated_at = excluded.updated_at", status_name, activity_type_name, activity_text, actor_id, dt.datetime.utcnow().isoformat(), ) async def _presence_embed(self) -> discord.Embed: current_activity = getattr(self.bot, "activity", None) activity_text = getattr(current_activity, "name", None) or "CYBER // GRID" activity_type_name = self._presence_activity_label(getattr(current_activity, "type", None)) status_name = self._presence_status_label(getattr(self.bot, "status", None)) row = await self.bot.db.fetchone( "SELECT updated_by, updated_at FROM bot_presence_config WHERE id = 1" ) updated_by = f"<@{row[0]}>" if row and row[0] else "Unknown" updated_at = row[1] if row and row[1] else "Not saved yet" embed = info_embed( "Bot Presence", ( f"Status: **{status_name}**\n" f"Activity: **{activity_type_name}**\n" f"Text: **{activity_text}**\n\n" f"Updated by: {updated_by}\n" f"Updated at: `{updated_at}`" ), ) return embed @commands.hybrid_command(name="purge", description=get_cmd_desc("commands.admin.purge_desc")) @commands.has_permissions(manage_messages=True) async def purge(self, ctx: commands.Context, amount: int) -> None: """Delete a number of messages from the channel.""" if ctx.interaction and not ctx.interaction.response.is_done(): await ctx.interaction.response.defer(ephemeral=True, thinking=True) guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) amount = max(1, min(amount, 100)) deleted = await ctx.channel.purge(limit=amount + 1) if lang == "ar": desc = f"📝 تم حذف **{len(deleted) - 1}** رسالة.\n👤 المشرف: {ctx.author.mention}" else: desc = f"📝 Deleted **{len(deleted) - 1}** messages.\n👤 Moderator: {ctx.author.mention}" embed = success_embed("🗑️ Messages Purged", f"{panel_divider('green')}\n{desc}\n{panel_divider('green')}") await self._safe_ctx_send(ctx, embed=embed, ephemeral=bool(ctx.interaction), delete_after=5 if not ctx.interaction else None) @commands.hybrid_command(name="admin_panel", description=get_cmd_desc("commands.admin.admin_panel_desc"), with_app_command=False) @commands.has_permissions(administrator=True) async def admin_panel(self, ctx: commands.Context) -> None: if ctx.interaction and not ctx.interaction.response.is_done(): await ctx.interaction.response.defer(ephemeral=True, thinking=True) await self._safe_ctx_send( ctx, embed=await self.build_admin_home_embed(ctx.guild), view=AdminMasterPanel(self), ephemeral=bool(ctx.interaction), ) @commands.hybrid_group(name="admin", fallback="panel", description="Administrative grouped controls") @commands.has_permissions(administrator=True) async def admin_group(self, ctx: commands.Context) -> None: await self.admin_panel(ctx) @admin_group.group(name="botstatus", invoke_without_command=True) @commands.is_owner() async def admin_botstatus_group(self, ctx: commands.Context) -> None: embed = await self._presence_embed() await self._safe_ctx_send(ctx, embed=embed, ephemeral=bool(ctx.interaction)) @admin_botstatus_group.command(name="set") @commands.is_owner() async def admin_botstatus_set( self, ctx: commands.Context, status: str, activity_type: str, *, text: str, ) -> None: status_name = (status or "").strip().lower() activity_type_name = (activity_type or "").strip().lower() activity_text = (text or "").strip()[:128] if status_name not in PRESENCE_STATUS_MAP: await self._safe_ctx_send( ctx, content="Invalid status. Use: `online`, `idle`, `dnd`, `invisible`", ephemeral=bool(ctx.interaction), ) return if activity_type_name not in PRESENCE_ACTIVITY_MAP: await self._safe_ctx_send( ctx, content="Invalid activity type. Use: `playing`, `watching`, `listening`, `competing`", ephemeral=bool(ctx.interaction), ) return if not activity_text: await self._safe_ctx_send(ctx, content="Activity text cannot be empty.", ephemeral=bool(ctx.interaction)) return await self._set_bot_presence( ctx.author.id, status_name=status_name, activity_type_name=activity_type_name, activity_text=activity_text, ) embed = success_embed( "Bot Status Updated", ( f"Status: **{status_name}**\n" f"Activity: **{activity_type_name}**\n" f"Text: **{activity_text}**" ), ) await self._safe_ctx_send(ctx, embed=embed, ephemeral=bool(ctx.interaction)) @admin_botstatus_group.command(name="reset") @commands.is_owner() async def admin_botstatus_reset(self, ctx: commands.Context) -> None: await self._set_bot_presence( ctx.author.id, status_name="online", activity_type_name="playing", activity_text="CYBER // GRID", ) embed = success_embed("Bot Status Reset", "Restored default presence.") await self._safe_ctx_send(ctx, embed=embed, ephemeral=bool(ctx.interaction)) @commands.hybrid_command(name="bot_status", description=get_cmd_desc("commands.admin.bot_status_desc"), with_app_command=True) @commands.is_owner() async def show_bot_status(self, ctx: commands.Context) -> None: embed = await self._presence_embed() await self._safe_ctx_send(ctx, embed=embed, ephemeral=bool(ctx.interaction)) @commands.hybrid_command(name="set_bot_status", description=get_cmd_desc("commands.admin.set_bot_status_desc"), with_app_command=True) @commands.is_owner() async def update_bot_status( self, ctx: commands.Context, status: str, activity_type: str, *, text: str, ) -> None: await self.admin_botstatus_set(ctx, status, activity_type, text=text) @commands.hybrid_command(name="reset_bot_status", description=get_cmd_desc("commands.admin.reset_bot_status_desc"), with_app_command=True) @commands.is_owner() async def restore_bot_status(self, ctx: commands.Context) -> None: await self.admin_botstatus_reset(ctx) @admin_group.group(name="shield", invoke_without_command=True) @commands.has_permissions(administrator=True) async def admin_shield_group(self, ctx: commands.Context) -> None: await self._safe_ctx_send( ctx, content="Use `/admin shield state`, `/admin shield set_level `, or `/admin shield add_image `", ephemeral=bool(ctx.interaction), ) @admin_shield_group.command(name="set_level") @commands.has_permissions(administrator=True) async def admin_shield_set_level(self, ctx: commands.Context, level: str) -> None: await self.shield_level(ctx, level) @admin_shield_group.command(name="state") @commands.has_permissions(administrator=True) async def admin_shield_state(self, ctx: commands.Context) -> None: await self.shield_state(ctx) @admin_shield_group.command(name="add_image") @commands.has_permissions(manage_messages=True) async def admin_shield_add_image(self, ctx: commands.Context, image: discord.Attachment) -> None: if not ctx.guild: await self._safe_ctx_send(ctx, content="Server only.", ephemeral=bool(ctx.interaction)) return if not (image.content_type or "").startswith("image/"): await self._safe_ctx_send(ctx, content="Please attach an image file.", ephemeral=bool(ctx.interaction)) return digest = hashlib.sha256(await image.read()).hexdigest() await self.bot.db.execute( "INSERT OR IGNORE INTO scam_images(guild_id, image_hash, created_by) VALUES (?, ?, ?)", ctx.guild.id, digest, ctx.author.id, ) await self._safe_ctx_send(ctx, content="✅ Scam image signature saved.", ephemeral=bool(ctx.interaction)) @commands.hybrid_command(name="shield_level", description=get_cmd_desc("commands.admin.shield_level_desc"), hidden=True, with_app_command=False) @commands.has_permissions(administrator=True) async def shield_level(self, ctx: commands.Context, level: str) -> None: if ctx.interaction and not ctx.interaction.response.is_done(): await ctx.interaction.response.defer(ephemeral=True, thinking=True) normalized = level.strip().lower() if normalized not in {"low", "medium", "high"}: await self._safe_ctx_send(ctx, content="Use one of: low, medium, high", ephemeral=bool(ctx.interaction)) return await self.bot.db.execute( "INSERT INTO shield_settings(guild_id, level) VALUES (?, ?) ON CONFLICT(guild_id) DO UPDATE SET level = excluded.level", ctx.guild.id if ctx.guild else 0, normalized, ) await self._safe_ctx_send(ctx, content=f"🛡️ Shield level set to `{normalized}`", ephemeral=bool(ctx.interaction)) @commands.hybrid_command(name="shield_state", description=get_cmd_desc("commands.admin.shield_state_desc"), with_app_command=True) @commands.has_permissions(administrator=True) async def shield_state(self, ctx: commands.Context) -> None: if not ctx.guild: await self._safe_ctx_send(ctx, content="Server only.", ephemeral=bool(ctx.interaction)) return row = await self.bot.db.fetchone( "SELECT level FROM shield_settings WHERE guild_id = ?", ctx.guild.id, ) level = str(row[0]).strip().lower() if row and row[0] else "medium" if level not in {"low", "medium", "high"}: level = "medium" if level == "high": profile = "Very strict. Fast context checks, highest sensitivity, strongest anti-scam posture." elif level == "low": profile = "Relaxed. Fewer interventions and lower sensitivity." else: profile = "Balanced. Good protection with moderate sensitivity." embed = info_embed( "🛡️ Shield State", f"Current level: **`{level}`**\n" f"Profile: {profile}\n\n" "Use `/shield_level low|medium|high` to change it.\n" "Use `/shield_restrict ` to add custom restrictions.\n" "Use `/shield_keywords ` to add strict trigger keywords." ) await self._safe_ctx_send(ctx, embed=embed, ephemeral=bool(ctx.interaction)) @commands.hybrid_command(name="shield_restrict", description="Set custom AI shield restrictions") @commands.has_permissions(manage_guild=True) async def shield_restrict(self, ctx: commands.Context, *, restrictions: str) -> None: """Tell the AI shield what to be strict on. Example: 'no crypto links, no dm requests'.""" if not ctx.guild: await ctx.send("Server only.", ephemeral=True) return await self.bot.db.execute( "INSERT INTO shield_settings(guild_id, level, custom_restrictions) VALUES (?, ?, ?) " "ON CONFLICT(guild_id) DO UPDATE SET custom_restrictions = excluded.custom_restrictions", ctx.guild.id, "medium", restrictions[:500], ) embed = discord.Embed( title="🛡️ Shield Restrictions Updated", description=f"Custom restrictions set to:\n`{restrictions[:200]}`", color=discord.Color.green(), ) await ctx.send(embed=embed, ephemeral=True) @commands.hybrid_command(name="shield_keywords", description="Add keywords that trigger instant shield action") @commands.has_permissions(manage_guild=True) async def shield_keywords(self, ctx: commands.Context, *, keywords: str) -> None: """Add custom keywords for the shield. Example: 'crypto, investment, dm me, wallet, airdrop'.""" if not ctx.guild: await ctx.send("Server only.", ephemeral=True) return await self.bot.db.execute( "INSERT INTO shield_settings(guild_id, level, strict_keywords) VALUES (?, ?, ?) " "ON CONFLICT(guild_id) DO UPDATE SET strict_keywords = excluded.strict_keywords", ctx.guild.id, "medium", keywords[:500], ) embed = discord.Embed( title="🔑 Shield Keywords Updated", description=f"Strict keywords set to:\n`{keywords[:200]}`", color=discord.Color.green(), ) await ctx.send(embed=embed, ephemeral=True) @commands.hybrid_command(name="econ_admin", description=get_cmd_desc("commands.admin.econ_admin_desc")) @commands.has_permissions(administrator=True) async def econ_admin(self, ctx: commands.Context, member: discord.Member, action: str, amount: int) -> None: if ctx.interaction and not ctx.interaction.response.is_done(): await ctx.interaction.response.defer(ephemeral=True, thinking=True) action_n = action.strip().lower() if action_n not in {"add", "remove"}: await self._safe_ctx_send(ctx, content="Use action: add/remove", ephemeral=bool(ctx.interaction)) return amount = max(0, amount) row = await self.bot.db.fetchone("SELECT coins, xp FROM economy WHERE user_id = ?", member.id) coins = int(row[0]) if row else 0 xp = int(row[1]) if row else 0 new_coins = coins + amount if action_n == "add" else max(0, coins - amount) await self.bot.db.execute( "INSERT INTO economy(user_id, coins, xp) VALUES (?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET coins = excluded.coins, xp = excluded.xp", member.id, new_coins, xp, ) await self._safe_ctx_send(ctx, content=f"💰 {member.mention} coins: `{coins}` → `{new_coins}`", ephemeral=bool(ctx.interaction)) @commands.hybrid_command(name="warn", description=get_cmd_desc("commands.admin.warn_desc")) @commands.has_permissions(manage_roles=True) async def warn(self, ctx: commands.Context, member: discord.Member, *, reason: str = "No reason provided") -> None: """Warn a member.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) await self.bot.db.execute( "INSERT INTO user_warnings(guild_id, user_id, moderator_id, reason, timestamp) VALUES (?, ?, ?, ?, ?)", ctx.guild.id, member.id, ctx.author.id, reason, dt.datetime.utcnow().isoformat(), ) if lang == "ar": title = "⚠️ تحذير للعضو" desc = ( f"{panel_divider('orange')}\n" f"👤 **المستخدم:** {member.mention}\n" f"🛡️ **المشرف:** {ctx.author.mention}\n" f"📝 **السبب:** {reason}\n" f"{panel_divider('orange')}" ) dm_title = "⚠️ تم تحذيرك" dm_desc = f"🏛️ **السيرفر:** {ctx.guild.name}\n📝 **السبب:** {reason}\n🛡️ **المشرف:** {ctx.author}" else: title = "⚠️ Member Warned" desc = ( f"{panel_divider('orange')}\n" f"👤 **User:** {member.mention}\n" f"🛡️ **Moderator:** {ctx.author.mention}\n" f"📝 **Reason:** {reason}\n" f"{panel_divider('orange')}" ) dm_title = "⚠️ You have been warned" dm_desc = f"🏛️ **Server:** {ctx.guild.name}\n📝 **Reason:** {reason}\n🛡️ **Moderator:** {ctx.author}" embed = warning_embed(title, desc) await ctx.reply(embed=embed) try: dm_embed = warning_embed(dm_title, dm_desc) await member.send(embed=dm_embed) except discord.Forbidden: pass @commands.hybrid_command(name="warnings", description=get_cmd_desc("commands.admin.warnings_desc")) @commands.has_permissions(manage_roles=True) async def warnings(self, ctx: commands.Context, member: discord.Member) -> None: """View warnings for a member.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) rows = await self.bot.db.fetchall( "SELECT moderator_id, reason, timestamp FROM user_warnings WHERE guild_id = ? AND user_id = ? ORDER BY timestamp DESC LIMIT 10", ctx.guild.id, member.id, ) if not rows: if lang == "ar": embed = info_embed("📋 التحذيرات", f"{member.mention} لا توجد تحذيرات. ✅") else: embed = info_embed("📋 Warnings", f"{member.mention} has no warnings. ✅") await ctx.reply(embed=embed) return lines = [] for idx, (mod_id, reason, timestamp) in enumerate(rows, 1): mod = ctx.guild.get_member(mod_id) mod_name = mod.display_name if mod else f"<@{mod_id}>" lines.append(f"`{idx}.` 📝 {reason[:50]}{'...' if len(reason) > 50 else ''}\n 🛡️ by {mod_name}") title = f"📋 تحذيرات {member.display_name}" if lang == "ar" else f"📋 Warnings for {member.display_name}" embed = discord.Embed( title=title, description=f"{panel_divider('orange')}\n" + "\n".join(lines) + f"\n{panel_divider('orange')}", color=NEON_ORANGE, ) embed.set_thumbnail(url=member.display_avatar.url) await ctx.reply(embed=embed) @commands.hybrid_command(name="clearwarn", description=get_cmd_desc("commands.admin.clearwarn_desc")) @commands.has_permissions(manage_roles=True) async def clearwarn(self, ctx: commands.Context, member: discord.Member) -> None: """Clear all warnings for a member.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) await self.bot.db.execute( "DELETE FROM user_warnings WHERE guild_id = ? AND user_id = ?", ctx.guild.id, member.id, ) if lang == "ar": embed = success_embed("✅ تم مسح التحذيرات", f"🗑️ تم إزالة جميع تحذيرات {member.mention}.") else: embed = success_embed("✅ Warnings Cleared", f"🗑️ All warnings for {member.mention} have been removed.") await ctx.reply(embed=embed) @commands.hybrid_command(name="kick", description=get_cmd_desc("commands.admin.kick_desc")) @commands.has_permissions(kick_members=True) async def kick(self, ctx: commands.Context, member: discord.Member, *, reason: str = "No reason provided") -> None: """Kick a member from the server.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) if member.top_role >= ctx.author.top_role and ctx.author.id != ctx.guild.owner_id: if lang == "ar": await ctx.reply("❌ لا يمكنك طرد شخص برتبة أعلى أو مساوية.") else: await ctx.reply("❌ You cannot kick someone with a higher or equal role.") return try: if lang == "ar": dm_embed = warning_embed("👢 تم طردك", f"🏛️ **السيرفر:** {ctx.guild.name}\n📝 **السبب:** {reason}\n🛡️ **المشرف:** {ctx.author}") else: dm_embed = warning_embed("👢 You have been kicked", f"🏛️ **Server:** {ctx.guild.name}\n📝 **Reason:** {reason}\n🛡️ **Moderator:** {ctx.author}") await member.send(embed=dm_embed) except discord.Forbidden: pass await member.kick(reason=reason) if lang == "ar": embed = success_embed( "👢 تم طرد العضو", f"{panel_divider('green')}\n" f"👤 **المستخدم:** {member.mention}\n" f"🛡️ **المشرف:** {ctx.author.mention}\n" f"📝 **السبب:** {reason}\n" f"{panel_divider('green')}" ) else: embed = success_embed( "👢 Member Kicked", f"{panel_divider('green')}\n" f"👤 **User:** {member.mention}\n" f"🛡️ **Moderator:** {ctx.author.mention}\n" f"📝 **Reason:** {reason}\n" f"{panel_divider('green')}" ) await ctx.reply(embed=embed) @commands.hybrid_command(name="ban", description=get_cmd_desc("commands.admin.ban_desc")) @commands.has_permissions(ban_members=True) async def ban(self, ctx: commands.Context, member: discord.Member, *, reason: str = "No reason provided") -> None: """Ban a member from the server.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) if member.top_role >= ctx.author.top_role and ctx.author.id != ctx.guild.owner_id: if lang == "ar": await ctx.reply("❌ لا يمكنك حظر شخص برتبة أعلى أو مساوية.") else: await ctx.reply("❌ You cannot ban someone with a higher or equal role.") return try: if lang == "ar": dm_embed = error_embed("🔨 تم حظرك", f"🏛️ **السيرفر:** {ctx.guild.name}\n📝 **السبب:** {reason}\n🛡️ **المشرف:** {ctx.author}") else: dm_embed = error_embed("🔨 You have been banned", f"🏛️ **Server:** {ctx.guild.name}\n📝 **Reason:** {reason}\n🛡️ **Moderator:** {ctx.author}") await member.send(embed=dm_embed) except discord.Forbidden: pass await member.ban(reason=reason) if lang == "ar": embed = success_embed( "🔨 تم حظر العضو", f"{panel_divider('pink')}\n" f"👤 **المستخدم:** {member.mention}\n" f"🛡️ **المشرف:** {ctx.author.mention}\n" f"📝 **السبب:** {reason}\n" f"{panel_divider('pink')}" ) else: embed = success_embed( "🔨 Member Banned", f"{panel_divider('pink')}\n" f"👤 **User:** {member.mention}\n" f"🛡️ **Moderator:** {ctx.author.mention}\n" f"📝 **Reason:** {reason}\n" f"{panel_divider('pink')}" ) await ctx.reply(embed=embed) await self.bot.log_to_guild( ctx.guild, "🔨 Moderation: Ban", f"{member.mention} was banned by {ctx.author.mention}\nReason: {reason}", color=discord.Color.red(), ) @commands.hybrid_command(name="unban", description=get_cmd_desc("commands.admin.unban_desc")) @commands.has_permissions(ban_members=True) async def unban(self, ctx: commands.Context, user_id: int, *, reason: str = "No reason provided") -> None: """Unban a user by their ID.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) try: user = await self.bot.fetch_user(user_id) except discord.NotFound: if lang == "ar": await ctx.reply("❌ المستخدم غير موجود.") else: await ctx.reply("❌ User not found.") return except discord.HTTPException: if lang == "ar": await ctx.reply("❌ تعذر جلب المستخدم.") else: await ctx.reply("❌ Could not fetch user.") return bans = [b async for b in ctx.guild.bans()] if user not in [b.user for b in bans]: if lang == "ar": await ctx.reply(f"❌ {user.mention} غير محظور.") else: await ctx.reply(f"❌ {user.mention} is not banned.") return await ctx.guild.unban(user, reason=reason) if lang == "ar": embed = success_embed( "🔓 تم فك الحظر", f"{panel_divider('green')}\n" f"👤 **المستخدم:** {user.mention}\n" f"🛡️ **المشرف:** {ctx.author.mention}\n" f"📝 **السبب:** {reason}\n" f"{panel_divider('green')}" ) else: embed = success_embed( "🔓 Member Unbanned", f"{panel_divider('green')}\n" f"👤 **User:** {user.mention}\n" f"🛡️ **Moderator:** {ctx.author.mention}\n" f"📝 **Reason:** {reason}\n" f"{panel_divider('green')}" ) await ctx.reply(embed=embed) await self.bot.log_to_guild( ctx.guild, "🔓 Moderation: Unban", f"{user.mention} was unbanned by {ctx.author.mention}\nReason: {reason}", color=discord.Color.green(), ) @commands.hybrid_command(name="mute", description=get_cmd_desc("commands.admin.mute_desc")) @commands.has_permissions(moderate_members=True) async def mute(self, ctx: commands.Context, member: discord.Member, duration: int = 10, *, reason: str = "No reason provided") -> None: """Timeout a member for a specified duration in minutes.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) if member.top_role >= ctx.author.top_role and ctx.author.id != ctx.guild.owner_id: if lang == "ar": await ctx.reply("❌ لا يمكنك كتم شخص برتبة أعلى أو مساوية.") else: await ctx.reply("❌ You cannot mute someone with a higher or equal role.") return duration = max(1, min(duration, 40320)) # Max 28 days until = dt.datetime.utcnow() + dt.timedelta(minutes=duration) await member.timeout(until, reason=reason) if lang == "ar": embed = success_embed( "🔇 تم كتم العضو", f"{panel_divider('orange')}\n" f"👤 **المستخدم:** {member.mention}\n" f"⏰ **المدة:** {duration} دقيقة\n" f"🛡️ **المشرف:** {ctx.author.mention}\n" f"📝 **السبب:** {reason}\n" f"{panel_divider('orange')}" ) else: embed = success_embed( "🔇 Member Muted", f"{panel_divider('orange')}\n" f"👤 **User:** {member.mention}\n" f"⏰ **Duration:** {duration} minutes\n" f"🛡️ **Moderator:** {ctx.author.mention}\n" f"📝 **Reason:** {reason}\n" f"{panel_divider('orange')}" ) await ctx.reply(embed=embed) await self.bot.log_to_guild( ctx.guild, "🔇 Moderation: Mute", f"{member.mention} was muted by {ctx.author.mention} for {duration} minute(s)\nReason: {reason}", color=discord.Color.red(), ) @commands.hybrid_command(name="unmute", description=get_cmd_desc("commands.admin.unmute_desc")) @commands.has_permissions(moderate_members=True) async def unmute(self, ctx: commands.Context, member: discord.Member, *, reason: str = "No reason provided") -> None: """Remove timeout from a member.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) await member.timeout(None, reason=reason) if lang == "ar": embed = success_embed("🔊 تم فك الكتم", f"👤 **المستخدم:** {member.mention}\n🛡️ **المشرف:** {ctx.author.mention}\n📝 **السبب:** {reason}") else: embed = success_embed("🔊 Member Unmuted", f"👤 **User:** {member.mention}\n🛡️ **Moderator:** {ctx.author.mention}\n📝 **Reason:** {reason}") await ctx.reply(embed=embed) async def _clone_emoji_to_guild( self, guild: discord.Guild, actor: discord.abc.User, *, source: str, name: str | None = None, ) -> tuple[discord.Emoji | None, str]: source_clean = (source or "").strip() if not source_clean: return None, "Provide a valid emoji tag, emoji ID, emoji URL, or known emoji name." emoji_url: str | None = None detected_name: str | None = None tag_match = re.fullmatch(r"<(a?):([A-Za-z0-9_]{2,32}):(\d+)>", source_clean) if tag_match: is_animated = bool(tag_match.group(1)) detected_name = tag_match.group(2) eid = tag_match.group(3) ext = "gif" if is_animated else "png" emoji_url = f"https://cdn.discordapp.com/emojis/{eid}.{ext}?quality=lossless" elif source_clean.isdigit(): emoji_url = f"https://cdn.discordapp.com/emojis/{source_clean}.png?quality=lossless" elif source_clean.startswith(("http://", "https://")): emoji_url = source_clean else: found = discord.utils.get(self.bot.emojis, name=source_clean) if found: detected_name = found.name emoji_url = str(found.url) if not emoji_url: return None, "Provide a valid emoji tag, emoji ID, emoji URL, or known emoji name." target_name_raw = (name or detected_name or "cloned_emoji").strip().lower() target_name = re.sub(r"[^a-z0-9_]", "_", target_name_raw) target_name = re.sub(r"_+", "_", target_name).strip("_") if len(target_name) < 2: target_name = "cloned_emoji" target_name = target_name[:32] try: async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=20)) as session: async with session.get(emoji_url) as resp: if resp.status != 200: return None, f"Failed to fetch emoji image (HTTP {resp.status})." image_bytes = await resp.read() except Exception as exc: return None, f"Failed to fetch emoji image: {exc}" if not image_bytes: return None, "Fetched emoji image is empty." if len(image_bytes) > 256 * 1024: return None, "Emoji image is too large. Discord custom emojis must be <= 256 KB." try: created = await guild.create_custom_emoji( name=target_name, image=image_bytes, reason=f"Emoji cloned by {actor} via /admin emoji clone", ) except discord.Forbidden: return None, "Missing permission to create emojis." except discord.HTTPException as exc: return None, f"Could not create emoji: {exc}" return created, "" @admin_group.group(name="emoji", invoke_without_command=True) @commands.has_permissions(manage_emojis=True) async def admin_emoji_group(self, ctx: commands.Context) -> None: await self.admin_emoji_panel(ctx) @admin_emoji_group.command(name="panel") @commands.has_permissions(manage_emojis=True) async def admin_emoji_panel(self, ctx: commands.Context) -> None: if not ctx.guild: await self._safe_ctx_send(ctx, content="Server only.", ephemeral=bool(ctx.interaction)) return embed = info_embed( "Emoji Clone Panel", "Quick actions:\n- Clone from tag/ID/URL/name\n- View server emojis\n- Open bot emoji picker", ) await self._safe_ctx_send(ctx, embed=embed, view=EmojiClonePanelView(self), ephemeral=bool(ctx.interaction)) @commands.hybrid_command(name="emoji_clone_panel", description=get_cmd_desc("commands.admin.emoji_clone_panel_desc"), with_app_command=True) @commands.has_permissions(manage_emojis=True) async def emoji_clone_panel(self, ctx: commands.Context) -> None: await self.admin_emoji_panel(ctx) @admin_emoji_group.command(name="clone") @commands.has_permissions(manage_emojis=True) async def admin_emoji_clone(self, ctx: commands.Context, source: str | None = None, name: str | None = None) -> None: if not ctx.guild: await self._safe_ctx_send(ctx, content="Server only.", ephemeral=bool(ctx.interaction)) return if source: if ctx.interaction and not ctx.interaction.response.is_done(): try: await ctx.interaction.response.defer(ephemeral=True, thinking=True) except (discord.NotFound, discord.HTTPException, discord.InteractionResponded): pass created, error_text = await self._clone_emoji_to_guild( ctx.guild, ctx.author, source=source, name=name, ) if error_text: await self._safe_ctx_send(ctx, content=error_text, ephemeral=True) return embed = success_embed( "Emoji Cloned", f"Created {created} as `:{created.name}:`", ) await self._safe_ctx_send(ctx, embed=embed, ephemeral=bool(ctx.interaction)) return custom_names = sorted({emoji.name for emoji in getattr(self.bot, "emojis", []) if getattr(emoji, "name", None)}) if not custom_names: await self._safe_ctx_send(ctx, content="No bot emoji catalog available.", ephemeral=bool(ctx.interaction)) return lines = [] for em_name in custom_names[:40]: emoji_obj = discord.utils.get(self.bot.emojis, name=em_name) rendered = str(emoji_obj) if emoji_obj else em_name lines.append(f"{rendered} `{em_name}`") embed = info_embed( "Emoji Picker", "\n".join(lines) + "\n\nUse: `/admin emoji clone [name]`", ) await self._safe_ctx_send(ctx, embed=embed, ephemeral=bool(ctx.interaction)) @commands.hybrid_command(name="awesomeroles", description=get_cmd_desc("commands.admin.awesomeroles_desc")) @commands.has_permissions(manage_roles=True) async def awesomeroles(self, ctx: commands.Context) -> None: """Create awesome roles for the server with proper permissions.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) created = [] skipped = [] for role_data in AWESOME_ROLES: existing = discord.utils.get(ctx.guild.roles, name=role_data["name"]) if existing: skipped.append(role_data["name"]) continue # Build permissions permissions = discord.Permissions() for perm_name in role_data["permissions"]: setattr(permissions, perm_name, True) try: role = await ctx.guild.create_role( name=role_data["name"], color=role_data["color"], permissions=permissions, hoist=role_data["hoist"], reason=f"Created by {ctx.author} via /awesomeroles" ) created.append((role, role_data["description"])) except discord.Forbidden: continue embed = discord.Embed( title=f"✨ {'الأدوار الرائعة' if lang == 'ar' else 'Awesome Roles'}", description=f"{panel_divider('purple')}", color=NEON_PURPLE, ) if created: lines = [] for role, desc in created: lines.append(f"{role.mention}\n└─ *{desc}*") embed.add_field( name=f"✅ {'تم إنشاؤها' if lang == 'ar' else 'Created'} ({len(created)})", value="\n\n".join(lines), inline=False ) if skipped: embed.add_field( name=f"⏭️ {'موجودة مسبقاً' if lang == 'ar' else 'Already Exist'} ({len(skipped)})", value="\n".join(f"• {name}" for name in skipped), inline=False ) embed.add_field( name="💡 {'نصيحة' if lang == 'ar' else 'Tip'}", value="Drag roles in server settings to set their position!" if lang != "ar" else "اسحب الأدوار في إعدادات السيرفر لتغيير ترتيبها!", inline=False ) embed.set_footer(text=f"{E_STAR} Created by {ctx.author.display_name}") await ctx.reply(embed=embed) @commands.hybrid_command(name="backupserver", description=get_cmd_desc("commands.admin.backupserver_desc")) @commands.has_permissions(administrator=True) async def backupserver(self, ctx: commands.Context) -> None: """Create a backup of server structure.""" if not ctx.guild: await ctx.reply("Server only.") return if ctx.interaction and not ctx.interaction.response.is_done(): try: await ctx.interaction.response.defer(ephemeral=True, thinking=True) except (discord.NotFound, discord.HTTPException, discord.InteractionResponded): # If interaction token is already invalid, continue and fallback to channel sends via resilient ctx.reply. pass guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) # Collect server data roles = [] for r in ctx.guild.roles: if not r.is_default() and not r.managed: roles.append({ "name": r.name, "color": r.color.value, "permissions": r.permissions.value, "hoist": r.hoist, "mentionable": r.mentionable, "position": r.position }) channels = [] for c in ctx.guild.channels: channels.append({ "name": c.name, "type": str(c.type), "position": c.position, "category": c.category.name if c.category else None }) backup_data = { "roles": roles, "channels": channels, "name": ctx.guild.name, "icon_url": str(ctx.guild.icon.url) if ctx.guild.icon else None } await self.bot.db.execute( "INSERT INTO server_backups(guild_id, backup_data, created_by, created_at) VALUES (?, ?, ?, ?)", ctx.guild.id, json.dumps(backup_data, ensure_ascii=False), ctx.author.id, dt.datetime.utcnow().isoformat(), ) if lang == "ar": embed = success_embed( "💾 تم إنشاء نسخة احتياطية", f"{panel_divider('blue')}\n" f"🎭 **الأدوار:** {len(roles)}\n" f"📺 **القنوات:** {len(channels)}\n" f"👤 **أنشئت بواسطة:** {ctx.author.mention}\n" f"{panel_divider('blue')}" ) else: embed = success_embed( "💾 Server Backup Created", f"{panel_divider('blue')}\n" f"🎭 **Roles:** {len(roles)}\n" f"📺 **Channels:** {len(channels)}\n" f"👤 **Created by:** {ctx.author.mention}\n" f"{panel_divider('blue')}" ) await ctx.reply(embed=embed) @commands.hybrid_command(name="backup_panel", description=get_cmd_desc("commands.admin.backup_panel_desc")) @commands.has_permissions(administrator=True) async def backup_panel(self, ctx: commands.Context) -> None: """Open the interactive backup management panel.""" if not ctx.guild: await ctx.reply("Server only.") return guild_id = ctx.guild.id lang = await self.bot.get_guild_language(guild_id) rows = await self.bot.db.fetchall( "SELECT id, created_by, created_at, backup_data FROM server_backups WHERE guild_id = ? ORDER BY id DESC LIMIT 10", guild_id, ) embed = self._build_backup_embed(guild_id, lang, rows) await ctx.reply(embed=embed, view=BackupPanelView(self, guild_id)) def _build_backup_embed(self, guild_id: int, lang: str, rows: list) -> discord.Embed: if lang == "ar": title = "💾 إدارة النسخ الاحتياطية" desc = f"{panel_divider('blue')}\nإدارة النسخ الاحتياطية للسيرفر\n{panel_divider('blue')}" else: title = "💾 Backup Management" desc = f"{panel_divider('blue')}\nManage server backups\n{panel_divider('blue')}" embed = discord.Embed(title=title, description=desc, color=NEON_CYAN) if not rows: embed.add_field(name="📦" if lang == "ar" else "📦 No backups", value="No backups found." if lang != "ar" else "لا توجد نسخ احتياطية.", inline=False) else: lines = [] for bid, creator_id, created_at, data_str in rows: try: data = json.loads(data_str) roles_count = len(data.get("roles", [])) channels_count = len(data.get("channels", [])) except Exception: roles_count = "?" channels_count = "?" lines.append(f"**#{bid}** — {created_at[:19]} | 🎭{roles_count} 📺{channels_count}") embed.add_field( name="📦 Recent Backups" if lang != "ar" else "📦 النسخ الأخيرة", value="\n".join(lines[:10]), inline=False, ) return embed class BackupPanelView(discord.ui.View): def __init__(self, cog: "Admin", guild_id: int) -> None: super().__init__(timeout=None) self.cog = cog self.guild_id = guild_id @discord.ui.button(label="Refresh", style=discord.ButtonStyle.blurple, emoji=ui("refresh"), row=2) async def refresh(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: if not interaction.guild: await interaction.response.send_message("Server only.", ephemeral=True) return guild_id = interaction.guild.id lang = await self.cog.bot.get_guild_language(guild_id) rows = await self.cog.bot.db.fetchall( "SELECT id, created_by, created_at, backup_data FROM server_backups WHERE guild_id = ? ORDER BY id DESC LIMIT 10", guild_id, ) embed = self.cog._build_backup_embed(guild_id, lang, rows) await interaction.response.edit_message(embed=embed, view=self) @discord.ui.button(label="List Backups", style=discord.ButtonStyle.primary, emoji=ui("notebook"), row=0) async def list_backups(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: if not interaction.guild: await interaction.response.send_message("Server only.", ephemeral=True) return await interaction.response.defer(ephemeral=True) lang = await self.cog.bot.get_guild_language(interaction.guild.id) rows = await self.cog.bot.db.fetchall( "SELECT id, created_by, created_at, backup_data FROM server_backups WHERE guild_id = ? ORDER BY id DESC", interaction.guild.id, ) if not rows: await interaction.followup.send("No backups found." if lang != "ar" else "لا توجد نسخ احتياطية.", ephemeral=True) return lines = [] for bid, creator_id, created_at, data_str in rows: try: data = json.loads(data_str) roles_count = len(data.get("roles", [])) channels_count = len(data.get("channels", [])) except Exception: roles_count = "?" channels_count = "?" lines.append(f"**#{bid}** — {created_at[:19]} | 🎭{roles_count} 📺{channels_count} | By: <@{creator_id}>") embed = discord.Embed( title="📦 All Backups" if lang != "ar" else "📦 جميع النسخ", description="\n".join(lines[:20]), color=NEON_CYAN, ) await interaction.followup.send(embed=embed, ephemeral=True) @discord.ui.button(label="Restore Backup", style=discord.ButtonStyle.success, emoji=ui("ok"), row=0) async def restore_backup(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: if not interaction.guild: await interaction.response.send_message("Server only.", ephemeral=True) return if not interaction.user.guild_permissions.administrator: await interaction.response.send_message("Administrator permission required.", ephemeral=True) return await interaction.response.send_modal(RestoreBackupModal(self.cog, interaction.guild.id)) @discord.ui.button(label="Delete Backup", style=discord.ButtonStyle.danger, emoji=ui("trash"), row=0) async def delete_backup(self, interaction: discord.Interaction, _: discord.ui.Button) -> None: if not interaction.guild: await interaction.response.send_message("Server only.", ephemeral=True) return if not interaction.user.guild_permissions.administrator: await interaction.response.send_message("Administrator permission required.", ephemeral=True) return await interaction.response.send_modal(DeleteBackupModal(self.cog, interaction.guild.id)) class RestoreBackupModal(discord.ui.Modal, title="♻️ Restore Backup"): backup_id = discord.ui.TextInput( label="Backup ID", placeholder="1", required=True, max_length=16, ) def __init__(self, cog: "Admin", guild_id: int) -> None: super().__init__(timeout=None) self.cog = cog self.guild_id = guild_id async def on_submit(self, interaction: discord.Interaction) -> None: await interaction.response.defer(ephemeral=True) try: bid = int(self.backup_id.value.strip()) except ValueError: await interaction.followup.send("Invalid backup ID.", ephemeral=True) return row = await self.cog.bot.db.fetchone( "SELECT backup_data FROM server_backups WHERE guild_id = ? AND id = ?", self.guild_id, bid, ) if not row: await interaction.followup.send("Backup not found.", ephemeral=True) return try: data = json.loads(row[0]) except Exception: await interaction.followup.send("Corrupted backup data.", ephemeral=True) return guild = interaction.guild if not guild: await interaction.followup.send("Server only.", ephemeral=True) return lang = await self.cog.bot.get_guild_language(self.guild_id) created_roles = 0 created_channels = 0 for role_data in data.get("roles", []): name = role_data.get("name", "") if not name or guild.get_role_next_id() is None: continue existing = discord.utils.get(guild.roles, name=name) if existing: continue try: color = discord.Color(role_data.get("color", 0)) perms = discord.Permissions(role_data.get("permissions", 0)) await guild.create_role( name=name, color=color, permissions=perms, hoist=role_data.get("hoist", False), mentionable=role_data.get("mentionable", False), ) created_roles += 1 except Exception: pass for ch_data in data.get("channels", []): name = ch_data.get("name", "") ch_type = ch_data.get("type", "") if not name: continue existing = discord.utils.get(guild.channels, name=name) if existing: continue try: if "text" in ch_type.lower(): await guild.create_text_channel(name) elif "voice" in ch_type.lower(): await guild.create_voice_channel(name) elif "category" in ch_type.lower(): await guild.create_category(name) created_channels += 1 except Exception: pass if lang == "ar": embed = success_embed( "♻️ تم استعادة النسخة الاحتياطية", f"{panel_divider('blue')}\n🎭 **أدوار:** {created_roles}\n📺 **قنوات:** {created_channels}\n{panel_divider('blue')}" ) else: embed = success_embed( "♻️ Backup Restored", f"{panel_divider('blue')}\n🎭 **Roles:** {created_roles}\n📺 **Channels:** {created_channels}\n{panel_divider('blue')}" ) await interaction.followup.send(embed=embed, ephemeral=True) class DeleteBackupModal(discord.ui.Modal, title="🗑️ Delete Backup"): backup_id = discord.ui.TextInput( label="Backup ID", placeholder="1", required=True, max_length=16, ) def __init__(self, cog: "Admin", guild_id: int) -> None: super().__init__(timeout=None) self.cog = cog self.guild_id = guild_id async def on_submit(self, interaction: discord.Interaction) -> None: await interaction.response.defer(ephemeral=True) try: bid = int(self.backup_id.value.strip()) except ValueError: await interaction.followup.send("Invalid backup ID.", ephemeral=True) return await self.cog.bot.db.execute( "DELETE FROM server_backups WHERE guild_id = ? AND id = ?", self.guild_id, bid, ) lang = await self.cog.bot.get_guild_language(self.guild_id) msg = "✅ Backup deleted." if lang != "ar" else "✅ تم حذف النسخة الاحتياطية." await interaction.followup.send(msg, ephemeral=True) class SlowmodeCommandMixin: """Shared slowmode/lock/unlock commands.""" pass @commands.hybrid_command(name="slowmode", description=get_cmd_desc("commands.admin.slowmode_desc")) @commands.has_permissions(manage_channels=True) async def slowmode(self, ctx: commands.Context, seconds: int = 0) -> None: """Set slowmode for the current channel.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) seconds = max(0, min(seconds, 21600)) # Max 6 hours await ctx.channel.edit(slowmode_delay=seconds) if seconds > 0: if lang == "ar": embed = success_embed("⏱️ تم تفعيل الوضع البطيء", f"🕒 **المدة:** {seconds} ثانية\n📺 **القناة:** {ctx.channel.mention}") else: embed = success_embed("⏱️ Slowmode Enabled", f"🕒 **Duration:** {seconds} seconds\n📺 **Channel:** {ctx.channel.mention}") else: if lang == "ar": embed = success_embed("⏱️ تم إيقاف الوضع البطيء", f"📺 **القناة:** {ctx.channel.mention}") else: embed = success_embed("⏱️ Slowmode Disabled", f"📺 **Channel:** {ctx.channel.mention}") await ctx.reply(embed=embed) @commands.hybrid_command(name="lock", description=get_cmd_desc("commands.admin.lock_desc")) @commands.has_permissions(manage_channels=True) async def lock(self, ctx: commands.Context, channel: discord.TextChannel | None = None) -> None: """Lock a channel to prevent messages.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) channel = channel or ctx.channel overwrite = channel.overwrites_for(ctx.guild.default_role) overwrite.send_messages = False await channel.set_permissions(ctx.guild.default_role, overwrite=overwrite) if lang == "ar": embed = success_embed("🔒 تم قفل القناة", f"📺 **القناة:** {channel.mention}") else: embed = success_embed("🔒 Channel Locked", f"📺 **Channel:** {channel.mention}") await ctx.reply(embed=embed) @commands.hybrid_command(name="unlock", description=get_cmd_desc("commands.admin.unlock_desc")) @commands.has_permissions(manage_channels=True) async def unlock(self, ctx: commands.Context, channel: discord.TextChannel | None = None) -> None: """Unlock a channel to allow messages.""" guild_id = ctx.guild.id if ctx.guild else None lang = await self.bot.get_guild_language(guild_id) channel = channel or ctx.channel overwrite = channel.overwrites_for(ctx.guild.default_role) overwrite.send_messages = True await channel.set_permissions(ctx.guild.default_role, overwrite=overwrite) if lang == "ar": embed = success_embed("🔓 تم فتح القناة", f"📺 **القناة:** {channel.mention}") else: embed = success_embed("🔓 Channel Unlocked", f"📺 **Channel:** {channel.mention}") await ctx.reply(embed=embed) async def setup(bot: commands.Bot) -> None: await bot.add_cog(Admin(bot))