# Plugin System

Sonora v1.2.8 includes a powerful plugin system that allows you to extend functionality without modifying core code.

# Table of Contents

  • Overview
  • Creating Plugins
  • Plugin Lifecycle
  • Event System
  • Security
  • Examples
  • Best Practices

# Overview

Plugins in Sonora are Python modules that can:

  • Add new commands and features
  • Hook into audio processing
  • Modify queue behavior
  • Integrate with external services
  • Customize bot behavior

# Key Features

  • Sandboxed execution - Plugins run in isolated environments
  • Permission system - Granular access control
  • Hot reloading - Load/unload plugins without restarting
  • Event-driven - React to bot events
  • Dependency management - Handle plugin dependencies

# Creating Plugins

# Basic Plugin Structure

# my_plugin.py
from sonora.plugins import Plugin

class MyPlugin(Plugin):
    def __init__(self):
        super().__init__(
            name="my_plugin",
            version="1.0.0",
            description="My custom plugin",
            permissions=["track.read", "player.control"]
        )

    async def on_load(self):
        print("Plugin loaded!")

    async def on_enable(self):
        print("Plugin enabled!")

    async def on_track_start(self, event):
        track = event.data['track']
        print(f"Now playing: {track.title}")

# Plugin Registration

# In your bot code
from sonora.plugins import PluginManager
from my_plugin import MyPlugin

manager = PluginManager()

# Register and enable plugin
await manager.register_plugin(MyPlugin())
await manager.enable_plugin("my_plugin")

# Plugin Lifecycle

# Lifecycle Methods

class LifecyclePlugin(Plugin):
    async def on_load(self):
        """Called when plugin is first loaded"""
        # Initialize resources
        self.database = await create_database_connection()
        print("Plugin loaded and database connected")

    async def on_enable(self):
        """Called when plugin is enabled"""
        # Start background tasks
        self.monitor_task = asyncio.create_task(self.monitor_usage())
        print("Plugin enabled and monitoring started")

    async def on_disable(self):
        """Called when plugin is disabled"""
        # Stop background tasks
        if self.monitor_task:
            self.monitor_task.cancel()
        print("Plugin disabled")

    async def on_unload(self):
        """Called when plugin is unloaded"""
        # Clean up resources
        await self.database.close()
        print("Plugin unloaded and resources cleaned")

# States

  • Loaded: Plugin code loaded, not active
  • Enabled: Plugin active and running
  • Disabled: Plugin loaded but inactive
  • Unloaded: Plugin completely removed

# Event System

# Built-in Events

from sonora import event_manager, EventType

class EventPlugin(Plugin):
    async def on_enable(self):
        @event_manager.on(EventType.TRACK_START)
        async def on_track_start(event):
            track = event.data['track']
            player = event.data['player']
            guild = event.data['guild']

            # Log track plays
            await self.log_track_play(track, guild)

        @event_manager.on(EventType.QUEUE_EMPTY)
        async def on_queue_empty(event):
            guild = event.data['guild']

            # Auto-add recommended tracks
            recommendations = await self.get_recommendations(guild)
            for track in recommendations[:3]:
                player = await self.get_player(guild)
                await player.queue.add(track)

    async def log_track_play(self, track, guild):
        # Implement logging logic
        pass

    async def get_recommendations(self, guild):
        # Implement recommendation logic
        return []

# Custom Events

from sonora.events import Event

class AnalyticsPlugin(Plugin):
    async def on_enable(self):
        # Emit custom events
        @event_manager.on(EventType.TRACK_END)
        async def on_track_end(event):
            track = event.data['track']
            duration = event.data.get('duration', 0)

            # Emit custom analytics event
            analytics_event = Event(
                type="analytics.track_completed",
                data={
                    "track": track,
                    "duration": duration,
                    "timestamp": time.time()
                }
            )
            await event_manager.emit(analytics_event)

# Security

# Permission System

Plugins must declare required permissions:

class SecurePlugin(Plugin):
    def __init__(self):
        super().__init__(
            name="secure_plugin",
            permissions=[
                "track.read",      # Read track info
                "track.modify",    # Modify tracks
                "queue.read",      # Read queue
                "queue.modify",    # Modify queue
                "player.control",  # Control playback
                "filters.write",   # Apply filters
                "autoplay.read",   # Read autoplay settings
            ]
        )

    async def modify_track(self, track):
        # Check permission at runtime
        if not self.has_permission("track.modify"):
            raise PermissionError("Insufficient permissions")

        # Safe to modify
        track.title = f"[Modified] {track.title}"

# Sandboxing

class SandboxedPlugin(Plugin):
    async def execute_code(self, user_code):
        # Code is automatically sandboxed
        try:
            result = await self.safe_execute(user_code)
            return result
        except SecurityError as e:
            print(f"Security violation: {e}")
            return None

    async def safe_execute(self, code):
        # Plugin execution is sandboxed by Sonora
        # Dangerous operations are blocked
        exec_globals = {
            'print': print,
            'len': len,
            # ... safe builtins only
        }
        exec(code, exec_globals)

# Examples

# Analytics Plugin

class AnalyticsPlugin(Plugin):
    def __init__(self):
        super().__init__(
            name="analytics",
            permissions=["track.read"]
        )
        self.stats = {}

    async def on_enable(self):
        @event_manager.on(EventType.TRACK_START)
        async def track_started(event):
            track_id = event.data['track'].identifier
            if track_id not in self.stats:
                self.stats[track_id] = {'plays': 0, 'skips': 0}
            self.stats[track_id]['plays'] += 1

        @event_manager.on(EventType.TRACK_END)
        async def track_ended(event):
            if event.data.get('reason') == 'skipped':
                track_id = event.data['track'].identifier
                if track_id in self.stats:
                    self.stats[track_id]['skips'] += 1

    async def get_popular_tracks(self, limit=10):
        sorted_tracks = sorted(
            self.stats.items(),
            key=lambda x: x[1]['plays'],
            reverse=True
        )
        return sorted_tracks[:limit]

