Plugin Architecture

class NotifyBase(URLBase):
    def send(self, body, title='', notify_type=NotifyType.INFO, attach=None, **kwargs):
        """
        Send notification through this service.
        
        Parameters:
        - body (str): Notification message content
        - title (str, optional): Notification title
        - notify_type (NotifyType): Type of notification
        - attach (AppriseAttachment, optional): File attachments
        - **kwargs: Service-specific parameters
        
        Returns:
        bool: True if notification was sent successfully
        """

    def url(self, privacy=False, *args, **kwargs):
        """
        Generate service URL for this notification service.
        
        Parameters:
        - privacy (bool): Mask sensitive information
        - *args, **kwargs: Additional URL parameters
        
        Returns:
        str: Service URL string
        """

    def throttle(self, last_notification=None):
        """
        Handle rate limiting for this service.
        
        Parameters:
        - last_notification (datetime, optional): Time of last notification
        
        Returns:
        float: Delay in seconds before next notification allowed
        """

    @property
    def enabled(self):
        """
        Service availability status.
        
        Returns:
        bool: True if service is available and properly configured
        """

    @property
    def templates(self):
        """
        URL templates supported by this service.
        
        Returns:
        tuple: URL template patterns for service configuration
        """

    @property
    def requirements(self):
        """
        Required dependencies for this service.
        
        Returns:
        dict: Required Python packages and versions
        """

    @property
    def notify_format(self):
        """
        Supported body format for notifications.
        
        Returns:
        NotifyFormat: Preferred format (TEXT, HTML, MARKDOWN)
        """

class URLBase:
    def parse_url(self, url, verify_host=True):
        """
        Parse URL into components for service configuration.
        
        Parameters:
        - url (str): Service URL to parse
        - verify_host (bool): Verify host accessibility
        
        Returns:
        dict: Parsed URL components (scheme, host, path, params, etc.)
        """

    def parse_native_url(self, url):
        """
        Parse service-native URL format.
        
        Parameters:
        - url (str): Native service URL
        
        Returns:
        dict: Parsed native URL components
        """

    def url(self, privacy=False, *args, **kwargs):
        """
        Generate standardized URL for this service.
        
        Parameters:
        - privacy (bool): Mask sensitive information
        - *args, **kwargs: URL generation parameters
        
        Returns:
        str: Generated service URL
        """

    def __len__(self):
        """
        Get length representation of the service.
        
        Returns:
        int: Service length indicator
        """

    @property
    def app_id(self):
        """
        Application identifier for this service.
        
        Returns:
        str: Application ID
        """

    @property
    def app_desc(self):
        """
        Application description for this service.
        
        Returns:
        str: Application description
        """

    @property
    def service_name(self):
        """
        Human-readable service name.
        
        Returns:
        str: Service display name
        """

    @property
    def service_url(self):
        """
        Service homepage URL.
        
        Returns:
        str: URL to service documentation or homepage
        """

    @property
    def protocol(self):
        """
        URL protocol/scheme for this service.
        
        Returns:
        str: Protocol identifier (e.g., 'slack', 'discord', 'mailto')
        """

class NotificationManager:
    def load_modules(self, path=None, name=None):
        """
        Load notification plugin modules.
        
        Parameters:
        - path (str, optional): Path to plugin directory
        - name (str, optional): Specific plugin name to load
        
        Returns:
        bool: True if modules were loaded successfully
        """

    def plugins(self, include_disabled=True):
        """
        Get available notification plugins.
        
        Parameters:
        - include_disabled (bool): Include disabled/unavailable plugins
        
        Returns:
        dict: Available notification plugin classes keyed by protocol
        """

    def schemas(self):
        """
        Get supported notification URL schemas.
        
        Returns:
        list: Supported URL schemas for all loaded plugins
        """

import apprise
from apprise.plugins.base import NotifyBase
from apprise import NotifyType, NotifyFormat

class NotifyCustomService(NotifyBase):
    """
    Custom notification service plugin.
    """
    
    # Service metadata
    service_name = 'Custom Service'
    service_url = 'https://custom-service.com'
    protocol = 'custom'
    
    # URL templates this plugin supports
    templates = (
        '{schema}://{user}:{password}@{host}',
        '{schema}://{token}@{host}/{channel}',
    )
    
    # Required dependencies
    requirements = {
        'requests': '',  # Any version
        'custom-sdk': '>=1.0.0',  # Minimum version
    }
    
    # Supported notification format
    notify_format = NotifyFormat.MARKDOWN
    
    def __init__(self, user=None, password=None, token=None, 
                 host=None, channel=None, **kwargs):
        """
        Initialize custom service plugin.
        """
        super().__init__(**kwargs)
        
        self.user = user
        self.password = password  
        self.token = token
        self.host = host
        self.channel = channel
        
    def send(self, body, title='', notify_type=NotifyType.INFO, 
             attach=None, **kwargs):
        """
        Send notification through custom service.
        """
        try:
            # Implement actual notification sending logic
            # This would typically make HTTP requests or API calls
            
            # Example implementation structure:
            payload = {
                'message': body,
                'title': title,
                'type': notify_type,
                'channel': self.channel
            }
            
            # Send notification using service API
            # response = custom_service_api.send(payload)
            # return response.success
            
            return True  # Success
            
        except Exception as e:
            self.logger.warning(f'Failed to send notification: {e}')
            return False
    
    def url(self, privacy=False, *args, **kwargs):
        """
        Generate URL for this service configuration.
        """
        if self.token:
            # Token-based authentication
            if privacy:
                return f'{self.protocol}://***@{self.host}/{self.channel}'
            else:
                return f'{self.protocol}://{self.token}@{self.host}/{self.channel}'
        else:
            # User/password authentication
            if privacy:
                return f'{self.protocol}://***:***@{self.host}'
            else:
                return f'{self.protocol}://{self.user}:{self.password}@{self.host}'
    
    @staticmethod
    def parse_url(url):
        """
        Parse URL into configuration parameters.
        """
        # Implementation would parse the URL and return configuration dict
        # This is called when loading services from URL strings
        pass

