feat: Send nightly emails for YAML task failures (#347)

* Revert "Revert "fix: Silent errors in sumstats pipeline (#341)""

This reverts commit 712e2d1.

* feat: Check for files in staging FTP, not pub FTP

* feat: Send nightly emails for YAML task failures

Report only the ones with sumstats in the email

sumstats_service/app.py

@@ -14,6 +14,7 @@
 import sumstats_service.resources.globus as globus
 from sumstats_service import config, logger_config
 from sumstats_service.resources.error_classes import APIException
+from sumstats_service.resources.mongo_client import MongoClient
 from sumstats_service.resources.utils import send_mail
 
 try:
@@ -40,6 +41,7 @@
 app.config["BROKER_TRANSPORT_OPTIONS"] = {"confirm_publish": True}
 app.url_map.strict_slashes = False
 
+
 celery = Celery(
     "app",
     broker=app.config["CELERY_BROKER_URL"],
@@ -416,7 +418,27 @@ def task_failure_handler(sender=None, **kwargs) -> None:
               """.format(
         **kwargs
     )
-    send_mail(subject=subject, message=message)
+
+    if sender.name != "sumstats_service.app.convert_metadata_to_yaml":
+        send_mail(subject=subject, message=message)
+    else:
+        args = kwargs.get("args", [])
+        exception = kwargs.get("exception", "No exception info")
+        gcst_id = args[0] if args else "Unknown GCST ID"
+
+        # Save to MongoDB
+        mdb = MongoClient(
+            config.MONGO_URI,
+            config.MONGO_USER,
+            config.MONGO_PASSWORD,
+            config.MONGO_DB,
+        )
+        study_data = mdb.get_study(gcst_id=gcst_id)
+        if study_data.get("summaryStatisticsFile", "") != config.NR:
+            logger.info(f"Adding {gcst_id=} to the task failures collection")
+            mdb.insert_task_failure(gcst_id=gcst_id, exception=str(exception))
+        else:
+            logger.info(f"Skipping {gcst_id=} as it has no sumstats.")
 
 
 if __name__ == "__main__":

sumstats_service/config.py

@@ -1,3 +1,4 @@
+# flake8: noqa: W291,E501
 import os
 
 
@@ -13,6 +14,7 @@ def _env_variable_else(env_var_name, default):
 SW_PATH = _env_variable_else("SW_PATH", "./bin")
 DEPO_PATH = _env_variable_else("DEPO_PATH", "./depo_data")
 CONTAINERISE = _env_variable_else("CONTAINERISE", "./depo_data")
+FTP_STAGING_PATH = _env_variable_else("FTP_STAGING_PATH", "depo_ss_staging_ftp")
 
 
 # --- Rabbit and Celery --- #
@@ -45,14 +47,15 @@ def _env_variable_else(env_var_name, default):
 MONGO_USER = _env_variable_else("MONGO_USER", "")
 MONGO_PASSWORD = _env_variable_else("MONGO_PASSWORD", "")
 MONGO_DB = _env_variable_else("MONGO_DB", None)
-
+NR = "NR"
 
 # --- File transfer (FTP nad Globus) config --- #
 
 FTP_SERVER = _env_variable_else("FTP_SERVER", None)
 FTP_USERNAME = _env_variable_else("FTP_USERNAME", None)
 FTP_PASSWORD = _env_variable_else("FTP_PASSWORD", None)
 FTP_SERVER_EBI = _env_variable_else("FTP_SERVER_EBI", "ftp.ebi.ac.uk")
+FTP_PREFIX = _env_variable_else("FTP_PREFIX", "/pub/databases/gwas/summary_statistics")
 
 TOKEN_FILE = "refresh-tokens.json"
 REDIRECT_URI = "https://auth.globus.org/v2/web/auth-code"