from __future__ import annotations import difflib import json import traceback from pathlib import Path from typing import get_args, get_origin from bot.i18n import get_cmd_desc import discord from discord.ext import commands class Observability(commands.Cog): def __init__(self, bot: commands.Bot) -> None: self.bot = bot def _format_usage(self, ctx: commands.Context) -> str: if not ctx.command: return "" parts: list[str] = [] for name, param in ctx.command.clean_params.items(): required = param.default is param.empty parts.append(f"<{name}>" if required else f"[{name}]") return f"{ctx.prefix}{ctx.command.qualified_name} " + " ".join(parts) def _suggest_for_param(self, param: commands.Parameter) -> str: ann = param.annotation if ann is bool: return "Examples: `true`, `false`" if ann is int: return "Expected number (example: `10`)" if ann is float: return "Expected decimal number (example: `1.5`)" if ann is discord.Member: return "Expected member mention (example: `@User`)" if ann is discord.TextChannel: return "Expected channel mention (example: `#general`)" origin = get_origin(ann) if origin is not None: args = [str(a).replace("typing.", "") for a in get_args(ann)] if args: return "Choices: " + ", ".join(f"`{a}`" for a in args[:8]) return "Fill this argument in normal text form." def _closest_commands(self, name: str) -> list[str]: names = sorted({c.qualified_name for c in self.bot.commands} | {c.name for c in self.bot.commands}) return difflib.get_close_matches(name, names, n=5, cutoff=0.4) async def _safe_reply(self, ctx: commands.Context, message: str) -> None: try: await ctx.reply(message) except (discord.NotFound, discord.InteractionResponded): if ctx.channel: await ctx.channel.send(message) except discord.HTTPException as exc: if exc.code not in {10062, 40060}: raise if ctx.channel: await ctx.channel.send(message) @commands.Cog.listener() async def on_command_error(self, ctx: commands.Context, error: Exception) -> None: if hasattr(ctx.command, "on_error"): return if isinstance(error, commands.CommandNotFound): raw = (ctx.message.content or "").strip() prefix = (self.bot.settings.prefix or "!").strip() or "!" candidate = raw[len(prefix) :].split()[0] if raw.startswith(prefix) else "" if not candidate: return close = self._closest_commands(candidate) hint = "\n".join(f"• `{prefix}{name}`" for name in close) if close else "No similar commands found." await self._safe_reply(ctx, f"❓ Command not found: `{candidate}`\n{hint}") return if isinstance(error, commands.MissingPermissions): await self._safe_reply(ctx, "❌ ليس لديك صلاحية لتنفيذ هذا الأمر.") return if isinstance(error, commands.MissingRequiredArgument): usage = self._format_usage(ctx) suggestion = self._suggest_for_param(error.param) await self._safe_reply( ctx, f"⚠️ Missing argument: `{error.param.name}`\nUsage: `{usage.strip()}`\n{suggestion}", ) return if isinstance(error, commands.BadArgument): usage = self._format_usage(ctx) await self._safe_reply(ctx, f"⚠️ Invalid input type.\nUsage: `{usage.strip()}`") return tb = "".join(traceback.format_exception(type(error), error, error.__traceback__)) author = f"{ctx.author} ({ctx.author.id})" guild = f"{ctx.guild.name} ({ctx.guild.id})" if ctx.guild else "DM" command_name = ctx.command.qualified_name if ctx.command else "unknown" print("\n[COMMAND_ERROR]") print(f"guild={guild}") print(f"author={author}") print(f"command={command_name}") print(tb) await self._safe_reply(ctx, "❌ حدث خطأ أثناء تنفيذ الأمر. تم تسجيل السبب في السجلات.") if not ctx.guild: return row = await self.bot.db.fetchone( "SELECT log_channel_id FROM guild_config WHERE guild_id = ?", ctx.guild.id, ) if not row or not row[0]: return channel = ctx.guild.get_channel(row[0]) if not channel: return embed = discord.Embed(title="🚨 Command Error", color=discord.Color.red()) embed.add_field(name="Command", value=f"`{command_name}`", inline=False) embed.add_field(name="User", value=author, inline=False) embed.add_field(name="Error", value=f"```py\n{tb[-900:]}\n```", inline=False) await channel.send(embed=embed) @commands.hybrid_command(name="command_fill", description=get_cmd_desc("commands.tools.command_fill_desc")) async def command_fill(self, ctx: commands.Context, *, command_name: str) -> None: command = self.bot.get_command(command_name.strip()) if not command: close = self._closest_commands(command_name.strip()) hint = "\n".join(f"• `{ctx.prefix}{name}`" for name in close) if close else "No similar commands found." await ctx.reply(f"❓ Command not found: `{command_name}`\n{hint}") return pieces: list[str] = [] details: list[str] = [] for name, param in command.clean_params.items(): required = param.default is param.empty pieces.append(f"<{name}>" if required else f"[{name}]") details.append(f"• `{name}`: {self._suggest_for_param(param)}") usage = f"{ctx.prefix}{command.qualified_name} " + " ".join(pieces) embed = discord.Embed(title="🧩 Command Fill Helper", color=discord.Color.blurple()) embed.add_field(name="Usage", value=f"`{usage.strip()}`", inline=False) embed.add_field(name="Argument hints", value="\n".join(details) if details else "No arguments", inline=False) await ctx.reply(embed=embed) @commands.hybrid_command(name="system_audit", description=get_cmd_desc("commands.tools.system_audit_desc")) @commands.has_permissions(manage_guild=True) async def system_audit(self, ctx: commands.Context) -> None: await self._safe_reply(ctx, "Running system audit...") total_commands = len(self.bot.commands) loaded_cogs = len(self.bot.cogs) row = await self.bot.db.fetchone("PRAGMA table_info(guild_config)") has_guild_config = bool(row) cols = await self.bot.db.fetchall("PRAGMA table_info(guild_config)") col_names = {str(r[1]) for r in cols} if cols else set() has_banner_col = "custom_banner_url" in col_names has_lang_col = "guild_language" in col_names locale_dir = Path("bot/locales") locale_files = sorted(p.name for p in locale_dir.glob("*.json")) if locale_dir.exists() else [] locale_issues = 0 if locale_files: try: base = json.loads((locale_dir / "en.json").read_text(encoding="utf-8")) def flatten(obj, prefix=""): out = {} if isinstance(obj, dict): for k, v in obj.items(): key = f"{prefix}.{k}" if prefix else k out.update(flatten(v, key)) else: out[prefix] = obj return out base_keys = set(flatten(base).keys()) for name in locale_files: data = json.loads((locale_dir / name).read_text(encoding="utf-8")) keys = set(flatten(data).keys()) if base_keys - keys: locale_issues += 1 except Exception: locale_issues += 1 checks = [ ("Commands registered", total_commands >= 50, str(total_commands)), ("Cogs loaded", loaded_cogs >= 8, str(loaded_cogs)), ("guild_config table", has_guild_config, "ok" if has_guild_config else "missing"), ("guild_language column", has_lang_col, "ok" if has_lang_col else "missing"), ("custom_banner_url column", has_banner_col, "ok" if has_banner_col else "missing"), ("Locale files", bool(locale_files), f"{len(locale_files)} files"), ("Locale consistency", locale_issues == 0, "ok" if locale_issues == 0 else f"{locale_issues} issue(s)"), ] passed = sum(1 for _, ok, _ in checks if ok) status = "HEALTHY" if passed == len(checks) else "NEEDS ATTENTION" color = discord.Color.green() if passed == len(checks) else discord.Color.orange() lines = [f"{'✅' if ok else '⚠️'} {name}: `{detail}`" for name, ok, detail in checks] embed = discord.Embed( title=f"System Audit: {status}", description="\n".join(lines), color=color, ) embed.set_footer(text=f"Passed {passed}/{len(checks)} checks") await ctx.reply(embed=embed) async def setup(bot: commands.Bot) -> None: await bot.add_cog(Observability(bot))