stash

core/mail/migrations/0005_usernotificationchannel.py

@@ -0,0 +1,28 @@
+# Generated by Django 4.2.7 on 2024-04-26 13:08
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('mail', '0004_alter_emailattachment_file'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='UserNotificationChannel',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('channel_type', models.CharField(choices=[('telegram', 'telegram'), ('email', 'email')], max_length=255)),
+                ('channel_target', models.CharField(max_length=255)),
+                ('event_filter', models.CharField(max_length=255)),
+                ('active', models.BooleanField(default=True)),
+                ('created', models.DateTimeField(auto_now_add=True)),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+        ),
+    ]

core/mail/models.py

@@ -3,6 +3,7 @@ import random
 from django.db import models
 from django_softdelete.models import SoftDeleteModel
 
+from authentication.models import ExtendedUser
 from core.settings import MAIL_DOMAIN
 from files.models import AbstractFile
 from inventory.models import Event
@@ -38,3 +39,15 @@ class EventAddress(models.Model):
 class EmailAttachment(AbstractFile):
     email = models.ForeignKey(Email, models.CASCADE, related_name='attachments', null=True)
     name = models.CharField(max_length=255)
+
+
+class UserNotificationChannel(models.Model):
+    user = models.ForeignKey(ExtendedUser, models.CASCADE)
+    channel_type = models.CharField(choices=[('telegram', 'telegram'), ('email', 'email')], max_length=255)
+    channel_target = models.CharField(max_length=255)
+    event_filter = models.CharField(max_length=255)
+    active = models.BooleanField(default=True)
+    created = models.DateTimeField(auto_now_add=True)
+
+    def validate_constraints(self, exclude=None):  # TODO: email -> emailaddress, telegram -> chatid
+        return True

core/mail/notifications.py

@@ -1,5 +1,22 @@
+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:
@@ -13,7 +30,8 @@ class NotificationDispatcher:
 
     @database_sync_to_async
     def get_notification_targets(self):
-        return []
+        channels = UserNotificationChannel.objects.filter(active=True)
+        return list(channels)
 
     async def run_forever(self):
         # Infinite loop to continuously listen for messages
@@ -35,5 +53,11 @@ class NotificationDispatcher:
     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:
-            print("Sending message to target:", target)
+            if target.channel_type == 'telegram':
+                await telegram_notify(message, target.channel_target)
+            if target.channel_type == 'mail':
+                print("Sending mail to:", target.channel_target)
+            else:
+                print("Unknown channel type:", target.channel_type)

core/mail/tests/v2/test_user_notifications.py

@@ -0,0 +1,20 @@
+from django.contrib.auth.models import Permission
+from django.test import TestCase
+
+from authentication.models import ExtendedUser
+from mail.models import UserNotificationChannel
+
+
+class UserNotificationTestCase(TestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.user = ExtendedUser.objects.create_user('testuser', 'test', 'test')
+        self.user.user_permissions.add(*Permission.objects.all())
+        self.user.save()
+        self.channel = UserNotificationChannel.objects.create(user=self.user, channel_type='telegram',
+                                                              channel_target='123456789',
+                                                              event_filter='*', active=True)
+
+    async def test_telegram_notify(self):
+        pass