Leahnaya/Lacie
1
0
Fork 0
mirror of https://github.com/Lilac-Rose/Lacie.git synced 2026-08-25 10:14:32 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor: codebase cleanup across all modules

Browse Source
This commit is contained in:
Lilac-Rose
2026-02-18 02:45:11 +01:00
parent 831954a9b9
commit 461ff385d5
74 changed files with 891 additions and 629 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@@ -6,7 +6,5 @@ venv/
*/backups/
*/fonts/
.txt
*/databases/
*/old_backups/
logs/
data/
data/

81
PLAN.md Normal file
View File

@@ -0,0 +1,81 @@
# Lacie Discord Bot - Codebase Cleanup Plan
## Phase 1: Foundation (infrastructure changes everything else depends on)
### 1a. Set up centralized logging
- Create `utils/logger.py` with a configured logger factory
- Replace all ~154 `print()` statements with proper `logging` calls across every file
### 1b. Add `__init__.py` to all 13 packages
- `birthday/`, `commands/`, `embed/`, `events/`, `image/`, `lilac-tools/`, `moderation/`, `profiles/`, `reminders/`, `sparkle/`, `stats/`, `suggestion/`, `wordle/`, `xp/`, `utils/`
- Each `__init__.py` gets a module-level docstring explaining the package's purpose
### 1c. Create `utils/constants.py`
- Move hardcoded IDs (lilac_id, guild IDs, channel IDs) to a single constants file
- Update all files that reference these IDs
---
## Phase 2: File renames & import standardization
### 2a. Rename files for snake_case consistency
- `embed/embedcolor.py` → `embed/embed_color.py`
- `commands/roletrack.py` → `commands/role_track.py`
- `events/roletrack.py` → delete (already marked for deletion in git)
- Update all imports referencing renamed files
### 2b. Standardize import style
- Use absolute imports everywhere (e.g., `from embed.embed_color import ...`)
- Group imports: stdlib → third-party → local, separated by blank lines
- Remove unused imports (`import sys` in generate_role_color_images.py, botban.py, etc.)
### 2c. Standardize database path construction
- Use `pathlib.Path` consistently instead of mixed `os.path.join` / `Path`
---
## Phase 3: Code quality fixes
### 3a. Fix bare `except:` clauses (~40 instances)
- Replace with specific exception types (`except Exception as e:`, `except sqlite3.Error:`, etc.)
### 3b. Fix deprecated patterns
- Replace `bot.loop.create_task()` with `cog_load()` async method or `asyncio.create_task()`
### 3c. Remove dead code
- Clean up unused variables, unreachable code
- Remove the deleted `events/roletrack.py`
### 3d. Rename constants to UPPER_SNAKE_CASE
- e.g., `meow_list` → `MEOW_LIST`, `lilac_id` → move to constants
---
## Phase 4: Comments & docstrings (the big one)
### 4a. Add module-level docstrings to every `.py` file
- Brief description of what the module does
### 4b. Add class docstrings to all Cog classes
- Describe what the cog provides
### 4c. Add function/command docstrings where missing
- Focus on non-obvious logic, complex functions
- Add inline comments for complex algorithms (minesweeper flood fill, XP calculations, spam detection, etc.)
### 4d. Document database schemas
- Add comments in database initialization code describing table structures
---
## Phase 5: bot.py cleanup
### 5a. Clean up bot.py
- Add module docstring and section comments
- DRY up the duplicate cog folder lists (on_ready and reload command)
- Add proper comments for each section
---
## Order of operations
Phases 1-5 are sequential - each builds on the prior. Within each phase, tasks are independent and can be parallelized.

1
birthday/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Birthday tracking and celebration features."""

8
birthday/birthday.py
View File

@@ -1,11 +1,13 @@
"""Birthday tracking cog that sends birthday wishes and manages birthday roles."""
import discord
from discord.ext import commands, tasks
from discord import app_commands
import sqlite3
from datetime import datetime, timezone, timedelta
from embed.embed_color import get_embed_color
import asyncio
import pytz
import os
from pathlib import Path
from dotenv import load_dotenv
from moderation.loader import ModerationBase
@@ -15,7 +17,7 @@ BIRTHDAY_ROLE_ID = 1113751318918602762
class Birthday(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "birthdays.db")
self.db_path = Path(__file__).parent.parent / "data" / "birthdays.db"
self._init_db()
self.check_birthdays.start()
self.remove_birthday_roles.start()
@@ -239,7 +241,7 @@ class Birthday(commands.Cog):
embed = discord.Embed(
title=f"🎂 Birthdays in {month_name}",
description="\n".join(lines),
color=discord.Color.magenta()
color=get_embed_color(interaction.user.id)
)
await interaction.response.send_message(embed=embed)

126
bot.py
View File

@@ -1,143 +1,126 @@
"""Main entry point for the Lacie Discord bot."""
import os
from dotenv import load_dotenv
import discord
from discord.ext import commands
import asyncio
import glob
import traceback
from xp.database import get_db as get_xp_db
from xp.add_xp import add_xp
from moderation.loader import ModerationBase
from aiohttp import web
# Import sparkle DB to ensure it exists
from sparkle.database import get_db as get_sparkle_db
# Import and register the XP command groups onto the tree
from xp.groups import xp_group, xp_admin_group
from utils.logger import get_logger, setup_logging
# --- Startup ---
load_dotenv()
setup_logging()
logger = get_logger(__name__)
TOKEN = os.getenv("TOKEN")
# All cog folders to load on startup and reload
COG_FOLDERS = [
"commands",
"moderation",
"xp",
"sparkle",
"image",
"suggestion",
"birthday",
"embed",
"profiles",
"events",
"stats",
"wordle",
"reminders",
"lilac-tools",
]
# --- Bot setup ---
bot = commands.Bot(
command_prefix="!",
intents=discord.Intents.all(),
command_prefix="!",
intents=discord.Intents.all(),
help_command=None,
activity=discord.Activity(
type=discord.ActivityType.playing,
name="Paper Lily - Chapter 2"
))
# Add both groups to the command tree now, before any cogs load
# Register shared slash command groups defined in xp/groups.py
bot.tree.add_command(xp_group)
bot.tree.add_command(xp_admin_group)
# --- Cog loading ---
async def load_cogs(folder: str):
"""Load all cogs in the folder except utility files"""
non_cog_files = {"add_xp.py", "database.py", "utils.py", "__init__.py",
"import_old_data.py", "repair_db.py", "reset_db.py", "groups.py",
"loader.py"}
non_cog_files = {"add_xp.py", "database.py", "utils.py", "__init__.py", "groups.py", "loader.py"}
for file in glob.glob(f"{folder}/*.py"):
filename = os.path.basename(file)
if filename in non_cog_files:
print(f"Skipping {filename} (utility file)")
logger.debug(f"Skipping {filename} (utility file)")
continue
module_name = f"{folder}.{os.path.splitext(filename)[0]}"
try:
# Try to reload if already loaded, otherwise load fresh
if module_name in bot.extensions:
await bot.reload_extension(module_name)
print(f"Reloaded {module_name}")
logger.info(f"Reloaded {module_name}")
else:
await bot.load_extension(module_name)
print(f"Loaded {module_name}")
logger.info(f"Loaded {module_name}")
except Exception as e:
print(f"Failed to load {module_name}: {e}")
traceback.print_exc()
logger.exception(f"Failed to load {module_name}: {e}")
# --- Events and commands ---
@bot.event
async def on_ready():
print(f"Logged in as {bot.user}!")
logger.info(f"Logged in as {bot.user}!")
# Ensure XP database connection works
for lifetime in (True, False):
try:
conn, cur = get_xp_db(lifetime)
conn.close()
print(f"XP database connection successful (lifetime={lifetime})")
logger.info(f"XP database connection successful (lifetime={lifetime})")
except Exception as e:
print(f"XP database connection failed (lifetime={lifetime}): {e}")
logger.error(f"XP database connection failed (lifetime={lifetime}): {e}")
# Ensure Sparkle DB exists
try:
conn = get_sparkle_db()
conn.close()
print("Sparkle database initialized successfully.")
logger.info("Sparkle database initialized successfully.")
except Exception as e:
print(f"Failed to initialize sparkle database: {e}")
logger.error(f"Failed to initialize sparkle database: {e}")
for folder in COG_FOLDERS:
await load_cogs(folder)
# Load all cogs
await load_cogs("commands")
# await load_cogs("wordbomb")
await load_cogs("moderation")
await load_cogs("xp")
await load_cogs("sparkle")
await load_cogs("image")
await load_cogs("suggestion")
await load_cogs("birthday")
await load_cogs("embed")
await load_cogs("profiles")
await load_cogs("events")
await load_cogs("stats")
await load_cogs("wordle")
await load_cogs("reminders")
await load_cogs("lilac-tools")
# Sync slash commands after loading cogs
try:
synced = await bot.tree.sync()
print(f"Synced {len(synced)} slash commands")
logger.info(f"Synced {len(synced)} slash commands")
for cmd in synced:
print(f" - {cmd.name}")
logger.debug(f" - {cmd.name}")
except Exception as e:
print(f"Failed to sync slash commands: {e}")
traceback.print_exc()
logger.exception(f"Failed to sync slash commands: {e}")
@bot.event
async def on_command_error(ctx, error):
if isinstance(error, commands.CommandNotFound):
return
print(f"Command error: {error}")
traceback.print_exc()
logger.error(f"Command error: {error}", exc_info=True)
@bot.command(name="reload")
@ModerationBase.is_admin()
async def reload(ctx):
"""Reload commands cogs and sync slash commands"""
await load_cogs("commands")
await load_cogs("moderation")
await load_cogs("xp")
await load_cogs("sparkle")
await load_cogs("image")
await load_cogs("suggestion")
await load_cogs("birthday")
await load_cogs("embed")
await load_cogs("profiles")
await load_cogs("events")
await load_cogs("wordle")
await load_cogs("reminders")
await load_cogs("stats")
await load_cogs("lilac-tools")
for folder in COG_FOLDERS:
await load_cogs(folder)
try:
synced = await bot.tree.sync()
await ctx.send(f"Cogs reloaded successfully! Synced {len(synced)} slash commands.")
except Exception as e:
await ctx.send(f"Cogs reloaded but failed to sync slash commands: {e}")
# Hook XP into messages
from xp.add_xp import add_xp
@bot.event
async def on_message(message):
if message.author.bot:
@@ -145,17 +128,16 @@ async def on_message(message):
try:
await add_xp(message.author)
except Exception as e:
print(f"XP error: {e}")
traceback.print_exc()
logger.error(f"XP error: {e}", exc_info=True)
await bot.process_commands(message)
# --- Entry point ---
async def main():
try:
async with bot:
await bot.start(TOKEN)
except Exception as e:
print(f"Bot startup error: {e}")
traceback.print_exc()
logger.exception(f"Bot startup error: {e}")
if __name__ == "__main__":
asyncio.run(main())

1
commands/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""General-purpose bot commands (fun, utility, moderation helpers)."""

4
commands/bonk.py
View File

@@ -1,12 +1,14 @@
"""Bonk command cog that lets users bonk each other with an image."""
import discord
from discord import app_commands
from discord.ext import commands
import os
from pathlib import Path
class Bonk(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.bonk_path = os.path.join(os.path.dirname(__file__), "..", "media", "kat_bonk.png")
self.bonk_path = Path(__file__).parent.parent / "media" / "kat_bonk.png"
@app_commands.command(name="bonk", description="Bonk another user!")
@app_commands.describe(user="The user you want to bonk")

1
commands/connect4.py
View File

@@ -1,3 +1,4 @@
"""Connect 4 game implementation as a Discord slash command."""
import discord
from discord.ext import commands
from discord import app_commands

16
commands/emote_credits.py
View File

@@ -1,3 +1,4 @@
"""Emote credits cog for tracking and displaying custom emote artists."""
import discord
from discord import app_commands
from discord.ext import commands
@@ -5,6 +6,7 @@ from moderation.loader import ModerationBase
import re
import aiosqlite
from pathlib import Path
from embed.embed_color import get_embed_color
class EmoteCredits(ModerationBase, commands.Cog):
def __init__(self, bot):
@@ -12,8 +14,10 @@ class EmoteCredits(ModerationBase, commands.Cog):
self.bot = bot
self.db_path = Path(__file__).parent.parent / "data" / "emote_credits.db"
self.approval_channel_id = 1424145004976275617
bot.loop.create_task(self._init_db())
async def cog_load(self):
await self._init_db()
async def _init_db(self):
"""Initialize the database"""
async with aiosqlite.connect(self.db_path) as conn:
@@ -87,7 +91,7 @@ class EmoteCredits(ModerationBase, commands.Cog):
embed = discord.Embed(
title=f"🎨 Credit for: {emoji_name}",
description=f"**Artist:** {credit}",
color=discord.Color.blue()
color=get_embed_color(interaction.user.id)
)
embed.set_footer(text="Full credits document: https://docs.google.com/document/d/1o6dJS3G82rA03oHQn3Lu3ywK0SpepZnxnmP28R8Nnpc/edit?tab=t.0")
await interaction.followup.send(embed=embed)
@@ -137,7 +141,7 @@ class EmoteCredits(ModerationBase, commands.Cog):
approval_embed = discord.Embed(
title="🎨 New Credit Submission",
color=discord.Color.blue()
color=get_embed_color(interaction.user.id)
)
approval_embed.add_field(name="Emote Name", value=f"`{emoji_name}`", inline=False)
approval_embed.add_field(name="Artist", value=artist, inline=False)
@@ -343,7 +347,7 @@ class CreditApprovalView(discord.ui.View):
color=discord.Color.green()
)
await submitter.send(embed=notify_embed)
except:
except Exception:
pass
@discord.ui.button(label="Deny", style=discord.ButtonStyle.red, custom_id="deny_credit")
@@ -373,7 +377,7 @@ class CreditApprovalView(discord.ui.View):
color=discord.Color.red()
)
await submitter.send(embed=notify_embed)
except:
except Exception:
pass

10
commands/generate_role_color_images.py
View File

@@ -1,15 +1,17 @@
"""Admin command to generate the color role preview image."""
import discord
from discord import app_commands
from discord.ext import commands
from pathlib import Path
import traceback
from PIL import Image, ImageDraw, ImageFont
import os
import io
from math import floor, ceil
from moderation.loader import ModerationBase
import importlib
import sys
from utils.logger import get_logger
logger = get_logger(__name__)
class ColorImageGen(commands.Cog):
def __init__(self, bot):
@@ -43,7 +45,7 @@ class ColorImageGen(commands.Cog):
color_roles = []
# Styling
font_path_ttf = os.path.join(self.FONTS_PATH, "Renogare-Regular.otf")
font_path_ttf = self.FONTS_PATH / "Renogare-Regular.otf"
FONT_SIZE = 300
font = ImageFont.truetype(font_path_ttf, FONT_SIZE)
COLUMN_SIZE = 5
@@ -104,7 +106,7 @@ class ColorImageGen(commands.Cog):
)
except Exception as e:
print(f"[ERROR] /color list\n{traceback.format_exc()}")
logger.error(f"Error generating color image: {e}", exc_info=True)
msg = f"❌ Error in `/color list`: `{e}`" if self.DEBUG else "❌ Something went wrong loading color images."
await ctx.send(msg)

5
commands/infractions.py
View File

