Merge pull request 'commit, am too tired to add docs here' (#8) from feature/nsfw-only-filtering into main

Reviewed-on: #8
This commit was merged in pull request #8.
This commit is contained in:
2026-01-25 08:09:50 +00:00
15 changed files with 3336 additions and 61 deletions

371
MIGRATION.md Normal file
View File

@@ -0,0 +1,371 @@
# GuardDen Migration Guide: Discord Commands to File-Based Configuration
This guide explains how to migrate from Discord command-based configuration to the new file-based YAML configuration system.
## Why Migrate?
The new file-based configuration system offers several advantages:
- **✅ Version Control**: Track configuration changes with Git
- **✅ No Discord Dependencies**: Configure without being in Discord
- **✅ Backup & Restore**: Easy configuration backups and restoration
- **✅ Hot-Reloading**: Changes apply without bot restarts
- **✅ Better Organization**: Clean, structured configuration files
- **✅ Schema Validation**: Automatic error checking and prevention
- **✅ Bulk Operations**: Configure multiple servers efficiently
## Migration Overview
### Phase 1: Preparation
1. ✅ Update GuardDen to the latest version
2. ✅ Install new dependencies: `pip install -e ".[dev,ai]"`
3. ✅ Backup your current configuration (optional but recommended)
### Phase 2: Export Existing Settings
4. ✅ Run the migration tool to export Discord settings to files
5. ✅ Verify migration results
6. ✅ Review and customize exported configurations
### Phase 3: Switch to File-Based Configuration
7. ✅ Test the new configuration system
8. ✅ (Optional) Clean up database configurations
## Step-by-Step Migration
### Step 1: Update Dependencies
```bash
# Install new required packages
pip install -e ".[dev,ai]"
# Or if you prefer individual packages:
pip install pyyaml jsonschema watchfiles
```
### Step 2: Run Migration Tool
Export your existing Discord command settings to YAML files:
```bash
# Export all guild configurations from database to files
python -m guardden.cli.config migrate from-database
# This will create files like:
# config/guilds/guild-123456789.yml
# config/guilds/guild-987654321.yml
# etc.
```
**Migration Output Example:**
```
🔄 Starting migration from database to files...
📦 Existing files will be backed up
📊 Migration Results:
✅ Migrated: 3 guilds
❌ Failed: 0 guilds
⏭️ Skipped: 0 guilds
📝 Banned words migrated: 45
✅ Successfully migrated guilds:
• 123456789: My Gaming Server (12 banned words)
• 987654321: Friends Chat (8 banned words)
• 555666777: Test Server (0 banned words)
```
### Step 3: Verify Migration
Check that the migration was successful:
```bash
# Verify all guilds
python -m guardden.cli.config migrate verify
# Or verify specific guilds
python -m guardden.cli.config migrate verify 123456789 987654321
```
### Step 4: Review Generated Configurations
Examine the generated configuration files:
```bash
# List all configurations
python -m guardden.cli.config guild list
# Validate configurations
python -m guardden.cli.config guild validate
```
**Example Generated Configuration:**
```yaml
# config/guilds/guild-123456789.yml
guild_id: 123456789
name: "My Gaming Server"
owner_id: 987654321
premium: false
settings:
general:
prefix: "!"
locale: "en"
ai_moderation:
enabled: true
sensitivity: 80
nsfw_only_filtering: false # ← Your new NSFW-only feature!
confidence_threshold: 0.7
automod:
message_rate_limit: 5
scam_allowlist:
- "discord.com"
- "github.com"
# Migrated banned words (if any)
banned_words:
- pattern: "spam"
action: delete
is_regex: false
reason: "Anti-spam filter"
```
### Step 5: Customize Your Configuration
Now you can edit the YAML files directly or use the CLI:
```bash
# Enable NSFW-only filtering (only block sexual content)
python -m guardden.cli.config guild edit 123456789 ai_moderation.nsfw_only_filtering true
# Adjust AI sensitivity
python -m guardden.cli.config guild edit 123456789 ai_moderation.sensitivity 75
# Validate changes
python -m guardden.cli.config guild validate 123456789
```
**Or edit files directly:**
```yaml
# Edit config/guilds/guild-123456789.yml
ai_moderation:
enabled: true
sensitivity: 75 # Changed from 80
nsfw_only_filtering: true # Changed from false
confidence_threshold: 0.7
```
### Step 6: Test the New System
1. **Restart GuardDen** to load the file-based configuration:
```bash
python -m guardden
```
2. **Test hot-reloading** by editing a config file:
```bash
# Edit a setting in config/guilds/guild-123456789.yml
# Changes should apply within seconds (check bot logs)
```
3. **Verify settings in Discord** using read-only commands:
```
!config # View current settings
!ai # View AI moderation settings
!automod # View automod settings
```
### Step 7: Manage Wordlists (Optional)
Review and customize wordlist configurations:
```bash
# View wordlist status
python -m guardden.cli.config wordlist info
# Edit wordlists directly:
nano config/wordlists/banned-words.yml
nano config/wordlists/domain-allowlists.yml
nano config/wordlists/external-sources.yml
```
## Post-Migration Tasks
### Backup Your Configuration
```bash
# Create backups of specific guilds
python -m guardden.cli.config guild backup 123456789
# Or backup the entire config directory
cp -r config config-backup-$(date +%Y%m%d)
```
### Version Control Setup
Add configuration to Git for version tracking:
```bash
# Initialize Git repository (if not already)
git init
git add config/
git commit -m "Add GuardDen file-based configuration"
# Create .gitignore to exclude backups
echo "config/backups/" >> .gitignore
```
### Clean Up Database (Optional)
**⚠️ WARNING: Only do this AFTER verifying migration is successful!**
```bash
# This permanently deletes old configuration from database
python -c "
import asyncio
from guardden.services.config_migration import ConfigurationMigrator
from guardden.services.database import Database
from guardden.services.guild_config import GuildConfigService
from guardden.services.file_config import FileConfigurationManager
async def cleanup():
db = Database('your-db-url')
guild_service = GuildConfigService(db)
file_manager = FileConfigurationManager()
migrator = ConfigurationMigrator(db, guild_service, file_manager)
# ⚠️ This deletes all guild settings from database
results = await migrator.cleanup_database_configs(confirm=True)
print(f'Cleaned up: {results}')
await db.close()
asyncio.run(cleanup())
"
```
## Troubleshooting
### Common Issues
**1. Migration Failed for Some Guilds**
```bash
# Check the specific error messages
python -m guardden.cli.config migrate from-database
# Try migrating individual guilds if needed
# (This may require manual file creation)
```
**2. Configuration Validation Errors**
```bash
# Validate and see specific errors
python -m guardden.cli.config guild validate
# Common fixes:
# - Check YAML syntax (indentation, colons, quotes)
# - Verify Discord IDs are numbers, not strings
# - Ensure boolean values are true/false, not True/False
```
**3. Hot-Reload Not Working**
- Check bot logs for configuration errors
- Ensure YAML syntax is correct
- Verify file permissions are readable
- Restart bot if needed: `python -m guardden`
**4. Lost Configuration During Migration**
- Check `config/backups/` directory for backup files
- Database configurations are preserved during migration
- Re-run migration if needed: `python -m guardden.cli.config migrate from-database`
### Getting Help
**View CLI Help:**
```bash
python -m guardden.cli.config --help
python -m guardden.cli.config guild --help
python -m guardden.cli.config migrate --help
```
**Check Configuration Status:**
```bash
python -m guardden.cli.config guild list
python -m guardden.cli.config guild validate
python -m guardden.cli.config wordlist info
```
**Backup and Recovery:**
```bash
# Create backup before making changes
python -m guardden.cli.config guild backup <guild_id>
# Recovery from backup (manual file copy)
cp config/backups/guild-123456789_20260124_123456.yml config/guilds/guild-123456789.yml
```
## Configuration Examples
### NSFW-Only Filtering Setup
For gaming communities that want to allow violence but block sexual content:
```yaml
# config/guilds/guild-123456789.yml
ai_moderation:
enabled: true
sensitivity: 80
nsfw_only_filtering: true # Only block sexual content
confidence_threshold: 0.7
nsfw_detection_enabled: true
log_only: false
```
### High-Security Server Setup
For family-friendly or professional servers:
```yaml
ai_moderation:
enabled: true
sensitivity: 95 # Very strict
nsfw_only_filtering: false # Block all inappropriate content
confidence_threshold: 0.6 # Lower threshold = more sensitive
log_only: false
automod:
message_rate_limit: 3 # Stricter rate limiting
message_rate_window: 5
duplicate_threshold: 2 # Less tolerance for duplicates
```
### Development/Testing Server Setup
For development or testing environments:
```yaml
ai_moderation:
enabled: true
sensitivity: 50 # More lenient
nsfw_only_filtering: false
confidence_threshold: 0.8 # Higher threshold = less sensitive
log_only: true # Only log, don't take action
automod:
message_rate_limit: 10 # More relaxed limits
message_rate_window: 5
```
## Benefits of File-Based Configuration
After migration, you'll enjoy:
1. **Easy Bulk Changes**: Edit multiple server configs at once
2. **Configuration as Code**: Version control your bot settings
3. **Environment Management**: Different configs for dev/staging/prod
4. **Disaster Recovery**: Easy backup and restore of all settings
5. **No Discord Dependency**: Configure servers before bot joins
6. **Better Organization**: All settings in structured, documented files
7. **Hot-Reloading**: Changes apply instantly without restarts
8. **Schema Validation**: Automatic error checking prevents misconfigurations
**Welcome to the new GuardDen configuration system! 🎉**

292
README.md
View File

@@ -141,6 +141,162 @@ GuardDen is a comprehensive Discord moderation bot designed to protect your comm
## Configuration ## Configuration
GuardDen now supports **file-based configuration** as the primary method for managing bot settings. This replaces Discord commands for configuration, providing better version control, easier management, and more reliable deployments.
### File-Based Configuration (Recommended)
#### Directory Structure
```
config/
├── guilds/
│ ├── guild-123456789.yml # Per-server configuration
│ ├── guild-987654321.yml
│ └── default-template.yml # Template for new servers
├── wordlists/
│ ├── banned-words.yml # Custom banned words
│ ├── domain-allowlists.yml # Allowed domains whitelist
│ └── external-sources.yml # Managed wordlist sources
├── schemas/
│ ├── guild-schema.yml # Configuration validation
│ └── wordlists-schema.yml
└── templates/
└── guild-default.yml # Default configuration template
```
#### Quick Start with File Configuration
1. **Create your first server configuration:**
```bash
python -m guardden.cli.config guild create 123456789012345678 "My Discord Server"
```
2. **Edit the configuration file:**
```bash
nano config/guilds/guild-123456789012345678.yml
```
3. **Customize settings (example):**
```yaml
# Basic server information
guild_id: 123456789012345678
name: "My Discord Server"
settings:
# AI Moderation
ai_moderation:
enabled: true
sensitivity: 80 # 0-100 (higher = stricter)
nsfw_only_filtering: true # Only block sexual content, allow violence
# Automod settings
automod:
message_rate_limit: 5 # Max messages per 5 seconds
scam_allowlist:
- "discord.com"
- "github.com"
```
4. **Validate your configuration:**
```bash
python -m guardden.cli.config guild validate 123456789012345678
```
5. **Start the bot** (configurations auto-reload):
```bash
python -m guardden
```
#### Configuration Management CLI
**Guild Management:**
```bash
# List all configured servers
python -m guardden.cli.config guild list
# Create new server configuration
python -m guardden.cli.config guild create <guild_id> "Server Name"
# Edit specific settings
python -m guardden.cli.config guild edit <guild_id> ai_moderation.sensitivity 75
python -m guardden.cli.config guild edit <guild_id> ai_moderation.nsfw_only_filtering true
# Validate configurations
python -m guardden.cli.config guild validate
python -m guardden.cli.config guild validate <guild_id>
# Backup configuration
python -m guardden.cli.config guild backup <guild_id>
```
**Migration from Discord Commands:**
```bash
# Export existing Discord command settings to files
python -m guardden.cli.config migrate from-database
# Verify migration was successful
python -m guardden.cli.config migrate verify
```
**Wordlist Management:**
```bash
# View wordlist status
python -m guardden.cli.config wordlist info
# View available templates
python -m guardden.cli.config template info
```
#### Key Configuration Options
**AI Moderation Settings:**
```yaml
ai_moderation:
enabled: true # Enable AI content analysis
sensitivity: 80 # 0-100 scale (higher = stricter)
confidence_threshold: 0.7 # 0.0-1.0 confidence required
nsfw_only_filtering: false # true = only sexual content, false = all content
log_only: false # true = log only, false = take action
```
**NSFW-Only Filtering Guide:**
- `false` = Block ALL inappropriate content (sexual, violence, harassment, hate speech)
- `true` = Only block sexual/nude content, allow violence and other content types
**Automod Configuration:**
```yaml
automod:
message_rate_limit: 5 # Max messages per time window
message_rate_window: 5 # Time window in seconds
duplicate_threshold: 3 # Duplicate messages to trigger
scam_allowlist: # Domains that bypass scam detection
- "discord.com"
- "github.com"
```
**Banned Words Management:**
Edit `config/wordlists/banned-words.yml`:
```yaml
global_patterns:
- pattern: "badword"
action: delete
is_regex: false
category: profanity
guild_patterns:
123456789: # Specific server overrides
- pattern: "server-specific-rule"
action: warn
override_global: false
```
#### Hot-Reloading
Configuration changes are automatically detected and applied without restarting the bot:
- ✅ Edit YAML files directly
- ✅ Changes apply within seconds
- ✅ Invalid configs are rejected with error logs
- ✅ Automatic rollback on errors
### Environment Variables ### Environment Variables
| Variable | Description | Default | | Variable | Description | Default |
@@ -170,19 +326,29 @@ GuardDen is a comprehensive Discord moderation bot designed to protect your comm
### Per-Guild Settings ### Per-Guild Settings
Each server can configure: Each server can be configured via YAML files in `config/guilds/`:
- Command prefix
- Log channels (general and moderation) **General Settings:**
- Welcome channel - Command prefix and locale
- Mute role and verified role - Channel IDs (log, moderation, welcome)
- Automod toggles (spam, links, banned words) - Role IDs (mute, verified, moderator)
- Automod thresholds and scam allowlist
- Strike action thresholds **Content Moderation:**
- AI moderation settings (enabled, sensitivity, confidence threshold, log-only, NSFW detection, NSFW-only mode) - AI moderation (enabled, sensitivity, NSFW-only mode)
- Verification settings (type, enabled) - Automod thresholds and rate limits
- Banned words and domain allowlists
- Strike system and escalation actions
**Member Verification:**
- Verification challenges (button, captcha, math, emoji)
- Auto-role assignment
**All settings support hot-reloading** - edit files and changes apply immediately!
## Commands ## Commands
> **Note:** Configuration commands (`!config`, `!ai`, `!automod`, etc.) have been replaced with file-based configuration. See the [Configuration](#configuration) section above for managing settings via YAML files and the CLI tool.
### Moderation ### Moderation
| Command | Permission | Description | | Command | Permission | Description |
@@ -198,56 +364,49 @@ Each server can configure:
| `!purge <amount>` | Manage Messages | Delete multiple messages (max 100) | | `!purge <amount>` | Manage Messages | Delete multiple messages (max 100) |
| `!modlogs <user>` | Kick Members | View moderation history | | `!modlogs <user>` | Kick Members | View moderation history |
### Configuration (Admin only) ### Configuration Management
Configuration is now managed via **YAML files** instead of Discord commands. Use the CLI tool:
```bash
# Configuration Management CLI
python -m guardden.cli.config guild create <guild_id> "Server Name"
python -m guardden.cli.config guild list
python -m guardden.cli.config guild edit <guild_id> <setting> <value>
python -m guardden.cli.config guild validate [guild_id]
# Migration from old Discord commands
python -m guardden.cli.config migrate from-database
python -m guardden.cli.config migrate verify
# Wordlist management
python -m guardden.cli.config wordlist info
```
**Read-only Status Commands (Still Available):**
| Command | Description | | Command | Description |
|---------|-------------| |---------|-------------|
| `!config` | View current configuration | | `!config` | View current configuration (read-only) |
| `!config prefix <prefix>` | Set command prefix | | `!ai` | View AI moderation settings (read-only) |
| `!config logchannel [#channel]` | Set general log channel | | `!automod` | View automod status (read-only) |
| `!config modlogchannel [#channel]` | Set moderation log channel | | `!bannedwords` | List banned words (read-only) |
| `!config welcomechannel [#channel]` | Set welcome channel |
| `!config muterole [@role]` | Set mute role |
| `!config automod <true/false>` | Toggle automod |
| `!config antispam <true/false>` | Toggle anti-spam |
| `!config linkfilter <true/false>` | Toggle link filtering |
### Banned Words **Configuration Examples:**
| Command | Description | ```bash
|---------|-------------| # Set AI sensitivity to 75 (0-100 scale)
| `!bannedwords` | List all banned words | python -m guardden.cli.config guild edit 123456789 ai_moderation.sensitivity 75
| `!bannedwords add <word> [action] [is_regex]` | Add a banned word |
| `!bannedwords remove <id>` | Remove a banned word by ID |
Managed wordlists are synced weekly by default. You can override sources with # Enable NSFW-only filtering (only block sexual content)
`GUARDDEN_WORDLIST_SOURCES` (JSON array) or disable syncing entirely with python -m guardden.cli.config guild edit 123456789 ai_moderation.nsfw_only_filtering true
`GUARDDEN_WORDLIST_ENABLED=false`.
### Automod # Add domain to scam allowlist
Edit config/wordlists/domain-allowlists.yml
| Command | Description | # Add banned word pattern
|---------|-------------| Edit config/wordlists/banned-words.yml
| `!automod` | View automod status | ```
| `!automod test <text>` | Test text against filters |
| `!automod threshold <setting> <value>` | Update a single automod threshold |
| `!automod allowlist` | List allowlisted domains |
| `!automod allowlist add <domain>` | Add a domain to the allowlist |
| `!automod allowlist remove <domain>` | Remove a domain from the allowlist |
### AI Moderation (Admin only)
| Command | Description |
|---------|-------------|
| `!ai` | View AI moderation settings |
| `!ai enable` | Enable AI moderation |
| `!ai disable` | Disable AI moderation |
| `!ai sensitivity <0-100>` | Set AI sensitivity level |
| `!ai threshold <0.0-1.0>` | Set AI confidence threshold |
| `!ai logonly <true/false>` | Toggle AI log-only mode |
| `!ai nsfw <true/false>` | Toggle NSFW image detection |
| `!ai nsfwonly <true/false>` | Toggle NSFW-only filtering mode |
| `!ai analyze <text>` | Test AI analysis on text |
### Diagnostics (Admin only) ### Diagnostics (Admin only)
@@ -295,7 +454,7 @@ guardden/
│ ├── bot.py # Main bot class │ ├── bot.py # Main bot class
│ ├── config.py # Settings management │ ├── config.py # Settings management
│ ├── cogs/ # Discord command groups │ ├── cogs/ # Discord command groups
│ │ ├── admin.py # Configuration commands │ │ ├── admin.py # Configuration commands (read-only)
│ │ ├── ai_moderation.py # AI-powered moderation │ │ ├── ai_moderation.py # AI-powered moderation
│ │ ├── automod.py # Automatic moderation │ │ ├── automod.py # Automatic moderation
│ │ ├── events.py # Event logging │ │ ├── events.py # Event logging
@@ -304,18 +463,29 @@ guardden/
│ ├── models/ # Database models │ ├── models/ # Database models
│ │ ├── guild.py # Guild settings, banned words │ │ ├── guild.py # Guild settings, banned words
│ │ └── moderation.py # Logs, strikes, notes │ │ └── moderation.py # Logs, strikes, notes
── services/ # Business logic ── services/ # Business logic
├── ai/ # AI provider implementations ├── ai/ # AI provider implementations
├── automod.py # Content filtering ├── automod.py # Content filtering
├── database.py # DB connections ├── database.py # DB connections
├── guild_config.py # Config caching ├── guild_config.py # Config caching
├── ratelimit.py # Rate limiting ├── file_config.py # File-based configuration system
── verification.py # Verification challenges ── config_migration.py # Migration from DB to files
│ │ ├── ratelimit.py # Rate limiting
│ │ └── verification.py # Verification challenges
│ └── cli/ # Command-line tools
│ └── config.py # Configuration management CLI
├── config/ # File-based configuration
│ ├── guilds/ # Per-server configuration files
│ ├── wordlists/ # Banned words and allowlists
│ ├── schemas/ # Configuration validation schemas
│ └── templates/ # Configuration templates
├── tests/ # Test suite ├── tests/ # Test suite
├── migrations/ # Database migrations ├── migrations/ # Database migrations
├── dashboard/ # Web dashboard (FastAPI + React) ├── dashboard/ # Web dashboard (FastAPI + React)
├── docker-compose.yml # Docker deployment ├── docker-compose.yml # Docker deployment
── pyproject.toml # Dependencies ── pyproject.toml # Dependencies
├── README.md # This file
└── MIGRATION.md # Migration guide for file-based config
``` ```
## Verification System ## Verification System

View File

@@ -0,0 +1,149 @@
# Example Guild Configuration
# Copy this file to guild-{YOUR_GUILD_ID}.yml and customize
# Basic Guild Information
guild_id: 123456789012345678 # Replace with your Discord server ID
name: "Example Gaming Community" # Your server name
owner_id: 987654321098765432 # Guild owner's Discord user ID
premium: false # Set to true if you have premium features
settings:
# General Settings
general:
prefix: "!" # Command prefix (for read-only commands)
locale: "en" # Language code
# Discord Channel Configuration (use actual channel IDs or null)
channels:
log_channel_id: null # General event logging
mod_log_channel_id: 888999000111222333 # Moderation action logging
welcome_channel_id: null # New member welcome messages
# Discord Role Configuration (use actual role IDs or null)
roles:
mute_role_id: 444555666777888999 # Role for timed-out members
verified_role_id: 111222333444555666 # Role given after verification
mod_role_ids: # List of moderator role IDs
- 777888999000111222
- 333444555666777888
# Moderation System Configuration
moderation:
automod_enabled: true # Enable automatic moderation
anti_spam_enabled: true # Enable anti-spam protection
link_filter_enabled: true # Enable suspicious link detection
# Strike System - Actions at different strike thresholds
strike_actions:
"1": # At 1 strike: warn user
action: warn
"3": # At 3 strikes: 1 hour timeout
action: timeout
duration: 3600
"5": # At 5 strikes: kick from server
action: kick
"7": # At 7 strikes: ban from server
action: ban
# Automatic Moderation Thresholds
automod:
# Message Rate Limiting
message_rate_limit: 5 # Max messages per time window
message_rate_window: 5 # Time window in seconds
duplicate_threshold: 3 # Duplicate messages to trigger action
# Mention Spam Protection
mention_limit: 5 # Max mentions per single message
mention_rate_limit: 10 # Max mentions per time window
mention_rate_window: 60 # Mention time window in seconds
# Scam Protection - Domains that bypass scam detection
scam_allowlist:
- "discord.com" # Official Discord
- "github.com" # Code repositories
- "youtube.com" # Video platform
- "imgur.com" # Image hosting
- "steam.com" # Gaming platform
# Add your trusted domains here
# AI-Powered Content Moderation
ai_moderation:
enabled: true # Enable AI content analysis
sensitivity: 75 # AI sensitivity (0-100, higher = stricter)
confidence_threshold: 0.7 # Minimum confidence to take action (0.0-1.0)
log_only: false # Only log violations vs take action
nsfw_detection_enabled: true # Enable NSFW image detection
# NSFW-Only Filtering Mode (NEW FEATURE!)
nsfw_only_filtering: true # true = Only block sexual content
# false = Block all inappropriate content
# Member Verification System
verification:
enabled: false # Enable verification for new members
type: "captcha" # Verification type: button, captcha, math, emoji
# Guild-Specific Banned Words (optional)
# These are in addition to patterns in config/wordlists/banned-words.yml
banned_words:
- pattern: "guild-specific-word"
action: delete
is_regex: false
reason: "Server-specific rule"
category: harassment
- pattern: "sp[a4]m.*bot"
action: timeout
is_regex: true # This is a regex pattern
reason: "Spam bot detection"
category: spam
# Configuration Notes and Examples:
#
# === NSFW-ONLY FILTERING EXPLAINED ===
# This is perfect for gaming communities that discuss violence but want to block sexual content:
#
# nsfw_only_filtering: true
# ✅ BLOCKS: Sexual content, nude images, explicit material
# ❌ ALLOWS: Violence, gore, harassment, hate speech, self-harm discussions
#
# nsfw_only_filtering: false
# ✅ BLOCKS: All inappropriate content (sexual, violence, harassment, hate speech, etc.)
#
# === AI SENSITIVITY GUIDE ===
# 0-30 = Very lenient (only extreme violations)
# 31-50 = Lenient (clear violations only)
# 51-70 = Balanced (moderate detection) - RECOMMENDED
# 71-85 = Strict (catches most potential issues)
# 86-100 = Very strict (may have false positives)
#
# === VERIFICATION TYPES ===
# button = Simple button click (easiest for users)
# captcha = Text-based captcha entry (more secure)
# math = Solve simple math problem (educational)
# emoji = Select correct emoji from options (fun)
#
# === AUTOMOD ACTIONS ===
# warn = Send warning message to user
# delete = Delete the offending message
# timeout = Temporarily mute user (requires duration)
# kick = Remove user from server (can rejoin)
# ban = Permanently ban user from server
#
# === CONFIGURATION TIPS ===
# 1. Start with balanced settings and adjust based on your community
# 2. Use nsfw_only_filtering: true for gaming/mature discussion servers
# 3. Set higher sensitivity (80+) for family-friendly servers
# 4. Test settings with !ai analyze "test message" command
# 5. Monitor mod logs to tune your settings
# 6. Back up your config: python -m guardden.cli.config guild backup {guild_id}
#
# === HOT-RELOAD TESTING ===
# Edit this file and save - changes apply within seconds!
# Watch the bot logs to see configuration reload messages.
# Use "!config" in Discord to verify your settings loaded correctly.
#
# === GETTING HELP ===
# Run: python -m guardden.cli.config --help
# Validate: python -m guardden.cli.config guild validate {guild_id}
# Check status: python -m guardden.cli.config guild list

View File

@@ -0,0 +1,224 @@
# Guild Configuration Schema
# This defines the structure and validation rules for guild configurations
type: object
required:
- guild_id
- name
- settings
properties:
guild_id:
type: integer
description: "Discord guild (server) ID"
minimum: 1
name:
type: string
description: "Human-readable guild name"
maxLength: 100
owner_id:
type: integer
description: "Guild owner's Discord user ID"
minimum: 1
premium:
type: boolean
description: "Whether this guild has premium features"
default: false
settings:
type: object
required:
- general
- channels
- roles
- moderation
- automod
- ai_moderation
- verification
properties:
general:
type: object
properties:
prefix:
type: string
description: "Command prefix"
minLength: 1
maxLength: 10
default: "!"
locale:
type: string
description: "Language locale"
pattern: "^[a-z]{2}$"
default: "en"
channels:
type: object
description: "Channel configuration (Discord channel IDs)"
properties:
log_channel_id:
type: [integer, "null"]
description: "General event log channel"
minimum: 1
mod_log_channel_id:
type: [integer, "null"]
description: "Moderation action log channel"
minimum: 1
welcome_channel_id:
type: [integer, "null"]
description: "Welcome message channel"
minimum: 1
roles:
type: object
description: "Role configuration (Discord role IDs)"
properties:
mute_role_id:
type: [integer, "null"]
description: "Role for timed-out members"
minimum: 1
verified_role_id:
type: [integer, "null"]
description: "Role given after verification"
minimum: 1
mod_role_ids:
type: array
description: "Moderator roles"
items:
type: integer
minimum: 1
default: []
moderation:
type: object
properties:
automod_enabled:
type: boolean
description: "Enable automatic moderation"
default: true
anti_spam_enabled:
type: boolean
description: "Enable anti-spam protection"
default: true
link_filter_enabled:
type: boolean
description: "Enable link filtering"
default: false
strike_actions:
type: object
description: "Actions to take at strike thresholds"
patternProperties:
"^[0-9]+$":
type: object
required: [action]
properties:
action:
type: string
enum: [warn, timeout, kick, ban]
duration:
type: integer
minimum: 1
description: "Duration in seconds (for timeout/ban)"
default:
"1": {action: warn}
"3": {action: timeout, duration: 3600}
"5": {action: kick}
"7": {action: ban}
automod:
type: object
description: "Automatic moderation settings"
properties:
message_rate_limit:
type: integer
minimum: 1
maximum: 50
description: "Messages per time window"
default: 5
message_rate_window:
type: integer
minimum: 1
maximum: 300
description: "Time window in seconds"
default: 5
duplicate_threshold:
type: integer
minimum: 1
maximum: 20
description: "Duplicate messages to trigger action"
default: 3
mention_limit:
type: integer
minimum: 1
maximum: 50
description: "Maximum mentions per message"
default: 5
mention_rate_limit:
type: integer
minimum: 1
maximum: 100
description: "Mentions per time window"
default: 10
mention_rate_window:
type: integer
minimum: 1
maximum: 3600
description: "Mention time window in seconds"
default: 60
scam_allowlist:
type: array
description: "Domains allowed to bypass scam detection"
items:
type: string
pattern: "^[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"
default: []
ai_moderation:
type: object
description: "AI-powered moderation settings"
properties:
enabled:
type: boolean
description: "Enable AI moderation"
default: true
sensitivity:
type: integer
minimum: 0
maximum: 100
description: "AI sensitivity level (higher = stricter)"
default: 80
confidence_threshold:
type: number
minimum: 0.0
maximum: 1.0
description: "Minimum confidence to take action"
default: 0.7
log_only:
type: boolean
description: "Only log violations, don't take action"
default: false
nsfw_detection_enabled:
type: boolean
description: "Enable NSFW image detection"
default: true
nsfw_only_filtering:
type: boolean
description: "Only filter sexual content, allow violence/harassment"
default: false
verification:
type: object
description: "Member verification settings"
properties:
enabled:
type: boolean
description: "Enable verification for new members"
default: false
type:
type: string
enum: [button, captcha, math, emoji]
description: "Verification challenge type"
default: button

View File

@@ -0,0 +1,175 @@
# Wordlists Configuration Schema
# Defines structure for banned words and domain whitelists
banned_words:
type: object
description: "Banned words and patterns configuration"
properties:
global_patterns:
type: array
description: "Patterns applied to all guilds (unless overridden)"
items:
type: object
required: [pattern, action]
properties:
pattern:
type: string
description: "Word or regex pattern to match"
minLength: 1
maxLength: 200
action:
type: string
enum: [delete, warn, strike, timeout]
description: "Action to take when pattern matches"
is_regex:
type: boolean
description: "Whether pattern is a regular expression"
default: false
reason:
type: string
description: "Reason for this rule"
maxLength: 500
category:
type: string
description: "Category of banned content"
enum: [profanity, hate_speech, spam, scam, harassment, sexual, violence]
severity:
type: integer
minimum: 1
maximum: 10
description: "Severity level (1-10)"
default: 5
enabled:
type: boolean
description: "Whether this rule is active"
default: true
guild_patterns:
type: object
description: "Guild-specific pattern overrides"
patternProperties:
"^[0-9]+$": # Guild ID
type: array
items:
type: object
required: [pattern, action]
properties:
pattern:
type: string
minLength: 1
maxLength: 200
action:
type: string
enum: [delete, warn, strike, timeout]
is_regex:
type: boolean
default: false
reason:
type: string
maxLength: 500
category:
type: string
enum: [profanity, hate_speech, spam, scam, harassment, sexual, violence]
severity:
type: integer
minimum: 1
maximum: 10
default: 5
enabled:
type: boolean
default: true
override_global:
type: boolean
description: "Whether this rule overrides global patterns"
default: false
external_sources:
type: object
description: "External wordlist sources configuration"
properties:
sources:
type: array
items:
type: object
required: [name, url, category, action]
properties:
name:
type: string
description: "Unique identifier for this source"
pattern: "^[a-zA-Z0-9_-]+$"
url:
type: string
description: "URL to fetch wordlist from"
format: uri
category:
type: string
enum: [profanity, hate_speech, spam, scam, harassment, sexual, violence]
action:
type: string
enum: [delete, warn, strike, timeout]
reason:
type: string
description: "Default reason for words from this source"
enabled:
type: boolean
default: true
update_interval_hours:
type: integer
minimum: 1
maximum: 8760 # 1 year
description: "How often to update from source"
default: 168 # 1 week
applies_to_guilds:
type: array
description: "Guild IDs to apply this source to (empty = all guilds)"
items:
type: integer
minimum: 1
default: []
domain_allowlists:
type: object
description: "Domain whitelist configuration"
properties:
global_allowlist:
type: array
description: "Domains allowed for all guilds"
items:
type: object
required: [domain]
properties:
domain:
type: string
description: "Domain name to allow"
pattern: "^[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"
reason:
type: string
description: "Why this domain is allowed"
added_by:
type: string
description: "Who added this domain"
added_date:
type: string
format: date-time
description: "When this domain was added"
guild_allowlists:
type: object
description: "Guild-specific domain allowlists"
patternProperties:
"^[0-9]+$": # Guild ID
type: array
items:
type: object
required: [domain]
properties:
domain:
type: string
pattern: "^[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"
reason:
type: string
added_by:
type: string
added_date:
type: string
format: date-time

View File

@@ -0,0 +1,102 @@
# Default Guild Configuration Template
# Copy this file to config/guilds/guild-{GUILD_ID}.yml and customize
# Guild Information
guild_id: 123456789012345678 # Replace with your Discord server ID
name: "My Discord Server" # Replace with your server name
owner_id: 987654321098765432 # Replace with owner's Discord user ID
premium: false # Set to true if you have premium features
settings:
# General Settings
general:
prefix: "!" # Command prefix (if keeping read-only commands)
locale: "en" # Language code (en, es, fr, de, etc.)
# Channel Configuration (Discord Channel IDs)
# Set to null to disable, or use actual channel IDs
channels:
log_channel_id: null # General event logging
mod_log_channel_id: null # Moderation action logging
welcome_channel_id: null # New member welcome messages
# Role Configuration (Discord Role IDs)
roles:
mute_role_id: null # Role for timed-out members
verified_role_id: null # Role given after verification
mod_role_ids: [] # List of moderator role IDs
# Moderation Settings
moderation:
automod_enabled: true # Enable automatic moderation
anti_spam_enabled: true # Enable anti-spam protection
link_filter_enabled: false # Enable suspicious link filtering
# Strike System - Actions taken when users reach strike thresholds
strike_actions:
"1": # At 1 strike
action: warn
"3": # At 3 strikes
action: timeout
duration: 3600 # 1 hour timeout
"5": # At 5 strikes
action: kick
"7": # At 7 strikes
action: ban
# Automatic Moderation Thresholds
automod:
# Message Rate Limiting
message_rate_limit: 5 # Max messages per time window
message_rate_window: 5 # Time window in seconds
duplicate_threshold: 3 # Duplicate messages to trigger action
# Mention Spam Protection
mention_limit: 5 # Max mentions per message
mention_rate_limit: 10 # Max mentions per time window
mention_rate_window: 60 # Mention time window in seconds
# Scam Protection - Domains allowed to bypass scam detection
scam_allowlist:
- "discord.com" # Example: Allow Discord links
- "github.com" # Example: Allow GitHub links
# Add trusted domains here
# AI-Powered Moderation
ai_moderation:
enabled: true # Enable AI content analysis
sensitivity: 80 # AI sensitivity (0-100, higher = stricter)
confidence_threshold: 0.7 # Minimum confidence to take action (0.0-1.0)
log_only: false # Only log violations (true) or take action (false)
nsfw_detection_enabled: true # Enable NSFW image detection
nsfw_only_filtering: false # Only filter sexual content (true) vs all content (false)
# Member Verification System
verification:
enabled: false # Enable verification for new members
type: "button" # Verification type: button, captcha, math, emoji
# Configuration Notes:
#
# NSFW-Only Filtering:
# false = Block all inappropriate content (sexual, violence, harassment, hate speech)
# true = Only block sexual/nude content, allow violence and harassment
#
# AI Sensitivity Guide:
# 0-30 = Very lenient (only extreme violations)
# 31-50 = Lenient (clear violations)
# 51-70 = Balanced (moderate detection)
# 71-85 = Strict (catches most issues)
# 86-100 = Very strict (may have false positives)
#
# Verification Types:
# button = Simple button click (easiest)
# captcha = Text-based captcha entry
# math = Solve simple math problem
# emoji = Select correct emoji from options
#
# Strike Actions:
# warn = Send warning message
# timeout = Temporarily mute user (requires duration in seconds)
# kick = Remove user from server (can rejoin)
# ban = Permanently ban user from server

View File

@@ -0,0 +1,95 @@
# Banned Words Configuration
# Manage blocked words and patterns for content filtering
# Global patterns applied to all guilds (unless overridden)
global_patterns:
# Basic profanity filter
- pattern: "badword1"
action: delete
is_regex: false
reason: "Basic profanity filter"
category: profanity
severity: 5
enabled: true
- pattern: "badword2"
action: warn
is_regex: false
reason: "Mild profanity"
category: profanity
severity: 3
enabled: true
# Regex example for variations
- pattern: "sp[a4]mm*[i1]ng"
action: delete
is_regex: true
reason: "Spam pattern detection"
category: spam
severity: 7
enabled: true
# Hate speech prevention
- pattern: "hate.*speech.*example"
action: timeout
is_regex: true
reason: "Hate speech filter"
category: hate_speech
severity: 9
enabled: true
# Guild-specific pattern overrides
# Use your Discord server ID as the key
guild_patterns:
123456789012345678: # Replace with actual guild ID
- pattern: "guild-specific-word"
action: warn
is_regex: false
reason: "Server-specific rule"
category: harassment
severity: 4
enabled: true
override_global: false
- pattern: "allowed-here"
action: delete
is_regex: false
reason: "Disable global pattern for this guild"
category: profanity
severity: 1
enabled: false # Disabled = allows the word in this guild
override_global: true # Overrides global patterns
# Add more guild IDs as needed
# 987654321098765432:
# - pattern: "another-server-rule"
# action: strike
# [...]
# Configuration Notes:
#
# Actions Available:
# delete = Delete the message immediately
# warn = Send warning to user and log
# strike = Add strike to user (triggers escalation)
# timeout = Temporarily mute user
#
# Regex Patterns:
# is_regex: true allows advanced pattern matching
# Examples:
# - "hell+o+" matches "hello", "helllo", "helloooo"
# - "[a4]dmin" matches "admin" or "4dmin"
# - "spam.*bot" matches "spam bot", "spambot", "spam detection bot"
#
# Categories:
# profanity, hate_speech, spam, scam, harassment, sexual, violence
#
# Severity (1-10):
# 1-3 = Mild violations (warnings)
# 4-6 = Moderate violations (delete message)
# 7-8 = Serious violations (timeout)
# 9-10 = Severe violations (kick/ban)
#
# Override Global:
# false = Use this rule in addition to global patterns
# true = This rule replaces global patterns for this guild

View File

@@ -0,0 +1,99 @@
# Domain Allowlists Configuration
# Configure domains that bypass scam/phishing detection
# Global allowlist - applies to all guilds
global_allowlist:
- domain: "discord.com"
reason: "Official Discord domain"
added_by: "system"
added_date: "2026-01-24T00:00:00Z"
- domain: "github.com"
reason: "Popular code repository platform"
added_by: "admin"
added_date: "2026-01-24T00:00:00Z"
- domain: "youtube.com"
reason: "Popular video platform"
added_by: "admin"
added_date: "2026-01-24T00:00:00Z"
- domain: "youtu.be"
reason: "YouTube short links"
added_by: "admin"
added_date: "2026-01-24T00:00:00Z"
- domain: "imgur.com"
reason: "Popular image hosting"
added_by: "admin"
added_date: "2026-01-24T00:00:00Z"
- domain: "reddit.com"
reason: "Popular discussion platform"
added_by: "admin"
added_date: "2026-01-24T00:00:00Z"
- domain: "wikipedia.org"
reason: "Educational content"
added_by: "admin"
added_date: "2026-01-24T00:00:00Z"
# Guild-specific allowlists
# Use your Discord server ID as the key
guild_allowlists:
123456789012345678: # Replace with actual guild ID
- domain: "example-gaming-site.com"
reason: "Popular gaming community site"
added_by: "guild_admin"
added_date: "2026-01-24T00:00:00Z"
- domain: "guild-specific-forum.com"
reason: "Guild's official forum"
added_by: "guild_owner"
added_date: "2026-01-24T00:00:00Z"
# Add more guild IDs as needed
# 987654321098765432:
# - domain: "another-server-site.com"
# reason: "Server-specific trusted site"
# added_by: "admin"
# added_date: "2026-01-24T00:00:00Z"
# Configuration Notes:
#
# Domain Format:
# - Use base domain only (e.g., "example.com" not "https://www.example.com/path")
# - Subdomains are automatically included (allowing "example.com" also allows "www.example.com")
# - Do not include protocols (http/https) or paths
#
# Why Allowlist Domains:
# - Prevent false positives in scam detection
# - Allow trusted community sites and resources
# - Whitelist official platforms and services
# - Support educational and reference materials
#
# Security Considerations:
# - Only add domains you trust completely
# - Regularly review and update the list
# - Remove domains that become compromised
# - Be cautious with URL shorteners
#
# Common Domains to Consider:
# - Social platforms: twitter.com, instagram.com, tiktok.com
# - Gaming: steam.com, epicgames.com, battle.net, minecraft.net
# - Development: gitlab.com, stackoverflow.com, npm.org
# - Media: twitch.tv, spotify.com, soundcloud.com
# - Education: khan.org, coursera.org, edx.org
# - News: bbc.com, reuters.com, apnews.com
#
# Guild-Specific vs Global:
# - Global allowlist applies to all servers
# - Guild-specific allowlists are additional (don't override global)
# - Use guild-specific for community-specific trusted sites
# - Use global for widely trusted platforms
#
# Maintenance:
# - Review allowlist monthly for security
# - Document reasons for all additions
# - Track who added each domain for accountability
# - Monitor for changes in domain ownership or compromise

View File

@@ -0,0 +1,74 @@
# External Wordlist Sources Configuration
# Configure automatic wordlist updates from external sources
sources:
# Default profanity list (LDNOOBW)
- name: "ldnoobw_en"
url: "https://raw.githubusercontent.com/LDNOOBW/List-of-Dirty-Naughty-Obscene-and-Otherwise-Bad-Words/master/en"
category: profanity
action: warn
reason: "External profanity list (English)"
enabled: true
update_interval_hours: 168 # Update weekly
applies_to_guilds: [] # Empty = applies to all guilds
# Additional language support (uncomment and configure as needed)
# - name: "ldnoobw_es"
# url: "https://raw.githubusercontent.com/LDNOOBW/List-of-Dirty-Naughty-Obscene-and-Otherwise-Bad-Words/master/es"
# category: profanity
# action: warn
# reason: "External profanity list (Spanish)"
# enabled: false
# update_interval_hours: 168
# applies_to_guilds: []
# Custom external source example
# - name: "custom_hate_speech"
# url: "https://example.com/hate-speech-list.txt"
# category: hate_speech
# action: delete
# reason: "Custom hate speech prevention"
# enabled: false
# update_interval_hours: 24 # Update daily
# applies_to_guilds: [123456789012345678] # Only for specific guild
# Scam/phishing domains (if available)
# - name: "phishing_domains"
# url: "https://example.com/phishing-domains.txt"
# category: scam
# action: delete
# reason: "Known phishing domains"
# enabled: false
# update_interval_hours: 4 # Update every 4 hours
# applies_to_guilds: []
# Configuration Notes:
#
# Update Intervals:
# 1-6 hours = High-risk content (scams, phishing)
# 12-24 hours = Moderate risk content
# 168 hours = Weekly updates (default for profanity)
# 720 hours = Monthly updates (stable lists)
#
# Applies to Guilds:
# [] = Apply to all guilds
# [123, 456] = Only apply to specific guild IDs
# ["all_premium"] = Apply only to premium guilds (if implemented)
#
# Categories determine how content is classified and what AI moderation
# settings apply to the detected content.
#
# Actions determine the default action taken when words from this source
# are detected. Guild-specific overrides can modify this behavior.
#
# URL Requirements:
# - Must be publicly accessible
# - Should return plain text with one word/pattern per line
# - HTTPS URLs preferred for security
# - Consider rate limiting and source reliability
#
# Security Notes:
# - External sources are validated before applying
# - Malformed or suspicious content is logged but not applied
# - Sources that fail repeatedly are automatically disabled
# - All updates are logged for audit purposes

View File

@@ -33,6 +33,9 @@ dependencies = [
"authlib>=1.3.0", "authlib>=1.3.0",
"httpx>=0.27.0", "httpx>=0.27.0",
"itsdangerous>=2.1.2", "itsdangerous>=2.1.2",
"pyyaml>=6.0",
"jsonschema>=4.20.0",
"watchfiles>=0.21.0",
] ]
[project.optional-dependencies] [project.optional-dependencies]

View File

@@ -0,0 +1 @@
"""GuardDen CLI tools for configuration management."""

559
src/guardden/cli/config.py Normal file
View File

@@ -0,0 +1,559 @@
#!/usr/bin/env python3
"""GuardDen Configuration CLI Tool.
This CLI tool allows you to manage GuardDen bot configurations without
using Discord commands. You can create, edit, validate, and migrate
configurations using this command-line interface.
Usage:
python -m guardden.cli.config --help
python -m guardden.cli.config guild create 123456789 "My Server"
python -m guardden.cli.config guild edit 123456789 ai_moderation.sensitivity 75
python -m guardden.cli.config migrate from-database
python -m guardden.cli.config validate all
"""
import asyncio
import sys
import logging
from pathlib import Path
from typing import Optional, Dict, Any, List
import argparse
import yaml
# Add src to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from guardden.services.file_config import FileConfigurationManager, ConfigurationError
from guardden.services.config_migration import ConfigurationMigrator
from guardden.services.database import Database
from guardden.services.guild_config import GuildConfigService
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class ConfigurationCLI:
"""Command-line interface for GuardDen configuration management."""
def __init__(self, config_dir: str = "config"):
"""Initialize the CLI with configuration directory."""
self.config_dir = Path(config_dir)
self.file_manager: Optional[FileConfigurationManager] = None
self.database: Optional[Database] = None
self.migrator: Optional[ConfigurationMigrator] = None
async def initialize(self):
"""Initialize the configuration system."""
self.file_manager = FileConfigurationManager(str(self.config_dir))
await self.file_manager.initialize()
# Initialize database connection if available
try:
import os
database_url = os.getenv('GUARDDEN_DATABASE_URL', 'postgresql://guardden:guardden@localhost:5432/guardden')
self.database = Database(database_url)
guild_config_service = GuildConfigService(self.database)
self.migrator = ConfigurationMigrator(self.database, guild_config_service, self.file_manager)
logger.info("Database connection established")
except Exception as e:
logger.warning(f"Database not available: {e}")
async def cleanup(self):
"""Clean up resources."""
if self.file_manager:
await self.file_manager.shutdown()
if self.database:
await self.database.close()
# Guild management commands
async def guild_create(self, guild_id: int, name: str, owner_id: Optional[int] = None):
"""Create a new guild configuration."""
try:
file_path = await self.file_manager.create_guild_config(guild_id, name, owner_id)
print(f"✅ Created guild configuration: {file_path}")
print(f"📝 Edit the file to customize settings for {name}")
return True
except ConfigurationError as e:
print(f"❌ Failed to create guild configuration: {e.error_message}")
return False
except Exception as e:
print(f"❌ Unexpected error: {str(e)}")
return False
async def guild_list(self):
"""List all configured guilds."""
configs = self.file_manager.get_all_guild_configs()
if not configs:
print("📄 No guild configurations found")
print("💡 Use 'guild create <guild_id> <name>' to create a new configuration")
return
print(f"📋 Found {len(configs)} guild configuration(s):")
print()
for guild_id, config in configs.items():
status_icon = "" if config else ""
premium_icon = "" if config.premium else ""
print(f"{status_icon} {premium_icon} {guild_id}: {config.name}")
print(f" 📁 File: {config.file_path}")
print(f" 🕐 Updated: {config.last_updated.strftime('%Y-%m-%d %H:%M:%S')}")
# Show key settings
settings = config.settings
ai_enabled = settings.get("ai_moderation", {}).get("enabled", False)
nsfw_only = settings.get("ai_moderation", {}).get("nsfw_only_filtering", False)
automod_enabled = settings.get("moderation", {}).get("automod_enabled", False)
print(f" 🤖 AI: {'' if ai_enabled else ''} | "
f"🔞 NSFW-Only: {'' if nsfw_only else ''} | "
f"⚡ AutoMod: {'' if automod_enabled else ''}")
print()
async def guild_edit(self, guild_id: int, setting_path: str, value: Any):
"""Edit a guild configuration setting."""
config = self.file_manager.get_guild_config(guild_id)
if not config:
print(f"❌ Guild {guild_id} configuration not found")
return False
try:
# Load current configuration
with open(config.file_path, 'r', encoding='utf-8') as f:
file_config = yaml.safe_load(f)
# Parse setting path (e.g., "ai_moderation.sensitivity")
path_parts = setting_path.split('.')
current = file_config
# Navigate to the parent of the target setting
for part in path_parts[:-1]:
if part not in current:
print(f"❌ Setting path not found: {setting_path}")
return False
current = current[part]
# Set the value
final_key = path_parts[-1]
old_value = current.get(final_key, "Not set")
# Convert value to appropriate type
if isinstance(old_value, bool):
value = str(value).lower() in ('true', '1', 'yes', 'on')
elif isinstance(old_value, int):
value = int(value)
elif isinstance(old_value, float):
value = float(value)
elif isinstance(old_value, list):
value = value.split(',') if isinstance(value, str) else value
current[final_key] = value
# Write back to file
with open(config.file_path, 'w', encoding='utf-8') as f:
yaml.dump(file_config, f, default_flow_style=False, indent=2)
print(f"✅ Updated {setting_path} for guild {guild_id}")
print(f" 📝 Changed from: {old_value}")
print(f" 📝 Changed to: {value}")
print(f"🔄 Configuration will be hot-reloaded automatically")
return True
except Exception as e:
print(f"❌ Failed to edit configuration: {str(e)}")
return False
async def guild_validate(self, guild_id: Optional[int] = None):
"""Validate guild configuration(s)."""
if guild_id:
configs = {guild_id: self.file_manager.get_guild_config(guild_id)}
if not configs[guild_id]:
print(f"❌ Guild {guild_id} configuration not found")
return False
else:
configs = self.file_manager.get_all_guild_configs()
if not configs:
print("📄 No configurations to validate")
return True
all_valid = True
print(f"🔍 Validating {len(configs)} configuration(s)...")
print()
for guild_id, config in configs.items():
if not config:
continue
try:
# Load and validate configuration
with open(config.file_path, 'r', encoding='utf-8') as f:
file_config = yaml.safe_load(f)
errors = self.file_manager.validate_config(file_config)
if errors:
all_valid = False
print(f"❌ Guild {guild_id} ({config.name}) - INVALID")
for error in errors:
print(f" 🔸 {error}")
else:
print(f"✅ Guild {guild_id} ({config.name}) - VALID")
except Exception as e:
all_valid = False
print(f"❌ Guild {guild_id} - ERROR: {str(e)}")
print()
if all_valid:
print("🎉 All configurations are valid!")
else:
print("⚠️ Some configurations have errors. Please fix them before running the bot.")
return all_valid
async def guild_backup(self, guild_id: int):
"""Create a backup of guild configuration."""
try:
backup_path = await self.file_manager.backup_config(guild_id)
print(f"✅ Created backup: {backup_path}")
return True
except Exception as e:
print(f"❌ Failed to create backup: {str(e)}")
return False
# Migration commands
async def migrate_from_database(self, backup_existing: bool = True):
"""Migrate all configurations from database to files."""
if not self.migrator:
print("❌ Database not available for migration")
return False
print("🔄 Starting migration from database to files...")
print("⚠️ This will convert Discord command configurations to YAML files")
if backup_existing:
print("📦 Existing files will be backed up")
try:
results = await self.migrator.migrate_all_guilds(backup_existing)
print("\n📊 Migration Results:")
print(f" ✅ Migrated: {len(results['migrated_guilds'])} guilds")
print(f" ❌ Failed: {len(results['failed_guilds'])} guilds")
print(f" ⏭️ Skipped: {len(results['skipped_guilds'])} guilds")
print(f" 📝 Banned words migrated: {results['banned_words_migrated']}")
if results['migrated_guilds']:
print("\n✅ Successfully migrated guilds:")
for guild in results['migrated_guilds']:
print(f"{guild['guild_id']}: {guild['guild_name']} "
f"({guild['banned_words_count']} banned words)")
if results['failed_guilds']:
print("\n❌ Failed migrations:")
for guild in results['failed_guilds']:
print(f"{guild['guild_id']}: {guild['guild_name']} - {guild['error']}")
if results['skipped_guilds']:
print("\n⏭️ Skipped guilds:")
for guild in results['skipped_guilds']:
print(f"{guild['guild_id']}: {guild['guild_name']} - {guild['reason']}")
if results['errors']:
print("\n⚠️ Errors encountered:")
for error in results['errors']:
print(f"{error}")
return len(results['failed_guilds']) == 0
except Exception as e:
print(f"❌ Migration failed: {str(e)}")
return False
async def migrate_verify(self, guild_ids: Optional[List[int]] = None):
"""Verify migration by comparing database and file configurations."""
if not self.migrator:
print("❌ Database not available for verification")
return False
print("🔍 Verifying migration results...")
try:
results = await self.migrator.verify_migration(guild_ids)
print("\n📊 Verification Results:")
print(f" ✅ Verified: {len(results['verified_guilds'])} guilds")
print(f" ⚠️ Mismatches: {len(results['mismatches'])} guilds")
print(f" 📄 Missing files: {len(results['missing_files'])} guilds")
if results['verified_guilds']:
print("\n✅ Verified guilds:")
for guild in results['verified_guilds']:
print(f"{guild['guild_id']}: {guild['guild_name']}")
if results['mismatches']:
print("\n⚠️ Configuration mismatches:")
for guild in results['mismatches']:
print(f"{guild['guild_id']}: {guild['guild_name']}")
print(f" Mismatched fields: {', '.join(guild['mismatched_fields'])}")
if results['missing_files']:
print("\n📄 Missing configuration files:")
for guild in results['missing_files']:
print(f"{guild['guild_id']}: {guild['guild_name']}")
print(f" Expected: {guild['expected_file']}")
return len(results['mismatches']) == 0 and len(results['missing_files']) == 0
except Exception as e:
print(f"❌ Verification failed: {str(e)}")
return False
# Wordlist management
async def wordlist_info(self):
"""Show information about wordlist configurations."""
banned_words = self.file_manager.get_wordlist_config()
allowlists = self.file_manager.get_allowlist_config()
external_sources = self.file_manager.get_external_sources_config()
print("📝 Wordlist Configuration Status:")
print()
if banned_words:
global_patterns = len(banned_words.get('global_patterns', []))
guild_patterns = sum(
len(patterns) for patterns in banned_words.get('guild_patterns', {}).values()
)
print(f"🚫 Banned Words: {global_patterns} global, {guild_patterns} guild-specific")
else:
print("🚫 Banned Words: Not configured")
if allowlists:
global_allowlist = len(allowlists.get('global_allowlist', []))
guild_allowlists = sum(
len(domains) for domains in allowlists.get('guild_allowlists', {}).values()
)
print(f"✅ Domain Allowlists: {global_allowlist} global, {guild_allowlists} guild-specific")
else:
print("✅ Domain Allowlists: Not configured")
if external_sources:
sources = external_sources.get('sources', [])
enabled_sources = len([s for s in sources if s.get('enabled', False)])
print(f"🌐 External Sources: {len(sources)} total, {enabled_sources} enabled")
else:
print("🌐 External Sources: Not configured")
print()
print("📁 Configuration files:")
print(f"{self.config_dir / 'wordlists' / 'banned-words.yml'}")
print(f"{self.config_dir / 'wordlists' / 'domain-allowlists.yml'}")
print(f"{self.config_dir / 'wordlists' / 'external-sources.yml'}")
# Template management
async def template_create(self, guild_id: int, name: str):
"""Create a new guild configuration from template."""
return await self.guild_create(guild_id, name)
async def template_info(self):
"""Show available configuration templates."""
template_dir = self.config_dir / "templates"
templates = list(template_dir.glob("*.yml"))
if not templates:
print("📄 No configuration templates found")
return
print(f"📋 Available Templates ({len(templates)}):")
print()
for template in templates:
try:
with open(template, 'r', encoding='utf-8') as f:
content = yaml.safe_load(f)
description = "Default guild configuration template"
if '_description' in content:
description = content['_description']
print(f"📄 {template.name}")
print(f" {description}")
print(f" 📁 {template}")
print()
except Exception as e:
print(f"❌ Error reading template {template.name}: {str(e)}")
async def main():
"""Main CLI entry point."""
parser = argparse.ArgumentParser(
description="GuardDen Configuration CLI Tool",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Create a new guild configuration
python -m guardden.cli.config guild create 123456789 "My Server"
# List all guild configurations
python -m guardden.cli.config guild list
# Edit a configuration setting
python -m guardden.cli.config guild edit 123456789 ai_moderation.sensitivity 75
python -m guardden.cli.config guild edit 123456789 ai_moderation.nsfw_only_filtering true
# Validate configurations
python -m guardden.cli.config guild validate
python -m guardden.cli.config guild validate 123456789
# Migration from database
python -m guardden.cli.config migrate from-database
python -m guardden.cli.config migrate verify
# Wordlist management
python -m guardden.cli.config wordlist info
# Template management
python -m guardden.cli.config template info
"""
)
parser.add_argument(
'--config-dir', '-c',
default='config',
help='Configuration directory (default: config)'
)
subparsers = parser.add_subparsers(dest='command', help='Available commands')
# Guild management
guild_parser = subparsers.add_parser('guild', help='Guild configuration management')
guild_subparsers = guild_parser.add_subparsers(dest='guild_command')
# Guild create
create_parser = guild_subparsers.add_parser('create', help='Create new guild configuration')
create_parser.add_argument('guild_id', type=int, help='Discord guild ID')
create_parser.add_argument('name', help='Guild name')
create_parser.add_argument('--owner-id', type=int, help='Guild owner Discord user ID')
# Guild list
guild_subparsers.add_parser('list', help='List all guild configurations')
# Guild edit
edit_parser = guild_subparsers.add_parser('edit', help='Edit guild configuration setting')
edit_parser.add_argument('guild_id', type=int, help='Discord guild ID')
edit_parser.add_argument('setting', help='Setting path (e.g., ai_moderation.sensitivity)')
edit_parser.add_argument('value', help='New value')
# Guild validate
validate_parser = guild_subparsers.add_parser('validate', help='Validate guild configurations')
validate_parser.add_argument('guild_id', type=int, nargs='?', help='Specific guild ID (optional)')
# Guild backup
backup_parser = guild_subparsers.add_parser('backup', help='Backup guild configuration')
backup_parser.add_argument('guild_id', type=int, help='Discord guild ID')
# Migration
migrate_parser = subparsers.add_parser('migrate', help='Configuration migration')
migrate_subparsers = migrate_parser.add_subparsers(dest='migrate_command')
# Migrate from database
from_db_parser = migrate_subparsers.add_parser('from-database', help='Migrate from database to files')
from_db_parser.add_argument('--no-backup', action='store_true', help='Skip backing up existing files')
# Migrate verify
verify_parser = migrate_subparsers.add_parser('verify', help='Verify migration results')
verify_parser.add_argument('guild_ids', type=int, nargs='*', help='Specific guild IDs to verify')
# Wordlist management
wordlist_parser = subparsers.add_parser('wordlist', help='Wordlist management')
wordlist_subparsers = wordlist_parser.add_subparsers(dest='wordlist_command')
wordlist_subparsers.add_parser('info', help='Show wordlist information')
# Template management
template_parser = subparsers.add_parser('template', help='Template management')
template_subparsers = template_parser.add_subparsers(dest='template_command')
template_subparsers.add_parser('info', help='Show available templates')
args = parser.parse_args()
if not args.command:
parser.print_help()
return 1
# Initialize CLI
cli = ConfigurationCLI(args.config_dir)
try:
await cli.initialize()
success = True
# Execute command
if args.command == 'guild':
if args.guild_command == 'create':
success = await cli.guild_create(args.guild_id, args.name, args.owner_id)
elif args.guild_command == 'list':
await cli.guild_list()
elif args.guild_command == 'edit':
success = await cli.guild_edit(args.guild_id, args.setting, args.value)
elif args.guild_command == 'validate':
success = await cli.guild_validate(args.guild_id)
elif args.guild_command == 'backup':
success = await cli.guild_backup(args.guild_id)
else:
print("❌ Unknown guild command. Use --help for available commands.")
success = False
elif args.command == 'migrate':
if args.migrate_command == 'from-database':
success = await cli.migrate_from_database(not args.no_backup)
elif args.migrate_command == 'verify':
guild_ids = args.guild_ids if args.guild_ids else None
success = await cli.migrate_verify(guild_ids)
else:
print("❌ Unknown migrate command. Use --help for available commands.")
success = False
elif args.command == 'wordlist':
if args.wordlist_command == 'info':
await cli.wordlist_info()
else:
print("❌ Unknown wordlist command. Use --help for available commands.")
success = False
elif args.command == 'template':
if args.template_command == 'info':
await cli.template_info()
else:
print("❌ Unknown template command. Use --help for available commands.")
success = False
return 0 if success else 1
except KeyboardInterrupt:
print("\n⚠️ Interrupted by user")
return 1
except Exception as e:
print(f"❌ Unexpected error: {str(e)}")
logger.exception("CLI error")
return 1
finally:
await cli.cleanup()
if __name__ == '__main__':
sys.exit(asyncio.run(main()))

View File

@@ -0,0 +1,457 @@
"""Configuration migration system for GuardDen.
This module handles migration from database-based Discord command configuration
to file-based YAML configuration.
"""
import logging
import asyncio
from pathlib import Path
from typing import Dict, Any, List, Optional
from datetime import datetime
import yaml
from guardden.services.database import Database
from guardden.services.guild_config import GuildConfigService
from guardden.services.file_config import FileConfigurationManager
from guardden.models.guild import Guild, GuildSettings, BannedWord
logger = logging.getLogger(__name__)
class ConfigurationMigrator:
"""Handles migration from database to file-based configuration."""
def __init__(
self,
database: Database,
guild_config_service: GuildConfigService,
file_config_manager: FileConfigurationManager
):
"""Initialize the migration system.
Args:
database: Database instance
guild_config_service: Current guild configuration service
file_config_manager: File configuration manager
"""
self.database = database
self.guild_config_service = guild_config_service
self.file_config_manager = file_config_manager
async def migrate_all_guilds(self, backup_existing: bool = True) -> Dict[str, Any]:
"""Migrate all guild configurations from database to files.
Args:
backup_existing: Whether to backup existing configuration files
Returns:
Dictionary with migration results
"""
logger.info("Starting migration of all guild configurations...")
results = {
"migrated_guilds": [],
"failed_guilds": [],
"skipped_guilds": [],
"total_guilds": 0,
"banned_words_migrated": 0,
"errors": []
}
try:
async with self.database.session() as session:
# Get all guilds from database
from sqlalchemy import select
stmt = select(Guild)
result = await session.execute(stmt)
guilds = result.scalars().all()
results["total_guilds"] = len(guilds)
logger.info(f"Found {len(guilds)} guilds to migrate")
for guild in guilds:
try:
await self._migrate_single_guild(guild, backup_existing, results)
except Exception as e:
error_msg = f"Failed to migrate guild {guild.id}: {str(e)}"
logger.error(error_msg)
results["failed_guilds"].append({
"guild_id": guild.id,
"guild_name": guild.name,
"error": error_msg
})
results["errors"].append(error_msg)
# Migrate wordlists
await self._migrate_wordlists(results)
logger.info(f"Migration complete. Success: {len(results['migrated_guilds'])}, "
f"Failed: {len(results['failed_guilds'])}, "
f"Skipped: {len(results['skipped_guilds'])}")
except Exception as e:
error_msg = f"Migration failed with error: {str(e)}"
logger.error(error_msg)
results["errors"].append(error_msg)
return results
async def _migrate_single_guild(
self,
guild: Guild,
backup_existing: bool,
results: Dict[str, Any]
) -> None:
"""Migrate a single guild's configuration."""
# Check if file already exists
guild_file = self.file_config_manager.config_dir / "guilds" / f"guild-{guild.id}.yml"
if guild_file.exists():
if backup_existing:
backup_path = await self.file_config_manager.backup_config(guild.id)
logger.info(f"Backed up existing config for guild {guild.id}: {backup_path}")
else:
results["skipped_guilds"].append({
"guild_id": guild.id,
"guild_name": guild.name,
"reason": "Configuration file already exists"
})
return
# Get guild settings from database
async with self.database.session() as session:
from sqlalchemy import select
from sqlalchemy.orm import selectinload
stmt = select(Guild).where(Guild.id == guild.id).options(
selectinload(Guild.settings),
selectinload(Guild.banned_words)
)
result = await session.execute(stmt)
guild_with_settings = result.scalar_one_or_none()
if not guild_with_settings:
raise Exception(f"Guild {guild.id} not found in database")
# Convert to file configuration format
file_config = await self._convert_guild_to_file_config(guild_with_settings)
# Write to file
with open(guild_file, 'w', encoding='utf-8') as f:
yaml.dump(file_config, f, default_flow_style=False, indent=2, sort_keys=False)
logger.info(f"Migrated guild {guild.id} ({guild.name}) to {guild_file}")
results["migrated_guilds"].append({
"guild_id": guild.id,
"guild_name": guild.name,
"file_path": str(guild_file),
"banned_words_count": len(guild_with_settings.banned_words) if guild_with_settings.banned_words else 0
})
if guild_with_settings.banned_words:
results["banned_words_migrated"] += len(guild_with_settings.banned_words)
async def _convert_guild_to_file_config(self, guild: Guild) -> Dict[str, Any]:
"""Convert database guild model to file configuration format."""
settings = guild.settings if guild.settings else GuildSettings()
# Base guild information
config = {
"guild_id": guild.id,
"name": guild.name,
"owner_id": guild.owner_id,
"premium": guild.premium,
# Add migration metadata
"_migration_info": {
"migrated_at": datetime.now().isoformat(),
"migrated_from": "database",
"original_created_at": guild.created_at.isoformat() if guild.created_at else None,
"original_updated_at": guild.updated_at.isoformat() if guild.updated_at else None
},
"settings": {
"general": {
"prefix": settings.prefix,
"locale": settings.locale
},
"channels": {
"log_channel_id": settings.log_channel_id,
"mod_log_channel_id": settings.mod_log_channel_id,
"welcome_channel_id": settings.welcome_channel_id
},
"roles": {
"mute_role_id": settings.mute_role_id,
"verified_role_id": settings.verified_role_id,
"mod_role_ids": settings.mod_role_ids or []
},
"moderation": {
"automod_enabled": settings.automod_enabled,
"anti_spam_enabled": settings.anti_spam_enabled,
"link_filter_enabled": settings.link_filter_enabled,
"strike_actions": settings.strike_actions or {}
},
"automod": {
"message_rate_limit": settings.message_rate_limit,
"message_rate_window": settings.message_rate_window,
"duplicate_threshold": settings.duplicate_threshold,
"mention_limit": settings.mention_limit,
"mention_rate_limit": settings.mention_rate_limit,
"mention_rate_window": settings.mention_rate_window,
"scam_allowlist": settings.scam_allowlist or []
},
"ai_moderation": {
"enabled": settings.ai_moderation_enabled,
"sensitivity": settings.ai_sensitivity,
"confidence_threshold": settings.ai_confidence_threshold,
"log_only": settings.ai_log_only,
"nsfw_detection_enabled": settings.nsfw_detection_enabled,
"nsfw_only_filtering": getattr(settings, 'nsfw_only_filtering', False)
},
"verification": {
"enabled": settings.verification_enabled,
"type": settings.verification_type
}
}
}
# Add banned words if any exist
if guild.banned_words:
config["banned_words"] = []
for banned_word in guild.banned_words:
config["banned_words"].append({
"pattern": banned_word.pattern,
"action": banned_word.action,
"is_regex": banned_word.is_regex,
"reason": banned_word.reason,
"category": banned_word.category,
"source": banned_word.source,
"managed": banned_word.managed,
"added_by": banned_word.added_by,
"created_at": banned_word.created_at.isoformat() if banned_word.created_at else None
})
return config
async def _migrate_wordlists(self, results: Dict[str, Any]) -> None:
"""Migrate global banned words and allowlists to wordlist files."""
# Get all managed banned words (global wordlists)
async with self.database.session() as session:
from sqlalchemy import select
stmt = select(BannedWord).where(BannedWord.managed == True)
result = await session.execute(stmt)
managed_words = result.scalars().all()
if managed_words:
# Group by source and category
sources = {}
for word in managed_words:
source = word.source or "unknown"
if source not in sources:
sources[source] = []
sources[source].append(word)
# Update external sources configuration
external_config_path = self.file_config_manager.config_dir / "wordlists" / "external-sources.yml"
if external_config_path.exists():
with open(external_config_path, 'r', encoding='utf-8') as f:
external_config = yaml.safe_load(f)
else:
external_config = {"sources": []}
# Add migration info for discovered sources
for source_name, words in sources.items():
existing_source = next(
(s for s in external_config["sources"] if s["name"] == source_name),
None
)
if not existing_source:
# Add new source based on migrated words
category = words[0].category if words[0].category else "profanity"
action = words[0].action if words[0].action else "warn"
external_config["sources"].append({
"name": source_name,
"url": f"# MIGRATED: Originally from {source_name}",
"category": category,
"action": action,
"reason": f"Migrated from database source: {source_name}",
"enabled": False, # Disabled by default, needs manual URL
"update_interval_hours": 168,
"applies_to_guilds": [],
"_migration_info": {
"migrated_at": datetime.now().isoformat(),
"original_word_count": len(words),
"needs_url_configuration": True
}
})
# Write updated external sources
with open(external_config_path, 'w', encoding='utf-8') as f:
yaml.dump(external_config, f, default_flow_style=False, indent=2)
results["external_sources_updated"] = True
results["managed_words_found"] = len(managed_words)
logger.info(f"Updated external sources configuration with {len(sources)} discovered sources")
async def verify_migration(self, guild_ids: Optional[List[int]] = None) -> Dict[str, Any]:
"""Verify that migration was successful by comparing database and file configs.
Args:
guild_ids: Specific guild IDs to verify, or None for all
Returns:
Verification results
"""
logger.info("Verifying migration results...")
verification_results = {
"verified_guilds": [],
"mismatches": [],
"missing_files": [],
"errors": []
}
try:
async with self.database.session() as session:
from sqlalchemy import select
if guild_ids:
stmt = select(Guild).where(Guild.id.in_(guild_ids))
else:
stmt = select(Guild)
result = await session.execute(stmt)
guilds = result.scalars().all()
for guild in guilds:
try:
await self._verify_single_guild(guild, verification_results)
except Exception as e:
error_msg = f"Verification error for guild {guild.id}: {str(e)}"
logger.error(error_msg)
verification_results["errors"].append(error_msg)
logger.info(f"Verification complete. Verified: {len(verification_results['verified_guilds'])}, "
f"Mismatches: {len(verification_results['mismatches'])}, "
f"Missing: {len(verification_results['missing_files'])}")
except Exception as e:
error_msg = f"Verification failed: {str(e)}"
logger.error(error_msg)
verification_results["errors"].append(error_msg)
return verification_results
async def _verify_single_guild(self, guild: Guild, results: Dict[str, Any]) -> None:
"""Verify migration for a single guild."""
guild_file = self.file_config_manager.config_dir / "guilds" / f"guild-{guild.id}.yml"
if not guild_file.exists():
results["missing_files"].append({
"guild_id": guild.id,
"guild_name": guild.name,
"expected_file": str(guild_file)
})
return
# Load file configuration
with open(guild_file, 'r', encoding='utf-8') as f:
file_config = yaml.safe_load(f)
# Get database configuration
db_config = await self.guild_config_service.get_config(guild.id)
# Compare key settings
mismatches = []
if file_config.get("guild_id") != guild.id:
mismatches.append("guild_id")
if file_config.get("name") != guild.name:
mismatches.append("name")
if db_config:
file_settings = file_config.get("settings", {})
# Compare AI moderation settings
ai_settings = file_settings.get("ai_moderation", {})
if ai_settings.get("enabled") != db_config.ai_moderation_enabled:
mismatches.append("ai_moderation.enabled")
if ai_settings.get("sensitivity") != db_config.ai_sensitivity:
mismatches.append("ai_moderation.sensitivity")
# Compare automod settings
automod_settings = file_settings.get("automod", {})
if automod_settings.get("message_rate_limit") != db_config.message_rate_limit:
mismatches.append("automod.message_rate_limit")
if mismatches:
results["mismatches"].append({
"guild_id": guild.id,
"guild_name": guild.name,
"mismatched_fields": mismatches
})
else:
results["verified_guilds"].append({
"guild_id": guild.id,
"guild_name": guild.name
})
async def cleanup_database_configs(self, confirm: bool = False) -> Dict[str, Any]:
"""Clean up database configurations after successful migration.
WARNING: This will delete all guild settings and banned words from the database.
Only run after verifying migration is successful.
Args:
confirm: Must be True to actually perform cleanup
Returns:
Cleanup results
"""
if not confirm:
raise ValueError("cleanup_database_configs requires confirm=True to prevent accidental data loss")
logger.warning("STARTING DATABASE CLEANUP - This will delete all migrated configuration data!")
cleanup_results = {
"guild_settings_deleted": 0,
"banned_words_deleted": 0,
"errors": []
}
try:
async with self.database.session() as session:
# Delete all guild settings
from sqlalchemy import delete
# Delete banned words first (foreign key constraint)
banned_words_result = await session.execute(delete(BannedWord))
cleanup_results["banned_words_deleted"] = banned_words_result.rowcount
# Delete guild settings
guild_settings_result = await session.execute(delete(GuildSettings))
cleanup_results["guild_settings_deleted"] = guild_settings_result.rowcount
await session.commit()
logger.warning(f"Database cleanup complete. Deleted {cleanup_results['guild_settings_deleted']} "
f"guild settings and {cleanup_results['banned_words_deleted']} banned words.")
except Exception as e:
error_msg = f"Database cleanup failed: {str(e)}"
logger.error(error_msg)
cleanup_results["errors"].append(error_msg)
return cleanup_results

View File

@@ -0,0 +1,502 @@
"""File-based configuration system for GuardDen.
This module provides a complete file-based configuration system that replaces
Discord commands for bot configuration. Features include:
- YAML configuration files with schema validation
- Hot-reloading with file watching
- Migration from database settings
- Comprehensive error handling and rollback
"""
import logging
import asyncio
from pathlib import Path
from typing import Dict, Any, Optional, List, Callable
from datetime import datetime
from dataclasses import dataclass, field
import hashlib
try:
import yaml
import jsonschema
from watchfiles import watch, Change
except ImportError as e:
raise ImportError(f"Required dependencies missing: {e}. Install with 'pip install pyyaml jsonschema watchfiles'")
from guardden.models.guild import GuildSettings
from guardden.services.database import Database
logger = logging.getLogger(__name__)
@dataclass
class ConfigurationError(Exception):
"""Raised when configuration is invalid or cannot be loaded."""
file_path: str
error_message: str
validation_errors: List[str] = field(default_factory=list)
@dataclass
class FileConfig:
"""Represents a loaded configuration file."""
path: Path
content: Dict[str, Any]
last_modified: float
content_hash: str
is_valid: bool = True
validation_errors: List[str] = field(default_factory=list)
@dataclass
class GuildConfig:
"""Processed guild configuration."""
guild_id: int
name: str
owner_id: Optional[int]
premium: bool
settings: Dict[str, Any]
file_path: Path
last_updated: datetime
class FileConfigurationManager:
"""Manages file-based configuration with hot-reloading and validation."""
def __init__(self, config_dir: str = "config", database: Optional[Database] = None):
"""Initialize the configuration manager.
Args:
config_dir: Base directory for configuration files
database: Database instance for migration and fallback
"""
self.config_dir = Path(config_dir)
self.database = database
self.guild_configs: Dict[int, GuildConfig] = {}
self.wordlist_config: Optional[FileConfig] = None
self.allowlist_config: Optional[FileConfig] = None
self.external_sources_config: Optional[FileConfig] = None
# File watching
self._watch_task: Optional[asyncio.Task] = None
self._watch_enabled = True
self._callbacks: List[Callable[[int, GuildConfig], None]] = []
# Validation schemas
self._schemas: Dict[str, Dict[str, Any]] = {}
# Backup configurations (for rollback)
self._backup_configs: Dict[int, GuildConfig] = {}
# Ensure directories exist
self._ensure_directories()
def _ensure_directories(self) -> None:
"""Create configuration directories if they don't exist."""
dirs = [
self.config_dir / "guilds",
self.config_dir / "wordlists",
self.config_dir / "schemas",
self.config_dir / "templates",
self.config_dir / "backups"
]
for dir_path in dirs:
dir_path.mkdir(parents=True, exist_ok=True)
async def initialize(self) -> None:
"""Initialize the configuration system."""
logger.info("Initializing file-based configuration system...")
try:
# Load validation schemas
await self._load_schemas()
# Load all configuration files
await self._load_all_configs()
# Start file watching for hot-reload
if self._watch_enabled:
await self._start_file_watching()
logger.info(f"Configuration system initialized with {len(self.guild_configs)} guild configs")
except Exception as e:
logger.error(f"Failed to initialize configuration system: {e}")
raise
async def shutdown(self) -> None:
"""Shutdown the configuration system."""
logger.info("Shutting down configuration system...")
if self._watch_task and not self._watch_task.done():
self._watch_task.cancel()
try:
await self._watch_task
except asyncio.CancelledError:
pass
logger.info("Configuration system shutdown complete")
async def _load_schemas(self) -> None:
"""Load validation schemas from files."""
schema_dir = self.config_dir / "schemas"
schema_files = {
"guild": schema_dir / "guild-schema.yml",
"wordlists": schema_dir / "wordlists-schema.yml"
}
for schema_name, schema_path in schema_files.items():
if schema_path.exists():
try:
with open(schema_path, 'r', encoding='utf-8') as f:
self._schemas[schema_name] = yaml.safe_load(f)
logger.debug(f"Loaded schema: {schema_name}")
except Exception as e:
logger.error(f"Failed to load schema {schema_name}: {e}")
else:
logger.warning(f"Schema file not found: {schema_path}")
async def _load_all_configs(self) -> None:
"""Load all configuration files."""
# Load guild configurations
guild_dir = self.config_dir / "guilds"
if guild_dir.exists():
for config_file in guild_dir.glob("guild-*.yml"):
try:
await self._load_guild_config(config_file)
except Exception as e:
logger.error(f"Failed to load guild config {config_file}: {e}")
# Load wordlist configurations
await self._load_wordlist_configs()
async def _load_guild_config(self, file_path: Path) -> Optional[GuildConfig]:
"""Load a single guild configuration file."""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = yaml.safe_load(f)
# Validate against schema
if 'guild' in self._schemas:
try:
jsonschema.validate(content, self._schemas['guild'])
except jsonschema.ValidationError as e:
logger.error(f"Schema validation failed for {file_path}: {e}")
return None
# Extract guild information
guild_id = content.get('guild_id')
if not guild_id:
logger.error(f"Guild config missing guild_id: {file_path}")
return None
guild_config = GuildConfig(
guild_id=guild_id,
name=content.get('name', f"Guild {guild_id}"),
owner_id=content.get('owner_id'),
premium=content.get('premium', False),
settings=content.get('settings', {}),
file_path=file_path,
last_updated=datetime.now()
)
# Backup current config before updating
if guild_id in self.guild_configs:
self._backup_configs[guild_id] = self.guild_configs[guild_id]
self.guild_configs[guild_id] = guild_config
logger.debug(f"Loaded guild config for {guild_id}: {guild_config.name}")
# Notify callbacks of config change
await self._notify_config_change(guild_id, guild_config)
return guild_config
except Exception as e:
logger.error(f"Error loading guild config {file_path}: {e}")
return None
async def _load_wordlist_configs(self) -> None:
"""Load wordlist configuration files."""
wordlist_dir = self.config_dir / "wordlists"
configs = {
"banned-words.yml": "wordlist_config",
"domain-allowlists.yml": "allowlist_config",
"external-sources.yml": "external_sources_config"
}
for filename, attr_name in configs.items():
file_path = wordlist_dir / filename
if file_path.exists():
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = yaml.safe_load(f)
# Calculate content hash
content_hash = hashlib.md5(str(content).encode()).hexdigest()
file_config = FileConfig(
path=file_path,
content=content,
last_modified=file_path.stat().st_mtime,
content_hash=content_hash
)
setattr(self, attr_name, file_config)
logger.debug(f"Loaded {filename}")
except Exception as e:
logger.error(f"Failed to load {filename}: {e}")
async def _start_file_watching(self) -> None:
"""Start watching configuration files for changes."""
if self._watch_task and not self._watch_task.done():
return
self._watch_task = asyncio.create_task(self._file_watcher())
logger.info("Started file watching for configuration hot-reload")
async def _file_watcher(self) -> None:
"""Watch for file changes and reload configurations."""
try:
async for changes in watch(self.config_dir, recursive=True):
for change_type, file_path in changes:
file_path = Path(file_path)
# Only process YAML files
if file_path.suffix != '.yml':
continue
if change_type in (Change.added, Change.modified):
await self._handle_file_change(file_path)
elif change_type == Change.deleted:
await self._handle_file_deletion(file_path)
except asyncio.CancelledError:
logger.debug("File watcher cancelled")
except Exception as e:
logger.error(f"File watcher error: {e}")
async def _handle_file_change(self, file_path: Path) -> None:
"""Handle a file change event."""
try:
# Determine file type and reload appropriately
if file_path.parent.name == "guilds" and file_path.name.startswith("guild-"):
await self._load_guild_config(file_path)
logger.info(f"Reloaded guild config: {file_path}")
elif file_path.parent.name == "wordlists":
await self._load_wordlist_configs()
logger.info(f"Reloaded wordlist config: {file_path}")
except Exception as e:
logger.error(f"Error handling file change {file_path}: {e}")
await self._rollback_config(file_path)
async def _handle_file_deletion(self, file_path: Path) -> None:
"""Handle a file deletion event."""
try:
if file_path.parent.name == "guilds" and file_path.name.startswith("guild-"):
# Extract guild ID from filename
guild_id_str = file_path.stem.replace("guild-", "")
try:
guild_id = int(guild_id_str)
if guild_id in self.guild_configs:
del self.guild_configs[guild_id]
logger.info(f"Removed guild config for deleted file: {file_path}")
except ValueError:
logger.warning(f"Could not parse guild ID from filename: {file_path}")
except Exception as e:
logger.error(f"Error handling file deletion {file_path}: {e}")
async def _rollback_config(self, file_path: Path) -> None:
"""Rollback to previous configuration on error."""
try:
if file_path.parent.name == "guilds" and file_path.name.startswith("guild-"):
guild_id_str = file_path.stem.replace("guild-", "")
guild_id = int(guild_id_str)
if guild_id in self._backup_configs:
self.guild_configs[guild_id] = self._backup_configs[guild_id]
logger.info(f"Rolled back guild config for {guild_id}")
except Exception as e:
logger.error(f"Error during rollback for {file_path}: {e}")
async def _notify_config_change(self, guild_id: int, config: GuildConfig) -> None:
"""Notify registered callbacks of configuration changes."""
for callback in self._callbacks:
try:
callback(guild_id, config)
except Exception as e:
logger.error(f"Error in config change callback: {e}")
def register_change_callback(self, callback: Callable[[int, GuildConfig], None]) -> None:
"""Register a callback for configuration changes."""
self._callbacks.append(callback)
def get_guild_config(self, guild_id: int) -> Optional[GuildConfig]:
"""Get configuration for a specific guild."""
return self.guild_configs.get(guild_id)
def get_all_guild_configs(self) -> Dict[int, GuildConfig]:
"""Get all guild configurations."""
return self.guild_configs.copy()
def get_wordlist_config(self) -> Optional[Dict[str, Any]]:
"""Get wordlist configuration."""
return self.wordlist_config.content if self.wordlist_config else None
def get_allowlist_config(self) -> Optional[Dict[str, Any]]:
"""Get domain allowlist configuration."""
return self.allowlist_config.content if self.allowlist_config else None
def get_external_sources_config(self) -> Optional[Dict[str, Any]]:
"""Get external sources configuration."""
return self.external_sources_config.content if self.external_sources_config else None
async def create_guild_config(self, guild_id: int, name: str, owner_id: Optional[int] = None) -> Path:
"""Create a new guild configuration file from template."""
guild_file = self.config_dir / "guilds" / f"guild-{guild_id}.yml"
template_file = self.config_dir / "templates" / "guild-default.yml"
if guild_file.exists():
raise ConfigurationError(
str(guild_file),
"Guild configuration already exists"
)
# Load template
if template_file.exists():
with open(template_file, 'r', encoding='utf-8') as f:
template_content = yaml.safe_load(f)
else:
# Create basic template if file doesn't exist
template_content = await self._create_basic_template()
# Customize template
template_content['guild_id'] = guild_id
template_content['name'] = name
if owner_id:
template_content['owner_id'] = owner_id
# Write configuration file
with open(guild_file, 'w', encoding='utf-8') as f:
yaml.dump(template_content, f, default_flow_style=False, indent=2)
logger.info(f"Created guild configuration: {guild_file}")
# Load the new configuration
await self._load_guild_config(guild_file)
return guild_file
async def _create_basic_template(self) -> Dict[str, Any]:
"""Create a basic configuration template."""
return {
"guild_id": 0,
"name": "",
"premium": False,
"settings": {
"general": {
"prefix": "!",
"locale": "en"
},
"channels": {
"log_channel_id": None,
"mod_log_channel_id": None,
"welcome_channel_id": None
},
"roles": {
"mute_role_id": None,
"verified_role_id": None,
"mod_role_ids": []
},
"moderation": {
"automod_enabled": True,
"anti_spam_enabled": True,
"link_filter_enabled": False,
"strike_actions": {
"1": {"action": "warn"},
"3": {"action": "timeout", "duration": 3600},
"5": {"action": "kick"},
"7": {"action": "ban"}
}
},
"automod": {
"message_rate_limit": 5,
"message_rate_window": 5,
"duplicate_threshold": 3,
"mention_limit": 5,
"mention_rate_limit": 10,
"mention_rate_window": 60,
"scam_allowlist": []
},
"ai_moderation": {
"enabled": True,
"sensitivity": 80,
"confidence_threshold": 0.7,
"log_only": False,
"nsfw_detection_enabled": True,
"nsfw_only_filtering": False
},
"verification": {
"enabled": False,
"type": "button"
}
}
}
async def export_from_database(self, guild_id: int) -> Optional[Path]:
"""Export guild configuration from database to file."""
if not self.database:
raise ConfigurationError("", "Database not available for export")
try:
# Get guild settings from database
async with self.database.session() as session:
# This would need to be implemented based on your database service
# For now, return None to indicate not implemented
pass
logger.info(f"Exported guild {guild_id} configuration to file")
return None
except Exception as e:
logger.error(f"Failed to export guild {guild_id} from database: {e}")
raise ConfigurationError(
f"guild-{guild_id}.yml",
f"Database export failed: {str(e)}"
)
def validate_config(self, config_data: Dict[str, Any], schema_name: str = "guild") -> List[str]:
"""Validate configuration data against schema."""
errors = []
if schema_name in self._schemas:
try:
jsonschema.validate(config_data, self._schemas[schema_name])
except jsonschema.ValidationError as e:
errors.append(str(e))
else:
errors.append(f"Schema '{schema_name}' not found")
return errors
async def backup_config(self, guild_id: int) -> Path:
"""Create a backup of guild configuration."""
config = self.get_guild_config(guild_id)
if not config:
raise ConfigurationError(f"guild-{guild_id}.yml", "Guild configuration not found")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_file = self.config_dir / "backups" / f"guild-{guild_id}_{timestamp}.yml"
# Copy current configuration file
import shutil
shutil.copy2(config.file_path, backup_file)
logger.info(f"Created backup: {backup_file}")
return backup_file

294
tests/test_file_config.py Normal file
View File

@@ -0,0 +1,294 @@
"""Tests for file-based configuration system."""
import asyncio
import tempfile
import yaml
from pathlib import Path
from unittest.mock import patch
import pytest
from guardden.services.file_config import FileConfigurationManager, ConfigurationError
class TestFileConfigurationManager:
"""Tests for the file-based configuration manager."""
@pytest.fixture
def temp_config_dir(self):
"""Create a temporary configuration directory."""
with tempfile.TemporaryDirectory() as temp_dir:
yield temp_dir
@pytest.fixture
async def config_manager(self, temp_config_dir):
"""Create a configuration manager with temporary directory."""
manager = FileConfigurationManager(temp_config_dir)
await manager.initialize()
yield manager
await manager.shutdown()
def test_directory_creation(self, temp_config_dir):
"""Test that required directories are created."""
manager = FileConfigurationManager(temp_config_dir)
expected_dirs = [
"guilds",
"wordlists",
"schemas",
"templates",
"backups"
]
for dir_name in expected_dirs:
assert (Path(temp_config_dir) / dir_name).exists()
@pytest.mark.asyncio
async def test_guild_config_creation(self, config_manager, temp_config_dir):
"""Test creating a new guild configuration."""
guild_id = 123456789
name = "Test Guild"
owner_id = 987654321
file_path = await config_manager.create_guild_config(guild_id, name, owner_id)
assert file_path.exists()
assert file_path.name == f"guild-{guild_id}.yml"
# Verify content
with open(file_path, 'r') as f:
content = yaml.safe_load(f)
assert content["guild_id"] == guild_id
assert content["name"] == name
assert content["owner_id"] == owner_id
assert "settings" in content
@pytest.mark.asyncio
async def test_duplicate_guild_config_creation(self, config_manager):
"""Test that creating duplicate guild configs raises error."""
guild_id = 123456789
name = "Test Guild"
# Create first config
await config_manager.create_guild_config(guild_id, name)
# Attempt to create duplicate should raise error
with pytest.raises(ConfigurationError):
await config_manager.create_guild_config(guild_id, name)
@pytest.mark.asyncio
async def test_guild_config_loading(self, config_manager, temp_config_dir):
"""Test loading guild configuration from file."""
guild_id = 123456789
# Create config file manually
config_data = {
"guild_id": guild_id,
"name": "Test Guild",
"premium": False,
"settings": {
"general": {
"prefix": "!",
"locale": "en"
},
"ai_moderation": {
"enabled": True,
"sensitivity": 80,
"nsfw_only_filtering": True
}
}
}
guild_dir = Path(temp_config_dir) / "guilds"
guild_file = guild_dir / f"guild-{guild_id}.yml"
with open(guild_file, 'w') as f:
yaml.dump(config_data, f)
# Load config
await config_manager._load_guild_config(guild_file)
# Verify loaded config
config = config_manager.get_guild_config(guild_id)
assert config is not None
assert config.guild_id == guild_id
assert config.name == "Test Guild"
assert config.settings["ai_moderation"]["nsfw_only_filtering"] is True
@pytest.mark.asyncio
async def test_invalid_config_validation(self, config_manager, temp_config_dir):
"""Test that invalid configurations are rejected."""
guild_id = 123456789
# Create invalid config (missing required fields)
invalid_config = {
"name": "Test Guild",
# Missing guild_id
"settings": {}
}
guild_dir = Path(temp_config_dir) / "guilds"
guild_file = guild_dir / f"guild-{guild_id}.yml"
with open(guild_file, 'w') as f:
yaml.dump(invalid_config, f)
# Should return None for invalid config
result = await config_manager._load_guild_config(guild_file)
assert result is None
def test_configuration_validation(self, config_manager):
"""Test configuration validation against schema."""
valid_config = {
"guild_id": 123456789,
"name": "Test Guild",
"settings": {
"general": {"prefix": "!", "locale": "en"},
"channels": {"log_channel_id": None},
"roles": {"mod_role_ids": []},
"moderation": {"automod_enabled": True},
"automod": {"message_rate_limit": 5},
"ai_moderation": {"enabled": True},
"verification": {"enabled": False}
}
}
# Should return no errors for valid config
errors = config_manager.validate_config(valid_config)
assert len(errors) == 0
# Invalid config should return errors
invalid_config = {
"guild_id": "not-a-number", # Should be integer
"name": "Test Guild"
# Missing required settings
}
errors = config_manager.validate_config(invalid_config)
assert len(errors) > 0
@pytest.mark.asyncio
async def test_wordlist_config_loading(self, config_manager, temp_config_dir):
"""Test loading wordlist configurations."""
wordlist_dir = Path(temp_config_dir) / "wordlists"
# Create banned words config
banned_words_config = {
"global_patterns": [
{
"pattern": "badword",
"action": "delete",
"is_regex": False,
"category": "profanity"
}
]
}
with open(wordlist_dir / "banned-words.yml", 'w') as f:
yaml.dump(banned_words_config, f)
# Create allowlist config
allowlist_config = {
"global_allowlist": [
{
"domain": "discord.com",
"reason": "Official Discord domain"
}
]
}
with open(wordlist_dir / "domain-allowlists.yml", 'w') as f:
yaml.dump(allowlist_config, f)
# Load configs
await config_manager._load_wordlist_configs()
# Verify loaded configs
wordlist_config = config_manager.get_wordlist_config()
assert wordlist_config is not None
assert "global_patterns" in wordlist_config
assert len(wordlist_config["global_patterns"]) == 1
allowlist = config_manager.get_allowlist_config()
assert allowlist is not None
assert "global_allowlist" in allowlist
assert len(allowlist["global_allowlist"]) == 1
@pytest.mark.asyncio
async def test_config_backup(self, config_manager, temp_config_dir):
"""Test configuration backup functionality."""
guild_id = 123456789
# Create a guild config
await config_manager.create_guild_config(guild_id, "Test Guild")
# Create backup
backup_path = await config_manager.backup_config(guild_id)
assert backup_path.exists()
assert "backup" in backup_path.parent.name
assert f"guild-{guild_id}" in backup_path.name
# Verify backup content matches original
original_config = config_manager.get_guild_config(guild_id)
with open(backup_path, 'r') as f:
backup_content = yaml.safe_load(f)
assert backup_content["guild_id"] == original_config.guild_id
assert backup_content["name"] == original_config.name
@pytest.mark.asyncio
async def test_config_change_callbacks(self, config_manager, temp_config_dir):
"""Test that configuration change callbacks are triggered."""
callback_called = False
callback_guild_id = None
def test_callback(guild_id, config):
nonlocal callback_called, callback_guild_id
callback_called = True
callback_guild_id = guild_id
# Register callback
config_manager.register_change_callback(test_callback)
# Create config (should trigger callback)
guild_id = 123456789
await config_manager.create_guild_config(guild_id, "Test Guild")
# Wait a moment for callback to be called
await asyncio.sleep(0.1)
assert callback_called
assert callback_guild_id == guild_id
def test_all_guild_configs_retrieval(self, config_manager, temp_config_dir):
"""Test retrieving all guild configurations."""
# Initially should be empty
all_configs = config_manager.get_all_guild_configs()
assert len(all_configs) == 0
@pytest.mark.asyncio
async def test_nsfw_only_filtering_in_config(self, config_manager):
"""Test that NSFW-only filtering setting is properly handled."""
guild_id = 123456789
# Create config with NSFW-only filtering enabled
file_path = await config_manager.create_guild_config(guild_id, "NSFW Test Guild")
# Load and modify config to enable NSFW-only filtering
with open(file_path, 'r') as f:
config_data = yaml.safe_load(f)
config_data["settings"]["ai_moderation"]["nsfw_only_filtering"] = True
with open(file_path, 'w') as f:
yaml.dump(config_data, f)
# Reload config
await config_manager._load_guild_config(file_path)
# Verify NSFW-only filtering is enabled
config = config_manager.get_guild_config(guild_id)
assert config is not None
assert config.settings["ai_moderation"]["nsfw_only_filtering"] is True