Merge branch 'main' into feature/TED-1030

Author: duprijil

dags/notice_fetch_by_date_workflow.py

@@ -1,3 +1,5 @@
+from datetime import timedelta
+
 from airflow.decorators import dag, task
 from airflow.operators.dummy import DummyOperator
 from airflow.operators.python import BranchPythonOperator, PythonOperator
@@ -58,7 +60,9 @@ def validate_fetched_notices():
         from datetime import datetime
         from pymongo import MongoClient
 
-        publication_date = datetime.strptime(get_dag_param(key=WILD_CARD_DAG_KEY), "%Y%m%d*")
+        publication_date = datetime.strptime(get_dag_param(key=WILD_CARD_DAG_KEY,
+                                                           default_value=(datetime.now() - timedelta(days=1)).strftime(
+                                                               "%Y%m%d*")), "%Y%m%d*")
         mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
         validate_and_update_daily_supra_notice(notice_publication_day=publication_date,
                                                mongodb_client=mongodb_client)

ted_sws/data_manager/adapters/triple_store.py

@@ -13,9 +13,9 @@
 from ted_sws import config
 from ted_sws.data_manager.adapters.sparql_endpoint import TripleStoreEndpointABC, SPARQLTripleStoreEndpoint
 
-
 FUSEKI_REPOSITORY_ALREADY_EXIST_ERROR_MSG = 'A repository with this name already exists.'
 
+
 class TripleStoreABC:
     @abc.abstractmethod
     def create_repository(self, repository_name: str):

ted_sws/mapping_suite_processor/adapters/mapping_suite_hasher.py

@@ -8,6 +8,7 @@
 """ """
 import hashlib
 import pathlib
+import re
 from typing import Tuple, List, Union
 
 from ted_sws.data_manager.adapters.mapping_suite_repository import MS_TRANSFORM_FOLDER_NAME, \
@@ -35,7 +36,10 @@ def _hash_a_file(file_path: pathlib.Path) -> Tuple[str, str]:
             """
                 Return a tuple of the relative file path and the file hash.
             """
-            hashed_line = hashlib.sha256(file_path.read_bytes()).hexdigest()
+            # remove new-lines to align content generated on different operating systems
+            new_line_pattern = re.compile(b'\r\n|\r|\n')
+            file_content = re.sub(new_line_pattern, b'', file_path.read_bytes())
+            hashed_line = hashlib.sha256(file_content).hexdigest()
             relative_path = str(file_path).replace(str(self.mapping_suite_path), "")
             return relative_path, hashed_line
 
@@ -62,7 +66,7 @@ def hash_mapping_suite(self, with_version: str = "") -> str:
         """
             Returns a hash of the mapping suite.
             Only the critical resources are hashed in the mapping suite.
-            The decission which rescources are "critical" is implemented
+            The decision which resources are "critical" is implemented
             in self.hash_critical_mapping_files() function.
 
             If "with_version" parameter is used, then it computed the mapping

ted_sws/mapping_suite_processor/services/load_mapping_suite_output_into_triple_store.py

@@ -1,7 +1,7 @@
 import pathlib
 
 from ted_sws import config
-from ted_sws.data_manager.adapters.triple_store import AllegroGraphTripleStore, FusekiAdapter, TripleStoreABC
+from ted_sws.data_manager.adapters.triple_store import FusekiAdapter, TripleStoreABC
 
 
 def repository_exists(triple_store: TripleStoreABC, repository_name) -> bool:

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_subjects_by_cet_uri, get_rdf_fragment_by_cet_uri_from_notice
 
 MDR_TEMPORARY_FUSEKI_DATASET_NAME = "tmp_mdr_dataset"
 MDR_FUSEKI_DATASET_NAME = "mdr_dataset"
@@ -226,12 +226,14 @@ def deduplicate_entities_by_cet_uri(notices: List[Notice], cet_uri: str,
                                        alignment_graph=cet_alignment_links, inject_reflexive_links=True)
 
 
-def deduplicate_procedure_entities(notices: List[Notice], procedure_cet_uri: str, mongodb_client: MongoClient):
+def deduplicate_procedure_entities(notices: List[Notice], procedure_cet_uri: str, mongodb_client: MongoClient,
+                                   mdr_dataset_name: str = MDR_FUSEKI_DATASET_NAME):
     """
          This function deduplicate procedure entities for each notice from batch of notices.
     :param notices:
     :param procedure_cet_uri:
     :param mongodb_client:
+    :param mdr_dataset_name:
     :return:
     """
     notice_families = defaultdict(list)
@@ -243,6 +245,14 @@ def deduplicate_procedure_entities(notices: List[Notice], procedure_cet_uri: str
 
     parent_uries = {}
     notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+    triple_store = FusekiAdapter()
+    if mdr_dataset_name not in triple_store.list_repositories():
+        try:
+            triple_store.create_repository(repository_name=mdr_dataset_name)
+        except Exception as exception:
+            if str(exception) != FUSEKI_REPOSITORY_ALREADY_EXIST_ERROR_MSG:
+                log_error(message=str(exception))
+
     for parent_notice_id in notice_families.keys():
         parent_notice = notice_repository.get(reference=parent_notice_id)
         if parent_notice and parent_notice.rdf_manifestation and parent_notice.rdf_manifestation.object_data:
@@ -261,6 +271,11 @@ def deduplicate_procedure_entities(notices: List[Notice], procedure_cet_uri: str
             else:
                 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]}
+                register_new_cets_in_mdr(new_canonical_entities=parent_new_cet, triple_store=triple_store,
+                                         mdr_dataset_name=mdr_dataset_name)
 
     for parent_uri_key in parent_uries.keys():
         parent_uri = parent_uries[parent_uri_key]

ted_sws/notice_fetcher/adapters/ted_api.py

@@ -16,8 +16,10 @@
 DEFAULT_TED_API_QUERY_RESULT_FIELDS = {"fields": ["AA", "AC", "CY", "DD", "DI", "DS", "TVL", "TY",
                                                   "DT", "MA", "NC", "ND", "OC", "OJ", "OL", "OY",
                                                   "PC", "PD", "PR", "RC", "RN", "RP", "TD", "TVH",
-                                                  "CONTENT", "notice-type", "award-criterion-type", "corporate-body",
-                                                  "funding", "notice-identifier", "notice-version"
+                                                  "CONTENT",
+                                                  # INFO: This query result fields is not supported correctly by TED-API.
+                                                  #"notice-type", "award-criterion-type", "corporate-body",
+                                                  #"funding", "notice-identifier", "notice-version"
                                                   ]}
 
 TOTAL_DOCUMENTS_NUMBER = "total"

ted_sws/notice_metadata_processor/model/metadata.py

@@ -15,6 +15,7 @@ class ExtractedMetadata(Metadata):
     city_of_buyer: List[LanguageTaggedString] = None
     name_of_buyer: List[LanguageTaggedString] = None
     original_language: str = None
+    uri_list: str = None
     country_of_buyer: str = None
     type_of_buyer: EncodedValue = None
     eu_institution: str = None

ted_sws/notice_metadata_processor/services/xml_manifestation_metadata_extractor.py

@@ -86,6 +86,17 @@ def name_of_buyer(self):
     def eu_institution(self):
         return self.type_of_buyer.value if self.type_of_buyer.code == "5" else "-"
 
+    @property
+    def uri_list(self):
+        uri_elements = self.manifestation_root.findall(
+            self.xpath_registry.xpath_uri_elements,
+            namespaces=self.namespaces)
+
+        return [LanguageTaggedString(text=extract_text_from_element(element=uri.find(".")),
+                                     language=extract_attribute_from_element(element=uri.find("."),
+                                                                             attrib_key="LG")) for
+                uri in uri_elements]
+
     @property
     def country_of_buyer(self):
         return extract_attribute_from_element(element=self.manifestation_root.find(
@@ -219,14 +230,16 @@ def to_metadata(self) -> ExtractedMetadata:
          Creating extracted metadata
         :return:
         """