@@ -1,8 +1,9 @@
"""Command for users to view their own moderation infractions via DM."""
import discord
from discord import app_commands
from discord.ext import commands
import sqlite3
import os
from pathlib import Path
class InfractionsCommand(commands.Cog):
@@ -11,7 +12,7 @@ class InfractionsCommand(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
# Path to moderation database (go up to project root, then into moderation/)
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "moderation.db")
self.db_path = Path(__file__).parent.parent / "data" / "moderation.db"
@app_commands.command(name="infractions", description="View your infractions in this server")
async def infractions(self, interaction: discord.Interaction):

9
commands/meow.py
View File

@@ -1,18 +1,19 @@
"""Simple meow command cog that responds with a random cat sound."""
import discord
from discord.ext import commands
from discord import app_commands
import random
MEOW_LIST = ["Meowwwww~", "Purrrrrr", "Nyaaaaaa", "Meow Meow", "Nya!", "Meow :3"]
class Meow(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
@app_commands.command(name="meow")
async def meow(self, interaction: discord.Interaction):
meow_list=["Meowwwww~", "Purrrrrr", "Nyaaaaaa", "Meow Meow", "Nya!", "Meow :3"]
meow_index = random.randrange(0,5)
await interaction.response.send_message(meow_list[meow_index])
meow_index = random.randrange(0, len(MEOW_LIST))
await interaction.response.send_message(MEOW_LIST[meow_index])
async def setup(bot: commands.Bot):
await bot.add_cog(Meow(bot))

114
commands/minesweeper.py
View File

@@ -1,10 +1,58 @@
"""Minesweeper game cog with an interactive Discord UI."""
import discord
from discord.ext import commands
from discord import app_commands
import random
import asyncio
import aiosqlite
from pathlib import Path
from embed.embed_color import get_embed_color
import re
from datetime import datetime, timedelta
from utils.logger import get_logger
logger = get_logger(__name__)
DB_PATH = Path(__file__).parent.parent / "data" / "minesweeper.db"
async def _update_stats(user_id: int, win: bool):
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS minesweeper_stats (
user_id INTEGER PRIMARY KEY,
played INTEGER DEFAULT 0,
wins INTEGER DEFAULT 0,
current_streak INTEGER DEFAULT 0,
max_streak INTEGER DEFAULT 0
)
""")
cursor = await db.execute(
"SELECT played, wins, current_streak, max_streak FROM minesweeper_stats WHERE user_id=?",
(user_id,)
)
row = await cursor.fetchone()
if not row:
played, wins, streak, max_streak = 0, 0, 0, 0
else:
played, wins, streak, max_streak = row
played += 1
if win:
wins += 1
streak += 1
max_streak = max(max_streak, streak)
else:
streak = 0
await db.execute("""
INSERT INTO minesweeper_stats (user_id, played, wins, current_streak, max_streak)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(user_id) DO UPDATE SET
played=excluded.played, wins=excluded.wins,
current_streak=excluded.current_streak, max_streak=excluded.max_streak
""", (user_id, played, wins, streak, max_streak))
await db.commit()
class MinesweeperGame:
def __init__(self, rows: int = 13, cols: int = 13, mines: int = 20):
@@ -230,12 +278,13 @@ class MinesweeperView(discord.ui.View):
self.timed_out = True
self.game.game_over = True
await _update_stats(self.player.id, False)
for child in self.children:
child.disabled = True
self.stop()
if self.message:
embed = self.create_embed()
embed.color = discord.Color.orange()
@@ -248,7 +297,7 @@ class MinesweeperView(discord.ui.View):
# Message was deleted
pass
except Exception as e:
print(f"Error updating timed out game: {e}")
logger.error(f"Error updating timed out game: {e}")
def reset_timeout(self):
"""Reset the timeout timer - called when a move is made"""
@@ -270,11 +319,12 @@ class MinesweeperView(discord.ui.View):
for child in self.children:
child.disabled = True
self.stop()
await _update_stats(self.player.id, False)
embed = self.create_embed()
embed.color = discord.Color.red()
embed.title = "💀 Game Over - Forfeited"
await interaction.response.edit_message(embed=embed, view=self)
@discord.ui.button(label="How to Play", style=discord.ButtonStyle.secondary, emoji="❓")
@@ -282,7 +332,7 @@ class MinesweeperView(discord.ui.View):
help_embed = discord.Embed(
title="🎮 How to Play Minesweeper",
description="Send messages in this channel to make moves!",
color=discord.Color.blue()
color=get_embed_color(interaction.user.id)
)
help_embed.add_field(
name="📝 Move Format",
@@ -332,7 +382,7 @@ class MinesweeperView(discord.ui.View):
embed = discord.Embed(
title="💣 Minesweeper",
description=f"{self.player.mention}'s game",
color=discord.Color.blue()
color=get_embed_color(self.player.id)
)
board = self.game.render_board()
@@ -438,7 +488,7 @@ class Minesweeper(commands.Cog):
# Valid input - delete the player's message to reduce clutter
try:
await message.delete()
except:
except Exception:
pass
# Make the move
@@ -455,7 +505,9 @@ class Minesweeper(commands.Cog):
# Cancel timeout task since game is over
if view.timeout_task and not view.timeout_task.done():
view.timeout_task.cancel()
await _update_stats(player_id, view.game.won)
for child in view.children:
child.disabled = True
view.stop()
@@ -463,7 +515,7 @@ class Minesweeper(commands.Cog):
embed = view.create_embed()
try:
await view.message.edit(embed=embed, view=view)
except:
except Exception:
pass
def parse_move(self, text: str) -> tuple[int, int, bool] | None:
@@ -497,5 +549,45 @@ class Minesweeper(commands.Cog):
return (col, row, is_flag)
@app_commands.command(name="minesweeper_stats", description="View your Minesweeper stats")
async def minesweeper_stats(self, interaction: discord.Interaction):
user_id = interaction.user.id
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS minesweeper_stats (
user_id INTEGER PRIMARY KEY,
played INTEGER DEFAULT 0,
wins INTEGER DEFAULT 0,
current_streak INTEGER DEFAULT 0,
max_streak INTEGER DEFAULT 0
)
""")
cursor = await db.execute(
"SELECT played, wins, current_streak, max_streak FROM minesweeper_stats WHERE user_id=?",
(user_id,)
)
row = await cursor.fetchone()
if not row or row[0] == 0:
await interaction.response.send_message("You haven't played Minesweeper yet!", ephemeral=True)
return
played, wins, streak, max_streak = row
losses = played - wins
win_rate = round(wins / played * 100, 1) if played else 0
embed = discord.Embed(
title=f"{interaction.user.display_name}'s Minesweeper Stats",
color=get_embed_color(user_id)
)
embed.add_field(name="Games Played", value=str(played))
embed.add_field(name="Wins", value=str(wins))
embed.add_field(name="Losses", value=str(losses))
embed.add_field(name="Win Rate", value=f"{win_rate}%")
embed.add_field(name="Current Streak", value=str(streak))
embed.add_field(name="Best Streak", value=str(max_streak))
await interaction.response.send_message(embed=embed)
async def setup(bot):
await bot.add_cog(Minesweeper(bot))

8
commands/ping.py
View File

@@ -1,14 +1,16 @@
"""Ping command that shows WebSocket, API, and database latency."""
import discord
from discord.ext import commands
from discord import app_commands
import time
import aiosqlite
import os
from embed.embed_color import get_embed_color
from pathlib import Path
class Ping(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "suggestions.db")
self.db_path = Path(__file__).parent.parent / "data" / "suggestions.db"
@app_commands.command(name="ping", description="Check the bot's latency")
async def ping(self, interaction: discord.Interaction):
@@ -33,7 +35,7 @@ class Ping(commands.Cog):
embed = discord.Embed(
title="Pong!",
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user)
color=get_embed_color(interaction.user.id)
)
embed.add_field(name="WebSocket Latency", value=f"{ws_latency}ms", inline=True)
embed.add_field(name="API Latency", value=f"{api_latency}ms", inline=True)

25
commands/role_color.py
View File

@@ -1,13 +1,16 @@
"""Color role management cog allowing users to pick decorative color roles."""
import discord
from discord import app_commands
from discord.ext import commands
from pathlib import Path
import traceback
from PIL import Image, ImageDraw, ImageFont
import os
import io
from math import floor, ceil
from moderation.loader import ModerationBase
from embed.embed_color import get_embed_color
from utils.logger import get_logger
logger = get_logger(__name__)
DEBUG = False
@@ -18,7 +21,7 @@ COLOR_ROLE_NAMES = [
"Dark Air", "White-ish"
]
FONTS_PATH = os.path.join(os.path.dirname(__file__), "..", "fonts")
FONTS_PATH = Path(__file__).parent.parent / "fonts"
class ColorRoles(commands.Cog):
def __init__(self, bot):
@@ -87,7 +90,7 @@ class ColorRoles(commands.Cog):
ephemeral=True
)
except Exception as e:
print(f"[ERROR] /color set\n{traceback.format_exc()}")
logger.error(f"/color set: {e}", exc_info=True)
msg = f"Error: `{e}`" if DEBUG else "Something went wrong."
await interaction.followup.send(msg, ephemeral=True)
@@ -148,7 +151,7 @@ class ColorRoles(commands.Cog):
ephemeral=False
)
except Exception as e:
print(f"[ERROR] /color setfor\n{traceback.format_exc()}")
logger.error(f"/color setfor: {e}", exc_info=True)
msg = f"Error: `{e}`" if DEBUG else "Something went wrong."
await interaction.followup.send(msg, ephemeral=True)
@@ -199,7 +202,7 @@ class ColorRoles(commands.Cog):
ephemeral=True
)
except Exception as e:
print(f"[ERROR] /color remove\n{traceback.format_exc()}")
logger.error(f"/color remove: {e}", exc_info=True)
msg = f"Error: `{e}`" if DEBUG else "Something went wrong."
await interaction.followup.send(msg, ephemeral=True)
@@ -244,7 +247,7 @@ class ColorRoles(commands.Cog):
ephemeral=False
)
except Exception as e:
print(f"[ERROR] /color removefor\n{traceback.format_exc()}")
logger.error(f"/color removefor: {e}", exc_info=True)
msg = f"Error: `{e}`" if DEBUG else "Something went wrong."
await interaction.followup.send(msg, ephemeral=True)
@@ -280,7 +283,7 @@ class ColorRoles(commands.Cog):
embed1 = discord.Embed(
title="Available Color Roles (Part 1)",
description="Use `/color set` to pick one!",
color=discord.Color.purple()
color=get_embed_color(interaction.user.id)
)
embed1.set_image(url=f"attachment://{color_image_files[0].name}")
await interaction.followup.send(embed=embed1, file=file1, ephemeral=False)
@@ -290,7 +293,7 @@ class ColorRoles(commands.Cog):
embed2 = discord.Embed(
title="Available Color Roles (Part 2)",
description="More colors to choose from!",
color=discord.Color.purple()
color=get_embed_color(interaction.user.id)
)
embed2.set_image(url=f"attachment://{color_image_files[1].name}")
await interaction.followup.send(embed=embed2, file=file2, ephemeral=False)
@@ -300,13 +303,13 @@ class ColorRoles(commands.Cog):
embed = discord.Embed(
title="Available Color Roles",
description="Use `/color set` to pick one!",
color=discord.Color.purple()
color=get_embed_color(interaction.user.id)
)
embed.set_image(url="attachment://colorimage.png")
await interaction.followup.send(embed=embed, file=file, ephemeral=False)
except Exception as e:
print(f"[ERROR] /color list\n{traceback.format_exc()}")
logger.error(f"/color list: {e}", exc_info=True)
msg = f"Error in `/color list`: `{e}`" if DEBUG else "Something went wrong loading color images."
await interaction.followup.send(msg, ephemeral=True)

33
commands/roletrack.py → commands/role_track.py
View File

