import asyncio 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.protocol import send_smtp, make_notification from notifications.models import UserNotificationChannel from notifications.templates import render_notification_new_ticket_async, render_notification_reply_ticket_async from tickets.models import IssueThread 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) async def email_notify(message, title, email): mail = make_notification(message, email, title) await send_smtp(mail) 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) @database_sync_to_async def get_ticket(self, ticket_id): return IssueThread.objects.filter(id=ticket_id).select_related('event').first() 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 'ticket_id' in message and 'event_id' in message and 'new' in message: ticket = await self.get_ticket(message['ticket_id']) await self.dispatch(ticket, message['event_id'], message['new']) else: print("Error: Invalid message format") async def dispatch(self, ticket, event_id, new): message = await render_notification_new_ticket_async( ticket) if new else await render_notification_reply_ticket_async(ticket) title = f"[#{ticket.short_uuid()}] {ticket.name}" print("Dispatching message:", message, "with event_id:", event_id) targets = await self.get_notification_targets() jobs = [] jobs.append(telegram_notify(message, TELEGRAM_GROUP_CHAT_ID)) for target in targets: if target.channel_type == 'telegram': print("Sending telegram notification to:", target.channel_target) jobs.append(telegram_notify(message, target.channel_target)) elif target.channel_type == 'email': print("Sending email notification to:", target.channel_target) jobs.append(email_notify(message, title, target.channel_target)) else: print("Unknown channel type:", target.channel_type) await asyncio.gather(*jobs)