#470 Notification Task and Logic

core/migrations/0035_auto_20231108_1041.py

@@ -0,0 +1,37 @@
+# Generated by Django 3.2.20 on 2023-11-08 09:41
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0034_auto_20230515_1353"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="access",
+            name="user",
+            field=models.ForeignKey(
+                blank=True,
+                help_text="Use either `contact` or `user`",
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="accesses",
+                to=settings.AUTH_USER_MODEL,
+                verbose_name="User that has the access",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="project",
+            name="local_custodians",
+            field=models.ManyToManyField(
+                help_text="Custodians are the local responsibles for the project. This list must include a PI.",
+                related_name="project_set",
+                to=settings.AUTH_USER_MODEL,
+                verbose_name="Local custodians",
+            ),
+        ),
+    ]

core/models/access.py

@@ -1,21 +1,31 @@
 import logging
+import typing
 
-from datetime import datetime, date
+from datetime import datetime, date, timedelta
 from typing import List
 
+from django.conf import settings
+from django.contrib.contenttypes.models import ContentType
+from django.contrib.auth import get_user_model
 from django.core.exceptions import ValidationError
 from django.db import models
 from django.db.models import Q, ObjectDoesNotExist, Count, signals
 
 from enumchoicefield import EnumChoiceField, ChoiceEnum
 
-from .utils import CoreModel
+from .utils import CoreModel, CoreNotifyMeta
+from notification import NotifyMixin
+from notification.models import NotificationVerb, Notification
+from core.utils import DaisyLogger
 
 from auditlog.registry import auditlog
 from auditlog.models import AuditlogHistoryField
 
+if typing.TYPE_CHECKING:
+    User = settings.AUTH_USER_MODEL
 
-logger = logging.getLogger(__name__)
+
+logger = DaisyLogger(__name__)
 
 
 class StatusChoices(ChoiceEnum):
@@ -25,7 +35,7 @@ class StatusChoices(ChoiceEnum):
     terminated = "Terminated"
 
 
-class Access(CoreModel):
+class Access(CoreModel, NotifyMixin, metaclass=CoreNotifyMeta):
     """
     Represents the access given to an internal (LCSB) entity over data storage locations.
     """