@@ -1,15 +1,22 @@
"""Role tracking cog that monitors and records role assignment history."""
import discord
from discord import app_commands
from discord.ext import commands
import aiosqlite
import os
from pathlib import Path
from embed.embed_color import get_embed_color
from utils.logger import get_logger
logger = get_logger(__name__)
class RoleTrack(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "roletrack.db")
bot.loop.create_task(self.init_db())
self.db_path = Path(__file__).parent.parent / "data" / "roletrack.db"
async def cog_load(self):
await self.init_db()
async def init_db(self):
async with aiosqlite.connect(self.db_path) as db:
# role_ids stored as comma-separated string
@@ -63,7 +70,7 @@ class RoleTrack(commands.Cog):
embed = discord.Embed(
title="✅ Roles Synced",
description=f"Successfully saved {role_count} role(s) to the tracking system.",
color=discord.Color.green()
color=get_embed_color(interaction.user.id)
)
embed.add_field(
name="What does this do?",
@@ -85,7 +92,7 @@ class RoleTrack(commands.Cog):
embed = discord.Embed(
title="❌ Sync Failed",
description=f"An error occurred while syncing your roles: {str(e)}",
color=discord.Color.red()
color=get_embed_color(interaction.user.id)
)
await interaction.followup.send(embed=embed, ephemeral=True)
@@ -101,10 +108,10 @@ class RoleTrack(commands.Cog):
embed = discord.Embed(
title="📋 Saved Roles",
description="No roles are currently saved for you in the database.",
color=discord.Color.orange()
color=get_embed_color(interaction.user.id)
)
else:
roles = []
roles = []
for role_id in saved_role_ids:
role = interaction.guild.get_role(role_id)
if role:
@@ -115,7 +122,7 @@ class RoleTrack(commands.Cog):
embed = discord.Embed(
title="📋 Saved Roles",
description=f"Found {len(saved_role_ids)} role(s) in the database:",
color=discord.Color.blue()
color=get_embed_color(interaction.user.id)
)
embed.add_field(
name="Roles",
@@ -129,7 +136,7 @@ class RoleTrack(commands.Cog):
embed = discord.Embed(
title="❌ Check Failed",
description=f"An error occurred: {str(e)}",
color=discord.Color.red()
color=get_embed_color(interaction.user.id)
)
await interaction.followup.send(embed=embed, ephemeral=True)
@@ -162,7 +169,7 @@ class RoleTrack(commands.Cog):
embed = discord.Embed(
title="🎭 Roles Restored",
description=f"Welcome back to **{member.guild.name}**! Your roles have been automatically restored.",
color=discord.Color.blue()
color=get_embed_color(member.id)
)
embed.add_field(
name="Restored Roles",
@@ -170,7 +177,7 @@ class RoleTrack(commands.Cog):
inline=False
)
await member.send(embed=embed)
except:
except Exception:
# User has DMs disabled, that's fine
pass
@@ -178,7 +185,7 @@ class RoleTrack(commands.Cog):
# Bot doesn't have permission to add roles
pass
except Exception as e:
print(f"Error restoring roles for {member}: {e}")
logger.error(f"Error restoring roles for {member}: {e}")
@commands.Cog.listener()
async def on_member_update(self, before: discord.Member, after: discord.Member):

1
embed/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Embed color management for consistent bot embed styling."""

17
embed/embedcolor.py → embed/embed_color.py
View File

@@ -1,10 +1,23 @@
"""Embed color preference system allowing users to set custom embed colors."""
import discord
import sqlite3
import os
from pathlib import Path
from discord.ext import commands
from discord import app_commands
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "embed_colors.db")
DB_PATH = Path(__file__).parent.parent / "data" / "embed_colors.db"
def get_embed_color(user_id: int) -> discord.Color:
db = sqlite3.connect(DB_PATH)
cursor = db.cursor()
cursor.execute("SELECT color FROM user_embed_colors WHERE user_id = ?", (user_id,))
result = cursor.fetchone()
db.close()
if result and result[0]:
return discord.Color(int(result[0], 16))
return discord.Color.blurple()
class EmbedColor(commands.Cog):
def __init__(self, bot):

1
events/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Event listeners and automated monitoring (welcome, webhooks, status alerts)."""

22
events/archipelago_monitor.py
View File

@@ -8,15 +8,15 @@ import discord
from discord.ext import commands, tasks
import asyncio
import re
import logging
import os
from dotenv import load_dotenv
from typing import Optional, Set
from pathlib import Path
from utils.logger import get_logger
load_dotenv()
logger = logging.getLogger('ArchipelagoMonitor')
logger = get_logger(__name__)
class ArchipelagoMonitor(commands.Cog):
@@ -44,24 +44,16 @@ class ArchipelagoMonitor(commands.Cog):
'server_start': re.compile(r'Hosting game at (.+?) \(Password: (.+?)\)'),
}
# Print startup info
print(f"[Archipelago] Initializing log monitor...")
print(f"[Archipelago] Enabled: {self.enabled}")
print(f"[Archipelago] Channel ID: {self.channel_id}")
print(f"[Archipelago] Log directory: {self.log_directory}")
logger.info(f"Initializing log monitor - Enabled: {self.enabled}, Channel: {self.channel_id}, Log dir: {self.log_directory}")
if self.enabled and self.channel_id:
logger.info(f"Archipelago monitor enabled - Channel: {self.channel_id}, Log dir: {self.log_directory}")
print(f"[Archipelago] Monitor will start when bot is ready")
logger.info("Monitor will start when bot is ready")
self.monitor_log.start()
print(f"[Archipelago] Monitor task started")
else:
logger.info("Archipelago monitor disabled (check .env configuration)")
print(f"[Archipelago] Monitor DISABLED - check your .env file")
if not self.enabled:
print(f"[Archipelago] - ARCHIPELAGO_ENABLED is not true")
logger.info("Monitor DISABLED: ARCHIPELAGO_ENABLED is not true")
if not self.channel_id:
print(f"[Archipelago] - ARCHIPELAGO_CHANNEL_ID is not set")
logger.info("Monitor DISABLED: ARCHIPELAGO_CHANNEL_ID is not set")
async def cog_unload(self):
"""Called when the cog is unloaded."""

8
events/botban.py
View File

@@ -1,7 +1,10 @@
"""Auto-ban cog that bans users who receive a specific bot-trap role."""
import discord
from discord.ext import commands
from datetime import datetime, timedelta, timezone
import sys
from utils.logger import get_logger
logger = get_logger(__name__)
ROLE_ID_TO_BAN = 1439354601672282335
LOG_CHANNEL_ID = 1440055015711703242
@@ -95,8 +98,7 @@ class AutoBanOnRole(commands.Cog):
@commands.Cog.listener()
async def on_ready(self):
sys.stdout.write("[INFO] AutoBanOnRole cog loaded and ready!\n")
sys.stdout.flush()
logger.info("AutoBanOnRole cog loaded and ready!")
async def setup(bot: commands.Bot):
await bot.add_cog(AutoBanOnRole(bot))

1
events/chain.py
View File

@@ -1,3 +1,4 @@
"""Chain message event listener (reacts to chain messages in channels)."""
import discord
from discord.ext import commands

33
events/git_webhook.py
View File

@@ -1,3 +1,4 @@
"""Webhook server that listens for GitHub push events and posts to Discord."""
import discord
from discord.ext import commands
from aiohttp import web
@@ -5,9 +6,12 @@ import hmac
import hashlib
import os
from dotenv import load_dotenv
from utils.logger import get_logger
load_dotenv()
logger = get_logger(__name__)
COMMIT_CHANNEL_IDS = [876777562599194644, 1437941632849940563, 1470441786810826884]
WEBHOOK_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET", "")
# Updated port to be 8000 to prevent conflicts (should work)
@@ -29,9 +33,8 @@ class GitWebhook(commands.Cog):
await self.runner.setup()
self.site = web.TCPSite(self.runner, '0.0.0.0', WEBHOOK_PORT)
await self.site.start()
print(f"✅ Git webhook server started on port {WEBHOOK_PORT}")
print(f" Configure GitHub to send webhooks to: http://159.195.45.90:{WEBHOOK_PORT}/webhook")
print(f" Sending notifications to {len(COMMIT_CHANNEL_IDS)} channel(s)")
logger.info(f"Git webhook server started on port {WEBHOOK_PORT}")
logger.info(f"Sending notifications to {len(COMMIT_CHANNEL_IDS)} channel(s)")
async def cog_unload(self):
"""Stop the webhook server when cog unloads."""
@@ -39,7 +42,7 @@ class GitWebhook(commands.Cog):
await self.site.stop()
if self.runner:
await self.runner.cleanup()
print("🛑 Git webhook server stopped")
logger.info("Git webhook server stopped")
def verify_signature(self, payload_body, signature_header):
"""Verify GitHub webhook signature for security."""
@@ -73,9 +76,9 @@ class GitWebhook(commands.Cog):
# Handle GitHub ping event (test from GitHub)
if 'zen' in data and 'hook_id' in data:
print("✅ Received GitHub ping event - webhook is configured correctly!")
logger.info("Received GitHub ping event - webhook is configured correctly!")
return web.json_response({"status": "pong"}, status=200)
# Get all Discord channels
channels = []
for channel_id in COMMIT_CHANNEL_IDS:
@@ -83,10 +86,10 @@ class GitWebhook(commands.Cog):
if channel:
channels.append(channel)
else:
print(f"⚠️ Channel {channel_id} not found!")
logger.warning(f"Channel {channel_id} not found!")
if not channels:
print(f"❌ No valid channels found!")
logger.error("No valid channels found!")
return web.json_response({"error": "No channels found"}, status=500)
# Handle GitHub push events
@@ -99,13 +102,11 @@ class GitWebhook(commands.Cog):
await self.handle_gitlab_push(data, channels)
return web.json_response({"status": "success"}, status=200)
print(f"⚠️ Unknown webhook format. Keys in data: {list(data.keys())}")
logger.warning(f"Unknown webhook format. Keys in data: {list(data.keys())}")
return web.json_response({"error": "Unknown webhook format"}, status=400)
except Exception as e:
print(f"❌ Webhook error: {e}")
import traceback
traceback.print_exc()
logger.error(f"Webhook error: {e}", exc_info=True)
return web.json_response({"error": str(e)}, status=500)
async def handle_github_push(self, data, channels):
@@ -168,7 +169,7 @@ class GitWebhook(commands.Cog):
try:
await channel.send(embed=embed)
except Exception as e:
print(f"❌ Failed to send to channel {channel.id}: {e}")
logger.error(f"Failed to send to channel {channel.id}: {e}")
async def handle_gitlab_push(self, data, channels):
"""Handle GitLab push webhook."""
@@ -229,7 +230,7 @@ class GitWebhook(commands.Cog):
try:
await channel.send(embed=embed)
except Exception as e:
print(f"❌ Failed to send to channel {channel.id}: {e}")
logger.error(f"Failed to send to channel {channel.id}: {e}")
async def setup(bot):
await bot.add_cog(GitWebhook(bot))

1
events/goodbot.py
View File

@@ -1,3 +1,4 @@
"""Goodbot/badbot event listener that reacts to compliments or insults."""
import discord
from discord.ext import commands

140
events/roletrack.py
View File

@@ -1,140 +0,0 @@
import discord
from discord.ext import commands
import sqlite3
import os
import asyncio
class RoleTrackEvents(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(__file__), '..', 'roletrack.db')
def is_opted_in(self, user_id: int, guild_id: int) -> bool:
"""Check if a user is opted into role tracking"""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
c.execute('''
SELECT opted_in FROM role_tracking_users
WHERE user_id = ? AND guild_id = ?
''', (user_id, guild_id))
result = c.fetchone()
conn.close()
return result[0] == 1 if result else False
def get_tracked_roles(self, user_id: int, guild_id: int):
"""Retrieve tracked roles for a user"""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
c.execute('''
SELECT role_id, role_name FROM tracked_roles
WHERE user_id = ? AND guild_id = ?
''', (user_id, guild_id))
roles = c.fetchall()
conn.close()
return roles
def save_user_roles(self, member: discord.Member):
"""Save all roles for a user (excluding @everyone)"""
if not self.is_opted_in(member.id, member.guild.id):
return
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
# Clear existing roles for this user in this guild
c.execute('''
DELETE FROM tracked_roles
WHERE user_id = ? AND guild_id = ?
''', (member.id, member.guild.id))
# Save current roles (excluding @everyone)
for role in member.roles:
if role.id != member.guild.id: # Skip @everyone role
c.execute('''
INSERT INTO tracked_roles (user_id, guild_id, role_id, role_name)
VALUES (?, ?, ?, ?)
''', (member.id, member.guild.id, role.id, role.name))
conn.commit()
conn.close()
@commands.Cog.listener()
async def on_member_join(self, member: discord.Member):
"""Restore roles when a tracked user rejoins"""
if member.bot:
return
# Check if user is opted in
if not self.is_opted_in(member.id, member.guild.id):
return
# Get their tracked roles
tracked_roles = self.get_tracked_roles(member.id, member.guild.id)
if not tracked_roles:
return
# Wait a moment for Discord to fully process the join
await asyncio.sleep(1)
# Restore roles
roles_to_add = []
missing_roles = []
for role_id, role_name in tracked_roles:
role = member.guild.get_role(role_id)
if role:
# Check if bot can assign this role (bot's highest role must be higher)
if role < member.guild.me.top_role and not role.is_default():
roles_to_add.append(role)
else:
missing_roles.append(role_name)
# Add roles
if roles_to_add:
try:
await member.add_roles(*roles_to_add, reason="Role tracking: User rejoined server")
print(f"Restored {len(roles_to_add)} roles for {member.name} ({member.id})")
except discord.Forbidden:
print(f"Failed to restore roles for {member.name}: Missing permissions")
except discord.HTTPException as e:
print(f"Failed to restore roles for {member.name}: {e}")
# Log if there were missing roles
if missing_roles:
print(f"Could not restore the following roles for {member.name} (roles no longer exist): {', '.join(missing_roles)}")
@commands.Cog.listener()
async def on_member_update(self, before: discord.Member, after: discord.Member):
"""Update tracked roles when a user's roles change"""
if after.bot:
return
# Check if user is opted in
if not self.is_opted_in(after.id, after.guild.id):
return
# Check if roles actually changed
if before.roles == after.roles:
return
# Save updated roles
self.save_user_roles(after)
print(f"Updated tracked roles for {after.name} ({after.id})")
@commands.Cog.listener()
async def on_member_remove(self, member: discord.Member):
"""Optional: Log when a tracked user leaves"""
if member.bot:
return
if self.is_opted_in(member.id, member.guild.id):
print(f"Tracked user {member.name} ({member.id}) left the server. Roles are saved for restoration.")
async def setup(bot):
await bot.add_cog(RoleTrackEvents(bot))

19
events/status_alerts.py
View File

@@ -9,6 +9,9 @@ import os
from pathlib import Path
from datetime import datetime
from zoneinfo import ZoneInfo
from utils.logger import get_logger
logger = get_logger(__name__)
class StatusMonitor(commands.Cog):
def __init__(self, bot):
@@ -86,7 +89,7 @@ class StatusMonitor(commands.Cog):
try:
await admin.send(embed=embed)
except Exception as e:
print(f"Failed to send down alert for {service_name}: {e}")
logger.error(f"Failed to send down alert for {service_name}: {e}")
# Service went DEGRADED
elif current_status == "degraded" and previous_status == "up":
@@ -104,7 +107,7 @@ class StatusMonitor(commands.Cog):
try:
await admin.send(embed=embed)
except Exception as e:
print(f"Failed to send degraded alert for {service_name}: {e}")
logger.error(f"Failed to send degraded alert for {service_name}: {e}")
# Service RECOVERED
elif current_status == "up" and previous_status in ["down", "degraded"]:
@@ -146,20 +149,20 @@ class StatusMonitor(commands.Cog):
try:
await admin.send(embed=embed)
except Exception as e:
print(f"Failed to send recovery alert for {service_name}: {e}")
logger.error(f"Failed to send recovery alert for {service_name}: {e}")
# Update last known status
self.last_status[service_name] = current_status
except Exception as e:
print(f"[StatusMonitor] Error checking status: {e}")
logger.error(f"Error checking status: {e}", exc_info=True)
@check_status.before_loop
async def before_check_status(self):
"""Wait for bot to be ready before starting the loop"""
await self.bot.wait_until_ready()
print("[StatusMonitor] Starting status monitoring loop...")
logger.info("Starting status monitoring loop...")
# Initialize last_status with current state on startup
try:
if self.db_path.exists():
@@ -181,9 +184,9 @@ class StatusMonitor(commands.Cog):
self.last_status[row['service_name']] = row['status']
conn.close()
print(f"[StatusMonitor] Initialized with {len(self.last_status)} services")
logger.info(f"Initialized with {len(self.last_status)} services")
except Exception as e:
print(f"[StatusMonitor] Failed to initialize: {e}")
logger.error(f"Failed to initialize: {e}", exc_info=True)
@commands.command(name="statustest")
@commands.is_owner()

1
events/welcome.py
View File

@@ -1,3 +1,4 @@
"""Welcome event cog that greets new members and assigns default roles."""
import discord
from discord.ext import commands

1
image/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Image generation and avatar manipulation commands."""

41
image/avatar.py
View File

@@ -1,22 +1,27 @@
"""Avatar manipulation commands (bitcrush, edge detect, explode, filters, etc.)."""
import discord
from discord import app_commands
from discord.ext import commands
from PIL import Image, ImageOps, ImageSequence
import io
from embed.embed_color import get_embed_color
import asyncio
import aiohttp
import traceback
import os
from pathlib import Path
import numpy as np
from scipy.ndimage import uniform_filter
import cv2
from utils.logger import get_logger
logger = get_logger(__name__)
class AvatarCommands(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.session = None
self.explosion_path = os.path.join(os.path.dirname(__file__), "..", "media", "explosion-deltarune.gif")
self.obama_path = os.path.join(os.path.dirname(__file__), "..", "media", "obama.jpg")
self.explosion_path = Path(__file__).parent.parent / "media" / "explosion-deltarune.gif"
self.obama_path = Path(__file__).parent.parent / "media" / "obama.jpg"
async def cog_load(self):
self.session = aiohttp.ClientSession()
@@ -55,13 +60,9 @@ class AvatarCommands(commands.Cog):
avatar = self.get_avatar_url(target, avatar_type)
avatar_url = avatar.url
embed_color = discord.Color.blue()
if self.bot.get_cog("EmbedColor"):
embed_color = self.bot.get_cog("EmbedColor").get_user_color(interaction.user)
embed = discord.Embed(
title=f"{target.display_name}'s Avatar",
color=embed_color
color=get_embed_color(interaction.user.id)
)
embed.set_image(url=avatar_url)
embed.add_field(name="Direct Link", value=f"[Open Avatar]({avatar_url})")
@@ -108,9 +109,9 @@ class AvatarCommands(commands.Cog):
file=file
)
except Exception:
traceback.print_exc()
logger.exception("Error in avatar bitcrush")
await interaction.followup.send("An error occurred while processing the image.", ephemeral=True)
def _bitcrush_image(self, image_bytes: bytes, bits: int) -> bytes:
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
colors = 2 ** bits
@@ -169,9 +170,9 @@ class AvatarCommands(commands.Cog):
file=file
)
except Exception:
traceback.print_exc()
logger.exception("Error in avatar canny_edge")
await interaction.followup.send("An error occurred while processing the image.", ephemeral=True)
def _canny_edge_detection(self, image_bytes: bytes, threshold1: int, threshold2: int) -> bytes:
# Load image and convert to grayscale
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
@@ -226,7 +227,7 @@ class AvatarCommands(commands.Cog):
await interaction.followup.send(f"{user.display_name} just got exploded!", file=file)
except Exception:
traceback.print_exc()
logger.exception("Error in avatar explode")
await interaction.followup.send("An error occurred while processing the explosion.", ephemeral=True)
def _explode_avatar(self, avatar_bytes: bytes) -> bytes:
@@ -288,9 +289,9 @@ class AvatarCommands(commands.Cog):
file=file
)
except Exception:
traceback.print_exc()
logger.exception("Error in avatar grayscale")
await interaction.followup.send("An error occurred while processing the image.", ephemeral=True)
def _grayscale_image(self, image_bytes: bytes) -> bytes:
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
grayscaled = ImageOps.grayscale(img)
@@ -331,9 +332,9 @@ class AvatarCommands(commands.Cog):
file=file
)
except Exception:
traceback.print_exc()
logger.exception("Error in avatar inverse")
await interaction.followup.send("An error occurred while processing the image.", ephemeral=True)
def _invert_image(self, image_bytes: bytes) -> bytes:
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
inverted = ImageOps.invert(img)
@@ -382,9 +383,9 @@ class AvatarCommands(commands.Cog):
file=file
)
except Exception:
traceback.print_exc()
logger.exception("Error in avatar kuwahara")
await interaction.followup.send("An error occurred while processing the image.", ephemeral=True)
def _kuwahara_filter(self, image_bytes: bytes, kernel_size: int) -> bytes:
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
img_array = np.array(img, dtype=np.float32)
@@ -470,7 +471,7 @@ class AvatarCommands(commands.Cog):
await interaction.followup.send(file=discord.File(buf, filename="obama_mosaic.png"))
except Exception:
traceback.print_exc()
logger.exception("Error in avatar obamify")
await interaction.followup.send("An error occurred during mosaic generation.", ephemeral=True)
async def _fetch_avatar(self, url: str) -> Image.Image:

1
lilac-tools/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Administrative tools and bot management commands for the bot owner."""

45
lilac-tools/admin.py
View File

@@ -1,3 +1,4 @@
"""Owner-only admin tools: database browser, terminal, eval, file manager."""
import discord
from discord.ext import commands
import aiosqlite
@@ -5,8 +6,10 @@ import asyncio
import os
import glob
from typing import Optional
import traceback
import re
from utils.logger import get_logger
logger = get_logger(__name__)
ADMIN_USER_ID = 252130669919076352
@@ -100,7 +103,7 @@ class AdminCommands(commands.Cog):
db_files = await loop.run_in_executor(None, _glob)
self.db_connections = {os.path.basename(f).replace('.db', ''): f for f in db_files}
print(f"Discovered databases: {list(self.db_connections.keys())}")
logger.info(f"Discovered databases: {list(self.db_connections.keys())}")
@commands.command(name="admin_help", aliases=["adminhelp", "ahelp"])
@is_owner()
@@ -398,7 +401,7 @@ class AdminCommands(commands.Cog):
await ctx.send(embed=embed)
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@commands.command(name="dbinfo")
@is_owner()
@@ -438,7 +441,7 @@ class AdminCommands(commands.Cog):
await ctx.send(embed=embed)
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@commands.command(name="dbquery")
@is_owner()
@@ -476,7 +479,7 @@ class AdminCommands(commands.Cog):
await ctx.send(f"```\n{output}\n```")
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
def _write_file(self, filename, content):
"""Helper to write file synchronously in executor"""
@@ -500,7 +503,7 @@ class AdminCommands(commands.Cog):
await ctx.send(f"✅ Query executed successfully. Rows affected: {affected_rows}")
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@commands.command(name="terminal", aliases=["term", "sh"])
@is_owner()
@@ -544,7 +547,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@commands.command(name="eval")
@is_owner()
@@ -611,7 +614,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
error_msg = f"```python\n{type(e).__name__}: {str(e)}\n```"
await ctx.send(f"❌ Error:\n{error_msg}")
traceback.print_exc()
logger.exception("Error in admin command")
@commands.command(name="pwd")
@is_owner()
@@ -764,7 +767,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@code.command(name="search", aliases=["grep", "find-all"])
@is_owner()
@@ -805,7 +808,7 @@ class AdminCommands(commands.Cog):
})
if len(results) >= 100: # Limit results
return results, "truncated"
except:
except Exception:
continue
return results, None
@@ -860,7 +863,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@code.command(name="find")
@is_owner()
@@ -927,11 +930,11 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@code.command(name="edit")
@is_owner()
@@ -1079,7 +1082,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@code.command(name="insert")
@is_owner()
@@ -1124,7 +1127,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@code.command(name="delete", aliases=["del"])
@is_owner()
@@ -1172,7 +1175,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@code.command(name="replace")
@is_owner()
@@ -1233,7 +1236,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@code.command(name="backup")
@is_owner()
@@ -1265,7 +1268,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@code.command(name="diff")
@is_owner()
@@ -1319,7 +1322,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
@commands.command(name="grep")
@is_owner()
@@ -1356,7 +1359,7 @@ class AdminCommands(commands.Cog):
results.append(f"{filepath}:{i}: {line.rstrip()}")
if len(results) >= 50: # Limit results
return results, "truncated"
except:
except Exception:
continue
return results, None
@@ -1385,7 +1388,7 @@ class AdminCommands(commands.Cog):
except Exception as e:
await ctx.send(f"❌ Error: {str(e)}")
traceback.print_exc()
logger.exception("Error in admin command")
async def setup(bot):
await bot.add_cog(AdminCommands(bot))