# Auto-Moderation Plugin

class ModerationPlugin(Plugin):
    def __init__(self):
        super().__init__(
            name="moderation",
            permissions=["player.control", "track.read"]
        )
        self.banned_words = ["banned_word1", "banned_word2"]

    async def on_enable(self):
        @event_manager.on(EventType.TRACK_START)
        async def check_track(event):
            track = event.data['track']
            player = event.data['player']

            # Check for banned content
            if self.contains_banned_content(track):
                await player.skip()
                await self.notify_moderators(track, "Banned content detected")

    def contains_banned_content(self, track):
        title_lower = track.title.lower()
        return any(word in title_lower for word in self.banned_words)

    async def notify_moderators(self, track, reason):
        # Send notification to moderators
        print(f"Moderation: {track.title} - {reason}")

# Custom Filter Plugin

class CustomFilterPlugin(Plugin):
    def __init__(self):
        super().__init__(
            name="custom_filters",
            permissions=["filters.write"]
        )

    async def on_enable(self):
        # Register custom filter
        from sonora.filters import register_filter

        @register_filter("my_echo")
        class EchoFilter:
            def __init__(self, delay=0.3, feedback=0.4):
                self.delay = delay
                self.feedback = feedback

            def apply(self, audio_data):
                # Apply echo effect
                return self.apply_echo(audio_data, self.delay, self.feedback)

            def apply_echo(self, data, delay, feedback):
                # Echo implementation
                # This is a simplified example
                return data  # Placeholder

    async def apply_echo_command(self, ctx, delay: float = 0.3):
        """Apply echo filter"""
        player = await sonora.get_player(ctx.guild.id)
        await player.set_filter("my_echo", delay=delay)
        await ctx.send(f"Echo filter applied with {delay}s delay")

# Best Practices

# Plugin Development

  1. Request Minimal Permissions

    # Good: Only request what's needed
    permissions=["track.read"]
    
    # Bad: Request everything
    permissions=["*"]
  2. Handle Errors Gracefully

    async def risky_operation(self):
        try:
            result = await self.external_api_call()
            return result
        except Exception as e:
            self.logger.error(f"Operation failed: {e}")
            return None
  3. Clean Up Resources

    async def on_unload(self):
        # Always clean up
        if self.database:
            await self.database.close()
        if self.background_task:
            self.background_task.cancel()
  4. Document Your Plugin

    class DocumentedPlugin(Plugin):
        """
        This plugin does X, Y, and Z.
    
        Permissions required:
        - track.read: To analyze tracks
        - player.control: To skip inappropriate content
    
        Events handled:
        - TRACK_START: Analyze content
        - TRACK_END: Log statistics
        """
        pass

# Security Considerations

  1. Validate Input

    def process_user_input(self, user_input):
        if not isinstance(user_input, str):
            raise ValueError("Input must be string")
        if len(user_input) > 1000:
            raise ValueError("Input too long")
        # Sanitize input
        return user_input.strip()
  2. Limit Resource Usage

    async def rate_limited_operation(self):
        if time.time() - self.last_operation < 1.0:
            raise RateLimitError("Too many requests")
        self.last_operation = time.time()
        return await self.actual_operation()
  3. Log Appropriately

    # Good: Log useful information
    self.logger.info(f"Plugin {self.name} processed {count} items")
    
    # Bad: Log sensitive data
    self.logger.info(f"User {user_id} requested {sensitive_data}")

# Performance

  1. Use Async Operations

    # Good
    async def fetch_data(self):
        async with self.http_client.get(url) as response:
            return await response.json()
    
    # Bad
    def fetch_data(self):
        return requests.get(url).json()  # Blocking
  2. Cache Results

    async def get_expensive_data(self, key):
        if key in self.cache:
            return self.cache[key]
    
        result = await self.compute_expensive_data(key)
        self.cache[key] = result
        return result
  3. Batch Operations

    async def process_batch(self, items):
        # Process multiple items together
        results = []
        for batch in self.chunk_items(items, 10):
            batch_results = await self.process_batch_items(batch)
            results.extend(batch_results)
        return results

# Plugin Management

# CLI Commands

# List installed plugins
sonoractl plugin list

# Enable a plugin
sonoractl plugin enable my_plugin

# Disable a plugin
sonoractl plugin disable my_plugin

# Show plugin info
sonoractl plugin info my_plugin

# Programmatic Management

from sonora.plugins import PluginManager

manager = PluginManager()

# Load plugin from file
await manager.load_plugin('path/to/plugin.py')

# Enable plugin
await manager.enable_plugin('my_plugin')

# Get plugin instance
plugin = manager.get_plugin('my_plugin')

# Disable and unload
await manager.disable_plugin('my_plugin')
await manager.unload_plugin('my_plugin')

# Troubleshooting

# Common Issues

Plugin not loading

  • Check file syntax and imports
  • Verify required permissions
  • Check Sonora version compatibility

Permission denied

  • Plugin requesting permissions it doesn't need
  • Missing permission declarations
  • Runtime permission checks failing

Plugin crashes

  • Handle exceptions properly
  • Use try/catch blocks
  • Log errors for debugging

Memory leaks

  • Clean up resources in on_unload
  • Cancel background tasks
  • Remove event listeners

For more help, check the troubleshooting guide or visit our Discord server.