import discord from discord.ext import commands import os import logging from database import Database from ai_handler import AIHandler logger = logging.getLogger('AISupportBot.AdminCog') class AdminCog(commands.Cog): def __init__(self, bot): self.bot = bot self.db = Database() self.ai_handler = AIHandler() self.admin_role_id = int(os.getenv('ADMIN_ROLE_ID')) def is_admin(self, ctx): if isinstance(ctx.channel, discord.DMChannel): return False admin_role = discord.utils.get(ctx.author.roles, id=self.admin_role_id) return admin_role is not None @commands.command(name='admin') async def admin_help(self, ctx): if not self.is_admin(ctx): await ctx.send("You don't have permission to use admin commands.") return embed = discord.Embed( title="🛠️ Admin Commands", description="Available admin commands for managing the support bot:", color=discord.Color.gold() ) embed.add_field( name="📋 FAQ Management", value=( "`!addfaq | | | ` - Add a new FAQ\n" "`!listfaqs [category]` - List all FAQs or FAQs in a category\n" "`!updatefaq | | | | ` - Update an FAQ\n" "`!deletefaq ` - Delete an FAQ\n" "`!searchfaq ` - Search FAQs" ), inline=False ) embed.add_field( name="📊 Statistics", value="`!stats [days]` - View bot usage statistics (default: 7 days)", inline=False ) embed.add_field( name="🔄 Maintenance", value=( "`!clearhistory ` - Clear conversation history for a user\n" "`!resolveescalation ` - Mark an escalation as resolved" ), inline=False ) await ctx.send(embed=embed) @commands.command(name='addfaq') async def add_faq(self, ctx, *, args): if not self.is_admin(ctx): await ctx.send("You don't have permission to use this command.") return try: parts = args.split('|') if len(parts) != 4: await ctx.send("Usage: `!addfaq | | | `") return category = parts[0].strip() question = parts[1].strip() answer = parts[2].strip() keywords = parts[3].strip() faq_id = await self.db.add_faq(question, answer, keywords, category) embed = discord.Embed( title="✅ FAQ Added", description=f"FAQ #{faq_id} has been added successfully!", color=discord.Color.green() ) embed.add_field(name="Category", value=category, inline=False) embed.add_field(name="Question", value=question, inline=False) embed.add_field(name="Answer", value=answer[:1024], inline=False) embed.add_field(name="Keywords", value=keywords, inline=False) await ctx.send(embed=embed) logger.info(f'FAQ #{faq_id} added by {ctx.author}') except Exception as e: logger.error(f'Error adding FAQ: {e}', exc_info=True) await ctx.send("An error occurred while adding the FAQ.") @commands.command(name='listfaqs') async def list_faqs(self, ctx, category: str = None): if not self.is_admin(ctx): await ctx.send("You don't have permission to use this command.") return try: faqs = await self.db.get_all_faqs() if not faqs: await ctx.send("No FAQs found.") return if category: faqs = [faq for faq in faqs if faq[4] and faq[4].lower() == category.lower()] if not faqs: await ctx.send(f"No FAQs found in category '{category}'.") return embed = discord.Embed( title=f"📋 FAQs {f'- {category}' if category else ''}", color=discord.Color.blue() ) current_category = None faq_list = [] for faq in faqs: faq_id, question, answer, keywords, faq_category = faq if faq_category != current_category: if faq_list: embed.add_field( name=f"**{current_category or 'General'}**", value='\n'.join(faq_list), inline=False ) faq_list = [] current_category = faq_category faq_list.append(f"**#{faq_id}** - {question[:100]}{'...' if len(question) > 100 else ''}") if faq_list: embed.add_field( name=f"**{current_category or 'General'}**", value='\n'.join(faq_list), inline=False ) await ctx.send(embed=embed) except Exception as e: logger.error(f'Error listing FAQs: {e}', exc_info=True) await ctx.send("An error occurred while listing FAQs.") @commands.command(name='updatefaq') async def update_faq(self, ctx, *, args): if not self.is_admin(ctx): await ctx.send("You don't have permission to use this command.") return try: parts = args.split('|') if len(parts) != 5: await ctx.send("Usage: `!updatefaq | | | | `") return faq_id = int(parts[0].strip()) question = parts[1].strip() answer = parts[2].strip() keywords = parts[3].strip() category = parts[4].strip() await self.db.update_faq(faq_id, question, answer, keywords, category) embed = discord.Embed( title="✅ FAQ Updated", description=f"FAQ #{faq_id} has been updated successfully!", color=discord.Color.green() ) embed.add_field(name="Category", value=category, inline=False) embed.add_field(name="Question", value=question, inline=False) embed.add_field(name="Answer", value=answer[:1024], inline=False) await ctx.send(embed=embed) logger.info(f'FAQ #{faq_id} updated by {ctx.author}') except ValueError: await ctx.send("Invalid FAQ ID. Please provide a valid number.") except Exception as e: logger.error(f'Error updating FAQ: {e}', exc_info=True) await ctx.send("An error occurred while updating the FAQ.") @commands.command(name='deletefaq') async def delete_faq(self, ctx, faq_id: int): if not self.is_admin(ctx): await ctx.send("You don't have permission to use this command.") return try: await self.db.delete_faq(faq_id) await ctx.send(f"✅ FAQ #{faq_id} has been deleted successfully!") logger.info(f'FAQ #{faq_id} deleted by {ctx.author}') except Exception as e: logger.error(f'Error deleting FAQ: {e}', exc_info=True) await ctx.send("An error occurred while deleting the FAQ.") @commands.command(name='searchfaq') async def search_faq(self, ctx, *, query: str): if not self.is_admin(ctx): await ctx.send("You don't have permission to use this command.") return try: faqs = await self.db.search_faqs(query) if not faqs: await ctx.send(f"No FAQs found matching '{query}'.") return embed = discord.Embed( title=f"🔍 Search Results for '{query}'", color=discord.Color.blue() ) for faq in faqs[:10]: faq_id, question, answer, keywords, category = faq embed.add_field( name=f"#{faq_id} - {question[:100]}", value=f"**Category:** {category or 'General'}\n{answer[:200]}{'...' if len(answer) > 200 else ''}", inline=False ) if len(faqs) > 10: embed.set_footer(text=f"Showing 10 of {len(faqs)} results") await ctx.send(embed=embed) except Exception as e: logger.error(f'Error searching FAQs: {e}', exc_info=True) await ctx.send("An error occurred while searching FAQs.") @commands.command(name='stats') async def stats(self, ctx, days: int = 7): if not self.is_admin(ctx): await ctx.send("You don't have permission to use this command.") return try: stats = await self.db.get_interaction_stats(days) embed = discord.Embed( title=f"📊 Bot Statistics (Last {days} Days)", color=discord.Color.purple() ) embed.add_field(name="Total Interactions", value=stats['total_interactions'], inline=True) embed.add_field(name="Unique Users", value=stats['unique_users'], inline=True) embed.add_field(name="Escalations", value=stats['total_escalations'], inline=True) if stats['total_interactions'] > 0: escalation_rate = (stats['total_escalations'] / stats['total_interactions']) * 100 embed.add_field(name="Escalation Rate", value=f"{escalation_rate:.2f}%", inline=True) await ctx.send(embed=embed) except Exception as e: logger.error(f'Error fetching stats: {e}', exc_info=True) await ctx.send("An error occurred while fetching statistics.") @commands.command(name='clearhistory') async def clear_history(self, ctx, user_id: int): if not self.is_admin(ctx): await ctx.send("You don't have permission to use this command.") return try: self.ai_handler.clear_conversation_history(user_id) await ctx.send(f"✅ Conversation history cleared for user ID {user_id}.") logger.info(f'Conversation history cleared for user {user_id} by {ctx.author}') except Exception as e: logger.error(f'Error clearing history: {e}', exc_info=True) await ctx.send("An error occurred while clearing conversation history.") async def setup(bot): await bot.add_cog(AdminCog(bot))