1
moderation/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Moderation tools and admin permission management."""

3
moderation/ban.py
View File

@@ -1,3 +1,4 @@
"""Ban command with confirmation prompt and DM notification."""
import discord
from discord.ext import commands
from discord.ui import View, Button
@@ -62,7 +63,7 @@ class BanCommand(ModerationBase):
f"Reason: {reason or 'No reason provided'}\n\n"
f"If you believe this ban was unfiair and would like to appeal, join here: https://discord.gg/FYpfBzpjvq"
)
except:
except Exception:
await ctx.send("Could not DM the user.")
# Perform the ban

3
moderation/cleanban.py
View File

@@ -1,3 +1,4 @@
"""Cleanban command that bans and deletes recent messages."""
import discord
from discord.ext import commands
from discord.ui import View, Button
@@ -69,7 +70,7 @@ class CleanBanCommand(ModerationBase):
f"Messages from the past {days} day(s) have been deleted.\n"
f"Reason: {reason or 'No reason provided'}\n\n"
)
except:
except Exception:
await ctx.send("Could not DM the user.")
# Perform the ban with message deletion

18
moderation/infraction.py
View File

@@ -1,7 +1,11 @@
"""Infraction logging and retrieval system for moderation actions."""
import discord
from discord.ext import commands, tasks
from .loader import ModerationBase
from datetime import datetime, timedelta
from utils.logger import get_logger
logger = get_logger(__name__)
class InfractionCommand(ModerationBase):
@@ -69,7 +73,7 @@ class InfractionCommand(ModerationBase):
await self.check_user_eligibility(guild_id, user_id)
except Exception as e:
print(f"Error in auto-removal check: {e}")
logger.error(f"Error in auto-removal check: {e}", exc_info=True)
@check_auto_removals.before_loop
async def before_check_auto_removals(self):
@@ -136,7 +140,7 @@ class InfractionCommand(ModerationBase):
await self.send_removal_approval(guild_id, user_id, inf_id, inf_type, reason, timestamp_str, mod_id)
except Exception as e:
print(f"Error checking eligibility for user {user_id} in guild {guild_id}: {e}")
logger.error(f"Error checking eligibility for user {user_id} in guild {guild_id}: {e}", exc_info=True)
async def send_removal_approval(self, guild_id: int, user_id: int, inf_id: int,
inf_type: str, reason: str, timestamp: str, mod_id: int):
@@ -144,7 +148,7 @@ class InfractionCommand(ModerationBase):
try:
approval_channel = self.bot.get_channel(self.approval_channel_id)
if not approval_channel:
print(f"Approval channel {self.approval_channel_id} not found")
logger.error(f"Approval channel {self.approval_channel_id} not found")
return
# Get guild, user, and moderator info
@@ -155,13 +159,13 @@ class InfractionCommand(ModerationBase):
try:
user = await self.bot.fetch_user(user_id)
user_tag = f"{user.name}#{user.discriminator}"
except:
except Exception:
user_tag = f"Unknown User ({user_id})"
try:
moderator = await self.bot.fetch_user(mod_id)
mod_tag = f"{moderator.name}#{moderator.discriminator}"
except:
except Exception:
mod_tag = f"Unknown Mod ({mod_id})"
# Create approval embed
@@ -193,7 +197,7 @@ class InfractionCommand(ModerationBase):
self.conn.commit()
except Exception as e:
print(f"Error sending removal approval: {e}")
logger.error(f"Error sending removal approval: {e}", exc_info=True)
@commands.command(name="inf")
@ModerationBase.is_admin()
@@ -551,7 +555,7 @@ class InfractionRemovalView(discord.ui.View):
notify_embed.set_footer(text=f"You stayed clean for 4 months! Keep up the good behavior.")
await user.send(embed=notify_embed)
except:
except Exception:
pass # User has DMs disabled or bot can't reach them
except Exception as e:

3
moderation/kick.py
View File

@@ -1,3 +1,4 @@
"""Kick command with confirmation prompt and DM notification."""
import discord
from discord.ext import commands
from discord.ui import View, Button
@@ -41,7 +42,7 @@ class KickCommand(ModerationBase):
try:
await user.send(f"You have been **kicked** from **{ctx.guild.name}**.\nReason: {reason or 'No reason provided'}")
except:
except Exception:
await ctx.send("Could not DM the user.")
await ctx.guild.kick(user, reason=reason)

9
moderation/loader.py
View File

@@ -1,8 +1,11 @@
"""Shared ModerationBase cog with DB connection and is_admin() decorator."""
from discord.ext import commands
import os
import sqlite3
from pathlib import Path
from dotenv import load_dotenv
from datetime import datetime
from utils.constants import LILAC_ID
load_dotenv()
@@ -13,14 +16,12 @@ ADMIN_ROLE_IDS = {
if role_id.strip().isdigit()
}
lilac_id = 252130669919076352
class ModerationBase(commands.Cog):
"""Base cog for moderation commands with shared DB and utilities"""
def __init__(self, bot: commands.Bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(__file__), "moderation.db")
self.db_path = Path(__file__).parent / "moderation.db"
self.conn = sqlite3.connect(self.db_path)
self.conn.row_factory = sqlite3.Row
self.c = self.conn.cursor()
@@ -70,7 +71,7 @@ class ModerationBase(commands.Cog):
await send_message("Unable to check permissions in this context.", ephemeral=is_interaction)
return False
is_lilac = user.id == lilac_id
is_lilac = user.id == LILAC_ID
has_admin_role = any(
role.id in ADMIN_ROLE_IDS

33
moderation/lock.py
View File

@@ -1,18 +1,21 @@
"""Channel lock/unlock commands that save and restore permission overwrites."""
import discord
from discord.ext import commands
import traceback
from .loader import ModerationBase
import os
from pathlib import Path
import sqlite3
import json
import asyncio
from functools import partial
from utils.logger import get_logger
logger = get_logger(__name__)
class LockCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "moderation.db")
self.db_path = Path(__file__).parent.parent / "data" / "moderation.db"
self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
self.conn.row_factory = sqlite3.Row
self.c = self.conn.cursor()
@@ -195,7 +198,7 @@ class LockCog(commands.Cog):
failed_targets.append(target.name)
if failed_targets:
print(f"Warning: Could not clear permissions for: {', '.join(failed_targets)}")
logger.warning(f"Could not clear permissions for: {', '.join(failed_targets)}")
# Deny everyone from talking
try:
@@ -215,7 +218,7 @@ class LockCog(commands.Cog):
# If we can't set ritual member, at least try to undo the everyone change
try:
await channel.set_permissions(everyone_role, overwrite=None)
except:
except Exception:
pass
await ctx.message.add_reaction("❌")
return await ctx.send(f"Cannot modify {ritual_member_role.name} permissions! My role needs to be higher than that role.", allowed_mentions=discord.AllowedMentions.none())
@@ -226,23 +229,23 @@ class LockCog(commands.Cog):
# Try to send in channel
try:
await ctx.send(f"#{channel.name} has been locked! Only {ritual_member_role.name} can talk.", allowed_mentions=discord.AllowedMentions.none())
except:
except Exception:
pass
except discord.Forbidden as e:
await ctx.message.add_reaction("❌")
try:
await ctx.send(f"Missing permissions! My role needs to be higher than the roles I'm trying to modify.", allowed_mentions=discord.AllowedMentions.none())
except:
except Exception:
pass
traceback.print_exc()
logger.exception("Error in lock/unlock command")
except Exception as e:
await ctx.message.add_reaction("❌")
try:
await ctx.send(f"Error: {e}", allowed_mentions=discord.AllowedMentions.none())
except:
except Exception:
pass
traceback.print_exc()
logger.exception("Error in lock/unlock command")
@commands.command(name="unlock")
@ModerationBase.is_admin()
@@ -290,23 +293,23 @@ class LockCog(commands.Cog):
if len(failed_targets) > 5:
msg += f" and {len(failed_targets) - 5} more"
await ctx.send(msg, allowed_mentions=discord.AllowedMentions.none())
except:
except Exception:
pass
except discord.Forbidden as e:
await ctx.message.add_reaction("❌")
try:
await ctx.send(f"Missing permissions! My role needs to be higher than the roles I'm trying to modify.", allowed_mentions=discord.AllowedMentions.none())
except:
except Exception:
pass
traceback.print_exc()
logger.exception("Error in lock/unlock command")
except Exception as e:
await ctx.message.add_reaction("❌")
try:
await ctx.send(f"Error: {e}", allowed_mentions=discord.AllowedMentions.none())
except:
except Exception:
pass
traceback.print_exc()
logger.exception("Error in lock/unlock command")
async def setup(bot):

20
moderation/logger.py
View File

@@ -1,17 +1,20 @@
"""Moderation event logger that records Discord actions to the database."""
import discord
from discord.ext import commands
import sqlite3
import os
from pathlib import Path
from datetime import datetime, timezone
from typing import Optional
import traceback
from utils.logger import get_logger
_logger = get_logger(__name__)
class Logger(commands.Cog):
"""Core logging system that listens to Discord events and logs them"""
def __init__(self, bot: commands.Bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "moderation.db")
self.db_path = Path(__file__).parent.parent / "data" / "moderation.db"
self.initialize_db()
# Cache for deleted messages (for bulk delete context)
self.message_cache = {}
@@ -76,10 +79,9 @@ class Logger(commands.Cog):
try:
msg = await channel.send(embed=embed)
except discord.Forbidden as e:
print(f"[ERROR] Missing permissions to send log in channel {channel_id}: {e}")
_logger.error(f"Missing permissions to send log in channel {channel_id}: {e}")
except Exception as e:
print(f"[ERROR] Error sending log: {e}")
traceback.print_exc()
_logger.error(f"Error sending log: {e}", exc_info=True)
@commands.Cog.listener()
async def on_message_delete(self, message: discord.Message):
@@ -188,7 +190,7 @@ class Logger(commands.Cog):
except Exception as e:
error_text = f"⚠️ **Error while logging member join:**\n`{type(e).__name__}: {e}`"
traceback.print_exc()
_logger.error(f"Error logging member join: {e}", exc_info=True)
# Attempt to send the error to the designated debug channel
try:
@@ -196,9 +198,9 @@ class Logger(commands.Cog):
if channel:
await channel.send(error_text)
else:
print("[ERROR] Could not find error logging channel (1424145004976275617).")
_logger.error("Could not find error logging channel (1424145004976275617).")
except Exception as send_err:
print(f"[ERROR] Failed to send error message to debug channel: {send_err}")
_logger.error(f"Failed to send error message to debug channel: {send_err}")
@commands.Cog.listener()
async def on_member_remove(self, member: discord.Member):

66
moderation/modhelp.py Normal file
View File

@@ -0,0 +1,66 @@
import discord
from discord.ext import commands
from .loader import ModerationBase
class ModHelp(ModerationBase):
@commands.command(name="modhelp")
@ModerationBase.is_admin()
async def modhelp(self, ctx):
embeds = []
def embed(title, color=discord.Color.blurple()):
e = discord.Embed(title=title, color=color)
return e
e1 = embed("Moderation Commands — Member Actions", discord.Color.red())
e1.add_field(name="!ban <user> [reason]", value="Ban a user. Accepts mention, ID, or name. Works even if they're not in the server. Sends a DM.", inline=False)
e1.add_field(name="!cleanban <user> [days] [reason]", value="Ban a user and delete their recent messages. `days` is 1–7 (default 1). Sends a DM.", inline=False)
e1.add_field(name="!kick <user> [reason]", value="Kick a user from the server. Sends a DM.", inline=False)
e1.add_field(name="!unban <user> [reason]", value="Unban a user. Accepts mention, ID, or username.", inline=False)
e1.add_field(name="!warn <user> [reason]", value="Issue a warning to a user. Sends a DM and logs the infraction.", inline=False)
e1.add_field(name="!mute <user> <duration> [reason]", value="Mute a user for a set duration. Format: `1w`, `5d`, `12h`, `30m`. Auto-unmutes when time is up.", inline=False)
e1.add_field(name="!unmute <user>", value="Manually remove a mute from a user.", inline=False)
embeds.append(e1)
e2 = embed("Moderation Commands — Purge", discord.Color.orange())
e2.add_field(name="!purge <message_id>", value="Delete all messages up to (and including) the specified message ID in the current channel.", inline=False)
e2.add_field(name="!purgemember <user> <message_id>", value="Delete messages from a specific user up to the specified message ID. Aliases: `purgeuser`, `purgeu`, `purgem`", inline=False)
e2.add_field(name="!purgebot <message_id>", value="Delete bot messages up to the specified message ID. Aliases: `purgebots`, `purgeb`", inline=False)
e2.add_field(name="!purgecontains <message_id> <text>", value="Delete messages containing specific text up to the specified message ID. Aliases: `purgec`, `purgetext`", inline=False)
e2.add_field(name="!purgeembeds <message_id>", value="Delete messages that contain embeds up to the specified message ID. Aliases: `purgee`, `purgeembed`", inline=False)
e2.add_field(name="!purgememberall <user_id>", value="Delete **all** messages from a user across every text channel. Requires confirmation. Aliases: `purgeuserall`, `purgeua`, `purgeallm`", inline=False)
embeds.append(e2)
e3 = embed("Moderation Commands — Channel & Infractions", discord.Color.gold())
e3.add_field(name="!lock [channel]", value="Lock a channel so only staff can send messages. Defaults to the current channel. Saves original permissions for restore.", inline=False)
e3.add_field(name="!unlock [channel]", value="Unlock a previously locked channel and restore its original permissions. Defaults to the current channel.", inline=False)
e3.add_field(name="!checkperms [channel]", value="Check the bot's permissions in a channel and display role hierarchy info. Defaults to the current channel.", inline=False)
e3.add_field(name="\u200b", value="**Infractions**", inline=False)
e3.add_field(name="!inf search <user_id>", value="Show all active infractions for a user in this server.", inline=False)
e3.add_field(name="!inf search_full <user_id>", value="Show the full infraction history for a user, including removed infractions.", inline=False)
e3.add_field(name="!inf list", value="List all active infractions in this server.", inline=False)
e3.add_field(name="!inf delete <id>", value="Permanently delete an infraction by its ID. The user is notified via DM.", inline=False)
embeds.append(e3)
e4 = embed("Moderation Commands — Logging", discord.Color.blue())
e4.add_field(name="!log set <#channel> <type>", value="Route a log type to a channel. Use `!log types` to see all available types.", inline=False)
e4.add_field(name="!log remove <type>", value="Remove the channel assignment for a log type.", inline=False)
e4.add_field(name="!log list", value="Show all configured log types and their channels.", inline=False)
e4.add_field(name="!log types", value="List every available log type.", inline=False)
e4.add_field(name="!log exclude <channel_id>", value="Exclude a channel from being logged. Events in that channel won't appear in any log.", inline=False)
e4.add_field(name="!log unexclude <channel_id>", value="Remove a channel from the exclusion list.", inline=False)
e4.add_field(name="!log excluded", value="List all channels currently excluded from logging.", inline=False)
e4.add_field(name="!log clear", value="Remove **all** logging configurations for this server. Requires confirmation.", inline=False)
embeds.append(e4)
e5 = embed("Moderation Commands — Tools", discord.Color.green())
e5.add_field(name="!send_embed <#channel> <embed_string>", value="Send an embed (built with the embed builder) to a channel. Previews before sending. Requires confirmation.", inline=False)
e5.set_footer(text="Arguments in <> are required. Arguments in [] are optional.")
embeds.append(e5)
await ctx.send(embeds=embeds)
async def setup(bot: commands.Bot):
await bot.add_cog(ModHelp(bot))

18
moderation/mute.py
View File

@@ -1,13 +1,16 @@
"""Mute command with duration parsing and scheduled unmute task."""
import discord
from discord.ext import commands, tasks
from discord.ui import View, Button
import asyncio
import re
import sqlite3
import os
from pathlib import Path
from datetime import timedelta, datetime
from .loader import ModerationBase
DB_PATH = Path(__file__).parent.parent / "data" / "moderation.db"
MUTE_ROLE_ID = 982702037517090836
class MuteCommand(ModerationBase):
@@ -77,14 +80,13 @@ class MuteCommand(ModerationBase):
await user.add_roles(mute_role, reason=reason)
try:
await user.send(f"You have been muted in **{ctx.guild.name}** for **{duration}**.\nReason: {reason or 'No reason provided'}")
except:
except Exception:
await ctx.send("Could not DM the user.")
await self.log_infraction(ctx.guild.id, user.id, ctx.author.id, "mute", reason)
await ctx.send(f"{user.mention} has been muted for **{duration}**.")
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "moderation.db")
conn = sqlite3.connect(db_path)
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
c.execute("""
CREATE TABLE IF NOT EXISTS mutes (
@@ -111,8 +113,7 @@ class MuteCommand(ModerationBase):
async def schedule_unmute(self, user_id, guild_id, channel_id, delay):
await asyncio.sleep(delay)
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "moderation.db")
conn = sqlite3.connect(db_path)
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
c.execute("SELECT 1 FROM mutes WHERE user_id = ? AND guild_id = ?", (user_id, guild_id))
exists = c.fetchone()
@@ -140,13 +141,12 @@ class MuteCommand(ModerationBase):
await logger.log_moderation_action(
guild_id, "unmute", member, self.bot.user, "Mute duration expired"
)
except:
except Exception:
pass
@tasks.loop(minutes=1)
async def check_mutes(self):
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "moderation.db")
conn = sqlite3.connect(db_path)
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
now = datetime.utcnow().isoformat()
c.execute("SELECT user_id, guild_id, channel_id, unmute_time FROM mutes WHERE unmute_time <= ?", (now,))