-        metadata = ExtractedMetadata()
+        metadata: ExtractedMetadata = ExtractedMetadata()
         metadata.title = self.title
         metadata.notice_publication_number = self.notice_publication_number
         metadata.publication_date = self.publication_date
+        metadata.ojs_type = self.ojs_type
         metadata.ojs_issue_number = self.ojs_issue_number
         metadata.city_of_buyer = self.city_of_buyer
         metadata.name_of_buyer = self.name_of_buyer
         metadata.original_language = self.original_language
+        metadata.uri_list = self.uri_list
         metadata.country_of_buyer = self.country_of_buyer
         metadata.type_of_buyer = self.type_of_buyer
         metadata.eu_institution = self.eu_institution

ted_sws/notice_metadata_processor/services/xpath_registry.py

@@ -43,6 +43,10 @@ def xpath_name_of_buyer_elements(self):
     def xpath_country_of_buyer(self):
         return "manifestation_ns:CODED_DATA_SECTION/manifestation_ns:NOTICE_DATA/manifestation_ns:ISO_COUNTRY"
 
+    @property
+    def xpath_uri_elements(self):
+        return "manifestation_ns:CODED_DATA_SECTION/manifestation_ns:NOTICE_DATA/manifestation_ns:URI_LIST/"
+
     @property
     def xpath_original_language(self):
         return "manifestation_ns:CODED_DATA_SECTION/manifestation_ns:NOTICE_DATA/manifestation_ns:LG_ORIG"

ted_sws/notice_packager/entrypoints/cli/cmd_bulk_packager.py

@@ -11,17 +11,22 @@
 import base64
 import os
 from pathlib import Path
+from typing import List
 
 import click
+from pymongo import MongoClient
 
+from ted_sws import config
 from ted_sws.core.adapters.cmd_runner import CmdRunner as BaseCmdRunner
 from ted_sws.core.model.manifestation import XMLManifestation
 from ted_sws.core.model.notice import Notice
+from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
+from ted_sws.event_manager.adapters.log import LOG_WARN_TEXT
 from ted_sws.notice_metadata_processor.services.xml_manifestation_metadata_extractor import \
     XMLManifestationMetadataExtractor
-from ted_sws.notice_packager.services.metadata_transformer import MetadataTransformer
-from ted_sws.notice_packager.services.notice_packager import create_notice_package
 from ted_sws.notice_packager import DEFAULT_NOTICE_PACKAGE_EXTENSION
+from ted_sws.notice_packager.services.metadata_transformer import MetadataTransformer
+from ted_sws.notice_packager.services.notice_packager import create_notice_package, package_notice_and_save_to
 
 CMD_NAME = "CMD_BULK_PACKAGER"
 DEFAULT_FILES_COUNT: int = 3000
@@ -35,47 +40,60 @@ class PackageNotice(Notice):
 
 
 class CmdRunner(BaseCmdRunner):
-    def __init__(self, rdf_files_folder, output_folder, pkgs_count: int):
+    def __init__(self, rdf_files_folder, output_folder, pkgs_count: int, notice_ids: List = None,
+                 mongodb_client=MongoClient(config.MONGO_DB_AUTH_URL)):
         super().__init__(name=CMD_NAME)
-        self.rdf_files_path = Path(os.path.realpath(rdf_files_folder))
         self.output_path = Path(os.path.realpath(output_folder))
-        self.pkgs_count = pkgs_count
-        if not self.rdf_files_path.is_dir():
-            error_msg = f"No such folder :: [{rdf_files_folder}]"
-            self.log_failed_msg(error_msg)
-            raise FileNotFoundError(error_msg)
+        self.notices = None
+        if notice_ids:
+            self.log(LOG_WARN_TEXT.format("Notices: ") + str(notice_ids))
+            self.notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+            self.notices = []
+            for notice_id in notice_ids:
+                self.notices.append(self.notice_repository.get(reference=notice_id))
+        else:
+            self.rdf_files_path = Path(os.path.realpath(rdf_files_folder))
+            self.pkgs_count = pkgs_count
+            if not self.rdf_files_path.is_dir():
+                error_msg = f"No such folder :: [{rdf_files_folder}]"
+                self.log_failed_msg(error_msg)
+                raise FileNotFoundError(error_msg)
+
+        self.output_path.mkdir(parents=True, exist_ok=True)
 
     def run_cmd(self):
         error = None
         try:
