Merge pull request #64 from odissei-data/incremental-harvester

FjodorvRijsselberg · web-flow · commit cd323d47cdc4 · 2024-10-21T17:24:26.000+02:00
Incremental harvester
diff --git a/Dockerfile.server b/Dockerfile.server
@@ -1,4 +1,4 @@
-FROM python:3.12-slim
+FROM python:3.11-slim
 
 ENV PYTHONPATH="${PYTHONPATH}:/app/scripts/" \
     PYTHONDONTWRITEBYTECODE=1
diff --git a/Dockerfile.worker b/Dockerfile.worker
@@ -1,4 +1,4 @@
-FROM python:3.12-slim
+FROM python:3.11-slim
 
 ENV PYTHONPATH="${PYTHONPATH}:/app/scripts/" \
     PYTHONDONTWRITEBYTECODE=1
diff --git a/pyproject.toml b/pyproject.toml
@@ -5,10 +5,10 @@ description = "Workflow for metadata ingestion into Dataverse"
 authors = ["Fjodor van Rijsselberg"]
 
 [tool.poetry.dependencies]
-python = "^3.12.5"
-prefect = "^3.0.8"
+python = "^3.11.0"
+prefect = "^3.0.10"
 python-dotenv = "^1.0.1"
-boto3 = "^1.35.39"
+boto3 = "^1.35.44"
 dynaconf = "^3.2.6"
 jmespath = "^1.0.1"
 pydantic = "^2.9.2"
diff --git a/scripts/configuration/odissei_settings.toml b/scripts/configuration/odissei_settings.toml
@@ -72,7 +72,7 @@ CBS_TEMPLATE_FILE_PATH="/app/resources/templates/cbs_dataverse_template.json"
 CBS_MAPPING_FILE_PATH="/app/resources/mappings/cbs-mapping.json"
 
 LISS_TEMPLATE_FILE_PATH="/app/resources/templates/liss_dataverse_template.json"
-LISS_MAPPING_FILE_PATH="/resources/mappings/liss-mapping.json"
+LISS_MAPPING_FILE_PATH="/app/resources/mappings/liss-mapping.json"
 
 CID_TEMPLATE_FILE_PATH="/app/resources/templates/cid_dataverse_template.json"
 CID_MAPPING_FILE_PATH="/app/resources/mappings/cid-mapping.json"
diff --git a/scripts/flows/dataset_workflows/cbs_ingestion.py b/scripts/flows/dataset_workflows/cbs_ingestion.py
@@ -1,7 +1,7 @@
 import jmespath
 
 from prefect import flow
-from prefect.server.schemas.states import Completed, Failed
+from prefect.states import Completed, Failed
 from queries import DIST_DATE_QUERY, CBS_ID_QUERY
 from tasks.base_tasks import xml2json, dataverse_mapper, \
     dataverse_import, update_publication_date, add_workflow_versioning_url, \
diff --git a/scripts/flows/dataset_workflows/cid_ingestion.py b/scripts/flows/dataset_workflows/cid_ingestion.py
@@ -2,7 +2,7 @@
 import jmespath
 
 from prefect import flow, get_run_logger
-from prefect.server.schemas.states import Completed, Failed
+from prefect.states import Completed, Failed
 from queries import DIST_DATE_QUERY
 from tasks.base_tasks import dataverse_mapper, \
     dataverse_import, update_publication_date, add_workflow_versioning_url, \