@@ -161,7 +171,7 @@ def clean(self):
 
     user = models.ForeignKey(
         "core.User",
-        related_name="user",
+        related_name="accesses",
         verbose_name="User that has the access",
         on_delete=models.CASCADE,
         null=True,
@@ -240,5 +250,69 @@ def is_active(self):
 
         return True
 
+    @staticmethod
+    def get_notification_recipients():
+        """
+        Get distinct users that are local custodian of a dataset or a project.
+        """
+
+        return (
+            get_user_model()
+            .objects.filter(Q(datasets__isnull=False) | Q(project_set__isnull=False))
+            .distinct()
+        )
+
+    @classmethod
+    def make_notifications_for_user(
+        cls, day_offset: timedelta, exec_date: date, user: "User"
+    ):
+        # Considering users that are indirectly responsible for the dataset (through projects)
+        possible_datasets = set(user.datasets.all())
+        possible_datasets.update(
+            [
+                dataset
+                for project in user.project_set.all()
+                for dataset in project.datasets.all()
+            ]
+        )
+        # Fetch all necessary data at once before the loop
+        dataset_ids = [dataset.id for dataset in possible_datasets]
+        accesses = Access.objects.filter(
+            Q(dataset_id__in=dataset_ids) & Q(status=StatusChoices.active)
+        )
+
+        for access in accesses:
+            # Check if the dataset has an access that is about to expire
+            if (
+                access.grant_expires_on
+                and access.grant_expires_on - day_offset == exec_date
+            ):
+                cls.notify(user, access, NotificationVerb.expire)
+
+    @staticmethod
+    def notify(user: "User", obj: "Access", verb: "NotificationVerb"):
+        """
+        Notifies concerning users about the entity.
+        """
+        offset = user.notification_setting.notification_offset
+        dispatch_by_email = user.notification_setting.send_email
+        dispatch_in_app = user.notification_setting.send_in_app
+
+        msg = f"Access for {obj.dataset.title} of the user {obj.user or obj.contact} is ending in {offset} days."
+        on = obj.grant_expires_on
+
+        logger.info(f"Creating a notification for {user} : {msg}")
+
+        Notification.objects.create(
+            recipient=user,
+            verb=verb,
+            message=msg,
+            on=on,
+            dispatch_by_email=dispatch_by_email,
+            dispatch_in_app=dispatch_in_app,
+            content_type=ContentType.objects.get_for_model(obj),
+            object_id=obj.id,
+        ).save()
+
 
 auditlog.register(Access)

core/models/dataset.py

@@ -1,19 +1,34 @@
 import uuid
+import datetime
+from datetime import timedelta
+import typing
 
 from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.contenttypes.models import ContentType
 from django.db import models
+from django.db.models import Q
 from django.urls import reverse
 from django.utils.module_loading import import_string
-
 from guardian.models import GroupObjectPermissionBase, UserObjectPermissionBase
+
 from core import constants
+from core.utils import DaisyLogger
+from core.models import DataDeclaration
 from core.permissions.mapping import PERMISSION_MAPPING
-
-from .utils import CoreTrackedModel, TextFieldWithInputWidget
+from notification import NotifyMixin
+from notification.models import Notification, NotificationVerb
+from .utils import CoreTrackedModel, TextFieldWithInputWidget, CoreNotifyMeta
 from .partner import HomeOrganisation
 
+if typing.TYPE_CHECKING:
+    User = settings.AUTH_USER_MODEL
+
 
-class Dataset(CoreTrackedModel):
+logger = DaisyLogger(__name__)
+
+
+class Dataset(CoreTrackedModel, NotifyMixin, metaclass=CoreNotifyMeta):
     class Meta:
         app_label = "core"
         get_latest_by = "added"
@@ -227,6 +242,72 @@ def publish(self, save=True):
         for data_declaration in self.data_declarations.all():
             data_declaration.publish_subentities()
 
+    @staticmethod
+    def get_notification_recipients():
+        """
+        Get distinct users that are local custodian of a dataset or a project.
+        """
+
+        return (
+            get_user_model()
+            .objects.filter(Q(datasets__isnull=False) | Q(project_set__isnull=False))
+            .distinct()
+        )
+
+    @classmethod
+    def make_notifications_for_user(
+        cls, day_offset: timedelta, exec_date: datetime.date, user: "User"
+    ):
+        # Considering users that are indirectly responsible for the dataset (through projects)
+        possible_datasets = set(user.datasets.all())
+        for project in user.project_set.all():
+            possible_datasets.update(project.datasets.all())
+        for dataset in possible_datasets:
+            # Data Declaration (Embargo Date & End of Storage Duration)
+            for data_declaration in dataset.data_declarations.all():
+                if (
+                    data_declaration.embargo_date
+                    and data_declaration.embargo_date - day_offset == exec_date
+                ):
+                    cls.notify(user, data_declaration, NotificationVerb.embargo_end)
+                if (
+                    data_declaration.end_of_storage_duration
+                    and data_declaration.end_of_storage_duration - day_offset
+                    == exec_date
+                ):
+                    cls.notify(user, data_declaration, NotificationVerb.end)
+
+    @staticmethod
+    def notify(user: "User", obj: "DataDeclaration", verb: "NotificationVerb"):
+        """
+        Notifies concerning users about the entity.
+        """
+        offset = user.notification_setting.notification_offset
+        dispatch_by_email = user.notification_setting.send_email
+        dispatch_in_app = user.notification_setting.send_in_app
+
+        if verb == NotificationVerb.embargo_end:
+            msg = f"Embargo for {obj.dataset.title} is ending in {offset} days."
+            on = obj.embargo_date
+        else:
+            msg = (
+                f"Storage duration for {obj.dataset.title} is ending in {offset} days."
+            )
+            on = obj.end_of_storage_duration
+
+        logger.info(f"Creating a notification for {user} : {msg}")
+
+        Notification.objects.create(
+            recipient=user,
+            verb=verb,
+            message=msg,
+            on=on,
+            dispatch_by_email=dispatch_by_email,
+            dispatch_in_app=dispatch_in_app,
+            content_type=ContentType.objects.get_for_model(obj.dataset),
+            object_id=obj.dataset.id,
+        ).save()
+
 
 # faster lookup for permissions
 # https://django-guardian.readthedocs.io/en/stable/userguide/performance.html#direct-foreign-keys

core/models/document.py

@@ -1,14 +1,30 @@
 import os
+import typing
+import datetime
+from datetime import timedelta
+from model_utils import Choices
+
 from django.db import models
+from django.conf import settings
+from django.db.models import Q
+from django.contrib.auth import get_user_model
 from django.contrib.contenttypes.fields import GenericForeignKey
 from django.contrib.contenttypes.models import ContentType
 from django.utils import timezone
 from django.db.models.signals import post_delete
 from django.dispatch import receiver
 from django.core.files.storage import default_storage
-from model_utils import Choices
-from .utils import CoreModel
-from core import constants
+
+from .utils import CoreModel, CoreNotifyMeta
+from core.utils import DaisyLogger
+from notification.models import Notification, NotificationVerb
+from notification import NotifyMixin
+
+if typing.TYPE_CHECKING:
+    User = settings.AUTH_USER_MODEL
+
+
+logger = DaisyLogger(__name__)
 
 
 def get_file_name(instance, filename):
@@ -21,7 +37,7 @@ def get_file_name(instance, filename):
     )
 
 