-            self.output_path.mkdir(parents=True, exist_ok=True)
-            rdf_files = [Path(str(f_path)) for f in os.listdir(self.rdf_files_path) if
-                         os.path.isfile(f_path := os.path.join(self.rdf_files_path, f))]
-            rdf_files_count = len(rdf_files)
-            base_idx = 100000
-            year = 2021
-
-            for i in range(self.pkgs_count):
-                rdf_idx = i % rdf_files_count
-                rdf_file_path = rdf_files[rdf_idx]
-                notice_id = str(base_idx + i) + "_" + str(year)
-                pkg_name = notice_id
-                self.generate_package(notice_id, self.output_path, rdf_file_path, pkg_name)
+            if self.notices:
+                self.log("Saving packages to " + str(self.output_path))
+                for notice in self.notices:
+                    package_notice_and_save_to(notice=notice,
+                                               save_to=self.output_path)
+            else:
+                rdf_files = [Path(str(f_path)) for f in os.listdir(self.rdf_files_path) if
+                             os.path.isfile(f_path := os.path.join(self.rdf_files_path, f))]
+                rdf_files_count = len(rdf_files)
+                base_idx = 100000
+                year = 2021
+
+                for i in range(self.pkgs_count):
+                    rdf_idx = i % rdf_files_count
+                    rdf_file_path = rdf_files[rdf_idx]
+                    notice_id = str(base_idx + i) + "_" + str(year)
+                    self.generate_package(notice_id, self.output_path, rdf_file_path)
         except Exception as e:
             error = e
 
         return self.run_cmd_result(error)
 
     @classmethod
-    def generate_package(cls, notice_id, output_path, rdf_file_path, pkg_name):
+    def generate_package(cls, notice_id, output_path, rdf_file_path):
 
         with open(rdf_file_path, "r") as f:
             rdf_content = f.read()
 
             encoded_rdf_content = base64.b64encode(bytes(rdf_content, 'utf-8'))
 
-            output_file = output_path / (pkg_name + DEFAULT_NOTICE_PACKAGE_EXTENSION)
-
             notice = PackageNotice(ted_id=notice_id)
             notice_metadata = XMLManifestationMetadataExtractor(
                 xml_manifestation=notice.xml_manifestation).to_metadata()
@@ -84,24 +102,26 @@ def generate_package(cls, notice_id, output_path, rdf_file_path, pkg_name):
             create_notice_package(
                 notice_metadata,
                 rdf_content=encoded_rdf_content,
-                save_to=output_file
+                save_to=output_path
             )
 
 
-def run(rdf_files_count, output_folder, pkgs_count):
-    cmd = CmdRunner(rdf_files_count, output_folder, pkgs_count)
+def run(rdf_files_count=None, output_folder=None, pkgs_count=None, notice_id=None,
+        mongodb_client=MongoClient(config.MONGO_DB_AUTH_URL)):
+    cmd = CmdRunner(rdf_files_count, output_folder, pkgs_count, list(notice_id or []), mongodb_client)
     cmd.run()
 
 
 @click.command()
-@click.argument('rdf-files-folder', nargs=1)
-@click.argument('output-folder', nargs=1)
+@click.argument('rdf-files-folder', nargs=1, required=False)
 @click.argument('pkgs-count', nargs=1, type=click.INT, required=False, default=DEFAULT_FILES_COUNT)
-def main(rdf_files_folder, output_folder, pkgs_count):
+@click.option('--output-folder', required=False, default=".")
+@click.option('--notice-id', required=False, multiple=True, default=None)
+def main(rdf_files_folder, pkgs_count, output_folder, notice_id):
     """
-    Generates <PKGS_COUNT> test METS packages
+    Generates test METS packages
     """
-    run(rdf_files_folder, output_folder, pkgs_count)
+    run(rdf_files_folder, output_folder, pkgs_count, notice_id)
 
 
 if __name__ == '__main__':