notify-forwarder/notify_forwarder.py

#!/usr/bin/env python3
"""
GNOME Notification Forwarder to Bark
Monitors GNOME desktop notifications and forwards them to Bark API
"""

import dbus
import dbus.mainloop.glib
from gi.repository import GLib
import requests
import urllib.parse
import logging
from threading import Thread
import hashlib
import time
from collections import deque
import os
import sys
import configparser
from pathlib import Path

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Configuration file path
SCRIPT_DIR = Path(__file__).parent
CONFIG_FILE = SCRIPT_DIR / 'config.ini'

# Read configuration
config = configparser.ConfigParser()

if CONFIG_FILE.exists():
    logger.info(f"Loading configuration from {CONFIG_FILE}")
    config.read(CONFIG_FILE)

    # Notification service configuration
    NOTIFICATION_SERVICE = config.get('notification', 'service', fallback='bark').lower()

    # Bark API configuration
    BARK_URL = config.get('bark', 'url', fallback=None)

    # Gotify configuration
    GOTIFY_URL = config.get('gotify', 'url', fallback=None)
    GOTIFY_TOKEN = config.get('gotify', 'token', fallback=None)

    # Ntfy configuration
    NTFY_SERVER = config.get('ntfy', 'server', fallback='https://ntfy.sh')
    NTFY_TOPIC = config.get('ntfy', 'topic', fallback=None)
else:
    # Fallback to environment variables if config file doesn't exist
    logger.warning(f"Config file not found at {CONFIG_FILE}, falling back to environment variables")
    NOTIFICATION_SERVICE = os.environ.get('NOTIFICATION_SERVICE', 'bark').lower()
    BARK_URL = os.environ.get('BARK_URL')
    GOTIFY_URL = os.environ.get('GOTIFY_URL')
    GOTIFY_TOKEN = os.environ.get('GOTIFY_TOKEN')
    NTFY_SERVER = os.environ.get('NTFY_SERVER', 'https://ntfy.sh')
    NTFY_TOPIC = os.environ.get('NTFY_TOPIC')

# Parse notification services (support comma-separated or 'all')
if NOTIFICATION_SERVICE == 'all':
    ENABLED_SERVICES = ['bark', 'gotify', 'ntfy']
else:
    ENABLED_SERVICES = [s.strip() for s in NOTIFICATION_SERVICE.split(',')]

# Validate configuration based on enabled services
if 'bark' in ENABLED_SERVICES:
    if not BARK_URL:
        logger.error("Error: BARK_URL is not configured!")
        logger.error(f"Please set 'url' in the [bark] section of {CONFIG_FILE}")
        logger.error("Or set the BARK_URL environment variable")
        sys.exit(1)

if 'gotify' in ENABLED_SERVICES:
    if not GOTIFY_URL or not GOTIFY_TOKEN:
        logger.error("Error: Gotify URL and TOKEN are not configured!")
        logger.error(f"Please set 'url' and 'token' in the [gotify] section of {CONFIG_FILE}")
        logger.error("Or set the GOTIFY_URL and GOTIFY_TOKEN environment variables")
        sys.exit(1)

if 'ntfy' in ENABLED_SERVICES:
    if not NTFY_TOPIC:
        logger.error("Error: Ntfy TOPIC is not configured!")
        logger.error(f"Please set 'topic' in the [ntfy] section of {CONFIG_FILE}")
        logger.error("Or set the NTFY_TOPIC environment variable")
        sys.exit(1)

# Validate that at least one valid service is configured
valid_services = {'bark', 'gotify', 'ntfy'}
if not any(s in valid_services for s in ENABLED_SERVICES):
    logger.error(f"Error: Invalid NOTIFICATION_SERVICE '{NOTIFICATION_SERVICE}'")
    logger.error("Valid options are: bark, gotify, ntfy, or combinations like 'bark,gotify' or 'all'")
    sys.exit(1)

# Deduplication cache - stores hashes of recent notifications
# Format: (hash, timestamp)
recent_notifications = deque(maxlen=100)


def send_to_bark(title, body, app_name=""):
    """
    Send notification to Bark API (runs in separate thread to avoid blocking)

    Args:
        title: Notification title
        body: Notification body
        app_name: Application name that triggered the notification
    """
    try:
        # Construct message
        if app_name:
            message = f"[{app_name}] {body}" if body else f"[{app_name}]"
        else:
            message = body or "Notification"

        # URL encode the parameters
        title_encoded = urllib.parse.quote(title or "Notification")
        message_encoded = urllib.parse.quote(message)

        # Construct full URL
        url = f"{BARK_URL}/{title_encoded}/{message_encoded}"

        # Send request with timeout
        response = requests.get(url, timeout=10)

        if response.status_code == 200:
            logger.info(f"✓ [Bark] Forwarded: {title} - {message[:50]}...")
        else:
            logger.error(f"✗ [Bark] Failed to forward notification. Status: {response.status_code}")

    except requests.exceptions.RequestException as e:
        logger.error(f"✗ [Bark] Network error: {e}")
    except Exception as e:
        logger.error(f"✗ [Bark] Error sending: {e}")


