Merge remote-tracking branch 'origin/main' into feature/TED-677

Author: Kolea Plesco

Makefile

@@ -213,7 +213,7 @@ stop-metabase:
 init-rml-mapper:
 	@ echo -e "RMLMapper folder initialisation!"
 	@ mkdir -p ./.rmlmapper
-	@ wget -c https://api.bitbucket.org/2.0/repositories/Dragos0000/rml-mapper/src/master/rmlmapper.jar -P ./.rmlmapper
+	@ wget -c https://github.com/RMLio/rmlmapper-java/releases/download/v6.0.0/rmlmapper-6.0.0-r363-all.jar -O ./.rmlmapper/rmlmapper.jar
 
 init-limes:
 	@ echo -e "Limes folder initialisation!"

dags/daily_materialized_view_update.py

@@ -6,7 +6,7 @@
 from ted_sws.data_manager.services.create_batch_collection_materialised_view import \
     create_batch_collection_materialised_view
 from ted_sws.data_manager.services.create_notice_collection_materialised_view import \
-    create_notice_collection_materialised_view, update_notice_collection_materialised_view
+    create_notice_collection_materialised_view, create_notice_kpi_collection
 
 DAG_NAME = "daily_materialized_view_update"
 
@@ -22,12 +22,16 @@ def create_materialised_view():
         create_notice_collection_materialised_view(mongo_client=mongo_client)
 
     @task
-    def update_materialised_view_with_logs():
+    def create_kpi_collection_for_notices():
+        mongo_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        create_notice_kpi_collection(mongo_client=mongo_client)
+
+    @task
+    def aggregate_batch_logs():
         mongo_client = MongoClient(config.MONGO_DB_AUTH_URL)
-        update_notice_collection_materialised_view(mongo_client=mongo_client)
         create_batch_collection_materialised_view(mongo_client=mongo_client)
 
-    create_materialised_view() >> update_materialised_view_with_logs()
+    create_materialised_view() >> create_kpi_collection_for_notices() >> aggregate_batch_logs()
 
 
 dag = daily_materialized_view_update()

dags/operators/DagBatchPipelineOperator.py

@@ -8,15 +8,16 @@
     smart_xcom_push
 from dags.pipelines.pipeline_protocols import NoticePipelineCallable
 from ted_sws import config
+from ted_sws.core.model.notice import NoticeStatus
 from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
 from ted_sws.event_manager.model.event_message import EventMessage, NoticeEventMessage
-from ted_sws.event_manager.services.log import log_error
+from ted_sws.event_manager.services.log import log_notice_error
 from ted_sws.event_manager.services.logger_from_context import get_logger, handle_event_message_metadata_dag_context
 
 NOTICE_IDS_KEY = "notice_ids"
 START_WITH_STEP_NAME_KEY = "start_with_step_name"
 EXECUTE_ONLY_ONE_STEP_KEY = "execute_only_one_step"
-DEFAULT_NUBER_OF_CELERY_WORKERS = 144
+DEFAULT_NUBER_OF_CELERY_WORKERS = 144 #TODO: revise this config
 NOTICE_PROCESS_WORKFLOW_DAG_NAME = "notice_process_workflow"
 DEFAULT_START_WITH_TASK_ID = "notice_normalisation_pipeline"
 DEFAULT_PIPELINE_NAME_FOR_LOGS = "unknown_pipeline_name"
@@ -88,9 +89,14 @@ def execute(self, context: Any):
                     if result_notice_pipeline.processed:
                         processed_notice_ids.append(notice_id)
                     notice_event.end_record()
+                    if notice.normalised_metadata:
+                        notice_event.notice_form_number = notice.normalised_metadata.form_number
+                        notice_event.notice_eforms_subtype = notice.normalised_metadata.eforms_subtype
+                        notice_event.notice_status = str(notice.status)
                     logger.info(event_message=notice_event)
                 except Exception as e:
-                    log_error(message=str(e))
+                    log_notice_error(message=str(e), notice_id=notice_id, domain_action=pipeline_name)
+
         batch_event_message.end_record()
         logger.info(event_message=batch_event_message)
         if not processed_notice_ids:

dags/pipelines/notice_batch_processor_pipelines.py

