c3lf-system-3/core/notifications/dispatch.py

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)