-
Notifications
You must be signed in to change notification settings - Fork 8
Expand file tree
/
Copy pathnotification_service.py
More file actions
168 lines (138 loc) · 6.43 KB
/
notification_service.py
File metadata and controls
168 lines (138 loc) · 6.43 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
import asyncio
import logging
from typing import Optional
from telegram import Bot
from data_manager import DataChangeEvent, get_shared_data_manager
class BotNotificationHandler:
def __init__(self, bot_token: str = None, chat_ids: list = None):
self.bot_token = bot_token
self.chat_ids = chat_ids or []
self.bot: Optional[Bot] = None
self.enabled = False
if bot_token and chat_ids:
self.bot = Bot(token=bot_token)
self.enabled = True
shared_dm = get_shared_data_manager()
shared_dm.add_observer(self.handle_data_change)
def update_config(self, bot_token: str, chat_ids: list):
self.bot_token = bot_token
self.chat_ids = chat_ids
if bot_token and chat_ids:
self.bot = Bot(token=bot_token)
self.enabled = True
else:
self.enabled = False
def handle_data_change(self, event: DataChangeEvent):
if not self.enabled or not self.bot or not self.chat_ids:
return
try:
asyncio.create_task(self._send_notification(event))
except Exception as e:
logging.error(f"Error creating notification task: {e}")
async def _send_notification(self, event: DataChangeEvent):
try:
message = self._format_notification_message(event)
if not message:
return
for chat_id in self.chat_ids:
try:
await self.bot.send_message(
chat_id=chat_id, text=message, parse_mode="Markdown"
)
except Exception as e:
logging.error(f"Error sending notification to {chat_id}: {e}")
except Exception as e:
logging.error(f"Error in _send_notification: {e}")
def _format_notification_message(self, event: DataChangeEvent) -> str:
if event.data_type == "server":
custom_emoji = event.data.get("emoji", "🔹") if event.data else "🔹"
if custom_emoji == "🔹":
emoji = "👉"
else:
emoji = custom_emoji
if event.event_type == "add":
return (
f"✅ New Server Added\n\n"
f"{emoji} {event.item_name}\n"
f"• Expiration Date: `{event.data.get('date', 'Not specified')}`\n"
f"• Price: `{event.data.get('price', 'Not specified')}`\n"
f"• Datacenter: `{event.data.get('datacenter', 'Not specified')}`\n"
f"• Label: `{event.data.get('label', 'None') if event.data.get('label') else 'None'}`\n\n"
f"- Added via WebPanel"
)
elif event.event_type == "update":
return (
f"✅ Server Updated\n\n"
f"{emoji} {event.item_name}\n"
f"• Expiration Date: `{event.data.get('date', 'Not specified')}`\n"
f"• Price: `{event.data.get('price', 'Not specified')}`\n"
f"• Datacenter: `{event.data.get('datacenter', 'Not specified')}`\n"
f"• Label: `{event.data.get('label', 'None') if event.data.get('label') else 'None'}`\n\n"
f"- Updated via WebPanel"
)
elif event.event_type == "delete":
return (
f"🗑️ `{event.item_name}` has been removed from the system.\n\n"
f"- Deleted via WebPanel"
)
elif event.data_type == "domain":
custom_emoji = event.data.get("emoji", "🔹") if event.data else "🔹"
if custom_emoji == "🔹":
emoji = "👉"
else:
emoji = custom_emoji
if event.event_type == "add":
return (
f"✅ New Domain Added\n\n"
f"{emoji} {event.item_name}\n"
f"• Expiration Date: `{event.data.get('date', 'Not specified')}`\n"
f"• Price: `{event.data.get('price', 'Not specified')}`\n"
f"• Registrar: `{event.data.get('registrar', 'Not specified')}`\n\n"
f"- Added via WebPanel"
)
elif event.event_type == "update":
return (
f"✅ Domain Updated\n\n"
f"{emoji} {event.item_name}\n"
f"• Expiration Date: `{event.data.get('date', 'Not specified')}`\n"
f"• Price: `{event.data.get('price', 'Not specified')}`\n"
f"• Registrar: `{event.data.get('registrar', 'Not specified')}`\n\n"
f"- Updated via WebPanel"
)
elif event.event_type == "delete":
return (
f"🗑️ `{event.item_name}` has been removed from the system.\n\n"
f"- Deleted via WebPanel"
)
elif event.data_type == "settings":
return (
f"⚙️ Settings Updated\n\n"
f"• Warning Days: `{event.data.get('warning_days', 'Not specified')}`\n"
f"• Notification Time: `{event.data.get('notification_hour', 9):02d}:{event.data.get('notification_minute', 0):02d}`\n"
f"• Daily Notifications: `{'Enabled' if event.data.get('daily_notifications', True) else 'Disabled'}`\n\n"
f"- Updated via WebPanel"
)
elif event.data_type == "label":
if event.event_type == "add":
return (
f"🏷️ New Label Added\n\n"
f"• Name: `{event.item_name}`\n\n"
f"- Added via WebPanel"
)
elif event.event_type == "delete":
return (
f"🗑️ Label Removed\n\n"
f"• Name: `{event.item_name}`\n\n"
f"- Deleted via WebPanel"
)
return ""
_notification_handler = None
def get_notification_handler() -> BotNotificationHandler:
global _notification_handler
if _notification_handler is None:
_notification_handler = BotNotificationHandler()
return _notification_handler
def setup_bot_notifications(bot_token: str, chat_ids: list):
handler = get_notification_handler()
handler.update_config(bot_token, chat_ids)
logging.info("Bot notifications enabled")