343 lines
11 KiB
Python
Executable File
343 lines
11 KiB
Python
Executable File
#!/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()
|