63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
from aiohttp.client import ClientSession
|
|
from channels.layers import get_channel_layer
|
|
from channels.db import database_sync_to_async
|
|
from urllib.parse import quote as urlencode
|
|
|
|
from core.settings import TELEGRAM_BOT_TOKEN, TELEGRAM_GROUP_CHAT_ID
|
|
from mail.models import UserNotificationChannel
|
|
|
|
|
|
async def http_get(url):
|
|
async with ClientSession() as session:
|
|
async with session.get(url) as response:
|
|
return await response.text()
|
|
|
|
|
|
async def telegram_notify(message, chat_id):
|
|
encoded_message = urlencode(message)
|
|
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage?chat_id={chat_id}&text={encoded_message}"
|
|
return await http_get(url)
|
|
|
|
|
|
class NotificationDispatcher:
|
|
channel_layer = None
|
|
room_group_name = "general"
|
|
|
|
def __init__(self):
|
|
self.channel_layer = get_channel_layer('default')
|
|
if not self.channel_layer:
|
|
raise Exception("Could not get channel layer")
|
|
|
|
@database_sync_to_async
|
|
def get_notification_targets(self):
|
|
channels = UserNotificationChannel.objects.filter(active=True)
|
|
return list(channels)
|
|
|
|
async def run_forever(self):
|
|
# Infinite loop to continuously listen for messages
|
|
print("Listening for messages...")
|
|
channel_name = await self.channel_layer.new_channel()
|
|
await self.channel_layer.group_add(self.room_group_name, channel_name)
|
|
print("Channel name:", channel_name)
|
|
while True:
|
|
# Blocking receive to get the message from the channel layer
|
|
message = await self.channel_layer.receive(channel_name)
|
|
|
|
if (message and 'type' in message and message['type'] == 'generic.event' and 'name' in message and
|
|
message['name'] == 'user_notification'):
|
|
if 'message' in message and 'event_id' in message:
|
|
await self.dispatch(message['message'], message['event_id'])
|
|
else:
|
|
print("Error: Invalid message format")
|
|
|
|
async def dispatch(self, message, event_id):
|
|
print("Dispatching message:", message, "with event_id:", event_id)
|
|
targets = await self.get_notification_targets()
|
|
await telegram_notify(message, TELEGRAM_GROUP_CHAT_ID)
|
|
for target in targets:
|
|
if target.channel_type == 'telegram':
|
|
await telegram_notify(message, target.channel_target)
|
|
elif target.channel_type == 'email':
|
|
print("Sending mail to:", target.channel_target)
|
|
else:
|
|
print("Unknown channel type:", target.channel_type)
|