@@ -1,7 +1,10 @@
 from typing import List
 from pymongo import MongoClient
 
+from ted_sws.master_data_registry.services.entity_deduplication import deduplicate_procedure_entities
+
 CET_URIS = ["http://www.w3.org/ns/org#Organization"]
+PROCEDURE_CET_URI = "http://data.europa.eu/a4g/ontology#Procedure"
 
 
 def notices_batch_distillation_pipeline(notice_ids: List[str], mongodb_client: MongoClient) -> List[str]:
@@ -23,6 +26,7 @@ def notices_batch_distillation_pipeline(notice_ids: List[str], mongodb_client: M
         notices.append(notice)
     for cet_uri in CET_URIS:
         deduplicate_entities_by_cet_uri(notices=notices, cet_uri=cet_uri)
+    deduplicate_procedure_entities(notices=notices, procedure_cet_uri=PROCEDURE_CET_URI, mongodb_client=mongodb_client)
     for notice in notices:
         notice_repository.update(notice=notice)
     return notice_ids

dags/pipelines/notice_processor_pipelines.py

@@ -1,6 +1,7 @@
 from pymongo import MongoClient
 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
 
 
 def notice_normalisation_pipeline(notice: Notice, mongodb_client: MongoClient) -> NoticePipelineOutput:
@@ -9,7 +10,7 @@ def notice_normalisation_pipeline(notice: Notice, mongodb_client: MongoClient) -
     """
     from ted_sws.data_sampler.services.notice_xml_indexer import index_notice
     from ted_sws.notice_metadata_processor.services.metadata_normalizer import normalise_notice
-
+    notice.update_status_to(new_status=NoticeStatus.RAW)
     indexed_notice = index_notice(notice=notice)
     normalised_notice = normalise_notice(notice=indexed_notice)
 
@@ -26,10 +27,16 @@ def notice_transformation_pipeline(notice: Notice, mongodb_client: MongoClient)
     from ted_sws.notice_transformer.services.notice_transformer import transform_notice
     from ted_sws.notice_transformer.adapters.rml_mapper import RMLMapper
     from ted_sws.data_manager.adapters.mapping_suite_repository import MappingSuiteRepositoryMongoDB
-
+    notice.update_status_to(new_status=NoticeStatus.NORMALISED_METADATA)
     mapping_suite_repository = MappingSuiteRepositoryMongoDB(mongodb_client=mongodb_client)
     result = notice_eligibility_checker(notice=notice, mapping_suite_repository=mapping_suite_repository)
     if not result:
+        log_notice_error(
+            message=f"This notice {notice.ted_id} is not eligible for transformation. Notice info: "
+                    f"form_number=[{notice.normalised_metadata.form_number}],"
+                    f" eform_subtype=[{notice.normalised_metadata.eforms_subtype}], "
+                    f"xsd_version=[{notice.normalised_metadata.xsd_version}]. Check mapping suites!",
+            notice_id=notice.ted_id, domain_action=notice_transformation_pipeline.__name__)
         return NoticePipelineOutput(notice=notice, processed=False)
     notice_id, mapping_suite_id = result
     # TODO: Implement XML preprocessing
@@ -50,18 +57,18 @@ def notice_validation_pipeline(notice: Notice, mongodb_client: MongoClient) -> N
     from ted_sws.notice_validator.services.xpath_coverage_runner import validate_xpath_coverage_notice
     from ted_sws.data_manager.adapters.mapping_suite_repository import MappingSuiteRepositoryMongoDB
     from ted_sws.event_manager.services.log import log_notice_info
-
+    notice.update_status_to(new_status=NoticeStatus.DISTILLED)
     mapping_suite_id = notice.distilled_rdf_manifestation.mapping_suite_id
     mapping_suite_repository = MappingSuiteRepositoryMongoDB(mongodb_client=mongodb_client)
     mapping_suite = mapping_suite_repository.get(reference=mapping_suite_id)
     log_notice_info(message="Validation :: XPATH coverage :: START", notice_id=notice.ted_id)
     validate_xpath_coverage_notice(notice=notice, mapping_suite=mapping_suite, mongodb_client=mongodb_client)
     log_notice_info(message="Validation :: XPATH coverage :: END", notice_id=notice.ted_id)
     log_notice_info(message="Validation :: SPARQL :: START", notice_id=notice.ted_id)
-    validate_notice_with_sparql_suite(notice=notice, mapping_suite_package=mapping_suite)
+    validate_notice_with_sparql_suite(notice=notice, mapping_suite_package=mapping_suite, execute_full_validation=False)
     log_notice_info(message="Validation :: SPARQL :: END", notice_id=notice.ted_id)
     log_notice_info(message="Validation :: SHACL :: START", notice_id=notice.ted_id)
-    validate_notice_with_shacl_suite(notice=notice, mapping_suite_package=mapping_suite)
+    validate_notice_with_shacl_suite(notice=notice, mapping_suite_package=mapping_suite, execute_full_validation=False)
     log_notice_info(message="Validation :: SHACL :: END", notice_id=notice.ted_id)
     log_notice_info(message="Validation :: Summary :: START", notice_id=notice.ted_id)
     validation_summary_report_notice(notice=notice)
@@ -75,6 +82,7 @@ def notice_package_pipeline(notice: Notice, mongodb_client: MongoClient) -> Noti
     """
     from ted_sws.notice_packager.services.notice_packager import package_notice
 
+    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)
@@ -88,7 +96,7 @@ def notice_publish_pipeline(notice: Notice, mongodb_client: MongoClient) -> Noti
     from ted_sws.notice_publisher.services.notice_publisher import publish_notice, publish_notice_rdf_into_s3
     from ted_sws.event_manager.services.log import log_notice_error
     from ted_sws import config