def send_to_gotify(title, body, app_name=""):
    """
    Send notification to Gotify API (runs in separate thread to avoid blocking)

    Args:
        title: Notification title
        body: Notification body
        app_name: Application name that triggered the notification
    """
    try:
        # Construct message
        if app_name:
            message = f"[{app_name}] {body}" if body else f"[{app_name}]"
        else:
            message = body or "Notification"

        # Prepare Gotify message
        url = f"{GOTIFY_URL}/message"
        params = {'token': GOTIFY_TOKEN}
        data = {
            'title': title or "Notification",
            'message': message,
            'priority': 5
        }

        # Send request with timeout
        response = requests.post(url, params=params, json=data, timeout=10)

        if response.status_code == 200:
            logger.info(f"✓ [Gotify] Forwarded: {title} - {message[:50]}...")
        else:
            logger.error(f"✗ [Gotify] Failed to forward notification. Status: {response.status_code}")

    except requests.exceptions.RequestException as e:
        logger.error(f"✗ [Gotify] Network error: {e}")
    except Exception as e:
        logger.error(f"✗ [Gotify] Error sending: {e}")


def send_to_ntfy(title, body, app_name=""):
    """
    Send notification to Ntfy (runs in separate thread to avoid blocking)

    Args:
        title: Notification title
        body: Notification body
        app_name: Application name that triggered the notification
    """
    try:
        # Construct message
        if app_name:
            message = f"[{app_name}] {body}" if body else f"[{app_name}]"
        else:
            message = body or "Notification"

        # Prepare Ntfy request
        url = f"{NTFY_SERVER}/{NTFY_TOPIC}"
        headers = {
            'Title': title or "Notification",
            'Priority': '3',
            'Tags': 'computer'
        }

        # Send request with timeout
        response = requests.post(url, data=message.encode('utf-8'), headers=headers, timeout=10)

        if response.status_code == 200:
            logger.info(f"✓ [Ntfy] Forwarded: {title} - {message[:50]}...")
        else:
            logger.error(f"✗ [Ntfy] Failed to forward notification. Status: {response.status_code}")

    except requests.exceptions.RequestException as e:
        logger.error(f"✗ [Ntfy] Network error: {e}")
    except Exception as e:
        logger.error(f"✗ [Ntfy] Error sending: {e}")


def message_filter(bus, message):
    """
    Filter and handle D-Bus messages
    """
    try:
        # Check if this is a method call to Notify
        if message.get_member() != "Notify":
            return

        if message.get_interface() != "org.freedesktop.Notifications":
            return

        # Get the arguments
        args = message.get_args_list()
        if len(args) < 5:
            return

        app_name = str(args[0])
        replaces_id = args[1]
        app_icon = args[2]
        summary = str(args[3])
        body = str(args[4])

        # Skip empty notifications
        if not summary and not body:
            return

        # Create a hash of the notification for deduplication
        notification_str = f"{app_name}:{summary}:{body}"
        notification_hash = hashlib.md5(notification_str.encode()).hexdigest()
        current_time = time.time()

        # Check if this notification was recently processed (within 2 seconds)
        for cached_hash, cached_time in recent_notifications:
            if cached_hash == notification_hash and (current_time - cached_time) < 2:
                # Duplicate notification, skip it
                return

        # Add to cache
        recent_notifications.append((notification_hash, current_time))

        # Log the notification
        logger.info(f"Received: [{app_name}] {summary}")

        # Forward to notification service(s) in separate thread(s) to avoid blocking
        if 'bark' in ENABLED_SERVICES:
            thread = Thread(target=send_to_bark, args=(summary, body, app_name))
            thread.daemon = True
            thread.start()

        if 'gotify' in ENABLED_SERVICES:
            thread = Thread(target=send_to_gotify, args=(summary, body, app_name))
            thread.daemon = True
            thread.start()

        if 'ntfy' in ENABLED_SERVICES:
            thread = Thread(target=send_to_ntfy, args=(summary, body, app_name))
            thread.daemon = True
            thread.start()

    except Exception as e:
        logger.error(f"Error handling message: {e}")


def main():
    """
    Main function to set up D-Bus monitoring
    """
    logger.info("Starting GNOME Notification Forwarder...")
    logger.info(f"Enabled services: {', '.join(ENABLED_SERVICES)}")

    if 'bark' in ENABLED_SERVICES:
        logger.info(f"Bark URL: {BARK_URL}")

    if 'gotify' in ENABLED_SERVICES:
        logger.info(f"Gotify URL: {GOTIFY_URL}")

    if 'ntfy' in ENABLED_SERVICES:
        logger.info(f"Ntfy server: {NTFY_SERVER}")
        logger.info(f"Ntfy topic: {NTFY_TOPIC}")

    try:
        # Set up D-Bus main loop
        dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

        # Get session bus
        bus = dbus.SessionBus()

        # Become a monitor to intercept messages
        bus.call_blocking(
            'org.freedesktop.DBus',
            '/org/freedesktop/DBus',
            'org.freedesktop.DBus.Monitoring',
            'BecomeMonitor',
            'asu',
            ([
                "type='method_call',interface='org.freedesktop.Notifications',member='Notify'"
            ], 0)
        )

        # Add message filter
        bus.add_message_filter(message_filter)

        logger.info("✓ Monitoring notifications... Press Ctrl+C to stop.")

        # Run main loop
        loop = GLib.MainLoop()
        loop.run()

    except KeyboardInterrupt:
        logger.info("\nStopping notification forwarder...")
    except dbus.exceptions.DBusException as e:
        if "org.freedesktop.DBus.Monitoring" in str(e):
            logger.error("Error: Unable to become D-Bus monitor.")
            logger.error("This might require additional permissions or a policy update.")
            logger.error("Try running with: dbus-send or check D-Bus policies.")
        else:
            logger.error(f"D-Bus error: {e}")
        raise
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        raise


if __name__ == "__main__":
    main()