"""
Example Plugin: Discord Webhook Integration
Sends notifications to Discord when events occur.
"""
import sys
sys.path.append('../api')
from base_plugin import BasePlugin
import requests
import threading
import time
class DiscordWebhookPlugin(BasePlugin):
def on_load(self):
"""Initialize plugin."""
self.log("Discord Webhook Plugin loaded")
# Register custom macro type
self.register_macro_type('discord_webhook', self.handle_discord_webhook)
# Get settings
self.webhook_url = self.get_setting('webhook_url', '')
if not self.webhook_url:
self.log("Warning: Webhook URL not configured", level="warning")
# Notification settings
self.notify_on_connect = self.get_setting('notify_on_connect', True)
self.notify_on_button = self.get_setting('notify_on_button', False)
def on_unload(self):
"""Cleanup."""
self.log("Discord Webhook Plugin unloading")
def on_device_connect(self, port: str):
"""Device connected."""
if self.notify_on_connect and self.webhook_url:
self.send_webhook(
title="Device Connected",
message=f"Connected on port {port}",
color="00FF00"
)
def on_device_disconnect(self):
"""Device disconnected."""
if self.notify_on_connect and self.webhook_url:
self.send_webhook(
title="Device Disconnected",
message="Waiting for reconnection...",
color="FF0000"
)
def on_macro_button(self, index: int, page: int):
"""Macro button pressed."""
if self.notify_on_button and self.webhook_url:
self.log(f"Macro button {index+1} pressed on page {page+1}")
def on_spotify_track_change(self, track_info: dict):
"""Spotify track changed."""
if self.get_setting('notify_on_track_change', False) and self.webhook_url:
title = track_info.get('title', 'Unknown')
artist = track_info.get('artist', 'Unknown')
self.send_webhook(
title="🎵 Now Playing",
message=f"{title}\n{artist}",
color="1DB954"
)
def handle_discord_webhook(self, macro: dict):
"""Execute discord_webhook macro."""
if not self.webhook_url:
self.log("Cannot send webhook - URL not configured", level="error")
return
title = macro.get('title', 'Notification')
message = macro.get('message', '')
color = macro.get('color', 'FF5733')
self.send_webhook(title, message, color)
def send_webhook(self, title: str, message: str, color: str):
"""
Send Discord webhook message.
Args:
title: Embed title
message: Embed description
color: Hex color (without #)
"""
if not self.webhook_url:
return
# Convert hex to decimal
try:
color_int = int(color.lstrip('#'), 16)
except ValueError:
color_int = 0xFF5733 # Default orange
payload = {
'embeds': [{
'title': title,
'description': message,
'color': color_int,
'timestamp': time.strftime('%Y-%m-%dT%H:%M:%S')
}]
}
try:
response = requests.post(
self.webhook_url,
json=payload,
timeout=5
)
if response.status_code == 204:
self.log(f"Webhook sent: {title}")
else:
self.log(f"Webhook failed: {response.status_code}", level="error")
except Exception as e:
self.log(f"Webhook error: {e}", level="error")
