Advanced • Modular • Production-Ready
A production-ready, enterprise-grade Discord bot framework featuring atomic file operations, advanced permission systems, dynamic extension loading, comprehensive monitoring, modular framework cogs, and an integrated extension marketplace.
🌐 Website
Atomic File Operations
- Thread-safe file handling with built-in LRU caching (300s TTL, 1000 entry limit)
- Automatic lock management with cleanup threshold (500 locks)
- Zero data corruption through tempfile-based atomic writes
- Cache invalidation and TTL-based expiration
- Cross-platform compatibility (Windows/Linux)
Advanced Database Management
- Per-guild SQLite databases with automatic connection pooling
- WAL (Write-Ahead Logging) mode for concurrent access
- Automatic backup system on shutdown
- Orphaned connection cleanup
- Global and guild-specific data separation
- Command usage analytics and statistics tracking
Safe Log Rotation
- Automatic size-based rotation (10MB default)
- Configurable backup count (5 backups default)
- Age-based cleanup (30-day retention)
- Dual-mode logging: permanent + current session
- Structured logging with timestamps and levels
Hot-Reload System
- File-watching based auto-reload (30s interval)
- Extension modification timestamps tracked
- Zero downtime during development
- Load time tracking per extension
- Graceful error handling during reload
Intelligent Extension Loading
- Automatic whitespace handling (converts spaces to underscores)
- Extension blacklist support
- Load time tracking and diagnostics
- Conflict detection through Plugin Registry
- Dependency resolution system
Hybrid Command System
- Seamless prefix and slash command support
- Automatic command type indicators in help menu
- Slash command limit protection (100 command cap)
- Graceful degradation to prefix-only mode
- Visual indicators: ⚡ (hybrid), 🔸 (prefix-only), 🔹 (limit reached)
Event Hooks System (cogs/event_hooks.py)
- Internal event system for framework lifecycle events
- Priority-based callback execution
- Asynchronous queue processing:
- 1000 event capacity with backpressure (5s wait before drop)
- Queue delay tracking for performance monitoring
- Worker auto-restart on crash (up to 10 attempts)
- Circuit Breaker Pattern:
- Opens after 5 failures in 5 minutes
- Disables hook for 60 seconds
- Automatic recovery with manual override
- Prevents cascading failures
- Hook timeout protection:
- 10-second timeout per hook execution
- Prevents blocking the entire queue
- Timeout triggers circuit breaker
- Performance monitoring:
- Execution time tracking per hook
- Success/failure counts
- Average execution time calculations
- Queue delay metrics
- Hook management:
- Disable/enable hooks without code changes
- Status indicators (🔴 disabled,
⚠️ circuit open) - Manual circuit breaker reset
- Alert system:
- Queue full notifications
- Circuit breaker triggers
- Worker crash alerts
- Configurable alert channel
- Hook execution history (100 total, 20 per event)
- Built-in events: bot_ready, guild_joined, guild_left, command_executed, command_error
- Custom event emission support
Plugin Registry (cogs/plugin_registry.py)
- Automatic metadata extraction from extensions with error logging
- Dependency validation and enforcement:
- Version comparison support (
>=,>,==,<=,<) - Missing dependency detection
- Version mismatch detection
- Version comparison support (
- Conflict detection and prevention:
- Bidirectional conflict checking
- Automatic conflict alerts
- Circular dependency detection:
- Prevents infinite dependency loops
- Shows full dependency cycle path
- Command and cog enumeration
- Load time tracking per plugin
- Auto-registration on extension load via event hooks
- JSON-based registry persistence with enforcement settings
- Alert system:
- Configurable alert channel
- Validation issue warnings
- Scan error notifications
- Status indicators:
- ✅ Valid plugin
⚠️ Warnings (conflicts, scan errors)- ❌ Missing dependencies
- 🔄 Circular dependencies
- Toggleable enforcement:
- Enable/disable dependency checking
- Enable/disable conflict checking
Framework Diagnostics (cogs/framework_diagnostics.py)
- Real-time system health monitoring with three-tier status (healthy/degraded/critical)
- CPU, memory, threads, open files, and connections tracking
- Command/error rate metrics with automatic calculation
- Event loop lag monitoring (checks every second)
- Uptime and latency monitoring
- Extension load time analysis
- Automatic diagnostics generation every 5 minutes
- Alert system with configurable channel (
!fw_alert_channel) - Automatic alerts for:
- Critical health status (≥10% error rate)
- Degraded health status (≥5% error rate)
- Event loop blocking/lag
- Consecutive write failures (≥3 failures)
- Comprehensive diagnostics saved to JSON files:
framework_diagnostics.json- Full system reportframework_health.json- Real-time health metrics
Slash Command Limiter (cogs/slash_command_limiter.py)
- Intelligent Discord 100-command limit management with automatic conversion
- Thread-safe singleton pattern - Only one instance allowed per bot
- Configurable thresholds:
- Warning threshold (default: 90/100)
- Safe limit (default: 95/100)
- Hard limit (100/100)
- Multi-layered interception system:
tree.add_command()monkey-patching for dynamic command blockingCog._inject()interception for cog-level slash stripping- Prevents commands from ever reaching Discord's API
- Intelligent command type detection:
- ✅ Skips
HybridCommand- Already have prefix support, no conversion needed - ✅ Skips
commands.Command- Already prefix-only - ✅ Skips
app_commands.Group- Command groups cannot be converted - 🔄 Converts
app_commands.Command- Pure slash commands to prefix equivalents
- ✅ Skips
- Advanced parameter conversion system:
- Automatic type detection from function signatures
- Discord object conversion:
discord.Member/discord.User- Converts @mentions and IDs to Member objectsdiscord.Role- Converts role mentions and IDs to Role objectsdiscord.TextChannel/discord.VoiceChannel- Converts #channel mentions to Channel objects
- Primitive type handling:
int- String to integer conversionfloat- String to float conversionbool- Handles "true", "yes", "1", "y", "on" variations
- Intelligent error messages:
- Missing required parameters with usage examples
- Type conversion failures with helpful hints
- Guild context validation (DM protection)
- Production-ready MockInteraction wrapper:
- Emulates full
discord.Interactionobject for seamless compatibility - Handles
interaction.response.send_message(),interaction.followup.send() - Supports embeds, files, views (buttons/dropdowns), and all Discord features
- Proper attribute forwarding:
guild,user,channel,permissions,command - Modal support with graceful degradation message
- Emulates full
- Comprehensive tracking and logging:
- Blocked commands registry with timestamps and caller info
- Converted commands tracking with conversion type metadata
- Warning threshold system with automatic reset
- Debug logging to
botlogs/debug_slashlimiter.log - Caller tracking shows which file/line triggered blocking
- Self-protection mechanisms:
- Limiter's own commands (
slashlimit) exempted from blocking - Async startup to prevent bot hang during initialization
- Non-blocking background tasks for scanning
- Graceful degradation on conversion failures
- Limiter's own commands (
- User-friendly status display (
/slashlimit):- Visual progress bars showing command usage
- Color-coded status (green=safe, yellow=warning, red=critical)
- Lists blocked and converted commands
- Shows system status and verification state
- Integration with help system:
- Command type indicators: ⚡ (hybrid), 🔸 (prefix-only), 🔹 (converted from slash)
- Automatic updates when commands are converted
- Real-time command count tracking
Direct Extension Installation
- Browse official extensions from within Discord
- Search by keywords, categories, or status
- One-click installation with confirmation
- Automatic file writing to
./extensionsdirectory - Post-install instructions with load commands
ZygnalID System
- Unique 16-character bot identifier
- Automatic generation on first use
- Required for extension downloads
- Enables dedicated support and tracking
- Stored securely in
./data/marketplace/ZygnalID.txt
License Agreement
- Mandatory acceptance before marketplace access
- Per-user tracking of acceptance
- Usage restrictions enforcement
- Extension-specific licensing support
- Terms displayed in interactive embed
Dependency Management
!marketplace fixdepscommand- Automatic log parsing for
ModuleNotFoundError - pip-based dependency installation
- Success/failure reporting per package
- Duplicate installation prevention
Advanced Features
- Extension categories: Working, Beta, Broken
- Custom URL support for extensions
- Version tracking and display
- Author attribution
- Extension status indicators
- Pagination for large lists (5 per page)
- Dropdown menus for selection
- Cache system (300s TTL) for API calls
- Rate limit handling with retry logic
Interactive Help System
- Dropdown-based category navigation
- Pagination for large command lists (5 per page)
- Visual command type indicators
- Automatic cog organization
- User-specific interaction validation
- Credits button with framework info
- Dynamic back navigation
- Real-time command count per category
Rich Embeds Everywhere
- Color-coded responses (success=green, error=red, warning=yellow)
- Consistent styling across all commands
- Timestamp inclusion
- Footer information
- Thumbnail support
- Field-based organization
- Progress bars for visual feedback
Auto-Delete Messages
- Configurable delete timers (5-15s)
- Success message cleanup
- Error message cleanup
- Reduces channel clutter
- User-friendly experience
Multi-Tier Permission System
- Bot Owner: Full access (from BOT_OWNER_ID env variable)
- Guild Owner: Server management commands
- Role-Based: Per-command role requirements
- Discord Permissions: Native permission integration
- Hardcoded Restrictions: Owner-only command list
Interaction Security
- User-specific button/dropdown validation
- Interaction author verification
- Ephemeral error messages
- Timeout-based view expiration (180-300s)
- Session-based state management
Data Protection
- Atomic file operations prevent corruption
- Database WAL mode for ACID compliance
- Automatic backups before shutdown
- File locking mechanism
- Cache invalidation on writes
- Secure temporary file handling
Metrics Collection
- Real-time command tracking (LRU cache, 100 command limit)
- Message processing statistics
- Error rate monitoring
- Uptime tracking
- Top command analytics
- Per-command usage counts in database
Health Monitoring
- Error rate calculation
- Status determination (healthy/degraded)
- Latency tracking
- Extension load time analysis
- Database connection monitoring
- Cache performance metrics
Diagnostics Dashboard
- System information (Python version, platform, architecture)
- Bot statistics (guilds, users, channels)
- Extension analysis (count, load times)
- Command registration (prefix + slash)
- Performance metrics (CPU, memory, threads)
- Database health status
Atomic File Testing
- Built-in
!atomictestcommand - Write/read/cache performance benchmarking
- Concurrent operation testing (10 simultaneous writes)
- Data integrity verification
- Cache statistics display
Cache Management
!cachestatscommand for monitoring- File cache size tracking
- Lock count monitoring
- Prefix cache statistics
- Metrics cache info
- Database connection counts
Integrity Checks
!integritycheckcommand- File system validation
- Database connection testing
- Cache system verification
- Extension loading checks
- Shard status monitoring
- Memory usage analysis
System Cleanup
!cleanupcommand- pycache directory removal
- Expired prefix cache cleanup
- File lock cleanup
- Orphaned database connection removal
- Statistics reporting
✅ Atomic File Operations - Thread-safe file handling with LRU caching and retry logic
- Zero data corruption through tempfile-based writes
- 300s TTL, 1000 entry LRU cache
- Automatic lock cleanup (500 lock threshold)
- 3-attempt retry with exponential backoff
- Comprehensive metrics and health monitoring
- Diagnostic commands:
!atomicstats,!atomictest - Cache hit rate tracking and optimization
- Cross-platform compatibility (Windows/Linux)
✅ SQLite Database - Optimized with WAL mode and connection pooling
✅ Safe Log Rotation - Automatic management with size and age limits
✅ Hot-Reload System - File-watching based auto-reload (optional)
✅ Metrics Collection - Real-time command tracking and analytics
✅ Framework Cogs - Modular internal components with event system
✅ Auto-Sharding - Built-in support for large-scale deployments
✅ Hybrid Commands - Both prefix and slash command support
✅ Permission Framework - Multi-tier role-based access control
✅ Command Autocomplete - Dynamic suggestions for slash commands
✅ Cooldown Management - Built-in rate limiting
✅ Auto-Delete Messages - Automatic cleanup of responses
✅ Slash Limit Protection - Automatic prefix-only fallback
✅ Command Type Indicators - Visual markers in help menu
✅ Integrated Browser - Browse extensions from Discord
✅ Search & Filter - Find extensions by keywords or category
✅ One-Click Install - Direct installation to bot
✅ License Agreement - Mandatory acceptance system
✅ ZygnalID Support - Unique bot identification
✅ Dependency Auto-Fix - Automatic missing package installation
✅ Version Tracking - Extension versioning support
✅ Interactive Help - Dropdown navigation with pagination
✅ Rich Embeds - Modern Discord UI with color coding
✅ User Validation - Security-checked interactions
✅ Progress Bars - Visual feedback for operations
✅ Category Organization - Automatic command grouping
✅ Per-Guild Settings - Custom prefixes and configurations
✅ JSON Config System - Centralized configuration
✅ Command Permissions - Granular role requirements
✅ Extension Blacklist - Selective loading control
✅ Framework Cog Control - Enable/disable components
✅ Command Usage Stats - Track popular commands
✅ Error Tracking - Comprehensive error logging
✅ Uptime Monitoring - Real-time bot statistics
✅ Performance Metrics - Load times and query tracking
✅ Health Monitoring - System diagnostics and alerts
✅ Hook History - Event system execution tracking
Python: 3.8+ (Built and tested with 3.12.7)
discord.py: 2.0+ (Built with 2.6.3)
aiofiles==24.1.0
aiohttp==3.12.14
aiosqlite==0.21.0
discord.py==2.6.3
python-dotenv==1.0.0
psutil==5.9.6
rich==14.0.0
git clone https://github.com/TheHolyOneZ/discord-bot-framework.git
cd discord-bot-framework
docker build -t discord-bot-framework .
docker run -d --name mybot discord-bot-framework
git clone https://github.com/TheHolyOneZ/discord-bot-framework.git
cd discord-bot-frameworkpip install -r requirements.txtCreate .env in the root directory:
DISCORD_TOKEN=Bot_Token
BOT_OWNER_ID=Your_DiscordID
# Sharding Configuration
SHARD_COUNT=1
# Can be commented out depending on your needs. If you want to use auto-sharding, leave (SHARD_IDS) commented out.
# SHARD_IDS=0,1Getting Your User ID:
- Enable Developer Mode: User Settings → Advanced → Developer Mode
- Right-click your username → Copy ID
Getting Bot Token:
- Go to Discord Developer Portal
- Create New Application
- Go to Bot section → Reset Token → Copy token
In the Discord Developer Portal:
- Go to your application → Bot section
- Enable these Privileged Gateway Intents:
- ✅ Presence Intent
- ✅ Server Members Intent
- ✅ Message Content Intent
python main.pyYou should see a Rich console panel with bot statistics!
Generate invite URL in Developer Portal:
- OAuth2 → URL Generator
- Select Scopes:
bot,applications.commands - Select Permissions: Administrator (or specific permissions)
- Copy generated URL and open in browser
discord-bot-framework/
│
├── main.py # Core bot logic and built-in commands
├── atomic_file_system.py # Atomic operations and data management
│
├── extensions/ # Your extension modules (auto-loaded)
│ ├── Put_Cogs_Extensions_here.txt
│ ├── example_logger.py # Example extension
│ └── marketplace.py # Extension Marketplace cog
│
├── cogs/ # Framework internal cogs
│ ├── event_hooks.py # Internal event system
│ ├── plugin_registry.py # Plugin metadata & dependency tracking
│ ├── framework_diagnostics.py # Health monitoring
│ └── slash_command_limiter.py # Slash command protection
│
├── data/ # Auto-generated data directory
│ ├── main.db # Global SQLite database
│ ├── main.db-wal # WAL file for main DB
│ ├── main.db-shm # Shared memory for main DB
│ ├── [guild_id]/ # Per-guild databases
│ │ ├── guild.db
│ │ └── guild_backup_*.db
│ ├── plugin_registry.json # Plugin metadata cache
│ ├── framework_diagnostics.json # System diagnostics
│ ├── framework_health.json # Health monitoring data
│ └── marketplace/
│ ├── ZygnalID.txt # Unique bot identifier
│ └── license_accepted.json # License acceptance tracking
│
├── botlogs/ # Log files
│ ├── permanent.log # Persistent log (rotates at 10MB)
│ ├── permanent.log.1-5 # Backup logs
│ └── current_run.log # Current session only
│
├── images/ # Documentation images
│ ├── Terminal-1.png
│ ├── HelpMenu-Example.png
│ └── ... (showcase images)
│
├── config.json # Bot configuration (auto-generated)
├── .env # Environment variables (YOU CREATE THIS)
├── requirements.txt # Python dependencies
├── README.md # This file
├── LICENSE # MIT License
├── SECURITY.md # Security policy
├── CODE_OF_CONDUCT.md # Code of conduct
└── CONTRIBUTING.md # Contribution guidelines
| Command | Description | Cooldown | Hybrid |
|---|---|---|---|
!help / /help |
Interactive help menu with dropdown navigation | 10s | ✅ |
!stats / /stats |
Bot statistics, metrics, and framework info | 10s | ✅ |
!extensions / /extensions |
List loaded user extensions and framework cogs | 10s | ✅ |
!discordbotframework |
Framework information and feature list | 10s | ✅ |
!shardinfo / /shardinfo |
Shard information and distribution | 10s | ✅ |
!setprefix <prefix> |
Set custom prefix for your server (Admin only) | - | ✅ |
!config [cmd] [role] |
Configure command permissions (Owner only) | - | ✅ |
| Command | Description | Cooldown | Hybrid |
|---|---|---|---|
!marketplace / /marketplace |
Main marketplace menu with quick actions | 5s | ✅ |
!marketplace browse |
Browse all available extensions | 10s | ✅ |
!marketplace search <query> |
Search extensions by keywords | 10s | ✅ |
!marketplace install <id> |
Install extension by ID | 30s | ✅ |
!marketplace info <id> |
View detailed extension information | 5s | ✅ |
!marketplace refresh |
Refresh extension cache | 60s | ✅ |
!marketplace fixdeps |
Auto-install missing dependencies (Owner) | 60s | ✅ |
!marketplace myid |
View your ZygnalID (Owner only) | - | ✅ |
| Command | Description | Cooldown | Hybrid |
|---|---|---|---|
!atomicstats / /atomicstats |
Display atomic file system statistics (Owner) | - | Yes |
!atomictest / /atomictest |
Run comprehensive atomic operations test suite (Owner) | - | Yes |
| Command | Description | Cooldown | Hybrid |
|---|---|---|---|
!pr_list / /pr_list |
List all registered plugins with status indicators | 10s | ✅ |
!pr_info <name> / /pr_info |
Detailed information about a specific plugin | 10s | ✅ |
!pr_validate <name> |
Validate plugin dependencies and conflicts (Owner) | - | ✅ |
!pr_enforce <mode> |
Toggle dependency/conflict enforcement (Owner) | - | ✅ |
!pr_alert_channel [channel] |
Set alert channel for registry warnings (Owner) | - | ✅ |
| Command | Description | Cooldown | Hybrid |
|---|---|---|---|
!fw_diagnostics / /fw_diagnostics |
Display framework health and diagnostics (Owner) | - | ✅ |
!fw_alert_channel / /fw_alert_channel |
Set alert channel for framework diagnostics (Owner) | - | ✅ |
!slashlimit / /slashlimit |
Check slash command usage and limits | 10s | ✅ |
| Command | Description | Cooldown | Hybrid |
|---|---|---|---|
!eh_list / /eh_list |
Display registered hooks with metrics (Owner) | - | ✅ |
!eh_history [limit] / /eh_history |
Display hook execution history (Owner) | - | ✅ |
!eh_disable <hook_id> |
Disable a problematic hook (Owner) | - | ✅ |
!eh_enable <hook_id> |
Re-enable a disabled hook (Owner) | - | ✅ |
!eh_reset_circuit <hook_id> |
Reset circuit breaker for hook (Owner) | - | ✅ |
!eh_alert_channel [channel] |
Set alert channel for event hooks (Owner) | - | ✅ |
| Command | Description | Access |
|---|---|---|
!sync / /sync |
Force sync slash commands globally | Bot Owner |
!reload <extension> |
Hot-reload specific extension | Bot Owner |
!load <extension> |
Load extension | Bot Owner |
!unload <extension> |
Unload extension | Bot Owner |
!fw_diagnostics |
Display framework diagnostics and health | Bot Owner |
!fw_alert_channel <channel> |
Set alert channel for framework diagnostics | Bot Owner |
!cachestats |
Display cache statistics | Bot Owner |
!dbstats |
Display database connection stats | Bot Owner |
!integritycheck |
Run full system integrity check | Bot Owner |
!cleanup |
Clean up system cache and temp files | Bot Owner |
The integrated Extension Marketplace allows you to browse, search, and install official extensions directly from Discord without manual file downloads.
✅ Browse Extensions - View all available extensions with pagination
✅ Search Functionality - Find extensions by keywords
✅ Category Filtering - Filter by status (Working, Beta, Broken)
✅ One-Click Install - Direct installation to ./extensions directory
✅ License Agreement - Mandatory acceptance before use
✅ ZygnalID System - Unique bot identification for support
✅ Dependency Management - Automatic missing package installation
✅ Version Tracking - View extension versions and update dates
- Run marketplace command:
!marketplace-
Accept License Agreement:
- Read the terms carefully
- Click "✅ Accept License" button
- Acceptance is tracked per-user
-
Browse Extensions:
- Use dropdown menu or buttons
- Search by keywords
- Filter by categories
Method 1: Interactive Menu
!marketplace
# Click "Browse All" button
# Select extension from dropdown
# Click "📦 Install Extension"
# Confirm installationMethod 2: Direct Install
!marketplace install <extension_id>Post-Installation: After successful installation:
# Load the extension immediately
!load extension_name
# Or reload if already loaded
!reload extension_name
# Or restart bot for auto-loadWhat is ZygnalID?
- Unique 16-character identifier for your bot
- Auto-generated on first marketplace use
- Required for extension downloads
- Enables dedicated support and tracking
View Your ID:
!marketplace myid(Owner only, sent via DM for security)
Activation: If your ID is invalid or not activated:
- Join ZygnalBot Discord:
gg/sgZnXca5ts - Verify yourself
- Open ticket for "Zygnal ID Activation"
- Provide your ZygnalID from
!marketplace myid
Automatic Fix: If extensions fail to load due to missing Python packages:
!marketplace fixdepsThis command:
- Scans
botlogs/current_run.logforModuleNotFoundError - Extracts missing package names
- Automatically runs
pip install <package> - Reports success/failure per package
- Provides next steps for loading extensions
Manual Installation:
pip install package_name
!reload extension_nameWorking ✅
- Fully functional extensions
- Tested and verified
- Production-ready
Beta
- Experimental features
- May have minor issues
- Actively developed
Broken ❌
- Known issues
- Not recommended for use
- Awaiting fixes
View detailed information about any extension:
!marketplace info <extension_id>Displays:
- Full description
- Version number
- Status (Working/Beta/Broken)
- File type (PY/TXT)
- Upload date
- Custom URL (if available)
- Installation instructions
By Keyword:
!marketplace search moderationBy Title or Description: The search is case-insensitive and matches:
- Extension titles
- Description text
- Details field
The marketplace caches API responses for 300 seconds (5 minutes).
Force Refresh:
!marketplace refreshUse this when:
- New extensions are added
- Extension details are updated
- Cache shows outdated information
import discord
from discord.ext import commands
from discord import app_commands
class MyExtension(commands.Cog):
"""Description of your extension"""
def __init__(self, bot):
self.bot = bot
print(f"{self.__class__.__name__} initialized")
@commands.hybrid_command(
name="example",
help="An example command that works with both prefix and slash"
)
@commands.cooldown(1, 5, commands.BucketType.user)
async def example_command(self, ctx):
"""Command implementation"""
embed = discord.Embed(
title="✅ Success",
description="This is an example command!",
color=0x00ff00
)
await ctx.send(embed=embed)
@commands.Cog.listener()
async def on_message(self, message):
"""Event listener example"""
if message.author.bot:
return
# Your logic here
def cog_unload(self):
"""Cleanup when extension is unloaded"""
print(f"{self.__class__.__name__} unloaded")
async def setup(bot):
"""Required setup function"""
await bot.add_cog(MyExtension(bot))Basic File Operations:
from atomic_file_system import global_file_handler
class MyExtension(commands.Cog):
async def save_data(self, data: dict):
# Atomic write with cache invalidation
success = await global_file_handler.atomic_write_json(
"./data/my_extension_data.json",
data,
invalidate_cache_after=True
)
return success
async def load_data(self) -> dict:
# Read with caching enabled (300s TTL)
data = await global_file_handler.atomic_read_json(
"./data/my_extension_data.json",
use_cache=True
)
return data or {}Advanced Usage with Manual Cache Control:
class AdvancedExtension(commands.Cog):
async def update_config(self, key: str, value: str):
# Read current config (cached)
config = await global_file_handler.atomic_read_json(
"./data/config.json",
use_cache=True
) or {}
# Update config
config[key] = value
# Write and keep in cache (frequent reads)
await global_file_handler.atomic_write_json(
"./data/config.json",
config,
invalidate_cache_after=False # Keep in cache
)
async def force_reload(self):
# Bypass cache for critical reads
config = await global_file_handler.atomic_read_json(
"./data/config.json",
use_cache=False
)
# Or manually invalidate cache
global_file_handler.invalidate_cache("./data/config.json")Cache Statistics in Extensions:
class MonitoringExtension(commands.Cog):
@commands.command()
async def check_cache(self, ctx):
stats = global_file_handler.get_cache_stats()
await ctx.send(f"""
Cache Performance:
- Size: {stats['cache_size']}/{stats['max_cache_size']}
- Hit Rate: {stats['hit_rate']}%
- Total Ops: {stats['total_reads'] + stats['total_writes']}
- Failures: {stats['write_failures'] + stats['read_failures']}
""")import discord
from discord.ext import commands
# Plugin metadata (recommended for Plugin Registry)
__version__ = "1.0.0"
__author__ = "YourName"
__description__ = "A cool extension that does amazing things"
__dependencies__ = [] # List of required extensions
__conflicts__ = [] # List of incompatible extensions
class MyExtension(commands.Cog):
def __init__(self, bot):
self.bot = bot
# Register with Plugin Registry (if available)
if hasattr(bot, 'register_plugin'):
bot.register_plugin(
name="my_extension",
version=__version__,
author=__author__,
description=__description__,
dependencies=__dependencies__,
conflicts_with=__conflicts__
)
@commands.hybrid_command(name="example")
async def example_command(self, ctx):
await ctx.send("Hello from my extension!")
async def setup(bot):
await bot.add_cog(MyExtension(bot))from discord.ext import commands
import discord
class HookedExtension(commands.Cog):
def __init__(self, bot):
self.bot = bot
# Register hooks if Event Hooks system is available
if hasattr(bot, 'register_hook'):
# Priority: higher executes first (default: 0)
bot.register_hook("extension_loaded", self.on_ext_loaded, priority=5)
bot.register_hook("command_executed", self.on_cmd_used)
bot.register_hook("bot_ready", self.on_bot_ready)
async def on_ext_loaded(self, bot, extension_name: str, **kwargs):
"""Called when any extension is loaded"""
print(f"Extension loaded: {extension_name}")
async def on_cmd_used(self, bot, command_name: str, author, **kwargs):
"""Called when any command is executed"""
print(f"Command {command_name} used by {author}")
async def on_bot_ready(self, bot, bot_user, **kwargs):
"""Called when bot becomes ready"""
print(f"Bot is ready: {bot_user}")
def cog_unload(self):
"""Cleanup: Unregister hooks when cog unloads"""
if hasattr(self.bot, 'unregister_hook'):
self.bot.unregister_hook("extension_loaded", self.on_ext_loaded)
self.bot.unregister_hook("command_executed", self.on_cmd_used)
self.bot.unregister_hook("bot_ready", self.on_bot_ready)
async def setup(bot):
await bot.add_cog(HookedExtension(bot))from discord.ext import commands
import discord
class DatabaseExtension(commands.Cog):
def __init__(self, bot):
self.bot = bot
@commands.hybrid_command(name="savedata")
async def save_data(self, ctx, key: str, value: str):
"""Save data to guild database"""
# Get guild-specific database connection
conn = await self.bot.db._get_guild_connection(ctx.guild.id)
# Create table if not exists
await conn.execute("""
CREATE TABLE IF NOT EXISTS custom_data (
key TEXT PRIMARY KEY,
value TEXT,
user_id INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Insert or replace data
await conn.execute(
"INSERT OR REPLACE INTO custom_data (key, value, user_id) VALUES (?, ?, ?)",
(key, value, ctx.author.id)
)
await conn.commit()
await ctx.send(f"✅ Saved: {key} = {value}")
@commands.hybrid_command(name="getdata")
async def get_data(self, ctx, key: str):
"""Retrieve data from guild database"""
conn = await self.bot.db._get_guild_connection(ctx.guild.id)
async with conn.execute(
"SELECT value, user_id FROM custom_data WHERE key = ?",
(key,)
) as cursor:
row = await cursor.fetchone()
if row:
await ctx.send(f"📄 {key} = {row['value']} (saved by <@{row['user_id']}>)")
else:
await ctx.send(f"❌ No data found for key: {key}")
async def setup(bot):
await bot.add_cog(DatabaseExtension(bot))The database system uses the atomic file handler for:
- Configuration persistence -
config.jsonwritten atomically - Plugin registry storage - Registry JSON with ACID compliance
- Framework diagnostics - Diagnostic reports saved safely
- Extension data - Any JSON-based extension storage
Benefits:
- Zero data corruption even during crashes
- Concurrent access safety
- Automatic retry on transient failures
- Cache acceleration for frequently accessed files
- Lock-free reads (when cached)
Performance Optimization:
# High-frequency reads (use cache)
prefix = await bot.config.get("prefix") # Cached
# Critical writes (bypass cache)
await bot.config.file_handler.atomic_write_json(
path,
data,
invalidate_cache_after=True
)
# Batch operations (manual cache management)
for item in items:
await process_item(item)
global_file_handler.clear_all_cache() # Clear after batchfrom discord.ext import commands
import discord
from atomic_file_system import global_file_handler
from datetime import datetime
class FileExtension(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.data_file = "./data/my_extension_data.json"
@commands.hybrid_command(name="saveconfig")
async def save_config(self, ctx, setting: str, value: str):
"""Save configuration atomically"""
# Read existing data (with caching)
existing_data = await global_file_handler.atomic_read_json(
self.data_file,
use_cache=True
) or {}
# Update data
existing_data[setting] = {
"value": value,
"set_by": ctx.author.id,
"timestamp": datetime.now().isoformat()
}
# Atomic write (invalidates cache)
success = await global_file_handler.atomic_write_json(
self.data_file,
existing_data,
invalidate_cache_after=True
)
if success:
await ctx.send(f"✅ Configuration saved: {setting} = {value}")
else:
await ctx.send("❌ Failed to save configuration")
@commands.hybrid_command(name="getconfig")
async def get_config(self, ctx, setting: str = None):
"""Retrieve configuration"""
# Read with caching enabled (300s TTL)
data = await global_file_handler.atomic_read_json(
self.data_file,
use_cache=True
)
if not data:
await ctx.send("❌ No configuration found")
return
if setting:
if setting in data:
config = data[setting]
await ctx.send(
f"📄 {setting} = {config['value']}\n"
f"Set by: <@{config['set_by']}>\n"
f"Time: {config['timestamp']}"
)
else:
await ctx.send(f"❌ Setting not found: {setting}")
else:
# Display all settings
embed = discord.Embed(
title="⚙️ All Configurations",
color=0x5865f2
)
for key, config in data.items():
embed.add_field(
name=key,
value=f"Value: {config['value']}\nSet by: <@{config['set_by']}>",
inline=False
)
await ctx.send(embed=embed)
async def setup(bot):
await bot.add_cog(FileExtension(bot))import discord
from discord.ext import commands
from discord import app_commands
class ModerationExtension(commands.Cog, name="Moderation"):
"""Moderation commands for server management"""
def __init__(self, bot):
self.bot = bot
@commands.hybrid_command(
name="ban",
help="Ban a user from the server"
)
@commands.has_permissions(ban_members=True)
@commands.bot_has_permissions(ban_members=True)
@commands.guild_only()
async def ban_user(
self,
ctx,
member: discord.Member,
*,
reason: str = "No reason provided"
):
"""Ban a member with reason"""
# Safety check
if member.top_role >= ctx.author.top_role:
await ctx.send("❌ You cannot ban this user (role hierarchy)")
return
# Perform ban
await member.ban(reason=f"{reason} (By: {ctx.author})")
# Log to database
await self.bot.db.increment_command_usage("ban")
# Send confirmation
embed = discord.Embed(
title="🔨 User Banned",
description=f"**User:** {member.mention}\n**Reason:** {reason}",
color=0xff0000,
timestamp=discord.utils.utcnow()
)
embed.set_footer(text=f"Banned by {ctx.author}")
await ctx.send(embed=embed)
# Try to DM the user
try:
dm_embed = discord.Embed(
title="You were banned",
description=f"**Server:** {ctx.guild.name}\n**Reason:** {reason}",
color=0xff0000
)
await member.send(embed=dm_embed)
except:
pass # User has DMs disabled
@commands.hybrid_command(
name="kick",
help="Kick a user from the server"
)
@commands.has_permissions(kick_members=True)
@commands.bot_has_permissions(kick_members=True)
@commands.guild_only()
async def kick_user(
self,
ctx,
member: discord.Member,
*,
reason: str = "No reason provided"
):
"""Kick a member with reason"""
if member.top_role >= ctx.author.top_role:
await ctx.send("❌ You cannot kick this user (role hierarchy)")
return
await member.kick(reason=f"{reason} (By: {ctx.author})")
await self.bot.db.increment_command_usage("kick")
embed = discord.Embed(
title="👢 User Kicked",
description=f"**User:** {member.mention}\n**Reason:** {reason}",
color=0xff9900,
timestamp=discord.utils.utcnow()
)
embed.set_footer(text=f"Kicked by {ctx.author}")
await ctx.send(embed=embed)
async def setup(bot):
await bot.add_cog(ModerationExtension(bot))import discord
from discord.ext import commands, tasks
from datetime import datetime
class TaskExtension(commands.Cog):
"""Extension with background tasks"""
def __init__(self, bot):
self.bot = bot
self.message_count = 0
self.hourly_report.start()
def cog_unload(self):
"""Stop tasks when unloading"""
self.hourly_report.cancel()
@tasks.loop(hours=1)
async def hourly_report(self):
"""Send hourly statistics report"""
channel_id = self.bot.config.get("reports_channel_id")
if not channel_id:
return
channel = self.bot.get_channel(channel_id)
if not channel:
return
embed = discord.Embed(
title="📊 Hourly Report",
description=f"Messages tracked: {self.message_count}",
color=0x5865f2,
timestamp=datetime.utcnow()
)
stats = self.bot.metrics.get_stats()
embed.add_field(
name="Bot Statistics",
value=f"Commands: {stats['commands_processed']}\nErrors: {stats['error_count']}",
inline=False
)
await channel.send(embed=embed)
self.message_count = 0 # Reset counter
@hourly_report.before_loop
async def before_hourly_report(self):
"""Wait until bot is ready"""
await self.bot.wait_until_ready()
@commands.Cog.listener()
async def on_message(self, message):
"""Track messages"""
if not message.author.bot:
self.message_count += 1
async def setup(bot):
await bot.add_cog(TaskExtension(bot))import discord
from discord.ext import commands
def is_admin_or_owner():
"""Custom check for admin or bot owner"""
async def predicate(ctx):
if ctx.author.id == ctx.bot.bot_owner_id:
return True
if ctx.guild and ctx.author.guild_permissions.administrator:
return True
raise commands.CheckFailure("You must be an administrator or bot owner")
return commands.check(predicate)
def has_any_role(*role_names):
"""Custom check for any of the specified roles"""
async def predicate(ctx):
if not ctx.guild:
raise commands.CheckFailure("This command cannot be used in DMs")
member_roles = [role.name for role in ctx.author.roles]
if any(role in member_roles for role in role_names):
return True
raise commands.CheckFailure(f"You need one of these roles: {', '.join(role_names)}")
return commands.check(predicate)
class CustomChecksExtension(commands.Cog):
def __init__(self, bot):
self.bot = bot
@commands.hybrid_command(name="adminonly")
@is_admin_or_owner()
async def admin_only_command(self, ctx):
"""Only admins and bot owner can use this"""
await ctx.send("✅ You have admin privileges!")
@commands.hybrid_command(name="staffonly")
@has_any_role("Staff", "Moderator", "Admin")
async def staff_only_command(self, ctx):
"""Only staff members can use this"""
await ctx.send("✅ You are a staff member!")
async def setup(bot):
await bot.add_cog(CustomChecksExtension(bot))import discord
from discord.ext import commands
from discord import app_commands
from typing import Literal
class SlashOptionsExtension(commands.Cog):
def __init__(self, bot):
self.bot = bot
@commands.hybrid_command(name="color")
@app_commands.describe(
color="Choose a color",
intensity="Color intensity level"
)
async def color_command(
self,
ctx,
color: Literal["red", "green", "blue"],
intensity: int = 5
):
"""Command with slash options"""
color_map = {
"red": 0xff0000,
"green": 0x00ff00,
"blue": 0x0000ff
}
embed = discord.Embed(
title=f"{color.capitalize()} Color",
description=f"Intensity: {intensity}/10",
color=color_map[color]
)
await ctx.send(embed=embed)
@commands.hybrid_command(name="userinfo")
@app_commands.describe(member="The member to get info about")
async def userinfo_command(
self,
ctx,
member: discord.Member = None
):
"""Get information about a user"""
member = member or ctx.author
embed = discord.Embed(
title=f"User Info: {member}",
color=member.color
)
embed.set_thumbnail(url=member.display_avatar.url)
embed.add_field(name="ID", value=member.id, inline=True)
embed.add_field(name="Joined", value=member.joined_at.strftime("%Y-%m-%d"), inline=True)
embed.add_field(name="Roles", value=f"{len(member.roles)-1}", inline=True)
await ctx.send(embed=embed)
async def setup(bot):
await bot.add_cog(SlashOptionsExtension(bot))On first run, the bot creates config.json with default settings:
{
"prefix": "!",
"owner_ids": [],
"auto_reload": true,
"status": {
"type": "watching",
"text": "{guilds} servers"
},
"database": {
"base_path": "./data"
},
"logging": {
"level": "INFO",
"max_bytes": 10485760,
"backup_count": 5
},
"extensions": {
"auto_load": true,
"blacklist": []
},
"cooldowns": {
"default_rate": 3,
"default_per": 5.0
},
"command_permissions": {},
"slash_limiter": {
"max_limit": 100,
"warning_threshold": 90,
"safe_limit": 95
},
"framework": {
"load_cogs": true,
"enable_event_hooks": true,
"enable_plugin_registry": true,
"enable_framework_diagnostics": true,
"enable_slash_command_limiter": true
}
}Configuration Options Basic Settings prefix (string)
Default command prefix Default: "!" Per-guild overrides supported via !setprefix
owner_ids (array)
Additional bot owner IDs Default: [] Primary owner from BOT_OWNER_ID env variable
auto_reload (boolean)
Enable hot-reload for extensions Default: true Checks every 30 seconds for file changes
Status Configuration status.type (string)
Activity type shown in Discord Options: "playing", "watching", "listening", "competing" Default: "watching"
status.text (string)
Status message with variables Variables: {guilds}, {users}, {commands} Default: "{guilds} servers" Example: "with {users} users | {guilds} servers"
Database Configuration database.base_path (string)
Base directory for database files Default: "./data" Contains main.db and per-guild databases
Logging Configuration logging.level (string)
Logging level Options: "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL" Default: "INFO"
logging.max_bytes (integer)
Max log file size before rotation Default: 10485760 (10MB)
logging.backup_count (integer)
Number of backup log files to keep Default: 5
Extensions Configuration extensions.auto_load (boolean)
Automatically load all extensions on startup Default: true
extensions.blacklist (array)
Extension names to skip during auto-load Default: [] Example: ["debug_cog", "test_extension"]
Cooldowns Configuration cooldowns.default_rate (integer)
Default number of command uses Default: 3
cooldowns.default_per (float)
Default cooldown period in seconds Default: 5.0
Slash Command Limiter slash_limiter.max_limit (integer)
Discord's hard limit for slash commands Default: 100 Should not be changed
slash_limiter.warning_threshold (integer)
Command count to trigger warnings Default: 90 Logs warning at this count
slash_limiter.safe_limit (integer)
Command count to start prefix-only mode Default: 95 New extensions become prefix-only
Framework Cogs framework.load_cogs (boolean)
Load framework internal cogs Default: true Disable to use minimal framework
framework.enable_event_hooks (boolean)
Enable Event Hooks system Default: true
framework.enable_plugin_registry (boolean)
Enable Plugin Registry system Default: true
framework.enable_framework_diagnostics (boolean)
Enable Framework Diagnostics system Default: true
framework.enable_slash_command_limiter (boolean)
Enable Slash Command Limiter Default: true
Command Permissions Configure role-based command access:
# Grant @Moderator access to ban command
!config ban @Moderator
# Add multiple roles
!config kick @Moderator
!config kick @Admin
# View current permissions
!config
# Remove restrictions
!config ban none
{
"command_permissions": {
"ban": [123456789, 987654321],
"kick": [123456789]
}
}
Set per-guild prefixes:
# Set prefix to ?
!setprefix ?
# Now commands work with ?
?help
?stats
# Slash commands always work regardless of prefix
/help
/stats
Required .env file:
# Required
DISCORD_TOKEN=your_bot_token_here
BOT_OWNER_ID=your_discord_user_id
# Optional: Sharding (for large bots)
SHARD_COUNT=2
# SHARD_IDS can be commented out
SHARD_IDS=0,1
- SHARD_COUNT: Total number of shards
- SHARD_IDS: Comma-separated list of shard IDs to run
- Leave empty for auto-sharding
The framework uses a hybrid database approach:
- Main Database (./data/main.db)
- Global bot statistics
- Command usage tracking
- Framework-wide data
- Per-Guild Databases (./data/[guild_id]/guild.db)
- Guild-specific settings
- Custom prefixes
- Extension data per guild
✅ WAL Mode - Write-Ahead Logging for concurrent access ✅ Connection Pooling - Automatic per-guild connection management ✅ Automatic Backups - Created on bot shutdown ✅ Orphan Cleanup - Removes connections for left guilds ✅ Atomic Operations - ACID compliance
# Direct access to main database
async with self.bot.db.conn.execute(
"SELECT * FROM global_stats WHERE key = ?",
("some_key",)
) as cursor:
row = await cursor.fetchone()# Get guild-specific connection
conn = await self.bot.db._get_guild_connection(guild_id)
# Execute queries
await conn.execute(
"INSERT INTO custom_table (data) VALUES (?)",
(data,)
)
await conn.commit()# Set guild prefix
await bot.db.set_guild_prefix(guild_id, "?")
# Get guild prefix
prefix = await bot.db.get_guild_prefix(guild_id)
# Increment command usage
await bot.db.increment_command_usage("command_name")
# Get command statistics
stats = await bot.db.get_command_stats()
# Returns: [(command_name, count), ...]
# Cleanup guild data
await bot.db.cleanup_guild(guild_id)
# Backup databases
await bot.db.backup() # All guilds
await bot.db.backup(guild_id) # Specific guildglobal_stats
CREATE TABLE global_stats (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
CREATE TABLE guild_settings (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
CREATE TABLE command_stats (
command_name TEXT PRIMARY KEY,
usage_count INTEGER DEFAULT 0,
last_used TIMESTAMP
)
- Cleans up orphaned connections
- Expires prefix cache entries
- Removes pycache directories
- Logs maintenance actions
Manual Cleanup:
!cleanup
View Database Stats:
!dbstats
Framework cogs are internal system components that provide core functionality. They're located in ./cogs directory and are automatically loaded on startup (if enabled in config).
File: cogs/event_hooks.py
What It Does
- Internal event system for framework lifecycle events
- Priority-based callback execution
- Asynchronous queue processing (1000 event capacity)
- Execution history tracking (100 total, 20 per event)
- Worker task for async event processing
| Event Name | Description | Parameters |
|---|---|---|
| bot_ready | Bot becomes ready | — |
| guild_joined | Bot joins a guild | guild |
| guild_left | Bot leaves a guild | guild |
| command_executed | Command is executed | command_name, author, guild |
| command_error | Command error occurs | command_name, error, author, guild |
| extension_loaded | Extension is loaded (custom) | extension_name |
| extension_unloaded | Extension is unloaded (custom) | extension_name |
File: atomic_file_system.py
What It Does
- Thread-safe atomic file operations with zero data corruption
- LRU cache system (300s TTL, 1000 entry limit)
- Automatic file lock management with cleanup
- Retry logic for write operations (3 attempts)
- Comprehensive diagnostics and metrics tracking
Key Features:
Atomic Write Protection
- Tempfile-based writes prevent partial updates
- Automatic retry on permission errors
- Windows/Linux compatible file operations
- Cross-platform atomic move operations
LRU Cache System
- Configurable TTL (default: 300 seconds)
- Automatic eviction when full (max 1000 entries)
- Cache hit rate tracking
- Manual invalidation support
- Thread-safe cache access
Lock Management
- Per-file lock acquisition
- Automatic cleanup of inactive locks (threshold: 500)
- Lock age tracking for diagnostics
- Deadlock prevention
Performance Metrics
- Read/write operation counters
- Cache hit rate calculation
- Failure rate monitoring
- Lock cleanup statistics
Health Monitoring
- Real-time failure rate tracking
- Three-tier status: Healthy (<1%), Degraded (1-5%), Critical (>5%)
- Operation history tracking
View System Statistics:
!atomicstatsShows:
- Cache size and hit rate
- Active file locks
- Total read/write operations
- Failure rates and health status
- System uptime
Run Diagnostic Tests:
!atomictestPerforms:
- Atomic write test
- Read test without cache
- Read test with cache (verifies caching)
- Concurrent write test (10 simultaneous writes)
- Cache invalidation test
Performance Metrics:
- Cache hit rate >70% is healthy
- Active locks should be <100 normally
- Failure rate should be <1%
- Lock cleanup runs automatically at 500+ locks
Register a Hook:
# Register hook with priority
bot.register_hook("bot_ready", my_callback, priority=10)
# Hook callback signature
async def my_callback(bot, **kwargs):
# bot: Bot instance
# kwargs: Event-specific parameters
passHook ID Format:
- Format:
event_name:callback_name - Example:
bot_ready:on_bot_ready - Used for disabling/enabling specific hooks
View Registered Hooks:
!eh_listShows:
- All registered hooks by event
- Priority levels
- Execution counts and failure counts
- Average execution time
- Status indicators:
- 🔴 - Hook disabled manually
⚠️ - Circuit breaker open- No icon - Working normally
View Execution History:
# Last 10 executions
!eh_history
# Last 30 executions
!eh_history 30Manage Problematic Hooks:
# Disable a hook that's causing issues
!eh_disable bot_ready:on_bot_ready
# Re-enable when fixed
!eh_enable bot_ready:on_bot_ready
# Reset circuit breaker (if hook keeps failing)
!eh_reset_circuit command_executed:log_commandSet Alert Channel:
# Set current channel
!eh_alert_channel
# Set specific channel
!eh_alert_channel #hook-alertsAutomatic Alerts Sent For:
- Queue full (dropped events)
- Circuit breaker opens (hook disabled due to failures)
- Worker crashes (with restart attempt count)
- Max restart attempts reached
Built-in Events:
bot_ready- Bot becomes ready (kwargs:bot_user)guild_joined- Bot joins guild (kwargs:guild)guild_left- Bot leaves guild (kwargs:guild)command_executed- Command runs (kwargs:command_name,author,guild)command_error- Command error (kwargs:command_name,error,author,guild)
Custom Events:
# Emit custom event from your extension
await bot.emit_hook("custom_event", custom_param="value")Circuit Breaker Behavior:
- Hook fails 5 times within 5 minutes
- Circuit breaker opens (hook disabled)
- Hook disabled for 60 seconds
- Automatically closes after timeout
- Can manually reset with
!eh_reset_circuit
Performance Considerations:
- Each hook has 10-second timeout
- Queue processes sequentially (high-priority first)
- Average execution time tracked per hook
- Slow hooks should be optimized or disabled
Unregister Hooks:
# In cog_unload()
bot.unregister_hook("event_name", my_callback)View Current Health:
!fw_diagnosticsSet Alert Channel:
# Set current channel as alert channel
!fw_alert_channel
# Set specific channel
!fw_alert_channel #alertsWhat Gets Monitored:
-
Health Status:
- ✅ Healthy: Error rate < 5%
⚠️ Degraded: Error rate 5-9.9%- 🚨 Critical: Error rate ≥ 10%
-
System Metrics:
- Memory usage (MB)
- CPU percentage
- Active threads
- Open file handles
- Network connections
- Event loop lag (ms)
-
Bot Statistics:
- Uptime
- Latency
- Commands processed
- Error count
- Extensions loaded
- Servers/users/channels
Automatic Alerts: The system automatically sends alerts to your configured channel for:
- Health status changes (degraded/critical)
- Event loop lag above threshold (500ms)
- Repeated diagnostics write failures (3+ consecutive)
Diagnostics Files:
./data/framework_diagnostics.json- Complete system snapshot (updated every 5 min)./data/framework_health.json- Real-time health metrics (updated every 5 min)
Health Check Frequency:
- Event loop lag: Every 1 second
- Health monitoring: Every 5 minutes
- Diagnostics generation: Every 5 minutes
List All Plugins:
!pr_listShows all registered plugins with status indicators:
- ✅ - All checks passed
⚠️ - Has warnings (conflicts, scan errors)- ❌ - Missing dependencies
- 🔄 - Circular dependency detected
View Plugin Details:
!pr_info extension_nameShows:
- Version, author, description
- Commands and cogs provided
- Dependencies with version requirements
- Conflict status
- Circular dependency paths (if any)
- Scan errors (if any)
Validate Plugin:
!pr_validate extension_nameChecks if a plugin is safe to load by validating:
- All dependencies are met
- No conflicts with loaded plugins
- No circular dependencies
Toggle Enforcement:
# Toggle dependency enforcement
!pr_enforce deps
# Toggle conflict enforcement
!pr_enforce conflictsWhen enforcement is enabled (default):
- Plugins with issues trigger warnings
- Validation errors are logged
When enforcement is disabled:
- Plugins load regardless of issues
- Only informational warnings
Set Alert Channel:
# Set current channel
!pr_alert_channel
# Set specific channel
!pr_alert_channel #plugin-alertsAutomatic Alerts Sent For:
- Plugins loaded with missing dependencies
- Plugins loaded with conflicts
- Extension scan failures
- Circular dependency detection
Version Requirements:
In your extension, specify dependencies with versions:
# Supports: >=, >, ==, <=,
__dependencies__ = {
"other_extension": ">=1.0.0",
"another_ext": "==2.5.1"
}Conflict Declaration:
# List incompatible extensions
__conflicts__ = ["incompatible_ext", "old_version_ext"]Registry File:
- Location:
./data/plugin_registry.json - Auto-saved on changes
- Contains all plugin metadata
- Includes enforcement settings
- Higher priority = executes first
- Default priority: 0
- Range: any integer
- Example: priority=15 runs before priority=10
# View registered hooks (Owner only)
!hooks
# View execution history (Owner only)
!hookhistory [limit]
Data Integrity:
- Tempfile-based writes prevent partial updates
- Atomic move operations (all-or-nothing)
- Automatic retry on transient failures
- Lock-based concurrency control
- Cache invalidation on successful writes
Protection Against:
- Race conditions (per-file locks)
- Partial writes (tempfile staging)
- Data corruption (atomic moves)
- Concurrent access (lock management)
- Cache inconsistency (TTL + invalidation)
Monitoring:
- Real-time failure tracking
- Health status calculation
- Comprehensive operation logging
- Test suite for verification
- Full access to all commands
- Defined in
BOT_OWNER_IDenvironment variable - Cannot be overridden
- Access to server management commands
- Can configure command permissions
- Automatically detected via
guild.owner_id
- Per-command role requirements
- Set via
!configcommand - Stored in
config.json
- Uses native Discord permission checks
@commands.has_permissions()@commands.bot_has_permissions()
- No restrictions
- Available to all users
These commands cannot be configured for other users:
BOT_OWNER_ONLY_COMMANDS = [
"reload", "load", "unload", "sync",
"atomictest", "cachestats", "shardinfo",
"dbstats", "integritycheck", "cleanup"
]@commands.command()
@is_bot_owner()
async def owner_command(ctx):
await ctx.send("Owner only!")@commands.command()
@is_bot_owner_or_guild_owner()
async def management_command(ctx):
await ctx.send("Owner or guild owner!")@commands.command()
@has_command_permission()
async def configurable_command(ctx):
await ctx.send("Permission checked!")async def interaction_check(self, interaction):
if interaction.user != self.author:
await interaction.response.send_message(
"❌ Only the command user can use this!",
ephemeral=True
)
return False
return True- Default: 180–300 seconds
- Automatically disables interactions after timeout
- Prevents stale UI elements
Prevents:
- Data corruption during concurrent writes
- Partial file writes
- File locking issues
How it works:
- Write to temporary file
- Verify success
- Atomic move to final location
- Invalidate cache
- Write-Ahead Logging
- Concurrent reads during writes
- ACID compliant
- Crash recovery
- Per-guild isolation
- Automatic cleanup
- Lock-safe operation
- Automatic on shutdown
- Manual via
!dbstats - Timestamped backup files
- Per-file lock acquisition
- Automatic cleanup (500 lock threshold)
- Prevents concurrent write conflicts
- Unique bot identification
- Required for extension downloads
- Activation verification
- Mandatory acceptance
- Per-user tracking
- Enforces usage restrictions
- Extensions stored in isolated directory
- Never auto-executed
- Manual loading required
Default color scheme:
# Success
SUCCESS = 0x00ff00
# Error
ERROR = 0xff0000
# Warning
WARNING = 0xffff00
# Info
INFO = 0x5865f2
# Main Menu
MAIN_MENU = 0x2b2d31
# Credits
CREDITS = 0xffd700embed = discord.Embed(
title="Custom Title",
description="Custom Description",
color=0x9b59b6 # Purple
)Common Unicode emojis:
# Status
"✅" # Success
"❌" # Error
"⚠️" # Warning
"ℹ️" # Info
# Actions
"🔄" # Reload
"📊" # Stats
"⚙️" # Settings
"🔧" # Tools
# Navigation
"◀" # Previous
"▶" # Next
"🏠" # Home
# Categories
"📚" # Help
"📦" # Extensions
"🗄️" # Database
"📊" # StatisticsEdit config.json:
{
"status": {
"type": "watching",
"text": "{guilds} servers | {users} users"
}
}playingwatchinglisteningcompeting
{guilds}– Total server count{users}– Total user count{commands}– Total command count
{
"prefix": "?"
}!setprefix ?@bot.hybrid_command(name="myhelp")
async def custom_help(ctx):
embed = discord.Embed(
title="My Custom Help",
description="Custom help menu",
color=0x5865f2
)
# Add your fields
await ctx.send(embed=embed)Then disable the built-in help in main.py:
Then disable the built-in help in main.py:
bot = BotFrameWork(
command_prefix=lambda b, m: b.get_prefix(m),
intents=intents,
help_command=None, # Disable built-in help
# ... other options
)Error: DISCORD_TOKEN not found in .env!
Solution:
- Create
.envfile in root directory - Add:
DISCORD_TOKEN=your_token_here - Ensure no quotes around token
- Ensure file is named
.envnot.env.txt
Error: BOT_OWNER_ID not found in .env!
Solution:
- Add to
.env:BOT_OWNER_ID=your_user_id - Get ID by enabling Developer Mode in Discord
- Right-click your username → Copy ID
Error: discord.ext.commands.errors.ExtensionFailed
Solution:
- Check logs:
cat botlogs/current_run.log - Look for syntax errors in extension
- Ensure all imports are available
- Use
!marketplace fixdepsfor missing packages
High Cache Miss Rate (<50%):
Solution:
- Check if files are being modified externally
- Verify cache TTL isn't too short (default: 300s)
- Run
!atomicstatsto view statistics - Consider increasing
max_cache_sizein code
Many Active Locks (>100):
Solution:
- Check
!atomicstatsfor lock count - Locks cleanup automatically at 500 threshold
- Restart bot if locks are permanently stuck
- Review code for proper async/await usage
Write Failures:
Solution:
- Check file permissions in
./datadirectory - Ensure sufficient disk space
- Verify no external process is locking files
- Check
!atomicstatsfor failure patterns - Review logs for specific error messages
Cache Not Invalidating:
Solution:
# Manual invalidation in code
global_file_handler.invalidate_cache("./data/problematic_file.json")
# Or clear all cache
global_file_handler.clear_all_cache()Performance Degradation:
Solution:
- Check
!atomicstatsfor health status - If failure rate >5%, check disk I/O
- Monitor lock count (should be <50 normally)
- Run
!atomictestto verify system health - Consider restarting bot if critical
Check logs:
# Linux/Mac
cat botlogs/current_run.log | grep -i "extension"
# Windows
findstr /i "extension" botlogs\current_run.logCommon Issues:
- Missing
setupfunction
# Required in every extension
async def setup(bot):
await bot.add_cog(YourCog(bot))-
Syntax errors
- Check Python syntax
- Ensure proper indentation
- Validate imports
-
Extension in blacklist
- Check
config.json→extensions.blacklist - Remove extension name from list
- Check
-
File not in
extensions/directory- Move file to
./extensions/ - Ensure
.pyextension
- Move file to
-
Spaces in filename
- Framework auto-renames to underscores
- Use
!load extension_name(with underscores)
-
Missing dependencies
- Use
!marketplace fixdeps - Or manual:
pip install package_name
- Use
Manual sync:
!syncRate Limiting:
- Discord limits syncs to ~2 per hour
- Bot automatically waits and retries
- Check logs for retry messages
- Wait 1 hour if rate limited
Slash Command Limit Reached:
# Check current usage
!slashlimit
# If at limit (100 commands):
# - New extensions become prefix-only
# - Existing slash commands still work
# - Unload unused extensions to free slotsVerification:
# Check registered commands
!extensions
# View slash command count
!statsError: database is locked
Solution:
- Ensure only one bot instance running
- Wait for WAL checkpoint to complete
- Restart bot if persists
Backup and Reset:
# Backup current database
cp data/main.db data/main.db.backup
cp -r data/[guild_id] data/[guild_id].backup
# Delete database (will regenerate)
rm data/main.db
rm -rf data/[guild_id]
# Restart bot
python main.pyCheck Database Stats:
!dbstats"Missing Permissions" error:
Solution:
- Check bot has required Discord permissions
- Verify role hierarchy (bot role above target roles)
- Enable necessary intents in Developer Portal
- Check bot permissions in channel settings
"You don't have permission" error:
Solution:
- Check command permissions:
!config - Verify your roles match requirements
- Contact bot owner for access
- Check if command is owner-only
High Memory Usage:
Solution:
- Run
!cleanupto clear caches - Check
!cachestatsfor large caches - Reduce cache TTL in
atomic_file_system.py - Restart bot periodically
Slow Command Response:
Solution:
- Check
!statsfor latency - View
!dbstatsfor connection issues - Run
!integritycheckfor system health - Check database file sizes
- Consider sharding for large bots
"Invalid or not activated" ZygnalID:
Solution:
- Get ID:
!marketplace myid(Owner only) - Join ZygnalBot Discord:
gg/sgZnXca5ts - Verify yourself in server
- Open ticket for "Zygnal ID Activation"
- Provide your ZygnalID in ticket
Extension Install Failed:
Solution:
- Check error message for details
- Verify bot has write permissions to
./extensions - Ensure sufficient disk space
- Try
!marketplace refresh - Check logs:
botlogs/current_run.log
Missing Dependencies After Install:
Solution:
# Automatic fix
!marketplace fixdeps
# Manual fix
pip install package_name
!reload extension_nameIssue: File changes not detected
Solution:
- Check
config.json→auto_reload: true - Ensure file in
./extensionsdirectory - Wait 30 seconds for check cycle
- Use manual reload:
!reload extension_name - Check logs for reload errors
Commands not showing:
Solution:
- Check command has
helpparameter - Ensure command not marked
hidden=True - Verify cog loaded:
!extensions - Check command registered:
!stats
Dropdown not working:
Solution:
- Ensure only command requester can use menu
- Check view hasn't timed out (180s)
- Verify bot has "Use Application Commands" permission
- Try rerunning command
Cache Strategy:
# Frequently read files - USE cache
config = await global_file_handler.atomic_read_json(
"config.json",
use_cache=True # Fast repeated reads
)
# Rarely read files - BYPASS cache
log_data = await global_file_handler.atomic_read_json(
"large_log.json",
use_cache=False # Don't pollute cache
)
# After bulk updates - CLEAR cache
await process_many_files()
global_file_handler.clear_all_cache()Lock Optimization:
# Bad: Multiple separate operations
for file in files:
data = await read_file(file)
await write_file(file, process(data))
# Good: Batch operations
all_data = {}
for file in files:
all_data[file] = await read_file(file)
for file, data in all_data.items():
await write_file(file, process(data))Monitoring Health:
# Regular health checks
!atomicstats # View overall health
# Performance testing
!atomictest # Verify operation speedsBest Practices:
- Use cache for config files (read-heavy)
- Bypass cache for log files (write-heavy)
- Invalidate cache after critical writes
- Monitor failure rate (should be <1%)
- Run cleanup during low-activity periods
Use Transactions for Bulk Operations:
async with bot.db.conn.execute("BEGIN"):
for item in items:
await bot.db.conn.execute(
"INSERT INTO table VALUES (?)",
(item,)
)
await bot.db.conn.commit()Index Frequently Queried Columns:
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_user_id ON my_table(user_id)"
)Use LIMIT for Large Queries:
async with conn.execute(
"SELECT * FROM large_table LIMIT 100"
) as cursor:
rows = await cursor.fetchall()Use Caching for Read-Heavy Operations:
# Enable caching (300s TTL)
data = await global_file_handler.atomic_read_json(
filepath,
use_cache=True
)Disable Caching for Real-Time Data:
data = await global_file_handler.atomic_read_json(
filepath,
use_cache=False
)Batch File Operations:
# Bad: Multiple writes
for item in items:
await global_file_handler.atomic_write_json(f"data/{item}.json", item_data)
# Good: Single write
combined_data = {item: item_data for item in items}
await global_file_handler.atomic_write_json("data/all_items.json", combined_data)Batch Messages:
# Bad: Individual messages
for user in users:
await channel.send(f"Hello {user}")
# Good: Single message
await channel.send(f"Hello {', '.join(str(u) for u in users)}")Use Embeds for Rich Content:
# Embeds are more efficient than multiple messages
embed = discord.Embed(title="Data")
for key, value in data.items():
embed.add_field(name=key, value=value)
await channel.send(embed=embed)Cache Guild/Member Data:
# Use intents to cache data
intents = discord.Intents.all()
# Access cached data (no API call)
member = guild.get_member(user_id)Blacklist Unused Extensions:
{
"extensions": {
"blacklist": ["debug_cog", "test_extension"]
}
}Profile Extension Load Times:
# View load times
!extensions
# Check diagnostics
!diagnosticsRegular Cleanup:
# Run periodically
!cleanupMonitor Cache Usage:
!cachestatsLimit Background Tasks:
# Increase interval for non-critical tasks
@tasks.loop(hours=6) # Instead of minutes=5
async def background_task(self):
passFor bots in 2000+ guilds:
# Enable sharding
SHARD_COUNT=2
SHARD_IDS=0,1Benefits:
- Distributes load across processes
- Reduces per-process memory
- Improved stability
- Better rate limit handling
MIT License Copyright (c) 2025 TheHolyOneZ Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
IMPORTANT: The following components must remain intact in all versions:
- CreditsButton class in
main.py- Must remain visible and functional - Author attribution - "TheHolyOneZ" must not be removed or altered
- Repository link - GitHub link must remain in credits section
- Framework command -
!discordbotframeworkmust remain with original features
Marketplace License: Extensions downloaded through the marketplace have additional terms:
- Only usable within ZygnalBot ecosystem (including Zoryx Bot Framework)
- Cannot remove names: ZygnalBot, zygnalbot, TheHolyOneZ, TheZ
- Respect individual extension licenses
- No redistribution outside authorized systems
Violations result in ZygnalID deactivation and service ban.
Contributions are welcome! Here's how to contribute:
- Fork the Repository
git clone https://github.com/yourusername/discord-bot-framework.git
cd discord-bot-framework- Create Feature Branch
git checkout -b feature/amazing-feature-
Test Thoroughly
- Ensure bot starts without errors
- Test all affected commands
- Verify extensions load correctly
- Check for memory leaks
-
Follow Code Style
- Use type hints
- Add docstrings to functions
- Follow PEP 8 guidelines
- Use meaningful variable names
- Comment complex logic
-
Update Documentation
- Update README.md if adding features
- Add docstrings to new functions
- Update CONTRIBUTING.md if needed
-
Commit Changes
git add .
git commit -m "Add amazing feature"- Push to Branch
git push origin feature/amazing-feature- Open Pull Request
- Describe changes clearly
- Reference any related issues
- Include screenshots if UI changes
async def example_function(param: str) -> dict:
"""
Brief description of function.
Args:
param: Description of parameter
Returns:
Description of return value
Raises:
ValueError: When param is invalid
"""
if not param:
raise ValueError("param cannot be empty")
result = {"status": "success", "data": param}
return resultHigh Priority:
- Bug fixes
- Performance improvements
- Documentation improvements
- Extension examples
- Test coverage
Medium Priority:
- New framework features
- Additional framework cogs
- UI/UX improvements
- Logging enhancements
Low Priority:
- Code refactoring
- Style improvements
- Comment additions
Use GitHub Issues with this template:
**Bug Description:**
Clear description of the bug
**Steps to Reproduce:**
1. Step one
2. Step two
3. Error occurs
**Expected Behavior:**
What should happen
**Actual Behavior:**
What actually happens
**Environment:**
- Python version: 3.x.x
- discord.py version: 2.x.x
- OS: Windows/Linux/Mac
**Logs:**Relevant log output
Use GitHub Issues with this template:
**Feature Description:**
Clear description of the feature
**Use Case:**
Why this feature is needed
**Proposed Implementation:**
How it could be implemented
**Alternatives Considered:**
Other approaches consideredNeed help? Have questions?
- GitHub Issues: Report bugs or request features
- GitHub Discussions: Ask questions and discuss
- ZygnalBot Discord:
gg/sgZnXca5ts(For marketplace support)
- discord.py: https://discordpy.readthedocs.io/
- Discord Developer Portal: https://discord.com/developers/applications
- Python asyncio: https://docs.python.org/3/library/asyncio.html
- SQLite: https://www.sqlite.org/docs.html
- aiofiles: https://github.com/Tinche/aiofiles
- aiosqlite: https://aiosqlite.omnilib.dev/
- ZygnalBot Framework: https://zygnalbot.com/bot-framework/
- Extension Marketplace: https://zygnalbot.com/extension/
Discord Bot Development:
Python Async Programming:
Database Management:
Want to see the framework in action? Check the /images directory for screenshots:
- Terminal-1.png - Bot startup with Rich console
- HelpMenu-Example.png - Interactive help system
- Marketplace-Preview.png - Extension marketplace
- Diagnostics.png - Framework diagnostics dashboard
- And more!
TheHolyOneZ
- GitHub: @TheHolyOneZ
- Website: zygnalbot.com/bot-framework
- Discord: theholyonez
