#!/usr/bin/env python3 """Admin commands module for PetBot""" import sys import os # Add parent directory to path for config import sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from .base_module import BaseModule from config import ADMIN_USER # Import admin user from central config # ============================================================================= # ADMIN CONFIGURATION # ============================================================================= # To change the admin user, edit config.py in the project root # Current admin user: {ADMIN_USER} # ============================================================================= class Admin(BaseModule): """Handles admin-only commands like reload""" def get_commands(self): return ["reload", "rate_stats", "rate_user", "rate_unban", "rate_reset", "weather", "setweather"] async def handle_command(self, channel, nickname, command, args): if command == "reload": await self.cmd_reload(channel, nickname) elif command == "rate_stats": await self.cmd_rate_stats(channel, nickname) elif command == "rate_user": await self.cmd_rate_user(channel, nickname, args) elif command == "rate_unban": await self.cmd_rate_unban(channel, nickname, args) elif command == "rate_reset": await self.cmd_rate_reset(channel, nickname, args) elif command == "weather": await self.cmd_weather(channel, nickname, args) elif command == "setweather": await self.cmd_setweather(channel, nickname, args) async def cmd_reload(self, channel, nickname): """Reload bot modules (admin only)""" if not self.is_admin(nickname): self.send_message(channel, f"{nickname}: Access denied. Admin command.") return try: # Trigger module reload in main bot success = await self.bot.reload_modules() if success: self.send_message(channel, f"{nickname}: ✅ Modules reloaded successfully!") else: self.send_message(channel, f"{nickname}: ❌ Module reload failed!") except Exception as e: self.send_message(channel, f"{nickname}: ❌ Reload error: {str(e)}") def is_admin(self, nickname): """Check if user is admin""" return nickname.lower() == ADMIN_USER.lower() async def cmd_rate_stats(self, channel, nickname): """Show global rate limiting statistics""" if not self.is_admin(nickname): self.send_message(channel, f"{nickname}: Access denied. Admin command.") return if not self.bot.rate_limiter: self.send_message(channel, f"{nickname}: Rate limiter not available.") return try: stats = self.bot.rate_limiter.get_global_stats() response = f"{nickname}: 📊 Rate Limiter Stats:\n" response += f"â€ĸ Status: {'Enabled' if stats['enabled'] else 'Disabled'}\n" response += f"â€ĸ Requests this minute: {stats['requests_this_minute']}\n" response += f"â€ĸ Active users: {stats['active_users']}\n" response += f"â€ĸ Total requests: {stats['total_requests']}\n" response += f"â€ĸ Blocked requests: {stats['blocked_requests']}\n" response += f"â€ĸ Banned users: {stats['banned_users']}\n" response += f"â€ĸ Tracked users: {stats['tracked_users']}\n" response += f"â€ĸ Total violations: {stats['total_violations']}" self.send_message(channel, response) except Exception as e: self.send_message(channel, f"{nickname}: ❌ Error getting rate stats: {str(e)}") async def cmd_rate_user(self, channel, nickname, args): """Show rate limiting stats for a specific user""" if not self.is_admin(nickname): self.send_message(channel, f"{nickname}: Access denied. Admin command.") return if not args: self.send_message(channel, f"{nickname}: Usage: !rate_user ") return if not self.bot.rate_limiter: self.send_message(channel, f"{nickname}: Rate limiter not available.") return try: target_user = args[0] stats = self.bot.rate_limiter.get_user_stats(target_user) response = f"{nickname}: 👤 Rate Stats for {stats['user']}:\n" response += f"â€ĸ Admin exemption: {'Yes' if stats['admin_exemption'] else 'No'}\n" response += f"â€ĸ Currently banned: {'Yes' if stats['is_banned'] else 'No'}\n" if stats['ban_expires']: response += f"â€ĸ Ban expires: {stats['ban_expires']}\n" response += f"â€ĸ Total violations: {stats['violations']}\n" if stats['buckets']: response += f"â€ĸ Available tokens: {stats['buckets']['tokens']}\n" response += f"â€ĸ Last request: {stats['buckets']['last_request']}" else: response += "â€ĸ No recent activity" self.send_message(channel, response) except Exception as e: self.send_message(channel, f"{nickname}: ❌ Error getting user stats: {str(e)}") async def cmd_rate_unban(self, channel, nickname, args): """Manually unban a user from rate limiting""" if not self.is_admin(nickname): self.send_message(channel, f"{nickname}: Access denied. Admin command.") return if not args: self.send_message(channel, f"{nickname}: Usage: !rate_unban ") return if not self.bot.rate_limiter: self.send_message(channel, f"{nickname}: Rate limiter not available.") return try: target_user = args[0] success = self.bot.rate_limiter.unban_user(target_user) if success: self.send_message(channel, f"{nickname}: ✅ User {target_user} has been unbanned.") else: self.send_message(channel, f"{nickname}: ℹī¸ User {target_user} was not banned.") except Exception as e: self.send_message(channel, f"{nickname}: ❌ Error unbanning user: {str(e)}") async def cmd_rate_reset(self, channel, nickname, args): """Reset rate limiting violations for a user""" if not self.is_admin(nickname): self.send_message(channel, f"{nickname}: Access denied. Admin command.") return if not args: self.send_message(channel, f"{nickname}: Usage: !rate_reset ") return if not self.bot.rate_limiter: self.send_message(channel, f"{nickname}: Rate limiter not available.") return try: target_user = args[0] success = self.bot.rate_limiter.reset_user_violations(target_user) if success: self.send_message(channel, f"{nickname}: ✅ Violations reset for user {target_user}.") else: self.send_message(channel, f"{nickname}: ℹī¸ No violations found for user {target_user}.") except Exception as e: self.send_message(channel, f"{nickname}: ❌ Error resetting violations: {str(e)}") async def cmd_weather(self, channel, nickname, args): """Check current weather in locations (admin only)""" if not self.is_admin(nickname): self.send_message(channel, f"{nickname}: Access denied. Admin command.") return try: if args and args[0].lower() != "all": # Check weather for specific location location_name = " ".join(args) weather = await self.database.get_location_weather_by_name(location_name) if weather: self.send_message(channel, f"🌤ī¸ {nickname}: {location_name} - {weather['weather_type']} " f"(modifier: {weather['spawn_modifier']}x, " f"until: {weather['active_until'][:16]})") else: self.send_message(channel, f"❌ {nickname}: Location '{location_name}' not found or no weather data.") else: # Show weather for all locations all_weather = await self.database.get_all_location_weather() if all_weather: weather_info = [] for w in all_weather: weather_info.append(f"{w['location_name']}: {w['weather_type']} ({w['spawn_modifier']}x)") self.send_message(channel, f"🌤ī¸ {nickname}: Current weather - " + " | ".join(weather_info)) else: self.send_message(channel, f"❌ {nickname}: No weather data available.") except Exception as e: self.send_message(channel, f"{nickname}: ❌ Error checking weather: {str(e)}") async def cmd_setweather(self, channel, nickname, args): """Force change weather in a location or all locations (admin only)""" if not self.is_admin(nickname): self.send_message(channel, f"{nickname}: Access denied. Admin command.") return if not args: self.send_message(channel, f"{nickname}: Usage: !setweather [duration_minutes]\n" f"Weather types: sunny, rainy, storm, blizzard, earthquake, calm") return try: import json import random import datetime # Load weather patterns with open("config/weather_patterns.json", "r") as f: weather_data = json.load(f) weather_types = list(weather_data["weather_types"].keys()) # Smart argument parsing - check if any arg is a weather type location_arg = None weather_type = None duration = None for i, arg in enumerate(args): if arg.lower() in weather_types: weather_type = arg.lower() # Remove weather type from args for location parsing remaining_args = args[:i] + args[i+1:] break if not weather_type: self.send_message(channel, f"{nickname}: Please specify a valid weather type.") return # Parse location from remaining args if remaining_args: if remaining_args[0].lower() == "all": location_arg = "all" # Check if there's a duration after "all" if len(remaining_args) > 1: try: duration = int(remaining_args[1]) except ValueError: pass else: # Location name (might be multiple words) location_words = [] for arg in remaining_args: try: # If it's a number, it's probably duration duration = int(arg) break except ValueError: # It's part of location name location_words.append(arg) location_arg = " ".join(location_words) if location_words else "all" else: location_arg = "all" weather_config = weather_data["weather_types"][weather_type] # Calculate duration if not duration: duration_range = weather_config.get("duration_minutes", [90, 180]) duration = random.randint(duration_range[0], duration_range[1]) end_time = datetime.datetime.now() + datetime.timedelta(minutes=duration) if location_arg.lower() == "all": # Set weather for all locations success = await self.database.set_weather_all_locations( weather_type, end_time.isoformat(), weather_config.get("spawn_modifier", 1.0), ",".join(weather_config.get("affected_types", [])) ) if success: self.send_message(channel, f"🌤ī¸ {nickname}: Set {weather_type} weather for ALL locations! " f"Duration: {duration} minutes, Modifier: {weather_config.get('spawn_modifier', 1.0)}x") else: self.send_message(channel, f"❌ {nickname}: Failed to set weather for all locations.") else: # Set weather for specific location location_name = location_arg if len(args) == 2 else " ".join(args[:-1]) success = await self.database.set_weather_for_location( location_name, weather_type, end_time.isoformat(), weather_config.get("spawn_modifier", 1.0), ",".join(weather_config.get("affected_types", [])) ) if success: self.send_message(channel, f"🌤ī¸ {nickname}: Set {weather_type} weather for {location_name}! " f"Duration: {duration} minutes, Modifier: {weather_config.get('spawn_modifier', 1.0)}x") else: self.send_message(channel, f"❌ {nickname}: Failed to set weather for '{location_name}'. Location may not exist.") except FileNotFoundError: self.send_message(channel, f"{nickname}: ❌ Weather configuration file not found.") except ValueError as e: self.send_message(channel, f"{nickname}: ❌ Invalid duration: {str(e)}") except Exception as e: self.send_message(channel, f"{nickname}: ❌ Error setting weather: {str(e)}")