i left skid marks in the code
This commit is contained in:
Xargana
parent
fbfa27c979
commit
282ca4374a
13 changed files with 942 additions and 523 deletions
1
bot/cogs/__init__.py
Normal file
1
bot/cogs/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
# This makes the cogs directory a proper Python package
|
||||
129
bot/cogs/admin_commands.py
Normal file
129
bot/cogs/admin_commands.py
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
from bot.cogs.cog_manager import BaseCog
|
||||
import discord
|
||||
import os
|
||||
import importlib.util
|
||||
|
||||
class AdminCog(BaseCog):
|
||||
def __init__(self, bot):
|
||||
super().__init__(bot)
|
||||
self.bot = bot
|
||||
# Initialize tracked channels if not already present
|
||||
if not hasattr(self.bot, 'tracked_channels'):
|
||||
self.bot.tracked_channels = []
|
||||
|
||||
async def cmd_addcmd(self, message):
|
||||
"""
|
||||
Add a custom command
|
||||
Usage: .addcmd <name> <code>
|
||||
"""
|
||||
content = message.content.strip()
|
||||
parts = content.split(None, 2)
|
||||
|
||||
if len(parts) < 3:
|
||||
await message.edit(content="❌ Usage: `.addcmd <name> <code>`")
|
||||
return
|
||||
|
||||
cmd_name = parts[1]
|
||||
cmd_code = parts[2]
|
||||
|
||||
# Create commands directory if it doesn't exist
|
||||
commands_dir = os.path.join("bot", "commands", "custom")
|
||||
os.makedirs(commands_dir, exist_ok=True)
|
||||
|
||||
# Create command file
|
||||
cmd_path = os.path.join(commands_dir, f"{cmd_name}.py")
|
||||
|
||||
with open(cmd_path, "w") as f:
|
||||
f.write(f"""
|
||||
async def run(bot, message, args):
|
||||
\"\"\"
|
||||
Custom command: {cmd_name}
|
||||
\"\"\"
|
||||
try:
|
||||
{cmd_code.replace(chr(10), chr(10) + ' ' * 8)}
|
||||
except Exception as e:
|
||||
await message.channel.send(f"Error executing command: {{str(e)}}")
|
||||
""")
|
||||
|
||||
# Reload commands
|
||||
self.bot.reload_commands()
|
||||
|
||||
await message.edit(content=f"✅ Added command: `{cmd_name}`")
|
||||
|
||||
async def cmd_delcmd(self, message):
|
||||
"""
|
||||
Delete a custom command
|
||||
Usage: .delcmd <name>
|
||||
"""
|
||||
content = message.content.strip()
|
||||
parts = content.split()
|
||||
|
||||
if len(parts) != 2:
|
||||
await message.edit(content="❌ Usage: `.delcmd <name>`")
|
||||
return
|
||||
|
||||
cmd_name = parts[1]
|
||||
|
||||
# Check if command exists
|
||||
commands_dir = os.path.join("bot", "commands", "custom")
|
||||
cmd_path = os.path.join(commands_dir, f"{cmd_name}.py")
|
||||
|
||||
if not os.path.exists(cmd_path):
|
||||
await message.edit(content=f"❌ Command `{cmd_name}` does not exist")
|
||||
return
|
||||
|
||||
# Delete command
|
||||
os.remove(cmd_path)
|
||||
|
||||
# Reload commands
|
||||
self.bot.reload_commands()
|
||||
|
||||
await message.edit(content=f"✅ Deleted command: `{cmd_name}`")
|
||||
|
||||
async def cmd_listcmds(self, message):
|
||||
"""
|
||||
List all custom commands
|
||||
Usage: .listcmds
|
||||
"""
|
||||
commands_dir = os.path.join("bot", "commands", "custom")
|
||||
os.makedirs(commands_dir, exist_ok=True)
|
||||
|
||||
cmds = []
|
||||
for filename in os.listdir(commands_dir):
|
||||
if filename.endswith(".py"):
|
||||
cmds.append(filename[:-3])
|
||||
|
||||
if not cmds:
|
||||
await message.edit(content="No custom commands found.")
|
||||
return
|
||||
|
||||
cmd_list = "\n".join(f"• {cmd}" for cmd in sorted(cmds))
|
||||
await message.edit(content=f"**Custom Commands:**\n{cmd_list}")
|
||||
|
||||
async def cmd_trackmessages(self, message):
|
||||
"""
|
||||
Track message edits and deletions in a channel
|
||||
Usage: .trackmessages
|
||||
"""
|
||||
channel_id = message.channel.id
|
||||
|
||||
if channel_id in self.bot.tracked_channels:
|
||||
await message.edit(content="❌ This channel is already being tracked.")
|
||||
return
|
||||
|
||||
self.bot.tracked_channels.append(channel_id)
|
||||
await message.edit(content=f"✅ Now tracking message edits and deletions in <#{channel_id}>")
|
||||
|
||||
async def cmd_untrackmessages(self, message):
|
||||
"""
|
||||
Stop tracking message edits and deletions in a channel
|
||||
Usage: .untrackmessages
|
||||
"""
|
||||
channel_id = message.channel.id
|
||||
|
||||
if channel_id not in self.bot.tracked_channels:
|
||||
await message.edit(content="❌ This channel is not being tracked.")
|
||||
return
|
||||
|
||||
self.bot.tracked_channels.remove(channel_id)
|
||||
await message.edit(content=f"✅ No longer tracking message edits and deletions in <#{channel_id}>")
|
||||
96
bot/cogs/afk_commands.py
Normal file
96
bot/cogs/afk_commands.py
Normal file
|
|
@ -0,0 +1,96 @@
|
|||
from bot.cogs.cog_manager import BaseCog
|
||||
import discord
|
||||
import datetime
|
||||
import re
|
||||
|
||||
class AFKCog(BaseCog):
|
||||
def __init__(self, bot):
|
||||
super().__init__(bot)
|
||||
self.afk = False
|
||||
self.afk_since = None
|
||||
self.afk_reason = None
|
||||
|
||||
def cleanup(self):
|
||||
self.afk = False
|
||||
self.afk_since = None
|
||||
self.afk_reason = None
|
||||
|
||||
async def cmd_afk(self, message):
|
||||
"""
|
||||
Set AFK status
|
||||
Usage: .afk [reason]
|
||||
"""
|
||||
parts = message.content.split(' ', 1)
|
||||
reason = parts[1] if len(parts) > 1 else "AFK"
|
||||
|
||||
self.afk = True
|
||||
self.afk_since = datetime.datetime.now()
|
||||
self.afk_reason = reason
|
||||
|
||||
await message.edit(content=f"You are now AFK: {reason}")
|
||||
|
||||
async def cmd_unafk(self, message):
|
||||
"""
|
||||
Remove AFK status
|
||||
Usage: .unafk
|
||||
"""
|
||||
if not self.afk:
|
||||
await message.edit(content="You are not AFK.")
|
||||
return
|
||||
|
||||
self.afk = False
|
||||
afk_duration = datetime.datetime.now() - self.afk_since
|
||||
hours, remainder = divmod(afk_duration.seconds, 3600)
|
||||
minutes, seconds = divmod(remainder, 60)
|
||||
|
||||
duration_str = ""
|
||||
if hours > 0:
|
||||
duration_str += f"{hours} hour{'s' if hours != 1 else ''} "
|
||||
if minutes > 0:
|
||||
duration_str += f"{minutes} minute{'s' if minutes != 1 else ''} "
|
||||
if seconds > 0 or (hours == 0 and minutes == 0):
|
||||
duration_str += f"{seconds} second{'s' if seconds != 1 else ''}"
|
||||
|
||||
await message.edit(content=f"Welcome back! You were AFK for {duration_str.strip()}.")
|
||||
|
||||
self.afk_since = None
|
||||
self.afk_reason = None
|
||||
|
||||
async def handle_afk_dm(self, message):
|
||||
"""
|
||||
Handle incoming DMs when in AFK mode
|
||||
"""
|
||||
if not self.afk:
|
||||
return
|
||||
|
||||
# Don't respond to self
|
||||
if message.author == self.bot.user:
|
||||
return
|
||||
|
||||
# Check if we already replied recently
|
||||
# This prevents spam if someone sends multiple messages
|
||||
async for msg in message.channel.history(limit=10):
|
||||
if msg.author == self.bot.user and "I'm currently AFK" in msg.content:
|
||||
# Only reply once every 5 minutes to the same person
|
||||
time_diff = datetime.datetime.now(datetime.timezone.utc) - msg.created_at
|
||||
if time_diff.total_seconds() < 300: # 5 minutes
|
||||
return
|
||||
|
||||
# Calculate AFK duration
|
||||
afk_duration = datetime.datetime.now() - self.afk_since
|
||||
hours, remainder = divmod(afk_duration.seconds, 3600)
|
||||
minutes, seconds = divmod(remainder, 60)
|
||||
|
||||
duration_str = ""
|
||||
if hours > 0:
|
||||
duration_str += f"{hours} hour{'s' if hours != 1 else ''} "
|
||||
if minutes > 0:
|
||||
duration_str += f"{minutes} minute{'s' if minutes != 1 else ''} "
|
||||
if seconds > 0 or (hours == 0 and minutes == 0):
|
||||
duration_str += f"{seconds} second{'s' if seconds != 1 else ''}"
|
||||
|
||||
# Send AFK message
|
||||
await message.channel.send(
|
||||
f"I'm currently AFK ({duration_str.strip()}): {self.afk_reason}\n"
|
||||
f"I'll respond when I return."
|
||||
)
|
||||
85
bot/cogs/cog_commands.py
Normal file
85
bot/cogs/cog_commands.py
Normal file
|
|
@ -0,0 +1,85 @@
|
|||
from bot.cogs.cog_manager import BaseCog
|
||||
|
||||
class CogCommands(BaseCog):
|
||||
def __init__(self, bot):
|
||||
super().__init__(bot)
|
||||
|
||||
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]
|
||||
# Prevent unloading the cog_commands cog
|
||||
if cog_name == "cog_commands":
|
||||
await message.edit(content="❌ Cannot unload the cog management commands.")
|
||||
return
|
||||
|
||||
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_reloadall(self, message):
|
||||
"""
|
||||
Reload all cogs
|
||||
Usage: .reloadall
|
||||
"""
|
||||
self.bot.cog_manager.unload_all_cogs()
|
||||
num_loaded = self.bot.cog_manager.load_all_cogs()
|
||||
await message.edit(content=f"✅ Reloaded {num_loaded} cogs.")
|
||||
|
||||
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}")
|
||||
72
bot/cogs/fun_commands.py
Normal file
72
bot/cogs/fun_commands.py
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
from bot.cogs.cog_manager import BaseCog
|
||||
import discord
|
||||
import random
|
||||
import asyncio
|
||||
|
||||
class FunCog(BaseCog):
|
||||
def __init__(self, bot):
|
||||
super().__init__(bot)
|
||||
|
||||
async def cmd_horse(self, message):
|
||||
"""
|
||||
Toggle horse reactions in a channel
|
||||
Usage: .horse
|
||||
"""
|
||||
if message.channel.id in self.bot.horsin:
|
||||
self.bot.horsin.remove(message.channel.id)
|
||||
await message.reply("no longer horsin around D:")
|
||||
else:
|
||||
self.bot.horsin.append(message.channel.id)
|
||||
await message.reply(":D")
|
||||
|
||||
async def cmd_rps(self, message):
|
||||
"""
|
||||
Play Rock Paper Scissors
|
||||
Usage: .rps <rock|paper|scissors>
|
||||
"""
|
||||
choices = ["rock", "paper", "scissors"]
|
||||
content = message.content.strip().lower()
|
||||
parts = content.split(maxsplit=1)
|
||||
|
||||
if len(parts) != 2 or parts[1] not in choices:
|
||||
await message.edit(content="❌ Usage: `.rps <rock|paper|scissors>`")
|
||||
return
|
||||
|
||||
user_choice = parts[1]
|
||||
bot_choice = random.choice(choices)
|
||||
|
||||
# Determine winner
|
||||
if user_choice == bot_choice:
|
||||
result = "It's a tie! 🤝"
|
||||
elif (user_choice == "rock" and bot_choice == "scissors") or \
|
||||
(user_choice == "paper" and bot_choice == "rock") or \
|
||||
(user_choice == "scissors" and bot_choice == "paper"):
|
||||
result = "You win! 🎉"
|
||||
else:
|
||||
result = "I win! 🎮"
|
||||
|
||||
await message.edit(content=f"You chose {user_choice}, I chose {bot_choice}. {result}")
|
||||
|
||||
async def cmd_repeat(self, message):
|
||||
"""
|
||||
Repeat a message multiple times
|
||||
Usage: .repeat29 <text>
|
||||
"""
|
||||
content = message.content.strip()
|
||||
if " " not in content:
|
||||
await message.edit(content="❌ Usage: `.repeat29 <text>`")
|
||||
return
|
||||
|
||||
text = content.split(" ", 1)[1]
|
||||
|
||||
# Don't allow mentions in repeated text
|
||||
if "@" in text:
|
||||
text = text.replace("@", "@\u200b") # Insert zero-width space to break mentions
|
||||
|
||||
# Delete original command
|
||||
await message.delete()
|
||||
|
||||
# Send 29 times
|
||||
for _ in range(29):
|
||||
await message.channel.send(text)
|
||||
await asyncio.sleep(0.5) # Add small delay to avoid rate limiting
|
||||
104
bot/cogs/test_commands.py
Normal file
104
bot/cogs/test_commands.py
Normal file
|
|
@ -0,0 +1,104 @@
|
|||
from bot.cogs.cog_manager import BaseCog
|
||||
import discord
|
||||
import asyncio
|
||||
import traceback
|
||||
from unittest.mock import patch
|
||||
|
||||
class TestCog(BaseCog):
|
||||
def __init__(self, bot):
|
||||
super().__init__(bot)
|
||||
|
||||
async def cmd_test(self, message):
|
||||
"""
|
||||
Run tests to verify bot functionality
|
||||
Usage: .test
|
||||
"""
|
||||
await message.edit(content="🧪 Running tests...")
|
||||
|
||||
results = []
|
||||
total = 0
|
||||
passed = 0
|
||||
|
||||
# Test AFK feature
|
||||
total += 1
|
||||
try:
|
||||
# Check if AFK cog is loaded
|
||||
afk_cog = next((cog for cog in self.bot.cog_manager.cogs.values()
|
||||
if hasattr(cog, "afk")), None)
|
||||
|
||||
if afk_cog:
|
||||
original_afk = afk_cog.afk
|
||||
afk_cog.afk = False
|
||||
|
||||
# Create a test message
|
||||
test_msg = await message.channel.send(".afk Test AFK")
|
||||
|
||||
# Run the AFK command
|
||||
await self.bot.message_handler.handle_commands(test_msg)
|
||||
|
||||
# Check if AFK was set correctly
|
||||
if afk_cog.afk and afk_cog.afk_reason == "Test AFK":
|
||||
results.append("✅ AFK test passed")
|
||||
passed += 1
|
||||
else:
|
||||
results.append("❌ AFK test failed")
|
||||
|
||||
# Clean up
|
||||
await test_msg.delete()
|
||||
afk_cog.afk = original_afk
|
||||
else:
|
||||
results.append("⚠️ AFK test skipped - cog not loaded")
|
||||
except Exception as e:
|
||||
results.append(f"❌ AFK test error: {str(e)}")
|
||||
|
||||
# Test message detection
|
||||
total += 1
|
||||
try:
|
||||
reply_received = False
|
||||
|
||||
# Create a message that should trigger a reply
|
||||
test_msg = await message.channel.send("This is a test message")
|
||||
|
||||
# Mock the reply method
|
||||
original_reply = test_msg.reply
|
||||
|
||||
async def test_reply(content, **kwargs):
|
||||
nonlocal reply_received
|
||||
if content == "test success":
|
||||
reply_received = True
|
||||
return await original_reply(content, **kwargs)
|
||||
|
||||
# Patch the reply method
|
||||
with patch.object(test_msg, 'reply', side_effect=test_reply):
|
||||
# Create a handler specifically for this test
|
||||
test_handler = lambda m: asyncio.ensure_future(test_msg.reply("test success"))
|
||||
|
||||
# Add the handler
|
||||
self.bot.add_listener(test_handler, name='on_message')
|
||||
|
||||
# Trigger it
|
||||
await self.bot.message_handler.handle_message(test_msg)
|
||||
|
||||
# Wait a moment
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# Remove the handler
|
||||
self.bot.remove_listener(test_handler, name='on_message')
|
||||
|
||||
if reply_received:
|
||||
results.append("✅ Message handler test passed")
|
||||
passed += 1
|
||||
else:
|
||||
results.append("❌ Message handler test failed")
|
||||
|
||||
# Clean up
|
||||
await test_msg.delete()
|
||||
|
||||
except Exception as e:
|
||||
results.append(f"❌ Message handler test error: {traceback.format_exc()}")
|
||||
|
||||
# Show results
|
||||
result_text = "\n".join(results)
|
||||
summary = f"**Test Results:** {passed}/{total} passed"
|
||||
|
||||
await message.edit(content=f"{summary}\n\n{result_text}")
|
||||
98
bot/cogs/tracking.py
Normal file
98
bot/cogs/tracking.py
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
from bot.cogs.cog_manager import BaseCog
|
||||
import discord
|
||||
import difflib
|
||||
import io
|
||||
|
||||
class TrackingCog(BaseCog):
|
||||
def __init__(self, bot):
|
||||
super().__init__(bot)
|
||||
self.whitelist = set()
|
||||
|
||||
async def on_message_delete(self, message):
|
||||
"""Handle when messages are deleted"""
|
||||
if await self.process_log_whitelist(message):
|
||||
return
|
||||
|
||||
if message.channel.id in self.bot.tracked_channels:
|
||||
member = message.author
|
||||
if member != self.bot.user:
|
||||
await message.channel.send(
|
||||
f"<@{member.id}> deleted {message.content}",
|
||||
silent=True
|
||||
)
|
||||
|
||||
async def on_message_edit(self, before, after):
|
||||
"""Handle when messages are edited"""
|
||||
if await self.process_log_whitelist(before):
|
||||
return
|
||||
|
||||
if before.channel.id in self.bot.tracked_channels:
|
||||
member = after.author
|
||||
if member == self.bot.user:
|
||||
return
|
||||
|
||||
if before.content == after.content:
|
||||
return
|
||||
|
||||
diff = difflib.unified_diff(before.content.splitlines(), after.content.splitlines())
|
||||
diff_result = '\n'.join(diff)
|
||||
|
||||
with io.BytesIO(diff_result.encode('utf-8')) as diff_file:
|
||||
diff_file.seek(0)
|
||||
|
||||
await after.channel.send(
|
||||
f"<@{member.id}> edited a message",
|
||||
file=discord.File(diff_file, "diff.txt"),
|
||||
silent=True
|
||||
)
|
||||
|
||||
async def process_log_whitelist(self, message):
|
||||
"""Check if channel or author is in whitelist"""
|
||||
return (message.channel.id in self.whitelist or
|
||||
(hasattr(message, 'author') and message.author.id in self.whitelist))
|
||||
|
||||
async def cmd_addlogwhitelist(self, message):
|
||||
"""
|
||||
Add a channel or user to logging whitelist
|
||||
Usage: .addlogwhitelist [user/channel mention]
|
||||
"""
|
||||
if not message.mentions and not message.channel_mentions:
|
||||
# If no mentions, use current channel
|
||||
self.whitelist.add(message.channel.id)
|
||||
await message.edit(content=f"✅ Added <#{message.channel.id}> to log whitelist")
|
||||
return
|
||||
|
||||
for user in message.mentions:
|
||||
self.whitelist.add(user.id)
|
||||
|
||||
for channel in message.channel_mentions:
|
||||
self.whitelist.add(channel.id)
|
||||
|
||||
await message.edit(content="✅ Added mentioned entities to log whitelist")
|
||||
|
||||
async def cmd_removelogwhitelist(self, message):
|
||||
"""
|
||||
Remove a channel or user from logging whitelist
|
||||
Usage: .removelogwhitelist [user/channel mention]
|
||||
"""
|
||||
if not message.mentions and not message.channel_mentions:
|
||||
# If no mentions, use current channel
|
||||
self.whitelist.discard(message.channel.id)
|
||||
await message.edit(content=f"✅ Removed <#{message.channel.id}> from log whitelist")
|
||||
return
|
||||
|
||||
for user in message.mentions:
|
||||
self.whitelist.discard(user.id)
|
||||
|
||||
for channel in message.channel_mentions:
|
||||
self.whitelist.discard(channel.id)
|
||||
|
||||
await message.edit(content="✅ Removed mentioned entities from log whitelist")
|
||||
|
||||
async def cmd_clearlogwhitelist(self, message):
|
||||
"""
|
||||
Clear the logging whitelist
|
||||
Usage: .clearlogwhitelist
|
||||
"""
|
||||
self.whitelist.clear()
|
||||
await message.edit(content="✅ Cleared log whitelist")
|
||||
47
bot/cogs/user_management.py
Normal file
47
bot/cogs/user_management.py
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
from bot.cogs.cog_manager import BaseCog
|
||||
import discord
|
||||
import asyncio
|
||||
|
||||
class UserManagementCog(BaseCog):
|
||||
def __init__(self, bot):
|
||||
super().__init__(bot)
|
||||
|
||||
async def cmd_close(self, message):
|
||||
"""
|
||||
Close the DM channel
|
||||
Usage: .close
|
||||
"""
|
||||
try:
|
||||
# Only works in DM channels
|
||||
if isinstance(message.channel, discord.DMChannel):
|
||||
await message.edit(content="✅ Closing DM...")
|
||||
await asyncio.sleep(1) # Give a second for the message to be seen
|
||||
await message.channel.close()
|
||||
else:
|
||||
await message.edit(content="❌ This command only works in DM channels.")
|
||||
except Exception as e:
|
||||
await message.edit(content=f"❌ Error closing DM: {str(e)}")
|
||||
|
||||
async def cmd_block(self, message):
|
||||
"""
|
||||
Block a user using Discord's native block function
|
||||
Usage: .block @user or .block [in reply to a message]
|
||||
"""
|
||||
try:
|
||||
# Check if we have a mention or if it's a reply
|
||||
if message.mentions:
|
||||
user = message.mentions[0]
|
||||
elif message.reference and message.reference.resolved:
|
||||
user = message.reference.resolved.author
|
||||
else:
|
||||
await message.edit(content="❌ Usage: `.block @user` or reply to a message with `.block`")
|
||||
return
|
||||
|
||||
if user == self.bot.user:
|
||||
await message.edit(content="❌ Cannot block yourself.")
|
||||
return
|
||||
|
||||
await user.block()
|
||||
await message.edit(content=f"✅ Blocked {user.name}#{user.discriminator}")
|
||||
except Exception as e:
|
||||
await message.edit(content=f"❌ Error blocking user: {str(e)}")
|
||||
215
bot/cogs/utility_commands.py
Normal file
215
bot/cogs/utility_commands.py
Normal file
|
|
@ -0,0 +1,215 @@
|
|||
import discord
|
||||
import asyncio
|
||||
import re
|
||||
import io
|
||||
import traceback
|
||||
from bot.cogs.cog_manager import BaseCog
|
||||
from utils.time_parser import parse_time
|
||||
|
||||
class UtilityCog(BaseCog):
|
||||
def __init__(self, bot):
|
||||
super().__init__(bot)
|
||||
|
||||
async def cmd_remindme(self, message):
|
||||
"""
|
||||
Set a reminder
|
||||
Usage: .remindme <time> <text>
|
||||
Example: .remindme 5m Check the oven
|
||||
"""
|
||||
content = message.content.strip()
|
||||
match = re.match(r'\.remindme\s+(\d+[smhd])\s+(.*)', content)
|
||||
|
||||
if not match:
|
||||
await message.edit(content="❌ Usage: `.remindme <time> <text>`\nExample: `.remindme 5m Check the oven`")
|
||||
return
|
||||
|
||||
time_str = match.group(1)
|
||||
reminder_text = match.group(2)
|
||||
|
||||
duration = parse_time(time_str)
|
||||
if not duration:
|
||||
await message.edit(content="❌ Invalid time format. Use numbers followed by s (seconds), m (minutes), h (hours), or d (days).")
|
||||
return
|
||||
|
||||
await message.edit(content=f"✅ I'll remind you in {time_str}: {reminder_text}")
|
||||
|
||||
# Schedule the reminder
|
||||
async def reminder_task():
|
||||
await asyncio.sleep(duration)
|
||||
await message.reply(f"{message.author.mention} Reminder: {reminder_text.replace('@', 'at')}")
|
||||
|
||||
asyncio.create_task(reminder_task())
|
||||
|
||||
async def cmd_fmt(self, message):
|
||||
"""
|
||||
Format text nicely (code blocks, etc)
|
||||
Usage: .fmt <language> <code>
|
||||
Example: .fmt py print("Hello World")
|
||||
"""
|
||||
content = message.content.strip()
|
||||
if len(content.split()) < 3:
|
||||
await message.edit(content="❌ Usage: `.fmt <language> <code>`")
|
||||
return
|
||||
|
||||
_, lang, *code_parts = content.split()
|
||||
code = " ".join(code_parts)
|
||||
|
||||
formatted = f"{lang}\n{code}\n"
|
||||
await message.edit(content=formatted)
|
||||
|
||||
async def cmd_eval(self, message):
|
||||
"""
|
||||
Evaluate Python code (dangerous, use with caution)
|
||||
Usage: .eval <code>
|
||||
"""
|
||||
content = message.content.strip()
|
||||
if len(content.split()) < 2:
|
||||
await message.edit(content="❌ Usage: `.eval <code>`")
|
||||
return
|
||||
|
||||
code = content.split(" ", 1)[1]
|
||||
|
||||
# Create a safe execution environment
|
||||
local_vars = {
|
||||
'bot': self.bot,
|
||||
'message': message,
|
||||
'discord': discord,
|
||||
'asyncio': asyncio
|
||||
}
|
||||
|
||||
try:
|
||||
# Add return if it doesn't exist
|
||||
if not code.strip().startswith("return ") and "\n" not in code:
|
||||
code = f"return {code}"
|
||||
|
||||
# Wrap in async function
|
||||
code = f"async def __eval_func__():\n{' ' * 4}{code.replace(chr(10), chr(10) + ' ' * 4)}\nresult = asyncio.run_coroutine_threadsafe(__eval_func__(), bot.loop).result()"
|
||||
|
||||
# Execute code
|
||||
exec(code, globals(), local_vars)
|
||||
result = local_vars['result']
|
||||
|
||||
# Format result
|
||||
if result is None:
|
||||
await message.edit(content="✅ Code executed successfully (no output)")
|
||||
else:
|
||||
await message.edit(content=f"\n{result}\n")
|
||||
|
||||
except Exception as e:
|
||||
error = traceback.format_exc()
|
||||
await message.edit(content=f"❌ Error:\n\n{error}\n")
|
||||
|
||||
async def cmd_delrecent(self, message):
|
||||
"""
|
||||
Delete recent messages from the user
|
||||
Usage: .delrecent <count>
|
||||
"""
|
||||
content = message.content.strip()
|
||||
parts = content.split()
|
||||
|
||||
if len(parts) != 2:
|
||||
await message.edit(content="❌ Usage: `.delrecent <count>`")
|
||||
return
|
||||
|
||||
try:
|
||||
count = int(parts[1])
|
||||
if count < 1 or count > 100: # Discord limitation
|
||||
await message.edit(content="❌ Count must be between 1 and 100")
|
||||
return
|
||||
except ValueError:
|
||||
await message.edit(content="❌ Count must be a number")
|
||||
return
|
||||
|
||||
await message.edit(content=f"🗑️ Deleting {count} recent messages...")
|
||||
|
||||
# Delete messages
|
||||
deleted = 0
|
||||
async for msg in message.channel.history(limit=200):
|
||||
if msg.author == self.bot.user:
|
||||
await msg.delete()
|
||||
deleted += 1
|
||||
if deleted >= count:
|
||||
break
|
||||
await asyncio.sleep(0.5) # Avoid rate limiting
|
||||
|
||||
# The original message likely got deleted too, so we send a new one
|
||||
await message.channel.send(f"✅ Deleted {deleted} message(s)", delete_after=5)
|
||||
|
||||
async def cmd_nuke_server(self, message):
|
||||
"""
|
||||
Nuke a server (dangerous)
|
||||
Usage: .nuke
|
||||
"""
|
||||
try:
|
||||
# Extra safety check
|
||||
confirmation = await message.channel.send("⚠️ Are you sure you want to nuke this server? Reply 'yes' to confirm.")
|
||||
|
||||
def check(m):
|
||||
return m.author == message.author and m.content.lower() == "yes"
|
||||
|
||||
try:
|
||||
await self.bot.wait_for('message', check=check, timeout=15.0)
|
||||
except asyncio.TimeoutError:
|
||||
await message.edit(content="❌ Nuke cancelled.")
|
||||
await confirmation.delete()
|
||||
return
|
||||
|
||||
await confirmation.delete()
|
||||
await message.edit(content="💣 Nuking server...")
|
||||
|
||||
for channel in message.guild.channels:
|
||||
await channel.delete()
|
||||
|
||||
for member in message.guild.members:
|
||||
if member != self.bot.user and member != message.guild.owner:
|
||||
await member.kick(reason="Server nuke")
|
||||
|
||||
await message.edit(content="💥 Server nuked successfully")
|
||||
|
||||
except Exception as e:
|
||||
await message.edit(content=f"❌ Error nuking server: {str(e)}")
|
||||
|
||||
async def cmd_savechannel(self, message):
|
||||
"""
|
||||
Save a channel's messages to a file
|
||||
Usage: .savechannel [limit]
|
||||
"""
|
||||
content = message.content.strip()
|
||||
parts = content.split()
|
||||
|
||||
limit = 100 # Default limit
|
||||
if len(parts) >= 2:
|
||||
try:
|
||||
limit = int(parts[1])
|
||||
if limit < 1:
|
||||
await message.edit(content="❌ Limit must be positive")
|
||||
return
|
||||
except ValueError:
|
||||
await message.edit(content="❌ Limit must be a number")
|
||||
return
|
||||
|
||||
await message.edit(content=f"📥 Saving the last {limit} messages...")
|
||||
|
||||
# Get messages
|
||||
messages = []
|
||||
async for msg in message.channel.history(limit=limit):
|
||||
time_str = msg.created_at.strftime("%Y-%m-%d %H:%M:%S")
|
||||
author = f"{msg.author.name}#{msg.author.discriminator}"
|
||||
content = msg.content or "[No Text Content]"
|
||||
|
||||
# Handle attachments
|
||||
attachments = ""
|
||||
if msg.attachments:
|
||||
attachments = f" [Attachments: {', '.join(a.url for a in msg.attachments)}]"
|
||||
|
||||
messages.append(f"[{time_str}] {author}: {content}{attachments}")
|
||||
|
||||
# Reverse to get chronological order
|
||||
messages.reverse()
|
||||
|
||||
# Create file
|
||||
content = "\n".join(messages)
|
||||
file = discord.File(io.BytesIO(content.encode()), filename=f"channel_{message.channel.name}.txt")
|
||||
|
||||
await message.delete()
|
||||
await message.channel.send(f"📁 Here are the last {limit} messages from this channel:", file=file)
|
||||
Loading…
Add table
Add a link
Reference in a new issue