Fix multiple procedure CETs

Author: CaptainOfHacks

ted_sws/master_data_registry/resources/__init__.py

@@ -3,4 +3,5 @@
 MASTER_DATA_REGISTRY_RESOURCES_PATH = pathlib.Path(__file__).parent.resolve()
 
 TRIPLES_BY_CET_URI_SPARQL_QUERY_TEMPLATE_PATH = MASTER_DATA_REGISTRY_RESOURCES_PATH / "sparql_query_templates/get_by_cet_uri.rq"
+PROCEDURE_SUBJECTS_SPARQL_QUERY_TEMPLATE_PATH = MASTER_DATA_REGISTRY_RESOURCES_PATH / "sparql_query_templates/get_procedure_uris.rq"
 RDF_FRAGMENT_BY_URI_SPARQL_QUERY_TEMPLATE_PATH = MASTER_DATA_REGISTRY_RESOURCES_PATH / "sparql_query_templates/get_2_dependency_levels_for_a_uri_as_root.rq"

ted_sws/master_data_registry/resources/sparql_query_templates/get_procedure_uris.rq

@@ -0,0 +1,8 @@
+prefix epo:<http://data.europa.eu/a4g/ontology#>
+
+SELECT DISTINCT ?s
+WHERE {
+  ?s a epo:Procedure.
+  ?o ?po ?s.
+  filter not exists {?o epo:refersToPreviousProcedure ?s}
+}

ted_sws/master_data_registry/services/entity_deduplication.py

@@ -17,7 +17,7 @@
 from ted_sws.event_manager.services.log import log_error, log_notice_error
 from ted_sws.master_data_registry.services.rdf_fragment_processor import get_rdf_fragments_by_cet_uri_from_notices, \
     merge_rdf_fragments_into_graph, write_rdf_fragments_in_triple_store, RDF_FRAGMENT_FROM_NOTICE_PROPERTY, \
-    get_subjects_by_cet_uri, get_rdf_fragment_by_cet_uri_from_notice
+    get_procedure_subjects, get_rdf_fragment_by_root_uri_from_notice
 
 MDR_TEMPORARY_FUSEKI_DATASET_NAME = "tmp_mdr_dataset"
 MDR_FUSEKI_DATASET_NAME = "mdr_dataset"
