cogs fuckery (i have zero clue how this is supposed to work you're on your own for this one glitch)
This commit is contained in:
parent
c3114639a2
commit
fbfa27c979
173
bot/cogs/cog_manager.py
Normal file
173
bot/cogs/cog_manager.py
Normal file
|
@ -0,0 +1,173 @@
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import importlib
|
||||||
|
import importlib.util
|
||||||
|
import inspect
|
||||||
|
from typing import Dict, Any, Optional
|
||||||
|
|
||||||
|
class BaseCog:
|
||||||
|
"""Base cog class that all cogs should inherit from"""
|
||||||
|
def __init__(self, bot):
|
||||||
|
self.bot = bot
|
||||||
|
self.name = self.__class__.__name__
|
||||||
|
|
||||||
|
def cleanup(self):
|
||||||
|
"""Called when cog is unloaded"""
|
||||||
|
pass
|
||||||
|
|
||||||
|
class CogManager:
|
||||||
|
def __init__(self, bot):
|
||||||
|
self.bot = bot
|
||||||
|
self.cogs: Dict[str, BaseCog] = {}
|
||||||
|
self.cogs_dir = os.path.join("bot", "cogs")
|
||||||
|
os.makedirs(self.cogs_dir, exist_ok=True)
|
||||||
|
|
||||||
|
def load_cog(self, cog_name: str) -> bool:
|
||||||
|
"""
|
||||||
|
Load a cog by name
|
||||||
|
|
||||||
|
Args:
|
||||||
|
cog_name: Name of the cog file (without .py)
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
bool: True if loaded successfully, False otherwise
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
# If already loaded, unload first
|
||||||
|
if cog_name in self.cogs:
|
||||||
|
self.unload_cog(cog_name)
|
||||||
|
|
||||||
|
# Get full path to cog file
|
||||||
|
cog_path = os.path.join(self.cogs_dir, f"{cog_name}.py")
|
||||||
|
if not os.path.exists(cog_path):
|
||||||
|
print(f"Cog file not found: {cog_path}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
# Create a unique module name to avoid conflicts
|
||||||
|
module_name = f"bot.cogs.{cog_name}"
|
||||||
|
|
||||||
|
# Load the module
|
||||||
|
spec = importlib.util.spec_from_file_location(module_name, cog_path)
|
||||||
|
if not spec:
|
||||||
|
print(f"Failed to create spec for {cog_path}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
module = importlib.util.module_from_spec(spec)
|
||||||
|
sys.modules[module_name] = module
|
||||||
|
spec.loader.exec_module(module)
|
||||||
|
|
||||||
|
# Find and instantiate the cog class
|
||||||
|
# A cog class is a class that inherits from BaseCog
|
||||||
|
cog_class = None
|
||||||
|
for name, obj in inspect.getmembers(module):
|
||||||
|
if (inspect.isclass(obj) and
|
||||||
|
issubclass(obj, BaseCog) and
|
||||||
|
obj is not BaseCog):
|
||||||
|
cog_class = obj
|
||||||
|
break
|
||||||
|
|
||||||
|
if not cog_class:
|
||||||
|
print(f"No cog class found in {cog_path}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
# Instantiate the cog
|
||||||
|
cog = cog_class(self.bot)
|
||||||
|
self.cogs[cog_name] = cog
|
||||||
|
|
||||||
|
# Register commands
|
||||||
|
for name, method in inspect.getmembers(cog, inspect.ismethod):
|
||||||
|
if name.startswith("cmd_"):
|
||||||
|
cmd_name = name[4:] # Remove "cmd_" prefix
|
||||||
|
self.bot.loaded_commands[cmd_name] = method
|
||||||
|
|
||||||
|
print(f"Loaded cog: {cog_name}")
|
||||||
|
return True
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Error loading cog {cog_name}: {e}")
|
||||||
|
import traceback
|
||||||
|
traceback.print_exc()
|
||||||
|
return False
|
||||||
|
|
||||||
|
def unload_cog(self, cog_name: str) -> bool:
|
||||||
|
"""
|
||||||
|
Unload a cog by name
|
||||||
|
|
||||||
|
Args:
|
||||||
|
cog_name: Name of the cog
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
bool: True if unloaded successfully, False otherwise
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
if cog_name not in self.cogs:
|
||||||
|
print(f"Cog not loaded: {cog_name}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
cog = self.cogs[cog_name]
|
||||||
|
|
||||||
|
# Call cleanup method
|
||||||
|
cog.cleanup()
|
||||||
|
|
||||||
|
# Remove commands from loaded_commands
|
||||||
|
for name, method in inspect.getmembers(cog, inspect.ismethod):
|
||||||
|
if name.startswith("cmd_"):
|
||||||
|
cmd_name = name[4:]
|
||||||
|
if cmd_name in self.bot.loaded_commands:
|
||||||
|
del self.bot.loaded_commands[cmd_name]
|
||||||
|
|
||||||
|
# Remove cog
|
||||||
|
del self.cogs[cog_name]
|
||||||
|
|
||||||
|
# Remove module
|
||||||
|
module_name = f"bot.cogs.{cog_name}"
|
||||||
|
if module_name in sys.modules:
|
||||||
|
del sys.modules[module_name]
|
||||||
|
|
||||||
|
print(f"Unloaded cog: {cog_name}")
|
||||||
|
return True
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Error unloading cog {cog_name}: {e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
def reload_cog(self, cog_name: str) -> bool:
|
||||||
|
"""
|
||||||
|
Reload a cog by name
|
||||||
|
|
||||||
|
Args:
|
||||||
|
cog_name: Name of the cog
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
bool: True if reloaded successfully, False otherwise
|
||||||
|
"""
|
||||||
|
return self.load_cog(cog_name)
|
||||||
|
|
||||||
|
def load_all_cogs(self) -> int:
|
||||||
|
"""
|
||||||
|
Load all cogs from the cogs directory
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
int: Number of cogs loaded successfully
|
||||||
|
"""
|
||||||
|
count = 0
|
||||||
|
for filename in os.listdir(self.cogs_dir):
|
||||||
|
if filename.endswith(".py") and not filename.startswith("_"):
|
||||||
|
cog_name = filename[:-3]
|
||||||
|
if self.load_cog(cog_name):
|
||||||
|
count += 1
|
||||||
|
return count
|
||||||
|
|
||||||
|
def unload_all_cogs(self) -> int:
|
||||||
|
"""
|
||||||
|
Unload all cogs
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
int: Number of cogs unloaded successfully
|
||||||
|
"""
|
||||||
|
cog_names = list(self.cogs.keys())
|
||||||
|
count = 0
|
||||||
|
for cog_name in cog_names:
|
||||||
|
if self.unload_cog(cog_name):
|
||||||
|
count += 1
|
||||||
|
return count
|
33
bot/cogs/example_cog.py
Normal file
33
bot/cogs/example_cog.py
Normal file
|
@ -0,0 +1,33 @@
|
||||||
|
from bot.cogs.cog_manager import BaseCog
|
||||||
|
import discord
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
class ExampleCog(BaseCog):
|
||||||
|
def __init__(self, bot):
|
||||||
|
super().__init__(bot)
|
||||||
|
# Initialize any state for this cog
|
||||||
|
self.counter = 0
|
||||||
|
|
||||||
|
def cleanup(self):
|
||||||
|
# Clean up any resources when unloaded
|
||||||
|
print("Example cog is being unloaded and cleaned up")
|
||||||
|
|
||||||
|
async def cmd_hello(self, message):
|
||||||
|
"""
|
||||||
|
A simple hello command
|
||||||
|
Usage: .hello
|
||||||
|
"""
|
||||||
|
self.counter += 1
|
||||||
|
await message.edit(content=f"👋 Hello! I've been greeted {self.counter} times.")
|
||||||
|
|
||||||
|
async def cmd_echo(self, message):
|
||||||
|
"""
|
||||||
|
Echo the user's message
|
||||||
|
Usage: .echo <message>
|
||||||
|
"""
|
||||||
|
content = message.content
|
||||||
|
if ' ' in content:
|
||||||
|
text = content.split(' ', 1)[1]
|
||||||
|
await message.edit(content=f"🔊 {text}")
|
||||||
|
else:
|
||||||
|
await message.edit(content="❌ Usage: `.echo <message>`")
|
|
@ -62,4 +62,70 @@ class AdminCommands:
|
||||||
save_tracked_channels(self.bot.tracked_channels)
|
save_tracked_channels(self.bot.tracked_channels)
|
||||||
await message.reply(f"Stopped tracking messages in {message.channel.name}.", silent=True)
|
await message.reply(f"Stopped tracking messages in {message.channel.name}.", silent=True)
|
||||||
else:
|
else:
|
||||||
await message.reply("This channel is not being tracked.", silent=True)
|
await message.reply("This channel is not being tracked.", silent=True)
|
||||||
|
|
||||||
|
async def cmd_loadcog(self, message):
|
||||||
|
"""
|
||||||
|
Load a cog
|
||||||
|
Usage: .loadcog <cog_name>
|
||||||
|
"""
|
||||||
|
content = message.content.strip()
|
||||||
|
parts = content.split()
|
||||||
|
|
||||||
|
if len(parts) != 2:
|
||||||
|
await message.edit(content="❌ Usage: `.loadcog <cog_name>`")
|
||||||
|
return
|
||||||
|
|
||||||
|
cog_name = parts[1]
|
||||||
|
if self.bot.cog_manager.load_cog(cog_name):
|
||||||
|
await message.edit(content=f"✅ Loaded cog: `{cog_name}`")
|
||||||
|
else:
|
||||||
|
await message.edit(content=f"❌ Failed to load cog: `{cog_name}`")
|
||||||
|
|
||||||
|
async def cmd_unloadcog(self, message):
|
||||||
|
"""
|
||||||
|
Unload a cog
|
||||||
|
Usage: .unloadcog <cog_name>
|
||||||
|
"""
|
||||||
|
content = message.content.strip()
|
||||||
|
parts = content.split()
|
||||||
|
|
||||||
|
if len(parts) != 2:
|
||||||
|
await message.edit(content="❌ Usage: `.unloadcog <cog_name>`")
|
||||||
|
return
|
||||||
|
|
||||||
|
cog_name = parts[1]
|
||||||
|
if self.bot.cog_manager.unload_cog(cog_name):
|
||||||
|
await message.edit(content=f"✅ Unloaded cog: `{cog_name}`")
|
||||||
|
else:
|
||||||
|
await message.edit(content=f"❌ Failed to unload cog: `{cog_name}`")
|
||||||
|
|
||||||
|
async def cmd_reloadcog(self, message):
|
||||||
|
"""
|
||||||
|
Reload a cog
|
||||||
|
Usage: .reloadcog <cog_name>
|
||||||
|
"""
|
||||||
|
content = message.content.strip()
|
||||||
|
parts = content.split()
|
||||||
|
|
||||||
|
if len(parts) != 2:
|
||||||
|
await message.edit(content="❌ Usage: `.reloadcog <cog_name>`")
|
||||||
|
return
|
||||||
|
|
||||||
|
cog_name = parts[1]
|
||||||
|
if self.bot.cog_manager.reload_cog(cog_name):
|
||||||
|
await message.edit(content=f"✅ Reloaded cog: `{cog_name}`")
|
||||||
|
else:
|
||||||
|
await message.edit(content=f"❌ Failed to reload cog: `{cog_name}`")
|
||||||
|
|
||||||
|
async def cmd_listcogs(self, message):
|
||||||
|
"""
|
||||||
|
List all loaded cogs
|
||||||
|
Usage: .listcogs
|
||||||
|
"""
|
||||||
|
if not self.bot.cog_manager.cogs:
|
||||||
|
await message.edit(content="No cogs are currently loaded.")
|
||||||
|
return
|
||||||
|
|
||||||
|
cog_list = "\n".join(f"• {name}" for name in sorted(self.bot.cog_manager.cogs.keys()))
|
||||||
|
await message.edit(content=f"**Loaded Cogs:**\n{cog_list}")
|
|
@ -134,20 +134,22 @@ class MessageHandler:
|
||||||
"""Handle commands issued by the bot user"""
|
"""Handle commands issued by the bot user"""
|
||||||
content = message.content
|
content = message.content
|
||||||
|
|
||||||
# Check for custom commands first
|
# Skip if not a command
|
||||||
if content.startswith('.'):
|
if not content.startswith('.'):
|
||||||
cmd = content.split()[0][1:] # Remove the '.' and get the command name
|
return
|
||||||
if hasattr(self.bot, 'loaded_commands') and cmd in self.bot.loaded_commands:
|
|
||||||
# Execute the custom command
|
cmd_parts = content.split()
|
||||||
try:
|
cmd = cmd_parts[0][1:] # Get command name without the '.'
|
||||||
# Extract arguments from the command
|
|
||||||
args = content.split(' ')[1:] if ' ' in content else []
|
# Check for custom/loaded commands first
|
||||||
await self.bot.loaded_commands[cmd](self.bot, message, args)
|
if hasattr(self.bot, 'loaded_commands') and cmd in self.bot.loaded_commands:
|
||||||
return
|
try:
|
||||||
except Exception as e:
|
await self.bot.loaded_commands[cmd](message)
|
||||||
print(f"Error executing custom command {cmd}: {e}")
|
return
|
||||||
await message.channel.send(f"Error executing command: {e}")
|
except Exception as e:
|
||||||
return
|
print(f"Error executing command {cmd}: {e}")
|
||||||
|
await message.edit(content=f"❌ Error executing command: {e}")
|
||||||
|
return
|
||||||
|
|
||||||
# User Management Commands - only keeping close and block
|
# User Management Commands - only keeping close and block
|
||||||
if content.startswith(".close"):
|
if content.startswith(".close"):
|
||||||
|
|
|
@ -21,6 +21,13 @@ class Selfbot(discord.Client):
|
||||||
self.tracking_handler = TrackingHandler(self)
|
self.tracking_handler = TrackingHandler(self)
|
||||||
self.presence_handler = PresenceHandler(self)
|
self.presence_handler = PresenceHandler(self)
|
||||||
|
|
||||||
|
# Add to SelfBot.__init__
|
||||||
|
from bot.cogs.cog_manager import CogManager
|
||||||
|
|
||||||
|
# In the __init__ method:
|
||||||
|
self.cog_manager = CogManager(self)
|
||||||
|
self.cog_manager.load_all_cogs()
|
||||||
|
|
||||||
async def on_ready(self):
|
async def on_ready(self):
|
||||||
print(f"Logged in as {self.user}")
|
print(f"Logged in as {self.user}")
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue