Implement batch preparation methods for digitized theses workflow

Why these changes are being introduced:
* The batch preparation methods for the digitized theses workflow
must be capable of performing the following functions:

1. Sync the batch folder from the digitized theses workflow S3 bucket
to a minted batch folder in the DSC S3 bucket
2. Download metadata from Alma via SRU
3. Get an item from DSpace
4. Determine if an item is a valid 'replacement thesis', given DSpace item metadata
5. Organize the contents of the minted batch folder into 'new-theses/' and
'replacement-theses/' subfolders

How this addresses that need:
* Implement methods on DigitizedThesesWorkflow for batch preparation functions
* Expand ItemSubmissionStatus to `CREATE_SUCCESS`, `CREATE_FAILED`, `CREATE_SKIPPED`
* Add exceptions to handle item retrieval from DSpace

Side effects of this change:
* Though we've expanded ItemSubmissionStatus to include `CREATE_*` statuses,
`BATCH_CREATED` remains to support backwards compatibility with other workflows.
Updating all workflows to use `CREATE_*` will be in the DSC backlog for future
work.

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/IN-1626

Author: jonavellecuerdo

dsc/config.py

@@ -83,11 +83,13 @@ def warning_only_loggers(self) -> list:
 
     # Workflow-specific env vars
     @property
-    def dspace_credentials(self) -> str:
+    def dspace_credentials(self) -> dict:
         value = os.getenv("DSPACE_CREDENTIALS")
         if not value:
             raise OSError("Env var 'DSPACE_CREDENTIALS' must be defined")
-        return value
+        credentials = json.loads(value)
+
+        return {"IR-8": credentials["ir-8"], "DDC-8": credentials["ddc-8"]}
 
     @property
     def metadata_api_url(self) -> str:

dsc/db/models.py

@@ -21,6 +21,9 @@
 
 
 class ItemSubmissionStatus(StrEnum):
+    CREATE_SUCCESS = "create_success"
+    CREATE_FAILED = "create_failed"
+    CREATE_SKIPPED = "create_skipped"
     BATCH_CREATED = "batch_created"
     SUBMIT_SUCCESS = "submit_success"
     SUBMIT_FAILED = "submit_failed"

dsc/exceptions.py

@@ -30,6 +30,43 @@ class SQSMessageSendError(Exception):
     pass
 
 
+# Exceptions for DSpace client
+class DSpaceClientError(Exception):
+    """General exception raised when DSpace client action results in error."""
+
+
+class DSpaceClientCredentialsNotFoundError(DSpaceClientError):
+    """Raise when env var DSPACE_CREDENTIALS does not include submission system."""
+
+
+class DSpaceClientAuthenticationError(DSpaceClientError):
+    """Raise when DSpace client fails authentication.
+
+    Authentication may fail due to 401 Unauthorized or 403 Forbidden errors.
+    """
+
+    def __init__(
+        self,
+        dspace_url: str | float | None,
+        dspace_user: str | float | None,
+    ):
+        self.message = (
+            f"Failed to authenticate to DSpace server at '{dspace_url}' with user "
+            f"'{dspace_user}'. Please verify that the DSPACE_CREDENTIALS "
+            "environment variable is set correctly and that the DSpace server is "
+            "accessible."
+        )
+
+
+class DSpaceClientSearchError(DSpaceClientError):
+    """Raise when DSpace client search operation results in error.
+
+    Search is performed by dspace_rest_client.client.search_objects,
+    which returns None if the response from a GET request returns an
+    exit code other than 200.
+    """
+
+
 # Exceptions for 'create-batch' step
 class BatchCreationFailedError(Exception):
     def __init__(self, errors: list[tuple]) -> None:

dsc/workflows/digitized_theses/__init__.py

@@ -1,3 +1,4 @@
-from dsc.workflows.digitized_theses.transformer import DigitizedThesesTransformer, NSMAP
+from dsc.workflows.digitized_theses.transformer import NSMAP, DigitizedThesesTransformer
+from dsc.workflows.digitized_theses.workflow import DigitizedTheses
 
