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

Author: Kolea Plesco

dags/notice_validation_workflow.py

@@ -33,14 +33,16 @@ def notice_daily_validation_workflow():
         ))
     )
     def validate_fetched_notices():
+        """
+        :return:
+        """
         from ted_sws import config
         from ted_sws.supra_notice_manager.services.supra_notice_validator import validate_and_update_daily_supra_notice
 
         publication_date = get_notice_publication_date()
         mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
         validate_and_update_daily_supra_notice(notice_publication_day=publication_date,
-                                               mongodb_client=mongodb_client
-                                               )
+                                               mongodb_client=mongodb_client)
 
     @task
     @event_log(TechnicalEventMessage(
@@ -50,15 +52,19 @@ def validate_fetched_notices():
         ))
     )
     def summarize_validation_for_daily_supra_notice():
-        from ted_sws import config
-        from ted_sws.supra_notice_manager.services.supra_notice_validator import \
-            summary_validation_for_daily_supra_notice
-
-        publication_date = get_notice_publication_date()
-        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
-        summary_validation_for_daily_supra_notice(notice_publication_day=publication_date,
-                                                  mongodb_client=mongodb_client
-                                                  )
+        """
+        :return:
+        """
+        # Temporally disable DailySupraNotice validation summary
+        # from ted_sws import config
+        # from ted_sws.supra_notice_manager.services.supra_notice_validator import \
+        #     summary_validation_for_daily_supra_notice
+        #
+        # publication_date = get_notice_publication_date()
+        # mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        # summary_validation_for_daily_supra_notice(notice_publication_day=publication_date,
+        #                                           mongodb_client=mongodb_client
+        #                                           )
 
     @task
     @event_log(TechnicalEventMessage(
@@ -69,23 +75,13 @@ def summarize_validation_for_daily_supra_notice():
     )
     def validate_availability_of_notice_in_cellar():
         from ted_sws import config
-        from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
-        from ted_sws.data_manager.adapters.supra_notice_repository import DailySupraNoticeRepository
-        from ted_sws.notice_validator.services.check_availability_of_notice_in_cellar import \
-            validate_notice_availability_in_cellar
+        from ted_sws.supra_notice_manager.services.supra_notice_validator import \
+            validate_and_update_supra_notice_availability_in_cellar
 
         notice_publication_day = get_notice_publication_date()
         mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
-        repo = DailySupraNoticeRepository(mongodb_client=mongodb_client)
-        supra_notice = repo.get(reference=notice_publication_day)
-        if supra_notice:
-            notice_repository = NoticeRepository(mongodb_client=mongodb_client)
-            for notice_id in supra_notice.notice_ids:
-                notice = notice_repository.get(reference=notice_id)
-                old_notice_status = notice.status
-                notice = validate_notice_availability_in_cellar(notice=notice)
-                if notice.status != old_notice_status:
-                    notice_repository.update(notice=notice)
+        validate_and_update_supra_notice_availability_in_cellar(notice_publication_day=notice_publication_day,
+                                                                mongodb_client=mongodb_client)
 
     validate_fetched_notices() >> summarize_validation_for_daily_supra_notice() >> validate_availability_of_notice_in_cellar()
 

dags/operators/DagBatchPipelineOperator.py

@@ -19,7 +19,8 @@
 DEFAULT_NUBER_OF_CELERY_WORKERS = 144
 NOTICE_PROCESS_WORKFLOW_DAG_NAME = "notice_process_workflow"
 DEFAULT_START_WITH_TASK_ID = "notice_normalisation_pipeline"
-DEFAULT_PIPELINE_NAME_FOR_LOGS  = "unknown_pipeline_name"
+DEFAULT_PIPELINE_NAME_FOR_LOGS = "unknown_pipeline_name"
+
 
 class BatchPipelineCallable(Protocol):
 

dags/pipelines/notice_processor_pipelines.py

@@ -49,14 +49,23 @@ def notice_validation_pipeline(notice: Notice, mongodb_client: MongoClient) -> N
     from ted_sws.notice_validator.services.validation_summary_runner import validation_summary_report_notice
     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
 
     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)
+    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)
+    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)
+    log_notice_info(message="Validation :: Summary :: END", notice_id=notice.ted_id)
     return NoticePipelineOutput(notice=notice)
 
 
@@ -78,11 +87,13 @@ 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
 
