feat(ingest-upload): create new ingest xmlupload cli command (DEV-3019)

src/dsp_tools/cli/call_action.py

@@ -1,4 +1,5 @@
 import argparse
+from pathlib import Path
 
 from dsp_tools.commands.excel2json.lists import excel2lists, validate_lists_section_with_schema
 from dsp_tools.commands.excel2json.project import excel2json
@@ -9,6 +10,7 @@
 from dsp_tools.commands.fast_xmlupload.upload_files import upload_files
 from dsp_tools.commands.fast_xmlupload.upload_xml import fast_xmlupload
 from dsp_tools.commands.id2iri import id2iri
+from dsp_tools.commands.ingest_xmlupload.upload_xml import ingest_xmlupload
 from dsp_tools.commands.project.create.project_create import create_project
 from dsp_tools.commands.project.create.project_create_lists import create_lists
 from dsp_tools.commands.project.create.project_validate import validate_project
@@ -68,6 +70,8 @@ def call_requested_action(args: argparse.Namespace) -> bool:  # noqa: PLR0911 (T
             return _call_upload_files(args)
         case "fast-xmlupload":
             return _call_fast_xmlupload(args)
+        case "ingest-xmlupload":
+            return _call_ingest_xmlupload(args)
         case "template":
             return generate_template_repo()
         case "rosetta":
@@ -144,6 +148,17 @@ def _call_excel2json(args: argparse.Namespace) -> bool:
     )
 
 