28
moderation/spam_protection.py
View File

@@ -1,3 +1,4 @@
"""Automatic spam detection and rate-limiting with auto-mute."""
import discord
from discord.ext import commands, tasks
from discord.ui import View, Button
@@ -5,7 +6,10 @@ import asyncio
from datetime import datetime, timedelta, timezone
from collections import defaultdict, deque
import sqlite3
import os
from pathlib import Path
from utils.logger import get_logger
logger = get_logger(__name__)
MUTE_ROLE_ID = 982702037517090836
STAFF_CHANNEL_ID = 876780367296745493
@@ -17,7 +21,7 @@ class SpamProtection(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "moderation.db")
self.db_path = Path(__file__).parent.parent / "data" / "moderation.db"
# Track user message patterns
# user_id -> deque of (timestamp, channel_id, content)
@@ -149,7 +153,7 @@ class SpamProtection(commands.Cog):
f"Automatically muted for 1 day."
)
except Exception as e:
print(f"[ERROR] Failed to apply default spam action: {e}")
logger.error(f"Failed to apply default spam action: {e}", exc_info=True)
@commands.Cog.listener()
async def on_message(self, message: discord.Message):
@@ -192,7 +196,7 @@ class SpamProtection(commands.Cog):
except asyncio.QueueEmpty:
break
except Exception as e:
print(f"[ERROR] Error in message queue processing: {e}")
logger.error(f"Error in message queue processing: {e}", exc_info=True)
async def _process_message(self, message: discord.Message):
"""Actually process the message for spam detection"""
@@ -269,16 +273,16 @@ class SpamProtection(commands.Cog):
# Apply mute role
mute_role = guild.get_role(MUTE_ROLE_ID)
if not mute_role:
print(f"[ERROR] Mute role {MUTE_ROLE_ID} not found in guild {guild.id}")
logger.error(f"Mute role {MUTE_ROLE_ID} not found in guild {guild.id}")
return
try:
await member.add_roles(mute_role, reason="Automatic spam detection")
except discord.Forbidden:
print(f"[ERROR] Missing permissions to mute {member.id}")
logger.error(f"Missing permissions to mute {member.id}")
return
except Exception as e:
print(f"[ERROR] Failed to mute spammer: {e}")
logger.error(f"Failed to mute spammer: {e}", exc_info=True)
return
# Log the automatic mute action
@@ -295,13 +299,13 @@ class SpamProtection(commands.Cog):
f"You have been automatically muted in **{guild.name}** for spam detection. "
f"A staff member will review your case shortly."
)
except:
except Exception:
pass # Can't DM user
# Create staff alert
staff_channel = guild.get_channel(STAFF_CHANNEL_ID)
if not staff_channel:
print(f"[ERROR] Staff channel {STAFF_CHANNEL_ID} not found")
logger.error(f"Staff channel {STAFF_CHANNEL_ID} not found")
return
# Build embed
@@ -410,7 +414,7 @@ class SpamProtection(commands.Cog):
view.alert_message_id = msg.id
except Exception as e:
print(f"[ERROR] Failed to send spam alert: {e}")
logger.error(f"Failed to send spam alert: {e}", exc_info=True)
class SpamActionView(View):
"""Interactive buttons for staff to handle spam reports"""
@@ -592,7 +596,7 @@ class SpamActionView(View):
f"You have been **banned** from **{self.guild.name}** for spam.\n"
f"Reason: {reason}"
)
except:
except Exception:
pass
# Ban user

10
moderation/unmmute.py
View File

@@ -1,10 +1,13 @@
"""Unmute command with confirmation prompt."""
import discord
from discord.ext import commands
from discord.ui import View, Button
import sqlite3
import os
from pathlib import Path
from .loader import ModerationBase
DB_PATH = Path(__file__).parent.parent / "data" / "moderation.db"
MUTE_ROLE_ID = 982702037517090836
class UnmuteCommand(ModerationBase):
@@ -49,15 +52,14 @@ class UnmuteCommand(ModerationBase):
if mute_role in user.roles:
await user.remove_roles(mute_role, reason="Manual unmute issued")
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "moderation.db")
conn = sqlite3.connect(db_path)
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
c.execute("DELETE FROM mutes WHERE user_id = ? AND guild_id = ?", (user.id, ctx.guild.id))
conn.commit()
conn.close()
try:
await user.send(f"You have been **unmuted** in **{ctx.guild.name}**.")
except:
except Exception:
await ctx.send("Could not DM the user.")
await self.log_infraction(ctx.guild.id, user.id, ctx.author.id, "unmute", "Manual unmute issued")

3
moderation/warn.py
View File

@@ -1,3 +1,4 @@
"""Warn command that logs infractions and notifies the user."""
import discord
from discord.ext import commands
from discord.ui import View, Button
@@ -41,7 +42,7 @@ class WarnCommand(ModerationBase):
try:
await user.send(f"You have been **warned** in **{ctx.guild.name}**.\nReason: {reason or 'No reason provided'}")
except:
except Exception:
await ctx.send("Could not DM the user.")
await self.log_infraction(ctx.guild.id, user.id, ctx.author.id, "warn", reason)

1
profiles/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""User profile card generation and customization."""

5
profiles/database.py
View File

@@ -1,7 +1,8 @@
"""Profile database helpers for connecting to the profile SQLite database."""
import sqlite3
import os
from pathlib import Path
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "profile.db")
DB_PATH = Path(__file__).parent.parent / "data" / "profile.db"
def get_db():
db = sqlite3.connect(DB_PATH)

10
profiles/profiles.py
View File

@@ -1,3 +1,4 @@
"""User profile card generation with custom fonts and avatar overlays."""
import discord
from discord.ext import commands
from discord import app_commands
@@ -6,10 +7,11 @@ import aiohttp
import io
import os
import asyncio
from pathlib import Path
from textwrap import wrap
from .database import get_db, setup_db
FONTS_PATH = os.path.join(os.path.dirname(__file__), "fonts")
FONTS_PATH = Path(__file__).parent / "fonts"
def list_fonts():
fonts = []
@@ -243,7 +245,7 @@ class Profiles(commands.GroupCog, name="profile"):
# Validate it's a proper hex color
int(bg_color[1:], 16)
base_color = bg_color
except:
except Exception:
pass # Use default if invalid
img = Image.new("RGB", (img_width, img_height), base_color)
@@ -307,7 +309,7 @@ class Profiles(commands.GroupCog, name="profile"):
else:
username_font = ImageFont.load_default()
break
except:
except Exception:
username_font = ImageFont.load_default()
break
@@ -376,7 +378,7 @@ class Profiles(commands.GroupCog, name="profile"):
role_font = ImageFont.load_default()
else:
role_font = ImageFont.load_default()
except:
except Exception:
role_font = ImageFont.load_default()
for role in display_roles:

1
reminders/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Reminder scheduling and notification system."""

11
reminders/reminder.py
View File

@@ -1,3 +1,4 @@
"""Reminder cog that lets users set timed reminders via slash commands."""
import discord
from discord import app_commands
from discord.ext import commands, tasks
@@ -5,6 +6,10 @@ import aiosqlite
from datetime import datetime, timedelta, timezone
import re
from pathlib import Path
from embed.embed_color import get_embed_color
from utils.logger import get_logger
logger = get_logger(__name__)
def parse_timeframe(timeframe: str) -> timedelta:
@@ -85,7 +90,7 @@ class ReminderCog(commands.Cog):
ephemeral=True
)
except Exception as e:
print(f"Error in reminder_set: {e}")
logger.error(f"Error in reminder_set: {e}", exc_info=True)
if not interaction.response.is_done():
await interaction.response.send_message(f"❌ Error: {e}", ephemeral=True)
@@ -105,7 +110,7 @@ class ReminderCog(commands.Cog):
embed = discord.Embed(
title="📝 Your Reminders",
color=discord.Color.blurple(),
color=get_embed_color(interaction.user.id),
timestamp=datetime.now(timezone.utc)
)
@@ -218,7 +223,7 @@ class ReminderCog(commands.Cog):
embed = discord.Embed(
title="⏰ Reminder!",
description=message,
color=discord.Color.blue(),
color=get_embed_color(user_id),
timestamp=datetime.now(timezone.utc)
)
await user.send(embed=embed)

