Joindre les documents téléversés dans Monitorfish aux emails de préavis

datascience/src/pipeline/flows/distribute_pnos.py

@@ -11,11 +11,12 @@
 import prefect.engine
 import prefect.engine.signals
 import prefect.exceptions
+import sqlalchemy
 import weasyprint
 from jinja2 import Environment, FileSystemLoader, Template, select_autoescape
 from prefect import Flow, Parameter, case, flatten, task, unmapped
 from prefect.executors import LocalDaskExecutor
-from sqlalchemy import Executable, bindparam, text
+from sqlalchemy import Executable, Select, bindparam, select, text
 
 from config import (
     CNSP_CROSSA_CACEM_LOGOS_PATH,
@@ -60,11 +61,12 @@
 )
 from src.pipeline.shared_tasks.control_units import fetch_control_units_contacts
 from src.pipeline.shared_tasks.dates import get_utcnow, make_timedelta
-from src.pipeline.shared_tasks.infrastructure import execute_statement
+from src.pipeline.shared_tasks.infrastructure import execute_statement, get_table
 from src.pipeline.shared_tasks.pnos import (
     extract_pno_units_ports_and_segments_subscriptions,
     extract_pno_units_targeting_vessels,
 )
+from src.read_query import read_query
 
 
 @task(checkpoint=False)
@@ -121,6 +123,40 @@ def extract_facade_email_addresses() -> dict:
     return df.set_index("facade").email_address.to_dict()
 
 
+@task(checkpoint=False)
+def make_prior_notification_attachments_query(
+    pnos: List[RenderedPno], prior_notification_uploads_table: sqlalchemy.Table
+) -> Select | None:
+    pno_ids = sorted(set([pno.report_id for pno in pnos]))
+
+    if pno_ids:
+        res = select(
+            prior_notification_uploads_table.c.report_id,
+            prior_notification_uploads_table.c.file_name,
+            prior_notification_uploads_table.c.content,
+        ).where(prior_notification_uploads_table.c.report_id.in_(pno_ids))
+    else:
+        res = None
+
+    return res
+
+
+@task(checkpoint=False)
+def execute_prior_notification_attachments_query(
+    query: Select | None,
+) -> dict:
+    if isinstance(query, Select):
+        attachments = read_query(query, db="monitorfish_remote")
+        attachments["filename_content"] = attachments.apply(
+            lambda row: (row["file_name"], row["content"]), axis=1
+        )
+        res = attachments.groupby("report_id")["filename_content"].agg(list).to_dict()
+    else:
+        res = dict()
+
+    return res
+
+
 @task(checkpoint=False)
 def to_pnos_to_render(pnos: pd.DataFrame) -> List[PnoToRender]:
     records = pnos.to_dict(orient="records")