diff --git a/scripts/flows/dataset_workflows/dataverse_deletion.py b/scripts/flows/dataset_workflows/dataverse_deletion.py
@@ -0,0 +1,31 @@
+from prefect import flow
+from prefect.states import Failed, Completed
+
+from tasks.base_tasks import dataverse_dataset_check_status, \
+    delete_dataset
+from utils import generate_dv_flow_run_name, failed_dataverse_deletion_hook
+
+
+@flow(name="Deleting Dataverse metadata", flow_run_name= generate_dv_flow_run_name,
+      on_failure=[failed_dataverse_deletion_hook])
+def dataverse_metadata_deletion(pid, settings_dict):
+    """
+    Deletion flow for Dataverse to dataverse ingestion.
+
+    :param pid: pid of the dataset.
+    :param settings_dict: dict, contains settings for the current workflow.
+    :return: prefect.server.schemas.states Failed or Completed.
+    """
+    dv_response_status = dataverse_dataset_check_status(pid, settings_dict)
+    if not dv_response_status:
+        return Failed(message=f'No response from {pid}.')
+
+    if dv_response_status in (404, 403):
+        response = delete_dataset(pid, settings_dict)
+        if not response:
+            return Failed(message=f'Unable to delete dataset: {pid}.')
+
+        return Completed(message=pid + 'deleted successfully.')
+
+    # If the dataset is active: 200, it will not be deleted.
+    return Failed(message=f'{pid} will not deleted. Dataset is active.')
diff --git a/scripts/flows/dataset_workflows/dataverse_ingestion.py b/scripts/flows/dataset_workflows/dataverse_ingestion.py
@@ -1,9 +1,10 @@
 from prefect import flow
-from prefect.server.schemas.states import Failed, Completed
+from prefect.states import Failed, Completed
 
 from tasks.base_tasks import dataverse_metadata_fetcher, dataverse_import, \
     update_publication_date, add_workflow_versioning_url, refine_metadata, \
-    enrich_metadata, dataverse_mapper
+    enrich_metadata, dataverse_mapper, dataverse_dataset_check_status, \
+    delete_dataset
 from utils import generate_dv_flow_run_name, failed_dataverse_ingestion_hook
 
 
@@ -45,6 +46,20 @@ def dataverse_metadata_ingestion(pid, version, settings_dict):
     if not dataverse_json:
         return Failed(message='Unable to enrich metadata using ELSST.')
 
+    # The result will be 200 (Dataset exists)  or  404 (Dataset does not exist)
+    dv_response_status = dataverse_dataset_check_status(
+        pid,
+        settings_dict.DESTINATION_DATAVERSE_URL
+    )
+
+    if not dv_response_status:
+        return Failed(message=f'No response from {pid}.')
+    if dv_response_status == 200:
+        # 200 means that the dataset must be deleted and reingested.
+        deleted_response = delete_dataset(pid, settings_dict)
+        if not deleted_response:
+            return Failed(message=f'Unable to delete dataset: {pid}.')
+
     import_response = dataverse_import(dataverse_json, settings_dict, pid)
     if not import_response:
         return Failed(message='Unable to import dataset into Dataverse')
diff --git a/scripts/flows/dataset_workflows/liss_ingestion.py b/scripts/flows/dataset_workflows/liss_ingestion.py
@@ -2,7 +2,7 @@
 import jmespath
 
 from prefect import flow
-from prefect.server.schemas.states import Completed, Failed
+from prefect.states import Completed, Failed
 from queries import DIST_DATE_QUERY
 from tasks.base_tasks import dataverse_mapper, \
     dataverse_import, update_publication_date, get_doi_from_dv_json, \
diff --git a/scripts/flows/entry_workflows/main_dataverse_deletion.py b/scripts/flows/entry_workflows/main_dataverse_deletion.py
@@ -0,0 +1,59 @@
+from prefect import flow
+
+import utils
+from configuration.config import settings
+from flows.dataset_workflows.dataverse_deletion import dataverse_metadata_deletion
+from tasks.harvest_tasks import oai_harvest_metadata
+
+
+@flow(name="Dataverse Deleted Pipeline")
+def dataverse_deletion_pipeline(settings_dict_name: str,
+                                target_url: str = None,
+                                target_key: str = None,
+                                do_harvest: bool = True
+                                ):
+    """ Deletion pipeline dedicated to the Dataverse to Dataverse workflow.
+
+    :param do_harvest: Boolean stating if the dataset metadata should be
+     harvested before ingestion.
+    :param target_url: Optional target dataverse url.
+    :param target_key: API key of the optional target dataverse.
+    :param settings_dict_name: string, name of the settings you wish to use
+    """
+    settings_dict = getattr(settings, settings_dict_name)
+
+    if target_url:
+        settings_dict.DESTINATION_DATAVERSE_URL = target_url
+
+    if target_key:
+        settings_dict.DESTINATION_DATAVERSE_API_KEY = target_key
+
+
+    minio_client = utils.create_s3_client()
+
+    if hasattr(settings_dict,
+               'OAI_SET') and settings_dict.OAI_SET and do_harvest:
+        oai_harvest_metadata(
+            settings.METADATA_PREFIX,
+            f'{settings_dict.SOURCE_DATAVERSE_URL}/oai',
+            settings_dict.BUCKET_NAME,
+            'ListIdentifiers',
+            'start_harvest',
+            settings_dict.OAI_SET,
+            settings_dict.FROM
+        )
+
+    elif do_harvest:
+        oai_harvest_metadata(
+            settings.METADATA_PREFIX,
+            f'{settings_dict.SOURCE_DATAVERSE_URL}/oai',
+            settings_dict.BUCKET_NAME,
+            'ListIdentifiers',
+            'start_harvest'
+        )
+    utils.identifier_list_workflow_executor(
+        dataverse_metadata_deletion,
+        settings_dict,
+        minio_client,
+        "identifiers-deleted.json",
+    )
diff --git a/scripts/flows/entry_workflows/main_dataverse_ingestion.py b/scripts/flows/entry_workflows/main_dataverse_ingestion.py
@@ -9,7 +9,7 @@
 from tasks.harvest_tasks import oai_harvest_metadata
 
 