-class Document(CoreModel):
+class Document(CoreModel, NotifyMixin, metaclass=CoreNotifyMeta):
     """
     Represents a document
     """
@@ -77,6 +93,58 @@ def shortname(self):
     def size(self):
         return self.content.size
 
+    @staticmethod
+    def get_notification_recipients():
+        """
+        Get distinct users that are local custodian of a project or a contract.
+        """
+
+        return (
+            get_user_model()
+            .objects.filter(Q(project_set__isnull=False) | Q(contracts__isnull=False))
+            .distinct()
+        )
+
+    @classmethod
+    def make_notifications_for_user(
+        cls, day_offset: timedelta, exec_date: datetime.date, user: "User"
+    ):
+        docs = set()
+        [docs.update(p.legal_documents.all()) for p in user.project_set.all()]
+        [docs.update(c.legal_documents.all()) for c in user.contracts.all()]
+        # Also add all the documents of all contracts of all projects to address the indirect LCs of parent projects
+        for p in user.project_set.all():
+            _ = [docs.update(c.legal_documents.all()) for c in p.contracts.all()]
+
+        for doc in docs:
+            if doc.expiry_date and doc.expiry_date - day_offset == exec_date:
+                cls.notify(user, doc, NotificationVerb.expire)
+
+    @staticmethod
+    def notify(user: "User", obj: "Document", verb: "NotificationVerb"):
+        """
+        Notifies concerning users about the entity.
+        """
+        offset = user.notification_setting.notification_offset
+        dispatch_by_email = user.notification_setting.send_email
+        dispatch_in_app = user.notification_setting.send_in_app
+
+        msg = f"The Document {obj.shortname} is expiring in {offset} days."
+        on = obj.expiry_date
+
+        logger.info(f"Creating a notification for {user} : {msg}")
+
+        Notification.objects.create(
+            recipient=user,
+            verb=verb,
+            message=msg,
+            on=on,
+            dispatch_by_email=dispatch_by_email,
+            dispatch_in_app=dispatch_in_app,
+            content_type=ContentType.objects.get_for_model(obj),
+            object_id=obj.id,
+        ).save()
+
 
 @receiver(post_delete, sender=Document, dispatch_uid="document_delete")
 def document_cleanup(sender, instance, **kwargs):