+def _call_ingest_xmlupload(args: argparse.Namespace) -> bool:
+    ingest_xmlupload(
+        xml_file=Path(args.xml_file),
+        user=args.user,
+        password=args.password,
+        dsp_url=args.server,
+        sipi_url=args.sipi_url,
+    )
+    return True
+
+
 def _call_fast_xmlupload(args: argparse.Namespace) -> bool:
     return fast_xmlupload(
         xml_file=args.xml_file,

src/dsp_tools/cli/create_parsers.py

@@ -45,6 +45,8 @@ def make_parser(
 
     _add_xmlupload(subparsers, default_dsp_api_url, root_user_email, root_user_pw)
 
+    _add_ingest_xmlupload(subparsers, default_dsp_api_url, root_user_email, root_user_pw)
+
     _add_fast_xmlupload(subparsers, default_dsp_api_url, root_user_email, root_user_pw)
 
     _add_process_files(subparsers)
@@ -178,6 +180,23 @@ def _add_excel2json(subparsers: _SubParsersAction[ArgumentParser]) -> None:
     subparser.add_argument("project_definition", help="path to the output JSON file")
 
 
+def _add_ingest_xmlupload(
+    subparsers: _SubParsersAction[ArgumentParser],
+    default_dsp_api_url: str,
+    root_user_email: str,
+    root_user_pw: str,
+) -> None:
+    subparser = subparsers.add_parser(
+        name="ingest-xmlupload",
+        help="For internal use only: create resources with files already uploaded through dsp-ingest",
+    )
+    subparser.set_defaults(action="ingest-xmlupload")
+    subparser.add_argument("-s", "--server", default=default_dsp_api_url, help=dsp_server_text)
+    subparser.add_argument("-u", "--user", default=root_user_email, help=username_text)
+    subparser.add_argument("-p", "--password", default=root_user_pw, help=password_text)
+    subparser.add_argument("xml_file", help="path to XML file containing the data")
+
+
 def _add_fast_xmlupload(
     subparsers: _SubParsersAction[ArgumentParser],
     default_dsp_api_url: str,

src/dsp_tools/commands/fast_xmlupload/upload_xml.py

@@ -118,7 +118,7 @@ def fast_xmlupload(
         password=password,
         imgdir=".",
         sipi=sipi_url,
-        config=UploadConfig(preprocessing_done=True),
+        config=UploadConfig(media_previously_uploaded=True),
     )
 
     end_time = datetime.now()

src/dsp_tools/commands/ingest_xmlupload/apply_ingest_id.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from pathlib import Path
+from typing import cast
+
+import pandas as pd
+from lxml import etree
+
+from dsp_tools.commands.ingest_xmlupload.user_information import IngestInformation
+from dsp_tools.models.exceptions import InputError
+from dsp_tools.utils.create_logger import get_logger
+
+logger = get_logger(__name__)
+
+
+def get_mapping_dict_from_file(shortcode: str) -> dict[str, str]:
+    """
+    This functions returns the information to replace the original filepaths with the identifier from dsp-ingest.
+
+    Args:
+        shortcode: Shortcode of the project
+
+    Returns:
+        dictionary with original: identifier from dsp-ingest
+
+    Raises:
+        InputError: if no file was found
+    """
+    filepath = Path(f"mapping-{shortcode}.csv")
+    if not filepath.is_file():
+        raise InputError(f"No mapping CSV file was found at {filepath}.")
+    df = pd.read_csv(filepath)
+    msg = f"The file '{filepath}' is used to map the internal SIPI image IDs to the original filepaths."
+    print(msg)
+    logger.info(msg)
+    return dict(zip(df["original"].tolist(), df["derivative"].tolist()))
+
+
+def replace_filepath_with_sipi_id(
+    xml_tree: etree._ElementTree[etree._Element],
+    orig_path_2_id_filename: dict[str, str],
+) -> tuple[etree._ElementTree[etree._Element], IngestInformation]:
+    """
+    Replace the original filepaths in the <bitstream> tags by the id filenames of the uploaded files.
+
+    Args:
+        xml_tree: The parsed original XML tree
+        orig_path_2_id_filename: Mapping from original filenames to id filenames from the mapping.csv
+
+    Returns:
+        A copy of the XMl tree, with the replaced filepaths.
+        Message informing if all referenced files were uploaded or not.
+    """
+    no_id_found = []
+    used_media_file_paths = []
+    new_tree = deepcopy(xml_tree)
+    for elem in new_tree.iter():
+        if etree.QName(elem).localname.endswith("bitstream"):
+            if (img_path := elem.text) in orig_path_2_id_filename:
+                elem.text = orig_path_2_id_filename[img_path]
+                used_media_file_paths.append(img_path)
+            else:
+                no_id_found.append((cast("etree._Element", elem.getparent()).attrib["id"], str(elem.text)))
+    unused_media_paths = [x for x in orig_path_2_id_filename if x not in used_media_file_paths]
+    return new_tree, IngestInformation(unused_mediafiles=unused_media_paths, mediafiles_no_id=no_id_found)

src/dsp_tools/commands/ingest_xmlupload/upload_xml.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from lxml import etree
+
+from dsp_tools.commands.ingest_xmlupload.apply_ingest_id import (
+    get_mapping_dict_from_file,
+    replace_filepath_with_sipi_id,
+)
+from dsp_tools.commands.xmlupload.upload_config import UploadConfig
+from dsp_tools.commands.xmlupload.xmlupload import xmlupload
+from dsp_tools.models.exceptions import InputError
+from dsp_tools.utils.create_logger import get_logger
+from dsp_tools.utils.xml_utils import remove_comments_from_element_tree
+
+logger = get_logger(__name__)
+
+
+def ingest_xmlupload(
+    xml_file: Path,
+    user: str,
+    password: str,
+    dsp_url: str,
+    sipi_url: str,
+) -> None:
+    """
+    This function reads an XML file
+    and imports the data described in it onto the DSP server,
+    using the ingest XML upload method.
+    Before using this function,
+    the multimedia files must be ingested on the DSP server.
+    A mapping file with the internal IDs of the multimedia files must also be provided.
+
+    Args:
+        xml_file: path to XML file containing the resources
+        user: the user's e-mail for login into DSP
+        password: the user's password for login into DSP
+        dsp_url: URL to the DSP server
+        sipi_url: URL to the Sipi server
+
+    Raises:
+        InputError: if any media was not uploaded or uploaded media was not referenced.
+    """
+    xml_tree_orig = etree.parse(xml_file)
+    xml_tree_orig = remove_comments_from_element_tree(xml_tree_orig)
+
+    shortcode = xml_tree_orig.getroot().attrib["shortcode"]
+    orig_path_2_id_filename = get_mapping_dict_from_file(shortcode)
+    xml_tree_replaced, ingest_info = replace_filepath_with_sipi_id(
+        xml_tree=xml_tree_orig,
+        orig_path_2_id_filename=orig_path_2_id_filename,
+    )
+    if ok := ingest_info.ok_msg():
+        print(ok)
+        logger.info(ok)
+    else:
+        err_msg = ingest_info.execute_error_protocol()
+        logger.error(err_msg)
+        raise InputError(err_msg)
+
+    xmlupload(
+        input_file=xml_tree_replaced,
+        server=dsp_url,
+        user=user,
+        password=password,
+        imgdir=".",
+        sipi=sipi_url,
+        config=UploadConfig(media_previously_uploaded=True),
+    )

src/dsp_tools/commands/ingest_xmlupload/user_information.py

@@ -0,0 +1,120 @@
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import pandas as pd
+
+separator = "\n    "
+list_separator = "\n    - "
+
+
+@dataclass(frozen=True)
+class IngestInformation:
+    """
+    This class stores the information about the mapping of ids provided by the dsp-ingest service
+    and the filepaths used in the XML file.
+    """
+
+    unused_mediafiles: list[str]
+    mediafiles_no_id: list[tuple[str, str]]
+    maximum_prints: int = 20
+    csv_directory_path: Path = field(default=Path.cwd())
+    unused_mediafiles_csv: str = "UnusedMediaUploadedInSipi.csv"
+    mediafiles_no_id_csv: str = "FilesNotUploadedToSipi.csv"
+
+    def ok_msg(self) -> str | None:
+        """
+        This function checks if no media was unused or not uploaded.
+        If that is the case it returns the message,
+        if not, it ends without an effect.
+
+        Returns:
+            Message if all went well.
+        """
+        if not self.unused_mediafiles and not self.mediafiles_no_id:
+            return (
+                "All multimedia files referenced in the XML file were uploaded through dsp-ingest.\n"
+                "No multimedia files were uploaded through dsp-ingest that were not referenced in the XML file."
+            )
+        return None
+
+    def execute_error_protocol(self) -> str:
+        """
+        This function generates the user message and saves a file with the information
+        if a lot of resources are affected.
+
+        Returns:
+            User message
+        """
+        self._save_csv_if_applicable()
+        return self._get_error_msg()
+
+    def _get_error_msg(self) -> str:
+        msg_list = [
+            "The upload cannot continue as there are problems with the multimedia files referenced in the XML.",
+        ]
+        if has_msg := self._get_unused_mediafiles_msg():
+            msg_list.append(has_msg)
+        if has_msg := self._get_mediafiles_no_id_msg():
+            msg_list.append(has_msg)
+        return separator.join(msg_list)
+
+    def _get_mediafiles_no_id_msg(self) -> str | None:
+        if 0 < len(self.mediafiles_no_id) <= self.maximum_prints:
+            return (
+                "The data XML file contains references to the following multimedia files "
+                "which were not previously uploaded through dsp-ingest:"
+                + list_separator
+                + list_separator.join([f"Resource ID: '{x[0]}' | Filepath: '{x[1]}'" for x in self.mediafiles_no_id])
+            )
+        elif len(self.mediafiles_no_id) > self.maximum_prints:
+            return (
+                "The data XML file contains references to multimedia files "
+                "which were not previously uploaded through dsp-ingest:\n"
+                f"    The file with the resource IDs and problematic filenames was saved at "
+                f"'{Path(self.csv_directory_path/self.mediafiles_no_id_csv)}'."
+            )
+        return None
+
+    def _get_unused_mediafiles_msg(self) -> str | None:
+        if 0 < len(self.unused_mediafiles) <= self.maximum_prints:
+            return (
+                "The data XML file does not reference the following multimedia files which were previously "
+                "uploaded through dsp-ingest:" + list_separator + list_separator.join(self.unused_mediafiles)
+            )
+        elif len(self.unused_mediafiles) > self.maximum_prints:
+            return (
+                "The data XML file does not reference all the multimedia files which were previously "
+                "uploaded through dsp-ingest.\n"
+                f"    The file with the unused filenames was saved at "
+                f"'{Path(self.csv_directory_path/self.unused_mediafiles_csv)}'."
+            )
+        return None
+
+    def _save_csv_if_applicable(self) -> None:
+        if unused_mediafiles_df := self._unused_mediafiles_to_df():
+            _save_as_csv(unused_mediafiles_df, self.csv_directory_path, self.unused_mediafiles_csv)
+        if no_id_df := self._mediafiles_no_id_to_df():
+            _save_as_csv(no_id_df, self.csv_directory_path, self.mediafiles_no_id_csv)
+
+    def _unused_mediafiles_to_df(self) -> pd.DataFrame | None:
+        return (
+            pd.DataFrame({"Multimedia Filenames": self.unused_mediafiles})
+            if len(self.unused_mediafiles) > self.maximum_prints
+            else None
+        )
+
+    def _mediafiles_no_id_to_df(self) -> pd.DataFrame | None:
+        return (
+            pd.DataFrame(
+                {
+                    "Resource ID": [x[0] for x in self.mediafiles_no_id],
+                    "Filepath": [x[1] for x in self.mediafiles_no_id],
+                }
+            )
+            if len(self.mediafiles_no_id) > self.maximum_prints
+            else None
+        )
+
+
+def _save_as_csv(df: pd.DataFrame, directory_path: Path, filename: str) -> None:
+    df.to_csv(Path(directory_path, filename), index=False)

src/dsp_tools/commands/xmlupload/resource_create_client.py

@@ -156,7 +156,8 @@ def _make_bitstream_file_value(bitstream_info: BitstreamInfo) -> dict[str, Any]:
         case "mp4":
             prop = "knora-api:hasMovingImageFileValue"
             value_type = "MovingImageFileValue"
-        case "jpg" | "jpeg" | "jp2" | "png" | "tif" | "tiff":
+        # jpx is the extension of the files returned by dsp-ingest
+        case "jpg" | "jpeg" | "jp2" | "png" | "tif" | "tiff" | "jpx":
             prop = "knora-api:hasStillImageFileValue"
             value_type = "StillImageFileValue"
         case "odd" | "rng" | "txt" | "xml" | "xsd" | "xsl" | "xslt" | "csv":

src/dsp_tools/commands/xmlupload/upload_config.py

@@ -49,7 +49,7 @@ class DiagnosticsConfig:
 class UploadConfig:
     """Configuration for the upload process."""
 
-    preprocessing_done: bool = False
+    media_previously_uploaded: bool = False
     server: str = "unknown"
     shortcode: str = "unknown"
     diagnostics: DiagnosticsConfig = field(default_factory=DiagnosticsConfig)

src/dsp_tools/commands/xmlupload/xmlupload.py

@@ -74,7 +74,7 @@ def xmlupload(
     default_ontology, root, shortcode = validate_and_parse_xml_file(
         input_file=input_file,
         imgdir=imgdir,
-        preprocessing_done=config.preprocessing_done,
+        preprocessing_done=config.media_previously_uploaded,
     )
 
     config = config.with_server_info(
@@ -85,7 +85,12 @@ def xmlupload(
 
     # establish connection to DSP server
     con = login(server=server, user=user, password=password, dump=config.diagnostics.dump)
-    sipi_con = ConnectionLive(sipi, dump=config.diagnostics.dump, token=con.get_token())
+    if config.media_previously_uploaded:
+        sipi_con = ConnectionLive(
+            sipi, dump=config.diagnostics.dump, token=con.get_token(), headers={"X-Asset-Ingested": "true"}
+        )
+    else:
+        sipi_con = ConnectionLive(sipi, dump=config.diagnostics.dump, token=con.get_token())
     sipi_server = Sipi(sipi_con)
 
     ontology_client = OntologyClientLive(
@@ -338,7 +343,7 @@ def _upload_resources(
 
     for i, resource in enumerate(resources):
         success, media_info = handle_media_info(
-            resource, config.preprocessing_done, sipi_server, imgdir, permissions_lookup
+            resource, config.media_previously_uploaded, sipi_server, imgdir, permissions_lookup
         )
         if not success:
             failed_uploads.append(resource.res_id)