import apprise

# Get notification manager instance
manager = apprise.NotificationManager()

# Load all available plugins
manager.load_modules()

# Get available plugins
plugins = manager.plugins(include_disabled=False)
print(f"Available plugins: {len(plugins)}")

for protocol, plugin_class in plugins.items():
    print(f"  {protocol}: {plugin_class.service_name}")

# Get supported schemas
schemas = manager.schemas()
print(f"Supported URL schemas: {schemas}")

import apprise

# Get detailed plugin information
apobj = apprise.Apprise()
details = apobj.details(show_requirements=True, show_disabled=False)

for service_key, service_info in details.items():
    print(f"\nService: {service_info.get('service_name', service_key)}")
    print(f"  Protocols: {service_info.get('protocols', [])}")
    print(f"  Enabled: {service_info.get('enabled', False)}")
    
    # Show requirements if any
    requirements = service_info.get('requirements', {})
    if requirements:
        print(f"  Requirements:")
        for package, version in requirements.items():
            print(f"    {package} {version or '(any version)'}")
    
    # Show URL templates
    templates = service_info.get('templates', [])
    if templates:
        print(f"  URL Templates:")
        for template in templates:
            print(f"    {template}")

import apprise

# Register custom plugin (if not auto-discovered)
# This would typically be done in plugin initialization

# Use custom plugin through URL
apobj = apprise.Apprise()

# Add custom service using URL
apobj.add('custom://token@api.example.com/channel-123')

# Or add with parameters
custom_url = 'custom://user:pass@api.example.com'
apobj.add(custom_url)

# Send notification through custom plugin
success = apobj.notify(
    body='Test message through custom plugin',
    title='Custom Plugin Test',
    notify_type=apprise.NotifyType.INFO
)

if success:
    print("Custom plugin notification sent successfully")

import apprise

apobj = apprise.Apprise()

# Add various services to demonstrate plugin capabilities
services = [
    'mailto://user:pass@gmail.com',      # Email plugin
    'slack://TokenA/TokenB/TokenC/Ch',   # Slack plugin  
    'discord://webhook_id/webhook_token', # Discord plugin
    'telegram://bot_token/chat_id',      # Telegram plugin
]

for service_url in services:
    try:
        apobj.add(service_url)
    except Exception as e:
        print(f"Failed to add service {service_url}: {e}")

# Examine loaded services
for i, server in enumerate(apobj.servers):
    print(f"\nService {i + 1}: {server.service_name}")
    print(f"  Protocol: {server.protocol}")
    print(f"  Enabled: {server.enabled}")
    print(f"  Format: {server.notify_format}")
    print(f"  Requirements: {getattr(server, 'requirements', {})}")
    
    # Get service URL (with privacy)
    service_url = server.url(privacy=True)
    print(f"  URL: {service_url}")
    
    # Check rate limiting
    throttle_delay = server.throttle()
    if throttle_delay > 0:
        print(f"  Rate limit: {throttle_delay}s delay")

import apprise

def diagnose_plugin_issues():
    """Diagnose common plugin configuration issues."""
    apobj = apprise.Apprise()
    
    # Test problematic service URLs
    test_urls = [
        'slack://invalid/token/format',      # Invalid format
        'discord://missing-webhook',         # Incomplete URL
        'mailto://badformat@',               # Malformed email
        'unknown://unsupported/service',     # Unknown protocol
    ]
    
    for url in test_urls:
        try:
            success = apobj.add(url)
            if success:
                print(f"✓ Successfully added: {url}")
            else:
                print(f"✗ Failed to add: {url}")
        except Exception as e:
            print(f"✗ Exception adding {url}: {e}")
    
    # Check for disabled services
    details = apobj.details(show_disabled=True)
    disabled_services = [
        name for name, info in details.items() 
        if not info.get('enabled', True)
    ]
    
    if disabled_services:
        print(f"\nDisabled services: {disabled_services}")
        for service in disabled_services:
            service_info = details[service]
            requirements = service_info.get('requirements', {})
            if requirements:
                print(f"  {service} requirements: {requirements}")

diagnose_plugin_issues()

import apprise

# Configure services with advanced plugin features
def setup_advanced_notifications():
    apobj = apprise.Apprise()
    
    # Email with HTML support
    email_url = 'mailto://user:app_password@gmail.com'
    apobj.add(email_url)
    
    # Slack with custom formatting
    slack_url = 'slack://xoxb-token/channel?format=markdown'
    apobj.add(slack_url)
    
    # Discord with custom avatar
    discord_url = 'discord://webhook_id/webhook_token?avatar=no'
    apobj.add(discord_url)
    
    # Send with different formats to demonstrate plugin capabilities
    test_cases = [
        {
            'body': '<b>HTML formatted message</b>',
            'title': 'HTML Test',
            'body_format': apprise.NotifyFormat.HTML
        },
        {
            'body': '**Markdown** formatted message',
            'title': 'Markdown Test', 
            'body_format': apprise.NotifyFormat.MARKDOWN
        },
        {
            'body': 'Plain text message',
            'title': 'Text Test',
            'body_format': apprise.NotifyFormat.TEXT
        }
    ]
    
    for test_case in test_cases:
        success = apobj.notify(**test_case)
        print(f"Format {test_case['body_format']}: {'✓' if success else '✗'}")

setup_advanced_notifications()