@@ -258,7 +258,7 @@ def deduplicate_procedure_entities(notices: List[Notice], procedure_cet_uri: str
         if parent_notice and parent_notice.rdf_manifestation and parent_notice.rdf_manifestation.object_data:
             rdf_content = parent_notice.rdf_manifestation.object_data
             sparql_endpoint = SPARQLStringEndpoint(rdf_content=rdf_content)
-            result_uris = get_subjects_by_cet_uri(sparql_endpoint=sparql_endpoint, cet_uri=procedure_cet_uri)
+            result_uris = get_procedure_subjects(sparql_endpoint=sparql_endpoint)
             result_uris_len = len(result_uris)
             if result_uris_len != 1:
                 notice_normalised_metadata = parent_notice.normalised_metadata
@@ -269,11 +269,12 @@ def deduplicate_procedure_entities(notices: List[Notice], procedure_cet_uri: str
                     notice_status=parent_notice.status,
                     notice_eforms_subtype=notice_normalised_metadata.eforms_subtype if notice_normalised_metadata else None)
             else:
+                result_uri = result_uris[0]
                 parent_procedure_uri = rdflib.URIRef(result_uris[0])
                 parent_uries[parent_notice_id] = parent_procedure_uri
-                parent_procedure_rdf_fragments = get_rdf_fragment_by_cet_uri_from_notice(notice=parent_notice,
-                                                                                         cet_uri=procedure_cet_uri)
-                parent_new_cet = {parent_procedure_uri: parent_procedure_rdf_fragments[0]}
+                parent_procedure_rdf_fragment = get_rdf_fragment_by_root_uri_from_notice(notice=parent_notice,
+                                                                                         root_uri=result_uri)
+                parent_new_cet = {parent_procedure_uri: parent_procedure_rdf_fragment}
                 register_new_cets_in_mdr(new_canonical_entities=parent_new_cet, triple_store=triple_store,
                                          mdr_dataset_name=mdr_dataset_name)
 
@@ -282,7 +283,7 @@ def deduplicate_procedure_entities(notices: List[Notice], procedure_cet_uri: str
         for child_notice in notice_families[parent_uri_key]:
             rdf_content = child_notice.rdf_manifestation.object_data
             sparql_endpoint = SPARQLStringEndpoint(rdf_content=rdf_content)
-            result_uris = get_subjects_by_cet_uri(sparql_endpoint=sparql_endpoint, cet_uri=procedure_cet_uri)
+            result_uris = get_procedure_subjects(sparql_endpoint=sparql_endpoint)
             result_uris_len = len(result_uris)
             if result_uris_len != 1:
                 notice_normalised_metadata = child_notice.normalised_metadata

ted_sws/master_data_registry/services/rdf_fragment_processor.py

@@ -10,15 +10,15 @@
 """
 import pathlib
 from string import Template
-from typing import List, Tuple
+from typing import List, Tuple, Optional
 
 import rdflib
 
 from ted_sws.core.model.notice import Notice
 from ted_sws.data_manager.adapters.sparql_endpoint import SPARQLStringEndpoint
 from ted_sws.data_manager.adapters.triple_store import TripleStoreABC
 from ted_sws.master_data_registry.resources import RDF_FRAGMENT_BY_URI_SPARQL_QUERY_TEMPLATE_PATH, \
-    TRIPLES_BY_CET_URI_SPARQL_QUERY_TEMPLATE_PATH
+    TRIPLES_BY_CET_URI_SPARQL_QUERY_TEMPLATE_PATH, PROCEDURE_SUBJECTS_SPARQL_QUERY_TEMPLATE_PATH
 
 RDFTriple = Tuple[rdflib.term.Node, rdflib.term.Node, rdflib.term.Node]
 
@@ -39,6 +39,17 @@ def get_subjects_by_cet_uri(sparql_endpoint: SPARQLStringEndpoint, cet_uri: str)
     return query_table_result["s"].to_list()
 
 
+def get_procedure_subjects(sparql_endpoint: SPARQLStringEndpoint) -> List[str]:
+    """
+        This function return a list of procedure subjects.
+    :param sparql_endpoint:
+    :return:
+    """
+    sparql_query = PROCEDURE_SUBJECTS_SPARQL_QUERY_TEMPLATE_PATH.read_text(encoding="utf-8")
+    query_table_result = sparql_endpoint.with_query(sparql_query=sparql_query).fetch_tabular()
+    return query_table_result["s"].to_list()
+
+
 def get_rdf_fragment_by_root_uri(sparql_endpoint: SPARQLStringEndpoint, root_uri: str,
                                  inject_triples: List[RDFTriple] = None) -> rdflib.Graph:
     """
@@ -92,6 +103,25 @@ def get_rdf_fragments_by_cet_uri_from_file(rdf_file_path: pathlib.Path, cet_uri:
                                                    rdf_content_format=rdf_file_content_format)
 
 
+def get_rdf_fragment_by_root_uri_from_notice(notice: Notice, root_uri: str) -> Optional[rdflib.Graph]:
+    """
+         This function extracts from a Notice RDF content a RDFFragment dependent on a root URI.
+    :param notice:
+    :param root_uri:
+    :return:
+    """
+    sparql_endpoint = SPARQLStringEndpoint(rdf_content=notice.rdf_manifestation.object_data,
+                                          rdf_content_format=DEFAULT_RDF_FILE_FORMAT)
+    rdf_fragment = get_rdf_fragment_by_root_uri(sparql_endpoint=sparql_endpoint, root_uri=root_uri,
+                                                inject_triples=[(rdflib.URIRef(root_uri),
+                                                                 RDF_FRAGMENT_FROM_NOTICE_PROPERTY,
+                                                                 rdflib.Literal(notice.ted_id))
+                                                                ]
+                                                )
+    return rdf_fragment
+
+
+
 def get_rdf_fragment_by_cet_uri_from_notice(notice: Notice, cet_uri: str) -> List[rdflib.Graph]:
     """
         This function extracts from a Notice RDF content a list of RDFFragments dependent on a CET URI.