1
sparkle/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Sparkle currency system for rewarding community engagement."""

15
sparkle/database.py
View File

@@ -1,13 +1,12 @@
"""Sparkle currency database helpers for the sparkle economy."""
import sqlite3
import os
from pathlib import Path
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "sparkle.db")
DB_PATH = Path(__file__).parent.parent / "data" / "sparkle.db"
def get_db():
"""Return a SQLite3 connection with initialized tables."""
conn = sqlite3.connect(DB_PATH)
# Create sparkles table
conn.execute("""
CREATE TABLE IF NOT EXISTS sparkles (
server_id TEXT,
@@ -18,8 +17,7 @@ def get_db():
PRIMARY KEY (server_id, user_id)
)
""")
# Create sparkle_events table for tracking individual sparkle occurrences
conn.execute("""
CREATE TABLE IF NOT EXISTS sparkle_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -30,8 +28,7 @@ def get_db():
timestamp INTEGER NOT NULL
)
""")
# Create index for faster queries
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_sparkle_events_server
ON sparkle_events(server_id, timestamp)

18
sparkle/sparkle_commands.py
View File

@@ -1,3 +1,4 @@
"""Sparkle currency commands for giving, checking, and viewing sparkle stats."""
import discord
from discord.ext import commands
from discord import app_commands
@@ -5,6 +6,9 @@ from discord.utils import escape_markdown
from .database import get_db
import asyncio
import datetime
import sqlite3
from pathlib import Path
from embed.embed_color import get_embed_color
class SparkleCommands(commands.Cog):
@@ -54,7 +58,7 @@ class SparkleCommands(commands.Cog):
epic, rare, regular, total = result
embed = discord.Embed(
title=f"{user.display_name}'s Sparkles",
color=discord.Color.gold()
color=get_embed_color(interaction.user.id)
)
embed.set_thumbnail(url=user.display_avatar.url)
embed.add_field(
@@ -83,7 +87,7 @@ class SparkleCommands(commands.Cog):
"💫 **Epic Sparkle** – 1/100,000 chance\n\n"
"Track your sparkles using `/sparkle check` or `/sparkle leaderboard`."
),
color=discord.Color.purple()
color=get_embed_color(interaction.user.id)
)
embed.set_footer(text="Keep chatting to test your luck!")
await interaction.response.send_message(embed=embed)
@@ -125,7 +129,7 @@ class SparkleCommands(commands.Cog):
embed = discord.Embed(
title=f"✨ {escape_markdown(interaction.guild.name)} Sparkle Leaderboard",
color=discord.Color.gold()
color=get_embed_color(interaction.user.id)
)
medal = {1: "🥇", 2: "🥈", 3: "🥉"}
@@ -180,9 +184,7 @@ class SparkleCommands(commands.Cog):
conn.close()
# Get total message count from stats.db (located in ../stats/)
import os
import sqlite3
stats_db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "stats.db")
stats_db_path = Path(__file__).parent.parent / "data" / "stats.db"
total_messages = 0
try:
stats_conn = sqlite3.connect(stats_db_path)
@@ -193,7 +195,7 @@ class SparkleCommands(commands.Cog):
if result and result[0]:
total_messages = result[0]
stats_conn.close()
except:
except Exception:
pass # If stats.db doesn't exist or has issues, just use 0
return total_epic, total_rare, total_regular, events, total_messages
@@ -217,7 +219,7 @@ class SparkleCommands(commands.Cog):
embed = discord.Embed(
title=f"📊 Sparkle Stats for {interaction.guild.name}",
color=discord.Color.purple()
color=get_embed_color(interaction.user.id)
)
embed.add_field(

1
stats/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Server statistics tracking and reporting."""

30
stats/stats.py
View File

@@ -1,18 +1,24 @@
"""Statistics tracking cog: message counts, word frequencies, bot uptime."""
import discord
from discord.ext import commands
from discord import app_commands
from datetime import datetime, timezone
import sqlite3
from embed.embed_color import get_embed_color
import os
import json
from pathlib import Path
import asyncio
import aiofiles
import re
from collections import Counter
from typing import Dict, List, Tuple
from utils.logger import get_logger
BASE_DIR = os.path.dirname(__file__)
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "stats.db")
logger = get_logger(__name__)
BASE_DIR = Path(__file__).parent
DB_PATH = Path(__file__).parent.parent / "data" / "stats.db"
class Stats(commands.Cog):
def __init__(self, bot):
@@ -41,8 +47,10 @@ class Stats(commands.Cog):
}
self.init_db()
self.stats_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "bot_stats.json")
self.bot.loop.create_task(self.update_stats_file())
self.stats_file = Path(__file__).parent.parent / "data" / "bot_stats.json"
async def cog_load(self):
asyncio.create_task(self.update_stats_file())
def init_db(self):
db = sqlite3.connect(DB_PATH)
@@ -243,9 +251,9 @@ class Stats(commands.Cog):
file_path = os.path.join(stats_dir, filename)
if file_path != self.stats_file:
os.remove(file_path)
print(f"Removed old stats file: {filename}")
logger.info(f"Removed old stats file: {filename}")
except Exception as e:
print(f"Error cleaning up old stats files: {e}")
logger.error(f"Error cleaning up old stats files: {e}", exc_info=True)
async def update_stats_file(self):
await self.bot.wait_until_ready()
@@ -257,7 +265,7 @@ class Stats(commands.Cog):
async with aiofiles.open(self.stats_file, 'w') as f:
await f.write(json.dumps(stats, indent=2, default=str))
except Exception as e:
print(f"Error updating stats file: {e}")
logger.error(f"Error updating stats file: {e}", exc_info=True)
await asyncio.sleep(30) # Update every 30 seconds
@@ -346,7 +354,7 @@ class Stats(commands.Cog):
embed = discord.Embed(
title=f"🌟 {guild.name} Statistics 🌟",
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user),
color=get_embed_color(interaction.user.id),
timestamp=datetime.now(timezone.utc)
)
@@ -389,7 +397,7 @@ class Stats(commands.Cog):
embed = discord.Embed(
title="📊 Message Statistics",
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user),
color=get_embed_color(interaction.user.id),
timestamp=datetime.now(timezone.utc)
)
@@ -413,7 +421,7 @@ class Stats(commands.Cog):
embed = discord.Embed(
title="🔤 Word Frequency Statistics",
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user),
color=get_embed_color(interaction.user.id),
timestamp=datetime.now(timezone.utc)
)
@@ -437,7 +445,7 @@ class Stats(commands.Cog):
embed = discord.Embed(
title="📊 Most Active Channels",
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user),
color=get_embed_color(interaction.user.id),
timestamp=datetime.now(timezone.utc)
)

1
suggestion/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Suggestion submission and review system."""

78
suggestion/suggest.py
View File

@@ -1,11 +1,16 @@
"""Suggestion system cog with voting, approval, and completion workflow."""
import discord
from discord import app_commands
from discord.ext import commands
import aiosqlite
from datetime import datetime
import os
from embed.embed_color import get_embed_color
from pathlib import Path
from typing import Optional
import traceback
from utils.logger import get_logger
logger = get_logger(__name__)
ADMIN_ID = 252130669919076352
ADMIN_CHANNEL_ID = 1470441786810826884
@@ -32,7 +37,7 @@ class DenyModal(discord.ui.Modal, title="Reason for denying suggestion"):
reason_text = self.reason.value or None
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "suggestions.db")
db_path = Path(__file__).parent.parent / "data" / "suggestions.db"
async with aiosqlite.connect(db_path) as db:
await db.execute("UPDATE suggestions SET status = ?, reason = ? WHERE id = ?", ("Denied", reason_text, self.suggestion_id))
await db.commit()
@@ -72,7 +77,7 @@ class DenyModal(discord.ui.Modal, title="Reason for denying suggestion"):
)
await orig_msg.edit(embed=updated_embed, view=disabled_view)
except Exception as e:
print(f"Failed to edit admin message: {e}")
logger.error(f"Failed to edit admin message: {e}")
# Send DM to user
try:
@@ -82,7 +87,7 @@ class DenyModal(discord.ui.Modal, title="Reason for denying suggestion"):
dm_note += f"\n**Reason:** {reason_text}"
await user.send(dm_note)
except Exception as e:
print(f"Failed to DM user: {e}")
logger.error(f"Failed to DM user: {e}")
# Send message in original channel
channel = self.bot.get_channel(self.channel_id)
@@ -93,14 +98,14 @@ class DenyModal(discord.ui.Modal, title="Reason for denying suggestion"):
msg += f"\n**Reason:** {reason_text}"
await channel.send(msg)
except Exception as e:
print(f"Failed to send message in channel: {e}")
logger.error(f"Failed to send message in channel: {e}")
except Exception as e:
error_msg = f"❌ Error denying suggestion: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error denying suggestion: {e}", exc_info=True)
try:
await interaction.followup.send(error_msg[:2000], ephemeral=True)
except:
except Exception:
pass
@@ -152,7 +157,7 @@ class SuggestionButtons(discord.ui.View):
# Defer immediately to prevent timeout
await interaction.response.defer(ephemeral=True)
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "suggestions.db")
db_path = Path(__file__).parent.parent / "data" / "suggestions.db"
async with aiosqlite.connect(db_path) as db:
await db.execute("UPDATE suggestions SET status = ? WHERE id = ?", ("Approved", self.suggestion_id))
await db.commit()
@@ -189,14 +194,14 @@ class SuggestionButtons(discord.ui.View):
)
await orig_msg.edit(embed=updated_embed, view=complete_view)
except Exception as e:
print(f"Failed to edit admin message: {e}")
logger.error(f"Failed to edit admin message: {e}")
# Send DM to user
try:
user = await self.bot.fetch_user(self.user_id)
await user.send(f"✅ Your suggestion (ID: {self.suggestion_id}) — `{self.suggestion_text}` has been **approved!**")
except Exception as e:
print(f"Failed to DM user: {e}")
logger.error(f"Failed to DM user: {e}")
# Send message in original channel
channel = self.bot.get_channel(self.channel_id)
@@ -204,17 +209,17 @@ class SuggestionButtons(discord.ui.View):
try:
await channel.send(f"✅ Suggestion **#{self.suggestion_id}** (`{self.suggestion_text}`) has been **approved!**")
except Exception as e:
print(f"Failed to send message in channel: {e}")
logger.error(f"Failed to send message in channel: {e}")
except Exception as e:
error_msg = f"❌ Error approving suggestion: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error approving suggestion: {e}", exc_info=True)
try:
if not interaction.response.is_done():
await interaction.response.send_message(error_msg[:2000], ephemeral=True)
else:
await interaction.followup.send(error_msg[:2000], ephemeral=True)
except:
except Exception:
pass
async def deny(self, interaction: discord.Interaction):
@@ -239,7 +244,7 @@ class SuggestionButtons(discord.ui.View):
try:
orig_msg = await admin_channel.fetch_message(self.admin_message_id)
original_embed = orig_msg.embeds[0] if orig_msg.embeds else None
except:
except Exception:
original_embed = None
else:
original_embed = None
@@ -256,13 +261,13 @@ class SuggestionButtons(discord.ui.View):
await interaction.response.send_modal(modal)
except Exception as e:
error_msg = f"❌ Error opening deny modal: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error opening deny modal: {e}", exc_info=True)
try:
if not interaction.response.is_done():
await interaction.response.send_message(error_msg[:2000], ephemeral=True)
else:
await interaction.followup.send(error_msg[:2000], ephemeral=True)
except:
except Exception:
pass
async def complete(self, interaction: discord.Interaction):
@@ -284,7 +289,7 @@ class SuggestionButtons(discord.ui.View):
# Defer immediately to prevent timeout
await interaction.response.defer(ephemeral=True)
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "suggestions.db")
db_path = Path(__file__).parent.parent / "data" / "suggestions.db"
async with aiosqlite.connect(db_path) as db:
# Check if it's approved
async with db.execute("SELECT status FROM suggestions WHERE id = ?", (self.suggestion_id,)) as cursor:
@@ -333,14 +338,14 @@ class SuggestionButtons(discord.ui.View):
)
await orig_msg.edit(embed=updated_embed, view=disabled_view)
except Exception as e:
print(f"Failed to edit admin message: {e}")
logger.error(f"Failed to edit admin message: {e}")
# Send DM to user
try:
user = await self.bot.fetch_user(self.user_id)
await user.send(f"🎉 Your suggestion (ID: {self.suggestion_id}) — `{self.suggestion_text}` has been **implemented!**")
except Exception as e:
print(f"Failed to DM user: {e}")
logger.error(f"Failed to DM user: {e}")
# Send message in original channel
channel = self.bot.get_channel(self.channel_id)
@@ -348,17 +353,17 @@ class SuggestionButtons(discord.ui.View):
try:
await channel.send(f"🎉 Suggestion **#{self.suggestion_id}** (`{self.suggestion_text}`) has been marked as **completed!**")
except Exception as e:
print(f"Failed to send message in channel: {e}")
logger.error(f"Failed to send message in channel: {e}")
except Exception as e:
error_msg = f"❌ Error completing suggestion: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error completing suggestion: {e}", exc_info=True)
try:
if not interaction.response.is_done():
await interaction.response.send_message(error_msg[:2000], ephemeral=True)
else:
await interaction.followup.send(error_msg[:2000], ephemeral=True)
except:
except Exception:
pass
@@ -404,7 +409,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
def __init__(self, bot):
self.bot = bot
self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "suggestions.db")
self.db_path = Path(__file__).parent.parent / "data" / "suggestions.db"
self.db = None
async def cog_load(self):
@@ -438,7 +443,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
show_complete=(status == "Approved")
)
self.bot.add_view(view, message_id=admin_msg_id)
print(f"Re-registered view for suggestion #{sid} (message {admin_msg_id}, status: {status})")
logger.info(f"Re-registered view for suggestion #{sid} (message {admin_msg_id}, status: {status})")
async def cog_unload(self):
if self.db:
@@ -466,7 +471,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
embed = discord.Embed(
title=f"New Suggestion (ID: {suggestion_id})",
description=idea,
color=discord.Color.blurple(),
color=get_embed_color(interaction.user.id),
timestamp=datetime.utcnow()
)
embed.add_field(name="Suggested by", value=f"{interaction.user} ({interaction.user.id})")
@@ -488,15 +493,14 @@ class Suggestion(commands.GroupCog, name="suggest"):
admin_message_id=sent.id
)
self.bot.add_view(persistent_view, message_id=sent.id)
print(f"Registered persistent view for suggestion #{suggestion_id} (message {sent.id})")
logger.info(f"Registered persistent view for suggestion #{suggestion_id} (message {sent.id})")
except Exception as e:
print(f"Failed to send message to admin channel: {e}")
traceback.print_exc()
logger.error(f"Failed to send message to admin channel: {e}", exc_info=True)
except Exception as e:
error_msg = f"❌ An error occurred: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error in suggest command: {e}", exc_info=True)
await interaction.followup.send(error_msg[:2000])
@app_commands.command(name="view", description="View full details of a suggestion")
@@ -529,7 +533,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
try:
user = await self.bot.fetch_user(user_id)
embed.add_field(name="Suggested by", value=f"{user.mention} ({user})", inline=True)
except:
except Exception:
embed.add_field(name="Suggested by", value=f"<@{user_id}>", inline=True)
embed.add_field(name="Status", value=status, inline=True)
@@ -542,7 +546,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
except Exception as e:
error_msg = f"❌ An error occurred: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error in viewsuggestion command: {e}", exc_info=True)
await interaction.followup.send(error_msg[:2000])
@app_commands.command(name="complete", description="Mark an approved suggestion as completed (Admin only)")
@@ -612,12 +616,12 @@ class Suggestion(commands.GroupCog, name="suggest"):
)
await orig_msg.edit(embed=updated_embed, view=disabled_view)
except Exception as e:
print(f"Failed to edit admin message: {e}")
logger.error(f"Failed to edit admin message: {e}")
try:
user = await self.bot.fetch_user(user_id)
await user.send(f"🎉 Your suggestion (ID: {suggestion_id}) — `{suggestion_text}` has been **implemented!**")
except:
except Exception:
pass
channel = self.bot.get_channel(channel_id)
@@ -626,7 +630,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
except Exception as e:
error_msg = f"❌ An error occurred: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error in completesuggestion command: {e}", exc_info=True)
await interaction.followup.send(error_msg[:2000])
@app_commands.command(name="list", description="List suggestions with optional status filter")
@@ -674,7 +678,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
except Exception as e:
error_msg = f"❌ An error occurred: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error in listsuggestions command: {e}", exc_info=True)
await interaction.followup.send(error_msg[:2000])
@app_commands.command(name="todo", description="View your approved suggestions to-do list")
@@ -728,7 +732,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
user_approved.append((sid, uid, suggestion_text))
break
except Exception as e:
print(f"Failed to fetch message {admin_msg_id}: {e}")
logger.error(f"Failed to fetch message {admin_msg_id}: {e}")
continue
if not user_approved:
@@ -764,7 +768,7 @@ class Suggestion(commands.GroupCog, name="suggest"):
except Exception as e:
error_msg = f"❌ An error occurred: {str(e)}\n```{traceback.format_exc()}```"
print(error_msg)
logger.error(f"Error in todolist command: {e}", exc_info=True)
await interaction.followup.send(error_msg[:2000])

1
utils/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Shared utilities for the Lacie Discord bot."""

