282 lines
8.8 KiB
Python
282 lines
8.8 KiB
Python
"""
|
|
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'
|
|
]
|