-__all__ = ["DigitizedThesesTransformer", "NSMAP"]
+__all__ = ["NSMAP", "DigitizedTheses", "DigitizedThesesTransformer"]

dsc/workflows/digitized_theses/workflow.py

@@ -0,0 +1,321 @@
+# ruff: noqa: FIX002, TD002, TD003
+import logging
+from collections.abc import Iterator
+from enum import StrEnum
+from typing import Any
+
+import requests
+from dspace_rest_client.client import DSpaceClient
+from dspace_rest_client.models import Item as DSpaceItem
+from lxml import etree
+
+from dsc import exceptions
+from dsc.config import Config
+from dsc.item_submission import ItemSubmission
+from dsc.utils.aws.s3 import S3Client, run_aws_cli_sync
+from dsc.workflows.base import Workflow
+from dsc.workflows.digitized_theses import NSMAP, DigitizedThesesTransformer
+
+CONFIG = Config()
+logger = logging.getLogger(__name__)
+
+
+class MITThesesCommunityUUID(StrEnum):
+    dev = ""  # TODO: Update to uuid for 'MIT Theses' community in test
+    prod = "6fc02cc2-0d14-4023-8a6f-d9900d0c4302"
+
+
+class DigitizedTheses(Workflow):
+    """Workflow for digitized theses from the Imaging Lab."""
+
+    workflow_name: str = "digitized-theses"
+    metadata_transformer = DigitizedThesesTransformer
+
+    def __init__(self, batch_id: str):
+        self._minted_batch_id: str | None = None
+        self._dspace_client = None
+        self.community_uuid = (
+            MITThesesCommunityUUID.prod
+            if CONFIG.workspace == "prod"
+            else MITThesesCommunityUUID.dev
+        )
+        super().__init__(batch_id)
+
+    @property
+    def metadata_mapping_path(self) -> str:
+        raise NotImplementedError
+
+    @property
+    def minted_batch_id(self) -> str:
+        if not self._minted_batch_id:
+            self._minted_batch_id = (
+                f"{self.batch_id}-{self.run_date.strftime('%Y%m%dT%H%M%SZ')}"
+            )
+        return self._minted_batch_id
+
+    @property
+    def dspace_client(self) -> DSpaceClient:
+        if not self._dspace_client:
+            logger.debug(
+                f"Creating DSpace client for destination {self.submission_system}"
+            )
+
+            try:
+                credentials = CONFIG.dspace_credentials[self.submission_system]
+            except KeyError as exception:
+                raise exceptions.DSpaceClientCredentialsNotFoundError(
+                    f"No credentials for {self.submission_system}"
+                ) from exception
+
+            client = DSpaceClient(
+                api_endpoint=credentials["url"],
+                username=credentials["user"],
+                password=credentials["password"],
+                fake_user_agent=True,
+            )
+            authenticated = client.authenticate()
+            if not authenticated:
+                raise exceptions.DSpaceClientAuthenticationError(
+                    credentials["url"], credentials["user"]
+                )
+            self._dspace_client = client
+            logger.info(
+                f"Successfully authenticated to {credentials['url']} "
+                f"as {credentials['url']}"
+            )
+
+        return self._dspace_client
+
+    def get_batch_bitstream_uris(self) -> list[str]:
+        raise NotImplementedError
+
+    def item_metadata_iter(self) -> Iterator[dict[str, Any]]:
+        raise NotImplementedError
+
+    def prepare_batch(self, *, synced: bool = False) -> tuple[list, ...]:  # noqa: ARG002
+        """Prepare a batch folder in the DSC S3 bucket.
+
+        This method assumes that all the files synced from the batch folder in
+        the digitized-theses-workspace S3 bucket are intended for ingest. Each file
+        (a PDF) represents a digitized thesis and is assigned the filename
+        '<oclc-number>-MIT.pdf', which yield item identifiers formatted as
+        '<oclc-number>-MIT'.
+
+        Unlike other workflows, successes, failures, and skips are recorded in the
+        item submissions table in DynamoDB, as a means to collate all results for a batch
+        in the batch creation report. This means, the method will always return an empty
+        'errors' list.
+        """
+        item_submissions = []
+        errors: list[tuple] = []  # set but not used
+
+        # sync batch folder from DT S3 bucket to minted batch folder in DSC S3 bucket
+        run_aws_cli_sync(
+            source=f"s3://{CONFIG.s3_bucket_digitized_theses}/{self.batch_id}",
+            destination=f"s3://{CONFIG.s3_bucket_submission_assets}/{self.workflow_name}/{self.minted_batch_id}",
+        )
+
+        s3_client = S3Client()
+        for file in s3_client.files_iter(
+            bucket=self.s3_bucket, prefix=(f"{self.workflow_name}/{self.minted_batch_id}")
+        ):
+            item_submission = ItemSubmission(
+                batch_id=self.minted_batch_id,
+                item_identifier=self.parse_item_identifier(file),
+                workflow_name=self.workflow_name,
+            )
+
+            # get MARC XML metadata from Alma
+            try:
+                self._download_metadata_from_alma(item_submission)
+            except (
+                requests.exceptions.HTTPError,
+                exceptions.ItemMetadataNotFoundError,
+            ) as exception:
+                item_submission.status = "create_failed"
+                item_submission.status_details = str(exception)
+                item_submissions.append(item_submission)
+                continue
+
+            # check if item with the OCLC number exists
+            try:
+                dspace_item = self._get_item_from_dspace(item_submission)
+            except exceptions.DSpaceClientSearchError as exception:
+                item_submission.status = "create_skipped"
+                item_submission.status_details = str(exception)
+                item_submissions.append(item_submission)
+                continue
+
+            # check if item submission is a 'Replacement thesis'
+            if dspace_item and self._is_replacement_thesis(dspace_item):
+                item_submission.status = "create_success"
+                item_submission.status_details = "Replacement thesis"
+                item_submissions.append(item_submission)
+            else:
+                item_submission.status = "create_status"
+                item_submission.status_details = "New thesis"
+                item_submissions.append(item_submission)
+
+        self._move_batch_files_to_theses_subfolders(item_submissions)
+
+        return item_submissions, errors
+
+    def _download_metadata_from_alma(self, item_submission: ItemSubmission) -> bytes:
+        """Download MARC XML metadata for an item submission from Alma.
+
+        This method writes an XML file with MARC metadata to the
+        batch folder in S3 named: '<oclc-number>-MIT.xml'
+
+        For more information, see:
+        https://developers.exlibrisgroup.com/alma/integrations/sru/.
+        """
+        logger.debug(
+            f"Retrieving metadata from Alma for an item with alma.oclc_control_number_035_a={item_submission.item_identifier}"  # noqa: E501
+        )
+        s3_client = S3Client()
+
+        query_url = f"https://{CONFIG.metadata_api_url}"
+        response = requests.get(
+            query_url,
+            params={
+                "operation": "searchRetrieve",
+                "recordSchema": "marcxml",
+                "query": (
+                    f"alma.oclc_control_number_035_a={item_submission.item_identifier}"
+                ),
+            },
+            timeout=180,
+        )
+
+        try:
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as exception:
+            # TODO: Custom error includes HTTPError message
+            raise exceptions.ItemMetadataNotFoundError from exception
+
+        # get nested `<marc:record>` element from SRU response
+        record_element = self._parse_record_from_sru_response(response.content)
+
+        # write an XML file to batch folder
+        s3_client.put_file(
+            bucket=self.s3_bucket,
+            key=f"{self.workflow_name}/{self.minted_batch_id}/{item_submission.item_identifier}.xml",
+            file_content=etree.tostring(
+                record_element, xml_declaration=True, encoding="UTF-8"
+            ),
+        )
+        return response.content
+
+    def _get_item_from_dspace(self, item_submission: ItemSubmission) -> DSpaceItem:
+        logger.debug(
+            f"Searching DSpace for an item with dc.identifier.oclc={item_submission.item_identifier}"  # noqa: E501
+        )
+
+        dspace_objects = self.dspace_client.search_objects(
+            query=f"dc.identifier.oclc:{item_submission.item_identifier}",
+            scope=self.community_uuid,
+            dso_type="item",
+        )
+
+        if dspace_objects is None:
+            raise exceptions.DSpaceClientSearchError(
+                f"Failed search for item with dc.identifier.oclc={item_submission.item_identifier} "  # noqa: E501
+                f"in community {"<community>"}"
+            )
+
+        if len(dspace_objects) > 1:
+            raise exceptions.DSpaceClientSearchError(
+                f"Expecting one item with dc.identifier.oclc={item_submission.item_identifier} "  # noqa: E501
+                f"in community {"<community>"}; found {len(dspace_objects)} items"
+            )
+
+        if len(dspace_objects) == 0:
+            return None
+
+        return dspace_objects[0]
+
+    def _move_batch_files_to_theses_subfolders(
+        self, item_submissions: list[ItemSubmission]
+    ) -> None:
+        s3_client = S3Client()
+
+        for item_submission in item_submissions:
+            if (
+                item_submission.status_details
+                and "replacement" in item_submission.status_details.lower()
+            ):
+                prefix = "replacement-theses"
+            elif (
+                item_submission.status_details
+                and "new" in item_submission.status_details.lower()
+            ):
+                prefix = "new-theses"
+            else:
+                logger.warning(
+                    f"Cannot move files associated with item_identifier={item_submission.item_identifier} "  # noqa: E501
+                    "because thesis type (new vs. replacement) is unknown"
+                )
+                continue
+
+            logger.debug(
+                "Moving files associated with "
+                f"item_identifier={item_submission.item_identifier} to '{prefix}/'"
+            )
+            for file in s3_client.files_iter(
+                bucket=self.s3_bucket,
+                prefix=f"{self.workflow_name}/{self.minted_batch_id}",
+                item_identifier=item_submission.item_identifier,
+            ):
+                filename = file.rsplit("/", maxsplit=1)[-1]
+                s3_client.move_file(
+                    source_file=file,
+                    destination_file=f"s3://{self.s3_bucket}/{self.workflow_name}/{self.minted_batch_id}/{prefix}/{filename}",
+                )
+
+    @staticmethod
+    def _is_replacement_thesis(item: DSpaceItem) -> bool:
+        """Determine if a DSpace item is a replacement thesis.
+
+        If an item already exists in DSpace, the thesis (bitstream) should
+        only be replaced *if* it is not a student-submitted electronic PDF
+        file, which is checked via the presence of a target string in the
+        dc.description field.
+
+        This method returns True when the target string does *not* appear in
+        any dc.description value (if any), indicating that
+        the replacement is valid.
+        """
+        # skip the record if any dc_description value matches this string
+        target_string = "This electronic version was submitted by the student author."
+        if item.metadata.get("dc.description"):
+            dc_description_values = [
+                entry.get("value", "") for entry in item.metadata["dc.description"]
+            ]
+
+            if target_string in "|".join(dc_description_values):
+                return False
+
+        return True
+
+    @staticmethod
+    def _parse_record_from_sru_response(content: bytes) -> etree._Element:
+        root = etree.fromstring(content)
+        number_of_records = root.find("sru:numberOfRecords", namespaces=NSMAP)
+
+        if number_of_records is None:
+            # TODO: Custom error includes message "Unexpected response from Alma SRU"
+            raise exceptions.ItemMetadataNotFoundError
+
+        if int(number_of_records.text) == 0:
+            raise exceptions.ItemMetadataNotFoundError
+
+        if int(number_of_records.text) > 1:
+            # TODO: Custom error includes message
+            # "Unexpected response from Alma SRU, multiple records with OCLC"
+            raise exceptions.ItemMetadataNotFoundError
+
+        return root.xpath("//marc:record", namespaces=NSMAP)[0]
+
+    @staticmethod
+    def parse_item_identifier(filename: str) -> str:
+        return filename.rsplit("/", maxsplit=1)[-1].removesuffix("-MIT.pdf")