""" Lemontropia Suite - Notification System Send alerts to Discord, Telegram, or other services on important events. """ import json import logging import requests from decimal import Decimal from pathlib import Path from dataclasses import dataclass from typing import Optional, Dict, List from datetime import datetime logger = logging.getLogger(__name__) @dataclass class NotificationConfig: """Configuration for notifications.""" discord_webhook: Optional[str] = None telegram_bot_token: Optional[str] = None telegram_chat_id: Optional[str] = None # Event filters notify_on_global: bool = True notify_on_hof: bool = True notify_on_profit_threshold: bool = False profit_threshold: Decimal = Decimal("100") notify_on_loss_threshold: bool = True loss_threshold: Decimal = Decimal("-50") class DiscordNotifier: """Send notifications to Discord via webhook.""" def __init__(self, webhook_url: str): self.webhook_url = webhook_url def send_message(self, content: str, embeds: Optional[List[Dict]] = None) -> bool: """Send message to Discord.""" try: data = {"content": content} if embeds: data["embeds"] = embeds response = requests.post( self.webhook_url, json=data, timeout=10 ) return response.status_code == 204 except Exception as e: logger.error(f"Failed to send Discord notification: {e}") return False def send_loot_alert(self, item_name: str, value: Decimal, is_hof: bool = False): """Send loot alert.""" color = 0xFFD700 if is_hof else 0x00FF00 # Gold for HoF, Green for Global title = "🏆 HALL OF FAME!" if is_hof else "🌟 GLOBAL!" embed = { "title": title, "description": f"**{item_name}**", "color": color, "fields": [ { "name": "Value", "value": f"{value:.2f} PED", "inline": True } ], "timestamp": datetime.now().isoformat() } return self.send_message("", [embed]) def send_session_summary(self, session_data: Dict): """Send session summary.""" profit = session_data.get('profit_loss', Decimal("0")) color = 0x00FF00 if profit >= 0 else 0xFF0000 embed = { "title": "Hunting Session Complete", "color": color, "fields": [ { "name": "Duration", "value": session_data.get('duration', 'Unknown'), "inline": True }, { "name": "Profit/Loss", "value": f"{profit:+.2f} PED", "inline": True }, { "name": "Return %", "value": f"{session_data.get('return_pct', 0):.1f}%", "inline": True }, { "name": "Globals", "value": str(session_data.get('globals', 0)), "inline": True } ], "timestamp": datetime.now().isoformat() } return self.send_message("", [embed]) class TelegramNotifier: """Send notifications to Telegram via bot.""" def __init__(self, bot_token: str, chat_id: str): self.bot_token = bot_token self.chat_id = chat_id self.base_url = f"https://api.telegram.org/bot{bot_token}" def send_message(self, text: str) -> bool: """Send message to Telegram.""" try: url = f"{self.base_url}/sendMessage" data = { "chat_id": self.chat_id, "text": text, "parse_mode": "Markdown" } response = requests.post(url, json=data, timeout=10) return response.json().get("ok", False) except Exception as e: logger.error(f"Failed to send Telegram notification: {e}") return False def send_loot_alert(self, item_name: str, value: Decimal, is_hof: bool = False): """Send loot alert.""" emoji = "🏆" if is_hof else "🌟" title = "HALL OF FAME!" if is_hof else "GLOBAL!" text = f"{emoji} *{title}* {emoji}\n\n" text += f"*{item_name}*\n" text += f"Value: *{value:.2f} PED*" return self.send_message(text) def send_session_summary(self, session_data: Dict): """Send session summary.""" profit = session_data.get('profit_loss', Decimal("0")) emoji = "✅" if profit >= 0 else "❌" text = f"{emoji} *Session Complete*\n\n" text += f"Duration: {session_data.get('duration', 'Unknown')}\n" text += f"P/L: *{profit:+.2f} PED*\n" text += f"Return: {session_data.get('return_pct', 0):.1f}%\n" text += f"Globals: {session_data.get('globals', 0)}" return self.send_message(text) class NotificationManager: """ Central notification manager. Handles all notification services and event filtering. """ def __init__(self, config: Optional[NotificationConfig] = None): self.config = config or NotificationConfig() self.discord: Optional[DiscordNotifier] = None self.telegram: Optional[TelegramNotifier] = None if self.config.discord_webhook: self.discord = DiscordNotifier(self.config.discord_webhook) if self.config.telegram_bot_token and self.config.telegram_chat_id: self.telegram = TelegramNotifier( self.config.telegram_bot_token, self.config.telegram_chat_id ) def on_global(self, item_name: str, value: Decimal): """Handle global event.""" if not self.config.notify_on_global: return logger.info(f"Sending global notification: {item_name} - {value} PED") if self.discord: self.discord.send_loot_alert(item_name, value, is_hof=False) if self.telegram: self.telegram.send_loot_alert(item_name, value, is_hof=False) def on_hof(self, item_name: str, value: Decimal): """Handle HoF event.""" if not self.config.notify_on_hof: return logger.info(f"Sending HoF notification: {item_name} - {value} PED") if self.discord: self.discord.send_loot_alert(item_name, value, is_hof=True) if self.telegram: self.telegram.send_loot_alert(item_name, value, is_hof=True) def on_session_end(self, session_data: Dict): """Handle session end.""" profit = session_data.get('profit_loss', Decimal("0")) # Check thresholds if self.config.notify_on_profit_threshold and profit >= self.config.profit_threshold: pass # Will send below elif self.config.notify_on_loss_threshold and profit <= self.config.loss_threshold: pass # Will send below else: return logger.info(f"Sending session summary: {profit:+.2f} PED") if self.discord: self.discord.send_session_summary(session_data) if self.telegram: self.telegram.send_session_summary(session_data) def send_custom_message(self, message: str): """Send custom message to all channels.""" if self.discord: self.discord.send_message(message) if self.telegram: self.telegram.send_message(message) class SoundNotifier: """ Play sound alerts locally. """ def __init__(self): self.enabled = True self.global_sound = "global.wav" # Would need actual sound files self.hof_sound = "hof.wav" def play_global_sound(self): """Play global alert sound.""" if not self.enabled: return try: if sys.platform == 'win32': import winsound winsound.MessageBeep(winsound.MB_OK) except Exception as e: logger.error(f"Failed to play sound: {e}") def play_hof_sound(self): """Play HoF alert sound.""" if not self.enabled: return try: if sys.platform == 'win32': import winsound winsound.MessageBeep(winsound.MB_ICONEXCLAMATION) except Exception as e: logger.error(f"Failed to play sound: {e}") # Export main classes __all__ = [ 'NotificationManager', 'DiscordNotifier', 'TelegramNotifier', 'SoundNotifier', 'NotificationConfig' ]