add reprocess published in cellar notices feature

- add update mets package if mets package already was publicly available
- add published_in_cellar_counter in notice normalised metadata
- add new DAG for reprocess published in cellar notices

Author: CaptainOfHacks

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()

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)