33
utils/constants.py Normal file
View File

@@ -0,0 +1,33 @@
"""
Centralized constants for the Lacie Discord bot.
All hardcoded IDs (user IDs, guild IDs, channel IDs, role IDs)
are collected here so they can be managed in one place.
"""
# --- Owner / Admin ---
LILAC_ID = 252130669919076352
# --- Guild ---
GUILD_ID = 876772600704020530
# --- Channels ---
WELCOME_CHANNEL_ID = 876772600704020533
FALLBACK_CHANNEL_ID = 876772600704020533
LOG_CHANNEL_ID = 1440055015711703242
ADMIN_CHANNEL_ID = 1470441786810826884
APPROVAL_CHANNEL_ID = 1424145004976275617
BACKUP_CHANNEL_ID = 946421558778417172
NOTIFICATION_CHANNEL_ID = 1424145004976275617
COMMIT_CHANNEL_IDS = [876777562599194644, 1437941632849940563, 1470441786810826884]
# --- Roles ---
BIRTHDAY_ROLE_ID = 1113751318918602762
BOT_TRAP_ROLE_ID = 1439354601672282335
ADMIN_ROLE_ID = 1470439484549234866
# --- Emojis ---
SALT_EMOJI_ID = 1074583707459010560
# --- Thresholds ---
NEW_MEMBER_THRESHOLD_DAYS = 7

44
utils/logger.py Normal file
View File

@@ -0,0 +1,44 @@
"""
Centralized logging configuration for the Lacie Discord bot.
Provides a configured logger factory so all modules use consistent
logging instead of bare print() statements.
"""
import logging
import sys
def get_logger(name: str) -> logging.Logger:
"""Get a configured logger for the given module name.
Args:
name: The module name, typically __name__ from the calling module.
Returns:
A configured logging.Logger instance.
"""
logger = logging.getLogger(name)
return logger
def setup_logging(level: int = logging.INFO) -> None:
"""Configure the root logger with a console handler.
Call this once at bot startup (in bot.py) before any other logging.
Args:
level: The logging level to use. Defaults to INFO.
"""
root = logging.getLogger()
root.setLevel(level)
if not root.handlers:
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(level)
formatter = logging.Formatter(
"[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
handler.setFormatter(formatter)
root.addHandler(handler)

1
wordle/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Wordle game implementation for Discord."""

25
wordle/wordle.py
View File

@@ -1,16 +1,22 @@
"""Wordle game cog implementing the daily word-guessing game for Discord."""
import discord
from discord import app_commands
from discord.ext import commands
import aiosqlite
import aiohttp
from embed.embed_color import get_embed_color
import os
import datetime
from pathlib import Path
from utils.logger import get_logger
logger = get_logger(__name__)
# Configuration constants
WORD_LIST_URL = "https://raw.githubusercontent.com/tabatkins/wordle-list/main/words"
WORDLE_DIR = "wordle"
WORD_LIST_PATH = os.path.join(WORDLE_DIR, "words.txt")
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "wordle.db")
DB_PATH = Path(__file__).parent.parent / "data" / "wordle.db"
# Emoji squares for displaying guess results
SQUARES = {"green": "🟩", "yellow": "🟨", "gray": "⬛"}
@@ -22,9 +28,10 @@ class Wordle(commands.Cog):
self.bot = bot
# Create wordle directory if it doesn't exist
os.makedirs(WORDLE_DIR, exist_ok=True)
# Initialize database and word list asynchronously
self.bot.loop.create_task(self._init_db())
self.bot.loop.create_task(self._ensure_wordlist())
async def cog_load(self):
await self._init_db()
await self._ensure_wordlist()
async def _init_db(self):
async with aiosqlite.connect(DB_PATH) as db:
@@ -60,7 +67,7 @@ class Wordle(commands.Cog):
with open(WORD_LIST_PATH, "r", encoding="utf-8") as f:
self.words = [w.strip() for w in f.read().splitlines() if len(w.strip()) == 5]
print(f"[Wordle] Loaded {len(self.words)} words.")
logger.info(f"Loaded {len(self.words)} words.")
def get_daily_word(self):
"""
@@ -234,7 +241,7 @@ class Wordle(commands.Cog):
embed = discord.Embed(
title=f"{interaction.user.display_name}'s Wordle {datetime.date.today()} Result",
description="\n".join(["".join(SQUARES[c] for c in self.compare_guess(g, target)) for g in guesses]),
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user)
color=get_embed_color(interaction.user.id)
)
await interaction.channel.send(embed=embed)
return
@@ -249,7 +256,7 @@ class Wordle(commands.Cog):
embed = discord.Embed(
title=f"{interaction.user.display_name}'s Wordle {datetime.date.today()} Result",
description="\n".join(["".join(SQUARES[c] for c in self.compare_guess(g, target)) for g in guesses]),
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user)
color=get_embed_color(interaction.user.id)
)
await interaction.channel.send(embed=embed)
return
@@ -342,7 +349,7 @@ class Wordle(commands.Cog):
# Build and send stats embed
embed = discord.Embed(
title=f"{interaction.user.display_name}'s Wordle Stats",
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user)
color=get_embed_color(interaction.user.id)
)
embed.add_field(name="Games Played", value=str(played))
embed.add_field(name="Wins", value=str(wins))
@@ -374,7 +381,7 @@ class Wordle(commands.Cog):
# Build and send server stats embed
embed = discord.Embed(
title="Server Wordle Stats",
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user)
color=get_embed_color(interaction.user.id)
)
embed.add_field(name="Total Games Played", value=str(total_played))
embed.add_field(name="Total Wins", value=str(total_wins))

1
xp/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""XP and leveling system with role rewards and leaderboards."""

72
xp/backup_xp.py
View File

@@ -1,12 +1,17 @@
"""XP database backup cog with scheduled daily backups to Discord."""
import discord
from discord import app_commands
from discord.ext import commands, tasks
import os
import shutil
from pathlib import Path
from datetime import datetime, timedelta
import pytz
from moderation.loader import ModerationBase
from .groups import xp_admin_group
from utils.logger import get_logger
logger = get_logger(__name__)
BACKUP_CHANNEL_ID = 946421558778417172
NOTIFICATION_CHANNEL_ID = 1424145004976275617
@@ -18,11 +23,10 @@ BACKUP_HOUR = 10 # 10 AM EST
class BackupXP(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.base_dir = os.path.dirname(os.path.abspath(__file__))
self.db_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data")
self.backup_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "backups", "xp")
self.last_backup_file = os.path.join(self.backup_dir, "last_backup.txt")
self.last_auto_backup_file = os.path.join(self.backup_dir, "last_auto_backup.txt")
self.db_dir = Path(__file__).parent.parent / "data"
self.backup_dir = Path(__file__).parent.parent / "data" / "backups" / "xp"
self.last_backup_file = self.backup_dir / "last_backup.txt"
self.last_auto_backup_file = self.backup_dir / "last_auto_backup.txt"
os.makedirs(self.backup_dir, exist_ok=True)
# Register the manual backup command onto the shared xpadmin group
xp_admin_group.add_command(app_commands.command(name="backup", description="Backup both lifetime and annual XP databases")(self.backup_xp))
@@ -54,33 +58,33 @@ class BackupXP(commands.Cog):
# Get current time in EST
now_est = datetime.now(EST)
print(f"[Backup] Time check: {now_est.strftime('%Y-%m-%d %I:%M %p %Z')} (Hour: {now_est.hour}, Minute: {now_est.minute})")
logger.debug(f"Time check: {now_est.strftime('%Y-%m-%d %I:%M %p %Z')} (Hour: {now_est.hour}, Minute: {now_est.minute})")
# Check if it's between 10:00 AM and 10:15 AM EST
if now_est.hour == BACKUP_HOUR and now_est.minute < 15:
print(f"[Backup] Inside backup window - Checking for backup")
logger.debug("Inside backup window - Checking for backup")
await self.check_last_backup()
else:
print(f"[Backup] Outside backup window (need hour={BACKUP_HOUR} and minute<15)")
logger.debug(f"Outside backup window (need hour={BACKUP_HOUR} and minute<15)")
@auto_backup_task.before_loop
async def before_auto_backup(self):
"""Wait until the bot is ready before starting the loop."""
await self.bot.wait_until_ready()
print(f"[Backup] Auto backup task started. Will check every 15 minutes for 10 AM EST backup window.")
logger.info("Auto backup task started. Will check every 15 minutes for 10 AM EST backup window.")
now_est = datetime.now(EST)
print(f"[Backup] Current time: {now_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
logger.info(f"Current time: {now_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
async def check_last_backup(self):
"""Check if an auto backup has been done today at 10 AM EST"""
now = datetime.now()
now_est = datetime.now(EST)
print(f"[Backup] Checking auto backup status at {now_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
logger.info(f"Checking auto backup status at {now_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
if not os.path.exists(self.last_auto_backup_file):
# No previous auto backup — make initial backup
print("[Backup] No last_auto_backup.txt found, creating initial auto backup")
logger.info("No last_auto_backup.txt found, creating initial auto backup")
await self.create_backup(log_channel=True, reason="Auto daily backup (10 AM EST)", is_auto=True)
await self.cleanup_old_backups()
return
@@ -93,32 +97,28 @@ class BackupXP(commands.Cog):
# Check if we've already done an auto backup today
last_time_est = last_time.astimezone(EST)
print(f"[Backup] Last auto backup: {last_time_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
print(f"[Backup] Last auto backup date: {last_time_est.date()}, Today: {now_est.date()}")
logger.info(f"Last auto backup: {last_time_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
logger.debug(f"Last auto backup date: {last_time_est.date()}, Today: {now_est.date()}")
if last_time_est.date() != now_est.date():
# Haven't done auto backup today yet, so do it now
print("[Backup] Starting daily auto backup...")
logger.info("Starting daily auto backup...")
await self.create_backup(log_channel=True, reason="Auto daily backup (10 AM EST)", is_auto=True)
await self.cleanup_old_backups()
else:
print("[Backup] Already auto-backed up today, skipping")
logger.info("Already auto-backed up today, skipping")
async def create_backup(self, log_channel=False, reason=None, is_auto=False):
"""Handles the actual backup logic"""
lifetime_db = os.path.join(self.db_dir, "lifetime.db")
annual_db = os.path.join(self.db_dir, "annual.db")
print(f"[Backup] Looking for databases:")
print(f"[Backup] Lifetime: {lifetime_db}")
print(f"[Backup] Annual: {annual_db}")
print(f"[Backup] Lifetime exists: {os.path.exists(lifetime_db)}")
print(f"[Backup] Annual exists: {os.path.exists(annual_db)}")
logger.debug(f"Looking for databases: lifetime={lifetime_db} (exists={os.path.exists(lifetime_db)}), annual={annual_db} (exists={os.path.exists(annual_db)})")
missing = [db for db in [lifetime_db, annual_db] if not os.path.exists(db)]
if missing:
error_msg = f"❌ Missing database files: {', '.join(os.path.basename(m) for m in missing)}"
print(f"[Backup] {error_msg}")
logger.error(error_msg)
return False, error_msg
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
@@ -130,9 +130,7 @@ class BackupXP(commands.Cog):
shutil.copy2(lifetime_db, lifetime_backup)
shutil.copy2(annual_db, annual_backup)
print(f"[Backup] Files copied successfully")
print(f"[Backup] {os.path.basename(lifetime_backup)}")
print(f"[Backup] {os.path.basename(annual_backup)}")
logger.info(f"Files copied successfully: {os.path.basename(lifetime_backup)}, {os.path.basename(annual_backup)}")
# Record last backup time
with open(self.last_backup_file, "w") as f:
@@ -148,12 +146,12 @@ class BackupXP(commands.Cog):
annual_size = os.path.getsize(annual_db) / (1024 * 1024)
total_size = lifetime_size + annual_size
print(f"[Backup] Backup completed successfully")
logger.info("Backup completed successfully")
return True, f"✅ Backup complete! (`{lifetime_size:.2f}` MB lifetime, `{annual_size:.2f}` MB annual)"
except Exception as e:
error_msg = f"❌ Backup failed: `{e}`"
print(f"[Backup] {error_msg}")
logger.error(error_msg, exc_info=True)
return False, error_msg
async def cleanup_old_backups(self):
@@ -180,13 +178,13 @@ class BackupXP(commands.Cog):
if now - file_time > MAX_BACKUP_AGE:
os.remove(filepath)
deleted_count += 1
print(f"[Backup] Deleted old backup: {filename}")
logger.info(f"Deleted old backup: {filename}")
if deleted_count > 0:
print(f"[Backup] Cleaned up {deleted_count} old backup file(s)")
logger.info(f"Cleaned up {deleted_count} old backup file(s)")
except Exception as e:
print(f"[Backup] Error during cleanup: {e}")
logger.error(f"Error during cleanup: {e}", exc_info=True)
async def setup(bot: commands.Bot):
await bot.add_cog(BackupXP(bot))

8
xp/calculate.py
View File

@@ -1,3 +1,4 @@
"""XP calculation helpers: awarding XP, handling cooldowns, leveling up."""
import discord
from discord import app_commands
from discord.ext import commands, tasks
@@ -5,6 +6,7 @@ from .database import get_db
from .utils import xp_for_level, can_get_xp, get_multiplier, load_config
from .groups import xp_group
import time
from embed.embed_color import get_embed_color
class CalculateCommand(commands.Cog):
def __init__(self, bot):
@@ -116,11 +118,7 @@ class CalculateCommand(commands.Cog):
{bar} ({progress:.2f}%)"""
# Safe color handling
color_cog = self.bot.get_cog("EmbedColor")
color = color_cog.get_user_color(interaction.user) if color_cog else discord.Color.blurple()
embed = discord.Embed(description=response, color=color)
embed = discord.Embed(description=response, color=get_embed_color(interaction.user.id))
embed.set_author(name=user.display_name, icon_url=user.display_avatar.url)
await interaction.followup.send(embed=embed)