-
+    notice.update_status_to(new_status=NoticeStatus.PACKAGED)
     if config.S3_PUBLISH_ENABLED:
         published_into_s3 = publish_notice_rdf_into_s3(notice=notice)
         if not published_into_s3:
@@ -99,4 +107,5 @@ def notice_publish_pipeline(notice: Notice, mongodb_client: MongoClient) -> Noti
     if result:
         return NoticePipelineOutput(notice=notice)
     else:
+        notice.set_is_eligible_for_publishing(eligibility=False)
         return NoticePipelineOutput(notice=notice, processed=False)

dags/pipelines/notice_selectors_pipelines.py

@@ -7,11 +7,20 @@
 PUBLICATION_DATE = "normalised_metadata.publication_date"
 
 
-def build_selector_mongodb_filter(notice_status: str, form_number: str = None,
+def build_selector_mongodb_filter(notice_statuses: List[str], form_number: str = None,
                                   start_date: str = None, end_date: str = None,
                                   xsd_version: str = None) -> dict:
+    """
+
+    :param notice_statuses:
+    :param form_number:
+    :param start_date:
+    :param end_date:
+    :param xsd_version:
+    :return:
+    """
     from datetime import datetime
-    mongodb_filter = {NOTICE_STATUS: notice_status}
+    mongodb_filter = {NOTICE_STATUS: {"$in": notice_statuses}}
     if form_number:
         mongodb_filter[FORM_NUMBER] = form_number
     if start_date and end_date:
@@ -26,21 +35,26 @@ def build_selector_mongodb_filter(notice_status: str, form_number: str = None,
 def notice_ids_selector_by_status(notice_statuses: List[NoticeStatus], form_number: str = None,
                                   start_date: str = None, end_date: str = None,
                                   xsd_version: str = None) -> List[str]:
+    """
+
+    :param notice_statuses:
+    :param form_number:
+    :param start_date:
+    :param end_date:
+    :param xsd_version:
+    :return:
+    """
     from pymongo import MongoClient
     from ted_sws import config
     from ted_sws.data_manager.adapters.notice_repository import NoticeRepository, NOTICE_TED_ID
 
     mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
     notice_repository = NoticeRepository(mongodb_client=mongodb_client)
-    notice_ids = []
-    for notice_status in notice_statuses:
-        mongodb_filter = build_selector_mongodb_filter(notice_status=str(notice_status),
-                                                       form_number=form_number,
-                                                       start_date=start_date,
-                                                       end_date=end_date,
-                                                       xsd_version=xsd_version
-                                                       )
-        mongodb_result_iterator = notice_repository.collection.find(mongodb_filter, {NOTICE_TED_ID: 1})
-        notice_ids.extend([result_dict[NOTICE_TED_ID] for result_dict in mongodb_result_iterator])
-
-    return notice_ids
+    notice_statuses = [str(notice_status) for notice_status in notice_statuses]
+    mongodb_filter = build_selector_mongodb_filter(notice_statuses=notice_statuses,
+                                                   form_number=form_number,
+                                                   start_date=start_date,
+                                                   end_date=end_date,
+                                                   xsd_version=xsd_version)
+    mongodb_result_iterator = notice_repository.collection.find(mongodb_filter, {NOTICE_TED_ID: 1})
+    return [result_dict[NOTICE_TED_ID] for result_dict in mongodb_result_iterator]

dags/selector_repackage_process_orchestrator.py

@@ -13,6 +13,7 @@
 DAG_NAME = "selector_re_package_process_orchestrator"
 
 RE_PACKAGE_TARGET_NOTICE_STATES = [NoticeStatus.VALIDATED, NoticeStatus.INELIGIBLE_FOR_PACKAGING,
+                                   NoticeStatus.ELIGIBLE_FOR_PACKAGING,
                                    NoticeStatus.INELIGIBLE_FOR_PUBLISHING]
 TRIGGER_NOTICE_PROCESS_WORKFLOW_TASK_ID = "trigger_notice_process_workflow"
 FORM_NUMBER_DAG_PARAM = "form_number"

dags/selector_republish_process_orchestrator.py

@@ -13,7 +13,9 @@
 
 DAG_NAME = "selector_re_publish_process_orchestrator"
 
-RE_PUBLISH_TARGET_NOTICE_STATES = [NoticeStatus.PUBLICLY_UNAVAILABLE, NoticeStatus.ELIGIBLE_FOR_PUBLISHING]
+RE_PUBLISH_TARGET_NOTICE_STATES = [NoticeStatus.ELIGIBLE_FOR_PUBLISHING, NoticeStatus.INELIGIBLE_FOR_PUBLISHING,
+                                   NoticeStatus.PACKAGED
+                                   ]
 TRIGGER_NOTICE_PROCESS_WORKFLOW_TASK_ID = "trigger_notice_process_workflow"
 FORM_NUMBER_DAG_PARAM = "form_number"
 START_DATE_DAG_PARAM = "start_date"

dags/selector_retransform_process_orchestrator.py

@@ -13,7 +13,10 @@
 
 DAG_NAME = "selector_re_transform_process_orchestrator"
 
-RE_TRANSFORM_TARGET_NOTICE_STATES = [NoticeStatus.NORMALISED_METADATA, NoticeStatus.INELIGIBLE_FOR_TRANSFORMATION]
+RE_TRANSFORM_TARGET_NOTICE_STATES = [NoticeStatus.NORMALISED_METADATA, NoticeStatus.INELIGIBLE_FOR_TRANSFORMATION,
+                                     NoticeStatus.ELIGIBLE_FOR_TRANSFORMATION, NoticeStatus.PREPROCESSED_FOR_TRANSFORMATION,
+                                     NoticeStatus.TRANSFORMED, NoticeStatus.DISTILLED
+                                     ]
 TRIGGER_NOTICE_PROCESS_WORKFLOW_TASK_ID = "trigger_notice_process_workflow"
 FORM_NUMBER_DAG_PARAM = "form_number"
 START_DATE_DAG_PARAM = "start_date"

infra/airflow-cluster/Dockerfile

@@ -14,7 +14,7 @@ COPY requirements.txt /opt/airflow
 # working in the /opt/airflow
 WORKDIR /opt/airflow
 RUN mkdir -p ./.rmlmapper
-RUN wget -c https://api.bitbucket.org/2.0/repositories/Dragos0000/rml-mapper/src/master/rmlmapper.jar -P ./.rmlmapper
+RUN wget -c https://github.com/RMLio/rmlmapper-java/releases/download/v6.0.0/rmlmapper-6.0.0-r363-all.jar -O ./.rmlmapper/rmlmapper.jar
 
 
 RUN wget -c https://kumisystems.dl.sourceforge.net/project/saxon/Saxon-HE/10/Java/SaxonHE10-6J.zip -P .saxon/