-    published_into_s3 = publish_notice_rdf_into_s3(notice=notice)
-    if not published_into_s3:
-        log_notice_error(message="Can't load notice distilled rdf manifestation into S3 bucket!",
-                         notice_id=notice.ted_id)
+    if config.S3_PUBLISH_ENABLED:
+        published_into_s3 = publish_notice_rdf_into_s3(notice=notice)
+        if not published_into_s3:
+            log_notice_error(message="Can't load notice distilled rdf manifestation into S3 bucket!",
+                             notice_id=notice.ted_id)
     notice.set_is_eligible_for_publishing(eligibility=True)
     result = publish_notice(notice=notice)
     if result:

requirements.txt

@@ -25,3 +25,4 @@ ordered-set~=4.0.2
 json2html~=1.3.0
 minio~=7.1.1
 certifi
+networkx~=2.8.8

setup.py

@@ -95,7 +95,8 @@ def open_local(paths, mode="r", encoding="utf8"):
             "normalisation_resource_generator = ted_sws.data_manager.entrypoints.cli.cmd_generate_mapping_resources:main",
             "s3_rdf_publisher = ted_sws.notice_publisher.entrypoints.cli.cmd_s3_rdf_publisher:main",
             "bulk_packager = ted_sws.notice_packager.entrypoints.cli.cmd_bulk_packager:main",
-            "api-digest_service-start-server = ted_sws.notice_transformer.entrypoints.api.digest_service.server:api_server_start"
+            "api-digest_service-start-server = ted_sws.notice_transformer.entrypoints.api.digest_service.server:api_server_start",
+            "rdf_component_detector = ted_sws.rdf_component_detector.entrypoints.cli.cmd_rdf_component_detector:main"
         ],
     },
     include_package_data=True,

ted_sws/__init__.py

@@ -238,7 +238,7 @@ def S3_PUBLISH_REGION(self) -> str:
 
     @property
     def S3_PUBLISH_SSL_VERIFY(self) -> bool:
-        return True if AirflowAndEnvConfigResolver().config_resolve() == "1" else False
+        return AirflowAndEnvConfigResolver().config_resolve() == "1"
 
     @property
     def S3_PUBLISH_NOTICE_BUCKET(self) -> str:
@@ -248,6 +248,10 @@ def S3_PUBLISH_NOTICE_BUCKET(self) -> str:
     def S3_PUBLISH_NOTICE_RDF_BUCKET(self) -> str:
         return AirflowAndEnvConfigResolver().config_resolve()
 
+    @property
+    def S3_PUBLISH_ENABLED(self) -> bool:
+        return AirflowAndEnvConfigResolver().config_resolve() == "1"
+
 
 class TedConfigResolver(MongoDBConfig, RMLMapperConfig, XMLProcessorConfig, ELKConfig, LoggingConfig,
                         GitHubArtefacts, API, AllegroConfig, TedAPIConfig, SFTPConfig, FusekiConfig,

ted_sws/core/adapters/cmd_runner.py

@@ -17,7 +17,7 @@
 
 DEFAULT_MAPPINGS_PATH = 'mappings'
 DEFAULT_OUTPUT_PATH = 'output'
-EXIT_CODE_OK = os.EX_OK
+EXIT_CODE_OK = 0  # os.EX_OK
 DEFAULT_EXIT_CODE = EXIT_CODE_OK
 
 

ted_sws/core/model/supra_notice.py

@@ -35,9 +35,10 @@ class SupraNoticeValidationReport(Manifestation):
         Result of checking whether all the notices published in TED are present in the internal database.
     """
     missing_notice_ids: Optional[List[str]]
+    not_published_notice_ids: Optional[List[str]]
 
     def is_valid(self):
-        if not self.missing_notice_ids:
+        if not self.missing_notice_ids and not self.not_published_notice_ids:
             return True
         return False
 

ted_sws/notice_validator/adapters/validation_summary_runner.py

@@ -106,7 +106,7 @@ def notice_shacl_summary(self, notice: Notice, report: RDFManifestationValidatio
                     validation_results = shacl_report.validation_results
                     is_new, result_validation = self.shacl_summary_result(shacl_report, result_counts)
                     result_count: SHACLSummarySeverityCountReport = result_validation.result_severity.aggregate
-                    if validation_results:
+                    if validation_results and validation_results.results_dict:
                         bindings = validation_results.results_dict['results']['bindings']
                         for binding in bindings:
                             result_severity = binding['resultSeverity']