add deduplication report generation

Author: CaptainOfHacks

dags/pipelines/notice_processor_pipelines.py

@@ -2,6 +2,8 @@
 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_validator.services.entity_deduplication_validation import \
+    generate_rdf_manifestation_entity_deduplication_report
 
 
 def notice_normalisation_pipeline(notice: Notice, mongodb_client: MongoClient) -> NoticePipelineOutput:
@@ -75,6 +77,9 @@ def notice_validation_pipeline(notice: Notice, mongodb_client: MongoClient) -> N
     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)
+    log_notice_info(message="Validation :: Entity deduplication :: START", notice_id=notice.ted_id)
+    generate_rdf_manifestation_entity_deduplication_report(rdf_manifestation=notice.distilled_rdf_manifestation)
+    log_notice_info(message="Validation :: Entity deduplication :: END", notice_id=notice.ted_id)
     return NoticePipelineOutput(notice=notice)
 
 

ted_sws/core/model/manifestation.py

@@ -181,13 +181,21 @@ class SHACLTestSuiteValidationReport(RDFValidationManifestation):
     validation_results: Union[QueriedSHACLShapeValidationResult, str]
 
 
+class EntityDeduplicationReport(Manifestation):
+    object_data: Optional[str]
+    number_of_duplicates: int
+    number_of_cets: int
+    uries: List[str]
+
+
 class RDFManifestation(Manifestation):
     """
         Transformed manifestation in RDF format
     """
     mapping_suite_id = "unknown_mapping_suite_id"
     shacl_validations: List[SHACLTestSuiteValidationReport] = []
     sparql_validations: List[SPARQLTestSuiteValidationReport] = []
+    deduplication_report: Optional[EntityDeduplicationReport]
 
     def validation_exists(self, validation, validations):
         """

ted_sws/notice_validator/services/entity_deduplication_validation.py

@@ -0,0 +1,28 @@
+import rdflib
+from rdflib import OWL
+
+from ted_sws.core.model.manifestation import RDFManifestation, EntityDeduplicationReport
+from ted_sws.data_manager.adapters.sparql_endpoint import DEFAULT_RDF_FILE_FORMAT
+
+
+def generate_rdf_manifestation_entity_deduplication_report(rdf_manifestation: RDFManifestation):
+    """
+        This function generate entity deduplication report for an RDF manifestation.
+    :param rdf_manifestation:
+    :return:
+    """
+    rdf_content = rdflib.Graph()
+    rdf_content.parse(rdf_manifestation.object_data.encode(encoding="utf-8"), format=DEFAULT_RDF_FILE_FORMAT)
+    duplicate_entities = set()
+    new_entities = set()
+    for triple_sub, triple_pred, triple_obj in rdf_content.triples(triple=(None, OWL.sameAs, None)):
+        if triple_sub == triple_obj:
+            new_entities.add(str(triple_sub))
+        else:
+            duplicate_entities.add(str(triple_sub))
+
+    rdf_manifestation.deduplication_report = EntityDeduplicationReport(number_of_duplicates=len(duplicate_entities),
+                                                                       number_of_cets=len(new_entities),
+                                                                       uries=list(
+                                                                           duplicate_entities.union(new_entities))
+                                                                       )

tests/e2e/master_data_registry/test_entity_deduplication.py

@@ -5,6 +5,8 @@
 
 from ted_sws.master_data_registry.services.entity_deduplication import deduplicate_entities_by_cet_uri, \
     deduplicate_procedure_entities
+from ted_sws.notice_validator.services.entity_deduplication_validation import \
+    generate_rdf_manifestation_entity_deduplication_report
 
 TEST_MDR_REPOSITORY = "tmp_mdr_test_repository"
 TEST_QUERY_UNIQUE_NAMES = """SELECT distinct ?name
@@ -53,6 +55,12 @@ def test_deduplicate_entities_by_cet_uri(notice_with_rdf_manifestation, organisa
     for triple in canonical_cets_same_as_triples:
         assert str(triple[2]) in canonical_cets_set
 
+    assert notice_with_rdf_manifestation.distilled_rdf_manifestation.deduplication_report is None
+    generate_rdf_manifestation_entity_deduplication_report(notice_with_rdf_manifestation.distilled_rdf_manifestation)
+    assert notice_with_rdf_manifestation.distilled_rdf_manifestation.deduplication_report is not None
+    deduplication_report = notice_with_rdf_manifestation.distilled_rdf_manifestation.deduplication_report
+    assert deduplication_report.number_of_cets == len(canonical_cets_set)
+
     fuseki_triple_store.delete_repository(repository_name=TEST_MDR_REPOSITORY)