@@ -662,7 +698,9 @@ def attribute_addressees(
 
 
 @task(checkpoint=False)
-def create_email(pno: RenderedPno, test_mode: bool) -> PnoToSend:
+def create_email(
+    pno: RenderedPno, uploaded_attachments: dict, test_mode: bool
+) -> PnoToSend:
     if pno.emails:
         if test_mode:
             if PNO_TEST_EMAIL:
@@ -672,6 +710,13 @@ def create_email(pno: RenderedPno, test_mode: bool) -> PnoToSend:
         else:
             to = pno.emails
 
+        attachments = [
+            (
+                f"Preavis_{pno.vessel_name if pno.vessel_name else ''}.pdf",
+                pno.pdf_document,
+            )
+        ] + (uploaded_attachments.get(pno.report_id) or [])
+
         message = create_html_email(
             to=to,
             subject=f"Préavis de débarquement - {pno.vessel_name}",
@@ -682,9 +727,7 @@ def create_email(pno: RenderedPno, test_mode: bool) -> PnoToSend:
                 LIBERTE_EGALITE_FRATERNITE_LOGO_PATH,
                 MARIANNE_LOGO_PATH,
             ],
-            attachments={
-                f"Preavis_{pno.vessel_name if pno.vessel_name else ''}.pdf": pno.pdf_document
-            },
+            attachments=attachments,
             reply_to=CNSP_FRANCE_EMAIL_ADDRESS,
         )
         return PnoToSend(
@@ -1014,6 +1057,14 @@ def make_manual_prior_notifications_statement(
 
             with case(distribution_needed, True):
                 # Distribute PNOs
+                prior_notification_uploads_table = get_table(
+                    "prior_notification_uploads"
+                )
+                query = make_prior_notification_attachments_query(
+                    pnos_to_distribute, prior_notification_uploads_table
+                )
+                attachments = execute_prior_notification_attachments_query(query)
+
                 facade_email_addresses = extract_facade_email_addresses()
 
                 pnos_with_addressees = attribute_addressees.map(
@@ -1025,7 +1076,9 @@ def make_manual_prior_notifications_statement(
                 )
 
                 email = create_email.map(
-                    pnos_with_addressees, test_mode=unmapped(test_mode)
+                    pnos_with_addressees,
+                    uploaded_attachments=unmapped(attachments),
+                    test_mode=unmapped(test_mode),
                 )
                 email = filter_results(email)
 
@@ -1073,6 +1126,7 @@ def make_manual_prior_notifications_statement(
 
 flow.set_reference_tasks(
     [
+        attachments,
         pnos_to_generate,
         pnos_to_distribute,
         pnos_with_addressees,

datascience/src/pipeline/flows/notify_beacon_malfunctions.py

@@ -270,7 +270,7 @@ def create_email(
             subject=m.get_notification_subject(),
             html=html,
             images=[CNSP_LOGO_PATH],
-            attachments={"Notification.pdf": pdf},
+            attachments=[("Notification.pdf", pdf)],
             reply_to=CNSP_SIP_DEPARTMENT_EMAIL,
         )
 

datascience/src/pipeline/helpers/emails.py

@@ -14,7 +14,7 @@
     SMTPSenderRefused,
 )
 from time import sleep
-from typing import List, Union
+from typing import List, Tuple, Union
 
 import pypdf
 
@@ -45,7 +45,7 @@ def create_html_email(
     cc: Union[str, List[str]] = None,
     bcc: Union[str, List[str]] = None,
     images: List[Path] = None,
-    attachments: dict = None,
+    attachments: List[Tuple[str, bytes]] = None,
     reply_to: str = None,
 ) -> EmailMessage:
     """
@@ -75,9 +75,9 @@ def create_html_email(
             `<img src="cid:my_image_123.png">` in the html message.
 
           Defaults to None.
-        attachments (dict, optional): `dict` of attachments to add to the email.
-          Consists of {filename : bytes} value pairs. Defaults
-          to None.
+        attachments (List[Tuple[str, bytes]], optional): `list` of attachments to add
+          to the email. Elements of the list must be pairs of (filename, content).
+          Defaults to None.
         reply_to (str, optional): if given, added as `Reply-To` header. Defaults to
           None.
 
@@ -131,7 +131,7 @@ def create_html_email(
             msg.attach(img)
 
     if attachments:
-        for filename, filebytes in attachments.items():
+        for filename, filebytes in attachments:
             msg.add_attachment(
                 filebytes,
                 maintype="application",

datascience/tests/test_data/remote_database/V666.40__Reset_test_prior_notification_uploads.sql

@@ -0,0 +1,9 @@
+DELETE FROM prior_notification_uploads;
+INSERT INTO prior_notification_uploads (
+                                  report_id, is_manual_prior_notification,                    file_name,                                                                   mime_type,                        content,                                                           created_at,                                                           updated_at
+) VALUES
+    ('00000000-0000-4000-0000-000000000003',                         true,  'Uploaded document n°1.pdf',                                                           'application/pdf',                     '\x504446',                                                '2020-05-06 18:42:03',                                                '2020-05-06 18:42:03'),
+    ('00000000-0000-4000-0000-000000000003',                         true, 'Uploaded document n°2.docx',   'application/vnd.openxmlformats-officedocument.wordprocessingml.document', '\x5465787420446f63756d656e74',                                                '2020-05-06 18:45:23',                                                '2020-05-06 18:45:23'),
+    (                                  '11',                        false, 'Uploaded document n°3.xlsx',         'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',     '\x5370726561647368656574',   ((now() AT TIME ZONE 'UTC') - INTERVAL '1 year 3 days')::TIMESTAMP,   ((now() AT TIME ZONE 'UTC') - INTERVAL '1 year 3 days')::TIMESTAMP),
+    (                                  '12',                        false, 'Uploaded document n°4.jpeg',                                                                'image/jpeg',                     '\x4a5047', ((now() AT TIME ZONE 'UTC') - INTERVAL '1 month 2 hours')::TIMESTAMP, ((now() AT TIME ZONE 'UTC') - INTERVAL '1 month 2 hours')::TIMESTAMP),
+    (                     'Other-report-id',                        false, 'Uploaded document n°5.pptx', 'application/vnd.openxmlformats-officedocument.presentationml.presentation',   '\x50726573656e746174696f6e',  ((now() AT TIME ZONE 'UTC') - INTERVAL '1 week 2 hours')::TIMESTAMP,  ((now() AT TIME ZONE 'UTC') - INTERVAL '1 week 2 hours')::TIMESTAMP);