511 add reharvest dags for jagiellonian reprocessing

dags/elsevier/repository.py

@@ -21,7 +21,7 @@ def __init__(self):
 
     def get_all_raw_filenames(self):
         return [
-            f.key.removeprefix("raw/")
+            f.key.removeprefix(self.ZIPED_DIR)
             for f in self.s3.objects.filter(Prefix=self.ZIPED_DIR).all()
         ]
 

dags/jagiellonian/jagiellonian_process_file.py

@@ -2,18 +2,20 @@
 import logging
 import os
 from datetime import UTC, datetime
+from io import BytesIO
 
 import pendulum
-from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.providers.amazon.aws.transfers.http_to_s3 import HttpToS3Operator
 from airflow.sdk import dag, task
 from common.enhancer import Enhancer
 from common.enricher import Enricher
 from common.notification_service import FailedDagNotifier
 from common.utils import create_or_update_article
 from jagiellonian.parser import JagiellonianParser
+from jagiellonian.repository import JagiellonianRepository
 
 logger = logging.getLogger("airflow.task")
+JAGIELLONIAN_REPO = JagiellonianRepository()
 
 FILE_EXTENSIONS = {"pdf": ".pdf", "xml": ".xml", "pdfa": ".pdf"}
 
@@ -39,6 +41,16 @@ def jagiellonian_process_file():
     def parse(**kwargs):
         if "params" in kwargs and "article" in kwargs["params"]:
             article = kwargs["params"]["article"]
+
+            if (
+                isinstance(article, dict)
+                and isinstance(article.get("dois"), list)
+                and article.get("dois")
+                and isinstance(article["dois"][0], dict)
+                and article["dois"][0].get("value")
+            ):
+                return article
+
             parsed = JagiellonianParser().parse(article)
 
             if "dois" not in parsed:
@@ -105,21 +117,11 @@ def enrich(enhanced_file):
         return Enricher()(enhanced_file)
 
     @task(task_id="jagiellonian-save-to-s3")
-    def save_to_s3(enriched_file):
-        s3_bucket = os.getenv("JAGIELLONIAN_BUCKET_NAME", "jagiellonian")
-        aws_conn_id = os.getenv("AWS_CONN_ID", "aws_s3_minio")
-
+    def save_to_s3(enriched_file, repo=JAGIELLONIAN_REPO):
         doi = enriched_file["dois"][0]["value"]
         key = f"{doi}_metadata_{(datetime.now(UTC))}.json"
 
-        s3_hook = S3Hook(aws_conn_id=aws_conn_id)
-
-        s3_hook.load_string(
-            string_data=json.dumps(enriched_file, indent=2),
-            key=key,
-            bucket_name=s3_bucket,
-            replace=True,
-        )
+        repo.save(key=key, obj=BytesIO(json.dumps(enriched_file, indent=2).encode()))
 
     @task()
     def create_or_update(enriched_file):

dags/jagiellonian/jagiellonian_pull_api.py

@@ -3,33 +3,14 @@
 from datetime import timedelta
 
 import pendulum
-from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.providers.http.hooks.http import HttpHook
 from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
 from airflow.sdk import dag, task
 from common.notification_service import FailedDagNotifier
+from jagiellonian.repository import JagiellonianRepository
 
 logger = logging.getLogger("airflow.task")
-
-
-def get_latest_s3_file():
-    s3_bucket = os.getenv("JAGIELLONIAN_BUCKET_NAME", "jagiellonian")
-    aws_conn_id = os.getenv("AWS_CONN_ID", "aws_s3_minio")
-
-    s3_hook = S3Hook(aws_conn_id=aws_conn_id)
-    objects = s3_hook.list_keys(bucket_name=s3_bucket)
-    files = [obj for obj in objects if not obj.endswith("/")]
-    if not files:
-        return None
-
-    file_timestamps = []
-    for file_key in files:
-        object_info = s3_hook.get_key(key=file_key, bucket_name=s3_bucket)
-        file_timestamps.append((file_key, object_info.last_modified))
-
-    file_timestamps.sort(key=lambda x: x[1], reverse=True)
-    latest_timestamp = file_timestamps[0][1].strftime("%Y-%m-%d")
-    return latest_timestamp
+JAGIELLONIAN_REPO = JagiellonianRepository()
 
 
 default_args = {
@@ -53,15 +34,15 @@ def get_latest_s3_file():
 )
 def jagiellonian_pull_api(from_date=None):
     @task(task_id="jagiellonian_fetch_crossref_api")
-    def fetch_crossref_api(from_date_param=None):
+    def fetch_crossref_api(from_date_param=None, repo=JAGIELLONIAN_REPO):
         http_conn_id = os.getenv("HTTP_CONN_ID", "crossref_api")
         endpoint_filter = ""
 
         if from_date_param:
             logger.info("Using provided from_date: %s", from_date_param)
             endpoint_filter = f"from-created-date:{from_date_param}"
         else:
-            latest_s3_file = get_latest_s3_file()
+            latest_s3_file = repo.find_the_last_uploaded_file_date()
             if latest_s3_file:
                 logger.info("Using latest S3 file date: %s", latest_s3_file)
                 endpoint_filter = f"from-created-date:{latest_s3_file}"