-@flow
+@flow(name="Dataverse Ingestion Pipeline")
 def dataverse_ingestion_pipeline(settings_dict_name: str,
                                  target_url: str = "",
                                  target_key: str = "",
@@ -50,7 +50,8 @@ def dataverse_ingestion_pipeline(settings_dict_name: str,
             settings_dict.BUCKET_NAME,
             'ListIdentifiers',
             'start_harvest',
-            settings_dict.OAI_SET
+            settings_dict.OAI_SET,
+            settings_dict.FROM
         )
 
     elif do_harvest:
@@ -64,7 +65,8 @@ def dataverse_ingestion_pipeline(settings_dict_name: str,
 
     utils.identifier_list_workflow_executor(
         dataverse_metadata_ingestion,
-        version,
         settings_dict,
-        minio_client
-    )
+        minio_client,
+        "identifiers.json",
+        version
+    )
diff --git a/scripts/run_ingestion.py b/scripts/run_ingestion.py
@@ -1,4 +1,6 @@
 import argparse
+import re
+from datetime import datetime
 
 from configuration.config import settings
 from flows.entry_workflows.main_cbs_ingestion import cbs_ingestion_pipeline
@@ -32,6 +34,9 @@ def run_ingestion():
     parser.add_argument('--do_harvest', type=str, default="",
                         help='Bool that states if the metadata will'
                              ' be harvested.')
+    parser.add_argument('--harvest_from', type=str, default=None,
+                        help='Datestamps used as values of the optional '
+                             'argument from will be harvested.')
     args = parser.parse_args()
 
     print(f"args: {args}")
@@ -47,31 +52,24 @@ def run_ingestion():
               f" {data_providers}")
         return
 
+    if args.harvest_from and not validate_datestamp(args.harvest_from):
+        print(
+            f"Invalid datestamp specified, please use the format YYYY-MM-DD.")
+        return
+
+    settings_dict = getattr(settings, args.data_provider)
+    if args.harvest_from:
+        settings_dict["from"] = args.harvest_from
+
     if args.data_provider in provider_mapping:
         ingestion_function = provider_mapping[args.data_provider]
-        settings_dict = getattr(settings, args.data_provider)
         target_url = get_target_url(args.target_url, settings_dict)
-        if not prompt_for_confirmation(target_url):
-            print("Aborted")
-        else:
-            ingestion_function(target_url, args.target_key, do_harvest)
+        ingestion_function(target_url, args.target_key, do_harvest)
 
     else:
-        settings_dict = getattr(settings, args.data_provider)
         target_url = get_target_url(args.target_url, settings_dict)
-        if not prompt_for_confirmation(target_url):
-            print("Aborted")
-        else:
-            dataverse_ingestion_pipeline(args.data_provider, target_url,
-                                         args.target_key, do_harvest)
-
-
-def prompt_for_confirmation(target_url):
-    print("Current target configuration:")
-    print(f"Target URL: {target_url}")
-    confirmation = input(
-        "Do you want to proceed with this configuration? (yes/no): ").lower()
-    return confirmation == 'yes'
+        dataverse_ingestion_pipeline(args.data_provider, target_url,
+                                     args.target_key, do_harvest)
 
 
 def get_target_url(target_url, settings_dict):
