Merge pull request #477 from OP-TED/feature/TED-1364

add reprocess published in cellar notices feature

Author: CaptainOfHacks

.github/workflows/unit-tests.yml

@@ -25,6 +25,7 @@ jobs:
           python-version: 3.8
       - name: Install dependencies
         run: |
+          sudo apt-get install -y libssl-dev libcurl4-openssl-dev
           python -m pip install --upgrade setuptools pip wheel
           make install
           make install-dev

dags/pipelines/notice_processor_pipelines.py

@@ -2,6 +2,7 @@
 from dags.pipelines.pipeline_protocols import NoticePipelineOutput
 from ted_sws.core.model.notice import Notice, NoticeStatus
 from ted_sws.event_manager.services.log import log_notice_error
+from ted_sws.notice_packager.model.metadata import METS_TYPE_UPDATE
 from ted_sws.notice_validator.services.entity_deduplication_validation import \
     generate_rdf_manifestation_entity_deduplication_report
 
@@ -88,11 +89,15 @@ def notice_package_pipeline(notice: Notice, mongodb_client: MongoClient = None)
 
     """
     from ted_sws.notice_packager.services.notice_packager import package_notice
+    from ted_sws.notice_packager.model.metadata import METS_TYPE_CREATE
 
     notice.update_status_to(new_status=NoticeStatus.VALIDATED)
     # TODO: Implement notice package eligiblity
     notice.set_is_eligible_for_packaging(eligibility=True)
-    packaged_notice = package_notice(notice=notice)
+    package_action = METS_TYPE_CREATE
+    if notice.normalised_metadata.published_in_cellar_counter > 0:
+        package_action = METS_TYPE_UPDATE
+    packaged_notice = package_notice(notice=notice, action=package_action)
     return NoticePipelineOutput(notice=packaged_notice)
 
 

dags/reprocess_published_in_cellar_notices.py

@@ -0,0 +1,52 @@
+from airflow.decorators import dag, task
+
+from dags import DEFAULT_DAG_ARGUMENTS
+from dags.dags_utils import push_dag_downstream, get_dag_param
+from dags.notice_processing_pipeline import NOTICE_TRANSFORMATION_PIPELINE_TASK_ID
+from dags.operators.DagBatchPipelineOperator import NOTICE_IDS_KEY, TriggerNoticeBatchPipelineOperator, \
+    EXECUTE_ONLY_ONE_STEP_KEY
+from dags.pipelines.notice_selectors_pipelines import notice_ids_selector_by_status
+from ted_sws.core.model.notice import NoticeStatus
+from ted_sws.event_manager.adapters.event_log_decorator import event_log
+from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
+    EventMessageProcessType
+
+DAG_NAME = "reprocess_published_in_cellar_notices"
+
+RE_TRANSFORM_TARGET_NOTICE_STATES = [NoticeStatus.PUBLICLY_AVAILABLE]
+TRIGGER_NOTICE_PROCESS_WORKFLOW_TASK_ID = "trigger_notice_process_workflow"
+FORM_NUMBER_DAG_PARAM = "form_number"
+START_DATE_DAG_PARAM = "start_date"
+END_DATE_DAG_PARAM = "end_date"
+XSD_VERSION_DAG_PARAM = "xsd_version"
+
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS,
+     schedule_interval=None,
+     tags=['selector', 're-transform-publicly-available'])
+def reprocess_published_in_cellar_notices():
+    @task
+    @event_log(TechnicalEventMessage(
+        message="select_notices_for_re_transform",
+        metadata=EventMessageMetadata(
+            process_type=EventMessageProcessType.DAG, process_name=DAG_NAME
+        ))
+    )
+    def select_notices_for_re_transform():
+        form_number = get_dag_param(key=FORM_NUMBER_DAG_PARAM)
+        start_date = get_dag_param(key=START_DATE_DAG_PARAM)
+        end_date = get_dag_param(key=END_DATE_DAG_PARAM)
+        xsd_version = get_dag_param(key=XSD_VERSION_DAG_PARAM)
+        notice_ids = notice_ids_selector_by_status(notice_statuses=RE_TRANSFORM_TARGET_NOTICE_STATES,
+                                                   form_number=form_number, start_date=start_date,
+                                                   end_date=end_date, xsd_version=xsd_version)
+        push_dag_downstream(key=NOTICE_IDS_KEY, value=notice_ids)
+
+    trigger_notice_process_workflow = TriggerNoticeBatchPipelineOperator(
+        task_id=TRIGGER_NOTICE_PROCESS_WORKFLOW_TASK_ID,
+        start_with_step_name=NOTICE_TRANSFORMATION_PIPELINE_TASK_ID
+    )
+    select_notices_for_re_transform() >> trigger_notice_process_workflow
+
+
+dag = reprocess_published_in_cellar_notices()

requirements.txt

@@ -26,3 +26,4 @@ json2html~=1.3.0
 minio~=7.1.1
 certifi~=2022.12.7
 shortuuid~=1.0.11
+pycurl~=7.45.2

ted_sws/core/model/metadata.py

@@ -80,6 +80,7 @@ class NormalisedMetadata(Metadata):
     form_number: str
     eforms_subtype: str
     xsd_version: str
+    published_in_cellar_counter: int = Field(default=0)
 
 
 class NormalisedMetadataView(Metadata):
@@ -104,6 +105,7 @@ class NormalisedMetadataView(Metadata):
     form_number: str
     eforms_subtype: str
     xsd_version: str
+    published_in_cellar_counter: int = Field(default=0)
 
 
 

ted_sws/notice_validator/services/check_availability_of_notice_in_cellar.py

@@ -104,6 +104,7 @@ def validate_notices_availability_in_cellar(notice_statuses: List[NoticeStatus],
             for notice_uri, notice in selected_notices_map.items():
                 if notice_uri in available_notice_uries_in_cellar:
                     notice.update_status_to(new_status=NoticeStatus.PUBLICLY_AVAILABLE)
+                    notice.normalised_metadata.published_in_cellar_counter += 1
                 else:
                     notice.update_status_to(new_status=NoticeStatus.PUBLICLY_UNAVAILABLE)
                 notice_repository.update(notice=notice)

tests/e2e/mapping_suite_processor/conceptual_mapping_differ/test_conceptual_mapping_differ.py

@@ -51,7 +51,7 @@ def test_mapping_suite_diff_repo_conceptual_mappings(github_mapping_suite_id, pa
 
     diff = mapping_suite_diff_repo_conceptual_mappings(
         branch_or_tag_name=["main"],
-        mapping_suite_id=[github_mapping_suite_id, "package_F03_test"]
+        mapping_suite_id=[github_mapping_suite_id, "package_F06"]
     )
     assert diff['data']['original']
 

tests/e2e/mapping_suite_processor/test_mapping_suite_processor_from_github_expand_and_load_package_in_mongo_db.py

@@ -3,7 +3,9 @@
     mapping_suite_processor_from_github_expand_and_load_package_in_mongo_db
 
 MAPPING_SUITE_PACKAGE_NAME = "package_F03_test"
-MAPPING_SUITE_PACKAGE_ID = f"{MAPPING_SUITE_PACKAGE_NAME}_v2.3.0"
+MAPPING_SUITE_METADATA_IDENTIFIER = "package_F03"
+MAPPING_SUITE_METADATA_VERSION = "6.8.1"
+MAPPING_SUITE_PACKAGE_ID = f"{MAPPING_SUITE_METADATA_IDENTIFIER}_v{MAPPING_SUITE_METADATA_VERSION}"
 
 
 def test_mapping_suite_processor_from_github_expand_and_load_package_in_mongo_db(fake_mongodb_client):