SIENTIAPDE-1172

feat: enhance email notifications and orchestrator activities for reports

- Updated email HTML structure to clearly present errors, warnings, and info notifications.
- Introduced a new method in the Formatters class to filter notification reports based on group configurations.
- Enhanced the Reports workflow to integrate the new filtering functionality and manage notification packages effectively.
- Adjusted SlotManager to store timestamps with mail type specificity for better tracking.
- Improved test coverage for the new filtering functionality in the Formatters class.

orchestrator/activities/formatters.py

@@ -532,6 +532,41 @@ class Formatters(BaseActivity):
                         'mail_type': mail_type
                     }
                 else:
-                    data[key]['groups'].append(group_name)
+                    if group_name not in data[key]['groups']:
+                        data[key]['groups'].append(group_name)
 
         return DataFrame(list(data.values())).to_dict()
+
+    @activity.defn(name="filter_notification_reports")
+    async def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Filter notification reports.
+        """
+        metadata = input_data['metadata']
+        notification_package = input_data['notification_package']
+        sending_configs = input_data['sending_configs']
+
+        self.info("Filtering notification reports...", metadata=metadata)
+
+        receiver_groups = {}
+
+        for receiver_group in sending_configs:
+            group_name = receiver_group['group_name']
+            receiver_groups[group_name] = {
+                **receiver_group,
+                "notifications": []
+            }
+            receiver_groups[group_name]['notifications'] = []
+
+            ignore_list = receiver_group.get('ignore', [])
+
+            for notification in notification_package:
+                alert_type = "reports"
+                notification_id = notification['notification_id']
+
+                # Check if this group must be notified
+                if alert_type in receiver_group['contents'] and notification_id not in ignore_list:
+                    receiver_groups[group_name]["notifications"].append(
+                        notification)
+
+        return receiver_groups

orchestrator/activities/slot_manager.py

@@ -204,7 +204,7 @@ class SlotManager(Redis):
         Gets the last data timestamp from redis.
         """
         metadata = input_data['metadata']
-        key = "notification_last_timestamp"
+        key = f"notification_last_timestamp:{input_data['mail_type']}"
 
         try:
             data_hold = self.get(key)
@@ -235,7 +235,7 @@ class SlotManager(Redis):
         Puts the last data timestamp into redis.
         """
         metadata = input_data['metadata']
-        key = "notification_last_timestamp"
+        key = f"notification_last_timestamp:{input_data['mail_type']}"
 
         data = DataFrame(input_data['data'])
 

orchestrator/worker/worker.py

@@ -7,6 +7,7 @@ with workflow.unsafe.imports_passed_through():
     import sys
     import asyncio
     from orchestrator.workflows.alerts import Alerts
+    from orchestrator.workflows.reports import Reports
     from orchestrator.workflows.subworkflows.load_notification_package import LoadNotificationPackage
     from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications
     from orchestrator.workflows.orchestrator import Orchestrator
@@ -121,6 +122,27 @@ async def main():
                 # Store notification cache
                 activities.store_notification_cache
             ]
+        ),
+        Worker(
+            temporal_client,
+            task_queue='reports-queue',
+            workflows=[Reports, LoadNotificationPackage, ProcessNotifications],
+            activities=[
+                # Load notifications
+                activities.get_last_data_timestamp,
+                activities.find_documents_in_mongodb,
+                activities.load_latest_data,
+                activities.put_last_data_timestamp,
+
+                # Format and filter notifications
+                activities.filter_notification_reports,
+
+                # Send email and export data to postgres
+                activities.build_email_html,
+                activities.send_email,
+                activities.format_log_report,
+                activities.export_data_to_postgres
+            ]
         )
     ]
 

orchestrator/workflows/alerts.py

@@ -38,6 +38,7 @@ class Alerts:
         mail_type = "Alerts"
 
         input_data['metadata'] = metadata
+        input_data['mail_type'] = mail_type
 
         input_data['base_data_filter'] = {
             'level': 'ERROR'

orchestrator/workflows/reports.py

@@ -3,17 +3,76 @@ from temporalio import workflow
 with workflow.unsafe.imports_passed_through():
     from orchestrator.activities.activities import Activities
     from typing import Any
+    from datetime import timedelta
+    from sientia_do.temporal.utils.policies import retry_policy
 
 
 @workflow.defn(name="reports")
 class Reports:
     @workflow.run
     async def run(self, input_data: dict[str, Any]):
+        """
+        Workflow to send reports to the users
+
+        Args:
+            input_data (dict[str, Any]): Input data. It contains the following keys:
+                - schedule_name: str - Name of the schedule
+
+        Returns:
+            None
+
+        Raises:
+            Exception: If the workflow fails
+        """
+        metadata = {
+            'metadata': {
+                'schedule_name': input_data['schedule_name'],
+                'workflow_name': 'reports',
+                'model_name': '-',
+                'model_id': '-'
+            }
+        }
+
+        mail_type = "Reports"
+
+        input_data['metadata'] = metadata
+        input_data['mail_type'] = mail_type
+
+        input_data['base_data_filter'] = {}
+
         # Call subworkflow "load_notification_package" passing the static filters
         # (timestamp > last timestamp)
 
-        # Filter notification package by groups custom configs
+        package = await workflow.execute_child_workflow(
+            'load_notification_package',
+            input_data
+        )
+
+        if not package['notification_package'] or not package['sending_configs']:
+            return
+
+        # Filter notification package by groups custom configs, levels and
+        # timestamp cached
+
+        receiver_groups = await workflow.execute_local_activity_method(
+            Activities.filter_notification_reports,
+            {
+                **metadata,
+                'notification_package': package['notification_package'],
+                'sending_configs': package['sending_configs']
+            },
+            schedule_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy
+        )
 
         # Call subworkflow "process_notifications" passing the notification package
-
-        pass
+        await workflow.execute_child_workflow(
+            'process_notifications',
+            {
+                'metadata': metadata,
+                'mail_type': mail_type,
+                'notification_package': receiver_groups,
+                'schema': 'sientia_data',
+                'table_name': 'log_report'
+            }
+        )

orchestrator/workflows/subworkflows/load_notification_package.py

@@ -22,6 +22,7 @@ class LoadNotificationPackage:
             - last_timestamp (str): The last timestamp of the notification package.
             - notification_package (list[dict]): The notification package.
             - sending_configs (list[dict]): The sending configs.
+            - mail_type (str): The mail type.
         """
         metadata = input_data['metadata']
 
@@ -30,6 +31,7 @@ class LoadNotificationPackage:
             Activities.get_last_data_timestamp,
             {
                 **metadata,
+                'mail_type': input_data['mail_type']
             },
             start_to_close_timeout=timedelta(seconds=60),
             retry_policy=retry_policy
@@ -82,7 +84,8 @@ class LoadNotificationPackage:
             Activities.put_last_data_timestamp,
             {
                 **metadata,
-                'data': notification_package
+                'data': notification_package,
+                'mail_type': input_data['mail_type']
             },
             start_to_close_timeout=timedelta(seconds=60),
             retry_policy=retry_policy