@@ -81,5 +79,22 @@ def get_target_url(target_url, settings_dict):
         return target_url
 
 
+def validate_datestamp(datestamp):
+    """ Validates a datestamp in the format YYYY-MM-DD.
+
+    :param datestamp: The datestamp string to validate.
+    :return: True if the datestamp is valid, False otherwise.
+    """
+    date_pattern = re.compile(r'^\d{4}-\d{2}-\d{2}$')
+    if not date_pattern.match(datestamp):
+        return False
+
+    try:
+        datetime.strptime(datestamp, '%Y-%m-%d')
+        return True
+    except ValueError:
+        return False
+
+
 if __name__ == "__main__":
     run_ingestion()
diff --git a/scripts/tasks/base_tasks.py b/scripts/tasks/base_tasks.py
@@ -1,7 +1,6 @@
 import json
 from datetime import timedelta
 
-from pyDataverse.api import NativeApi
 from configuration.config import settings
 from prefect import task, get_run_logger
 import requests
@@ -204,6 +203,66 @@ def dataverse_metadata_fetcher(metadata_format, doi, settings_dict):
     return response.json()
 
 
+@task(name="Dataverse dataset check status", timeout_seconds=300, retries=1,
+      cache_expiration=timedelta(minutes=10))
+def dataverse_dataset_check_status(doi, dataverse_url):
+    """
+    Checks the status of a dataset in Dataverse.
+
+    This function sends a GET request to the Dataverse API to check the status
+    of a dataset identified by its DOI. The status codes returned by the API
+    indicate the following:
+    - 200: Dataset exists
+    - 404: Dataset does not exist
+    - 403: Dataset is deaccessioned
+
+    :param doi: The DOI of the dataset to check.
+    :param dataverse_url: The base URL of the Dataverse instance.
+    :return: The status code (200, 403, 404) or None on failure.
+    """
+    logger = get_run_logger()
+
+    url = f"{dataverse_url}/api/datasets/export?exporter=dcterms&" \
+          f"persistentId={doi}"
+    response = requests.get(url)
+
+    if response.status_code in (200, 403, 404):
+        return response.status_code
+
+    logger.info(f'response.text: {response.text}')
+    return None
+
+
+@task(name="Deleting dataset", timeout_seconds=300, retries=1,
+      cache_expiration=timedelta(minutes=10))
+def delete_dataset(pid, settings_dict):
+    """
+    Deletes a dataset from Dataverse.
+
+    This function sends a DELETE request to the Dataverse API to delete
+     a dataset identified by its persistent identifier (PID).
+
+    :param pid: The persistent identifier of the dataset to delete.
+    :param settings_dict: A dictionary containing settings for current task,
+                          including the Dataverse API key and URL.
+    :return: The response object if the deletion is successful, otherwise None.
+    """
+    headers = {
+        "X-Dataverse-key": settings_dict.DESTINATION_DATAVERSE_API_KEY
+    }
+
+    logger = get_run_logger()
+    url = f"{settings_dict.DESTINATION_DATAVERSE_URL}/api/datasets/" \
+          f":persistentId/destroy/?persistentId={pid}"
+    response = requests.delete(url, headers=headers)
+
+    if response and response.status_code == 200:
+        return response
+
+    logger.info(f'response.text: {response.text}')
+    return None
+
+
 @task(timeout_seconds=300, retries=1, cache_expiration=timedelta(minutes=10))
 def get_doi_from_dv_json(dataverse_json):
     """ Retrieves the DOI of a dataset from mapped Dataverse JSON
diff --git a/scripts/tasks/harvest_tasks.py b/scripts/tasks/harvest_tasks.py
diff --git a/scripts/utils.py b/scripts/utils.py

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-FROM python:3.12-slim`
	`1`	`+FROM python:3.11-slim`
`2`	`2`
`3`	`3`	`ENV PYTHONPATH="${PYTHONPATH}:/app/scripts/" \`
`4`	`4`	`PYTHONDONTWRITEBYTECODE=1`