7
xp/database.py
View File

@@ -1,9 +1,10 @@
"""XP database connection helpers for multiple leaderboard time periods."""
import sqlite3
import os
from pathlib import Path
DB_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data")
DB_DIR = Path(__file__).parent.parent / "data"
os.makedirs(DB_DIR, exist_ok=True)
DB_DIR.mkdir(parents=True, exist_ok=True)
def get_db(db_type="lifetime"):
if isinstance(db_type, bool):

7
xp/exclude_channels.py
View File

@@ -1,17 +1,18 @@
"""Utilities for managing channels excluded from XP gain."""
import discord
from discord.ext import commands
import json
import os
from pathlib import Path
from dotenv import load_dotenv
from moderation.loader import ModerationBase
load_dotenv()
EXCLUDED_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "excluded_channels.json")
EXCLUDED_FILE = Path(__file__).parent.parent / "data" / "excluded_channels.json"
def load_excluded_channels():
"""Load excluded channel IDs from JSON file."""
if not os.path.exists(EXCLUDED_FILE):
if not EXCLUDED_FILE.exists():
return []
with open(EXCLUDED_FILE, "r") as f:
return json.load(f)

1
xp/groups.py
View File

@@ -1,3 +1,4 @@
"""Shared slash command group definitions for the XP system."""
from discord import app_commands
# Public XP commands: /xp rank, /xp top, /xp calculate, /xp sync

78
xp/import_export.py
View File

@@ -1,3 +1,4 @@
"""XP import/export commands for backing up and restoring user XP data."""
import json
import asyncio
from io import BytesIO
@@ -8,6 +9,9 @@ from xp.utils import xp_for_level
from xp.database import get_db
from moderation.loader import ModerationBase
from .groups import xp_admin_group
from utils.logger import get_logger
logger = get_logger(__name__)
class XPImportExport(commands.Cog):
@@ -37,9 +41,9 @@ class XPImportExport(commands.Cog):
}
return users
except Exception as e:
print(f"Error in _db_work: {e}")
logger.error(f"Error in _db_work: {e}", exc_info=True)
raise
# Run everything in thread pool
users = await asyncio.to_thread(_db_work)
return {"users": users}
@@ -56,39 +60,31 @@ class XPImportExport(commands.Cog):
# Defer immediately
await interaction.response.defer()
print(f"Export started for {xp_type.value}")
logger.info(f"Export started for {xp_type.value}")
lifetime = xp_type.value == "lifetime"
# Get data asynchronously
print("Fetching data from database...")
logger.debug("Fetching data from database...")
data = await self._export_data(lifetime)
print(f"Data fetched: {len(data['users'])} users")
# Create file with pretty-printed JSON
print("Encoding JSON...")
logger.info(f"Data fetched: {len(data['users'])} users")
json_str = json.dumps(data, indent=2)
json_bytes = json_str.encode("utf-8")
print(f"JSON size: {len(json_bytes)} bytes")
print("Creating Discord file...")
logger.debug(f"JSON size: {len(json_bytes)} bytes")
file = discord.File(fp=BytesIO(json_bytes), filename=f"{xp_type.value}_xp_export.json")
# Public response
print("Sending response...")
await interaction.followup.send(
f"✅ Exported `{xp_type.value}` XP data ({len(data['users'])} users).",
file=file
)
print("Export complete!")
logger.info("Export complete!")
except Exception as e:
print(f"Error in export_xp: {e}")
import traceback
traceback.print_exc()
logger.error(f"Error in export_xp: {e}", exc_info=True)
try:
await interaction.followup.send(f"❌ An error occurred during export: {str(e)}")
except:
except Exception:
pass
# ==================== IMPORT ====================
@@ -130,9 +126,9 @@ class XPImportExport(commands.Cog):
conn.close()
return len(insert_data)
except Exception as e:
print(f"Error in _db_work: {e}")
logger.error(f"Error in _db_work: {e}", exc_info=True)
raise
# Run database work in thread pool
count = await asyncio.to_thread(_db_work)
return count
@@ -154,50 +150,44 @@ class XPImportExport(commands.Cog):
# Defer immediately
await interaction.response.defer()
print(f"Import started for {xp_type.value}")
logger.info(f"Import started for {xp_type.value}")
lifetime = xp_type.value == "lifetime"
if not attachment.filename.endswith(".json"):
await interaction.followup.send("❌ Please upload a valid `.json` file.")
return
# Download and parse JSON file
print("Downloading attachment...")
logger.debug("Downloading attachment...")
file_bytes = await attachment.read()
print(f"Downloaded {len(file_bytes)} bytes")
logger.debug(f"Downloaded {len(file_bytes)} bytes")
try:
print("Parsing JSON...")
data = json.loads(file_bytes.decode("utf-8"))
except json.JSONDecodeError as e:
await interaction.followup.send(f"❌ Invalid JSON file format: {str(e)}")
return
users_data = data.get("users", {})
if not users_data:
await interaction.followup.send("❌ No user data found in the JSON file.")
return
print(f"Found {len(users_data)} users to import")
# Import data asynchronously
print("Importing data...")
logger.info(f"Found {len(users_data)} users to import")
count = await self._import_data(users_data, lifetime)
print(f"Import complete: {count} users")
logger.info(f"Import complete: {count} users")
await interaction.followup.send(
f"✅ Imported `{xp_type.value}` XP data from `{attachment.filename}` — {count} users imported (existing data overwritten)."
)
except Exception as e:
print(f"Error in import_xp: {e}")
import traceback
traceback.print_exc()
logger.error(f"Error in import_xp: {e}", exc_info=True)
try:
await interaction.followup.send(f"❌ An error occurred during import: {str(e)}")
except:
except Exception:
pass
def cog_unload(self):

4
xp/leaderboard.py
View File

@@ -1,3 +1,4 @@
"""XP leaderboard commands showing top users by various time periods."""
import discord
from discord.ext import commands
from discord import app_commands
@@ -5,6 +6,7 @@ from discord.ui import View, Button
from .database import get_db
from .groups import xp_group
import math
from embed.embed_color import get_embed_color
class LeaderboardView(View):
def __init__(self, embed_pages):
@@ -110,7 +112,7 @@ class Leaderboard(commands.Cog):
embed = discord.Embed(
title=f"{board_display_name} Leaderboard (Page {page_num + 1}/{total_pages})",
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user)
color=get_embed_color(interaction.user.id)
)
if page_num == 0 and page_rows:

13
xp/rank.py
View File

@@ -1,12 +1,16 @@
"""XP rank card generation and rank display commands."""
import discord
import time
import math
import traceback
from discord.ext import commands
from discord import app_commands
from .database import get_db
from .utils import xp_for_level, get_multiplier, MULTIPLIERS, COOLDOWN
from .groups import xp_group
from embed.embed_color import get_embed_color
from utils.logger import get_logger
logger = get_logger(__name__)
class Rank(commands.Cog):
def __init__(self, bot):
@@ -109,7 +113,7 @@ class Rank(commands.Cog):
f"➡️ **Next level:** `{next_level_xp:,}` ({needed:,} more)\n"
f"🕒 **Cooldown:** {cooldown}"
),
color=self.bot.get_cog("EmbedColor").get_user_color(interaction.user)
color=get_embed_color(interaction.user.id)
)
if lifetime and multipliers_text:
@@ -128,8 +132,7 @@ class Rank(commands.Cog):
await interaction.followup.send(embed=embed)
except Exception as e:
print(f"Rank command error: {e}")
traceback.print_exc()
logger.exception(f"Rank command error: {e}")
try:
if interaction.response.is_done():
await interaction.followup.send(
@@ -139,7 +142,7 @@ class Rank(commands.Cog):
await interaction.response.send_message(
"An error occurred while fetching rank data.", ephemeral=True
)
except:
except Exception:
pass
def cog_unload(self):

10
xp/reset_task.py
View File

@@ -1,7 +1,11 @@
"""Scheduled XP reset tasks for daily, weekly, and monthly leaderboards."""
from discord.ext import tasks, commands
from datetime import datetime, timezone
from .database import reset_leaderboard, get_last_reset
import calendar
from utils.logger import get_logger
logger = get_logger(__name__)
class ResetTask(commands.Cog):
def __init__(self, bot):
@@ -23,7 +27,7 @@ class ResetTask(commands.Cog):
last_reset_date = datetime.fromtimestamp(last_reset, timezone.utc).date() if last_reset > 0 else None
if last_reset_date != now.date():
reset_leaderboard("daily")
print(f"[XP System] Daily leaderboard reset at {now}")
logger.info(f"Daily leaderboard reset at {now}")
# Weekly reset - Monday at midnight UTC
if self.should_reset_weekly(now):
@@ -32,7 +36,7 @@ class ResetTask(commands.Cog):
current_week = now.isocalendar()[1]
if last_reset_week != current_week:
reset_leaderboard("weekly")
print(f"[XP System] Weekly leaderboard reset at {now}")
logger.info(f"Weekly leaderboard reset at {now}")
# Monthly reset - First day of month at midnight UTC
if self.should_reset_monthly(now):
@@ -40,7 +44,7 @@ class ResetTask(commands.Cog):
last_reset_month = datetime.fromtimestamp(last_reset, timezone.utc).month if last_reset > 0 else None
if last_reset_month != now.month:
reset_leaderboard("monthly")
print(f"[XP System] Monthly leaderboard reset at {now}")
logger.info(f"Monthly leaderboard reset at {now}")
def should_reset_daily(self, now):
"""Check if it's time for daily reset (00:00 UTC)."""

7
xp/restoredb.py
View File

@@ -1,18 +1,19 @@
"""XP database restore commands for recovering from backups."""
import discord
from discord import app_commands
from discord.ext import commands
from discord.ui import View, Button
import os
import shutil
from pathlib import Path
from moderation.loader import ModerationBase
from .groups import xp_admin_group
class RestoreXP(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.base_dir = os.path.dirname(os.path.abspath(__file__))
self.db_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data")
self.backup_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "backups", "xp")
self.db_dir = Path(__file__).parent.parent / "data"
self.backup_dir = Path(__file__).parent.parent / "data" / "backups" / "xp"
# Register onto the shared xpadmin group
cmd = app_commands.command(name="restore", description="Restore a lifetime or annual XP database from backup")(self.restorebackup)
# Attach the autocompleters to the command object before adding

42
xp/sync.py
View File

@@ -1,13 +1,17 @@
"""XP role sync commands that assign Discord roles based on XP level."""
import discord
from discord.ext import commands
from discord import app_commands
from moderation.loader import ModerationBase, ADMIN_ROLE_IDS, lilac_id
from moderation.loader import ModerationBase, ADMIN_ROLE_IDS
from utils.constants import LILAC_ID
from .database import get_db
from .utils import load_config, xp_for_level
from .groups import xp_group
from discord.utils import get
import traceback
import asyncio
from utils.logger import get_logger
logger = get_logger(__name__)
class XPSync(commands.Cog):
"""Sync XP role rewards for users."""
@@ -19,7 +23,7 @@ class XPSync(commands.Cog):
def check_is_admin(self, user: discord.Member) -> bool:
"""Check if user has admin permissions."""
is_lilac = user.id == lilac_id
is_lilac = user.id == LILAC_ID
has_admin_role = any(role.id in ADMIN_ROLE_IDS for role in user.roles)
return has_admin_role or is_lilac
@@ -58,9 +62,9 @@ class XPSync(commands.Cog):
# Small delay between role additions to avoid rate limits
await asyncio.sleep(0.5)
except discord.Forbidden:
print(f"Cannot assign {role.name} to {member} - missing permissions")
logger.warning(f"Cannot assign {role.name} to {member} - missing permissions")
except discord.HTTPException as e:
print(f"HTTP error assigning {role.name} to {member}: {e}")
logger.error(f"HTTP error assigning {role.name} to {member}: {e}")
return (level, roles_added)
@@ -86,27 +90,27 @@ class XPSync(commands.Cog):
# NOW defer the interaction after permission checks pass
await interaction.response.defer(ephemeral=False)
print(f"[SYNC] Deferred interaction for user {interaction.user.id}")
logger.debug(f"Deferred interaction for user {interaction.user.id}")
# Fetch target member
try:
target_member = interaction.guild.get_member(target_user.id)
if not target_member:
print(f"[SYNC] Member not in cache, fetching from API")
logger.debug("Member not in cache, fetching from API")
target_member = await interaction.guild.fetch_member(target_user.id)
except discord.NotFound:
return await interaction.followup.send(
f"{target_user.mention} is not in this server.",
f"{target_user.mention} is not in this server.",
ephemeral=True
)
except discord.HTTPException as e:
print(f"[SYNC] HTTP error fetching member: {e}")
logger.error(f"HTTP error fetching member: {e}")
return await interaction.followup.send(
f"Error fetching member: {e}",
f"Error fetching member: {e}",
ephemeral=True
)
print(f"[SYNC] Starting role sync for {target_member.id}")
logger.debug(f"Starting role sync for {target_member.id}")
# Sync roles with timeout protection
try:
@@ -115,13 +119,13 @@ class XPSync(commands.Cog):
timeout=25.0 # 25 seconds to stay under Discord's 30s limit
)
except asyncio.TimeoutError:
print(f"[SYNC] Timeout during role sync for {target_member.id}")
logger.warning(f"Timeout during role sync for {target_member.id}")
return await interaction.followup.send(
"Role sync took too long. Please try again or contact an admin.",
"Role sync took too long. Please try again or contact an admin.",
ephemeral=True
)
print(f"[SYNC] Completed sync for {target_member.id}: Level {level}, Roles added: {roles_added}")
logger.info(f"Completed sync for {target_member.id}: Level {level}, Roles added: {roles_added}")
# Send response
if level == 0:
@@ -139,18 +143,16 @@ class XPSync(commands.Cog):
)
except discord.NotFound:
print(f"[SYNC] Interaction or message not found - may have timed out")
# Can't respond if interaction is gone
logger.warning("Interaction or message not found - may have timed out")
except Exception as e:
print(f"[SYNC] Unexpected error in sync command: {e}")
traceback.print_exc()
logger.error(f"Unexpected error in sync command: {e}", exc_info=True)
try:
await interaction.followup.send(
f"Error syncing roles: {str(e)[:100]}",
f"Error syncing roles: {str(e)[:100]}",
ephemeral=True
)
except Exception as followup_error:
print(f"[SYNC] Could not send error message: {followup_error}")
logger.error(f"Could not send error message: {followup_error}")
def cog_unload(self):
xp_group.remove_command("sync")

1
xp/utils.py
View File

@@ -1,3 +1,4 @@
"""XP system utilities: level math, multipliers, config loading."""
import math
import random
import time