Source code for pyPreservica.uploadAPI

"""
pyPreservica UploadAPI module definition

A client library for the Preservica Repository Upload API

author:     James Carr
licence:    Apache License 2.0

"""

import csv
import shutil
import tempfile
import uuid
import xml
from datetime import datetime, timedelta, timezone
from time import sleep
from typing import Callable
from xml.dom import minidom
from xml.etree import ElementTree
from xml.etree.ElementTree import Element, SubElement

import boto3
import s3transfer.tasks
import s3transfer.upload
from botocore.session import get_session
from boto3.s3.transfer import TransferConfig, S3Transfer
from botocore.config import Config
from botocore.credentials import RefreshableCredentials
from botocore.exceptions import ClientError, NoCredentialsError, PartialCredentialsError
from dateutil.tz import tzlocal
from pyrate_limiter import Rate, Duration, Limiter, InMemoryBucket
from s3transfer import S3UploadFailedError
from tqdm import tqdm

from pyPreservica.common import *
from pyPreservica.common import _make_stored_zipfile

logger = logging.getLogger(__name__)

MB = 1024 * 1024
GB = 1024 ** 3
transfer_config = TransferConfig(multipart_threshold=int(32 * MB))

CONTENT_FOLDER = "content"
PRESERVATION_CONTENT_FOLDER = "p1"
ACCESS_CONTENT_FOLDER = "a1"


def upload_file(self, filename, bucket, key, callback=None, extra_args=None):
    """Upload a file to an S3 object.

    Variants have also been injected into S3 client, Bucket and Object.
    You don't have to use S3Transfer.upload_file() directly.

    .. seealso::
        :py:meth:`S3.Client.upload_file`
        :py:meth:`S3.Client.upload_fileobj`
    """
    if not isinstance(filename, str):
        raise ValueError('Filename must be a string')

    subscribers = self._get_subscribers(callback)
    future = self._manager.upload(filename, bucket, key, extra_args, subscribers)
    try:
        return future.result()
    # If a client error was raised, add the backwards compatibility layer
    # that raises a S3UploadFailedError. These specific errors were only
    # ever thrown for upload_parts but now can be thrown for any related
    # client error.
    except ClientError as e:
        raise S3UploadFailedError("Failed to upload %s to %s: %s" % (filename, '/'.join([bucket, key]), e))


class PutObjectTask(s3transfer.tasks.Task):
    # Copied from s3transfer/upload.py, changed to return the result of client.put_object.
    def _main(self, client, fileobj, bucket, key, extra_args):
        with fileobj as body:
            response = client.put_object(Bucket=bucket, Key=key, Body=body, **extra_args)
            return response


class CompleteMultipartUploadTask(s3transfer.tasks.Task):
    # Copied from s3transfer/tasks.py, changed to return a result.
    def _main(self, client, bucket, key, upload_id, parts, extra_args):
        return client.complete_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id,
                                                MultipartUpload={"Parts": parts},
                                                **extra_args, )


s3transfer.upload.PutObjectTask = PutObjectTask
s3transfer.upload.CompleteMultipartUploadTask = CompleteMultipartUploadTask


def prettify(elem):
    """Return a pretty-printed XML string for the Element.
    """
    rough_string = xml.etree.ElementTree.tostring(elem, 'utf-8')
    re_parsed = minidom.parseString(rough_string)
    return re_parsed.toprettyxml(indent="  ")


def __create_io__(xip=None, file_name=None, parent_folder=None, **kwargs):
    if xip is None:
        xip = Element('xip:XIP')
        xip.set('xmlns:xip', 'http://preservica.com/XIP/v6.0')
    assert xip is not None
    io = SubElement(xip, 'xip:InformationObject')
    ref = SubElement(io, 'xip:Ref')

    if 'IO_Identifier_callback' in kwargs:
        ident_callback = kwargs.get('IO_Identifier_callback')
        ref.text = ident_callback()
    else:
        ref.text = str(uuid.uuid4())

    title = SubElement(io, 'xip:Title')
    title.text = kwargs.get('Title', file_name)
    description = SubElement(io, 'xip:Description')
    description.text = kwargs.get('Description', file_name)
    security = SubElement(io, 'xip:SecurityTag')
    security.text = kwargs.get('SecurityTag', "open")
    custom_type = SubElement(io, 'xip:CustomType')
    custom_type.text = kwargs.get('CustomType', "")
    parent = SubElement(io, 'xip:Parent')

    if hasattr(parent_folder, "reference"):
        parent.text = parent_folder.reference
    elif isinstance(parent_folder, str):
        parent.text = parent_folder

    return xip, ref.text


def __make_representation__(xip, rep_name, rep_type, io_ref):
    representation = SubElement(xip, 'xip:Representation')
    io_link = SubElement(representation, 'xip:InformationObject')
    io_link.text = io_ref
    access_name = SubElement(representation, 'xip:Name')
    access_name.text = rep_name
    access_type = SubElement(representation, 'xip:Type')
    access_type.text = rep_type
    content_objects = SubElement(representation, 'xip:ContentObjects')
    content_object = SubElement(content_objects, 'xip:ContentObject')
    content_object_ref = str(uuid.uuid4())
    content_object.text = content_object_ref
    return content_object_ref


def __make_content_objects__(xip, content_title, co_ref, io_ref, tag, content_description, content_type):
    content_object = SubElement(xip, 'xip:ContentObject')
    ref_element = SubElement(content_object, "xip:Ref")
    ref_element.text = co_ref
    title = SubElement(content_object, "xip:Title")
    title.text = content_title
    description = SubElement(content_object, "xip:Description")
    description.text = content_description
    security_tag = SubElement(content_object, "xip:SecurityTag")
    security_tag.text = tag
    custom_type = SubElement(content_object, "xip:CustomType")
    custom_type.text = content_type
    parent = SubElement(content_object, "xip:Parent")
    parent.text = io_ref


def __make_generation__(xip, filename, co_ref, generation_label, location=None):
    generation = SubElement(xip, 'xip:Generation', {"original": "true", "active": "true"})
    content_object = SubElement(generation, "xip:ContentObject")
    content_object.text = co_ref
    label = SubElement(generation, "xip:Label")
    if generation_label:
        label.text = generation_label
    else:
        label.text = os.path.splitext(filename)[0]
    effective_date = SubElement(generation, "xip:EffectiveDate")
    effective_date.text = datetime.now().isoformat()
    bitstreams = SubElement(generation, "xip:Bitstreams")
    bitstream = SubElement(bitstreams, "xip:Bitstream")
    bitstream.text = f"{location}/{filename}"
    SubElement(generation, "xip:Formats")
    SubElement(generation, "xip:Properties")


def __make_bitstream__(xip, file_name, full_path, callback, location=None):
    bitstream = SubElement(xip, 'xip:Bitstream')
    filename_element = SubElement(bitstream, "xip:Filename")
    filename_element.text = file_name
    filesize = SubElement(bitstream, "xip:FileSize")
    file_stats = os.stat(full_path)
    filesize.text = str(file_stats.st_size)
    physical_location = SubElement(bitstream, "xip:PhysicalLocation")
    physical_location.text = location
    fixities = SubElement(bitstream, "xip:Fixities")
    fixity_result = callback(file_name, full_path)
    if type(fixity_result) == tuple:
        fixity = SubElement(fixities, "xip:Fixity")
        fixity_algorithm_ref = SubElement(fixity, "xip:FixityAlgorithmRef")
        fixity_value = SubElement(fixity, "xip:FixityValue")
        fixity_algorithm_ref.text = fixity_result[0]
        fixity_value.text = fixity_result[1]
    elif type(fixity_result) == dict:
        for key, val in fixity_result.items():
            fixity = SubElement(fixities, "xip:Fixity")
            fixity_algorithm_ref = SubElement(fixity, "xip:FixityAlgorithmRef")
            fixity_value = SubElement(fixity, "xip:FixityValue")
            fixity_algorithm_ref.text = key
            fixity_value.text = val
    else:
        logger.error("Could Not Find Fixity Value")
        raise RuntimeError("Could Not Find Fixity Value")


def __make_representation_multiple_co__(xip, rep_name, rep_type, rep_files, io_ref):
    representation = SubElement(xip, 'xip:Representation')
    io_link = SubElement(representation, 'xip:InformationObject')
    io_link.text = io_ref
    access_name = SubElement(representation, 'xip:Name')
    access_name.text = rep_name
    access_type = SubElement(representation, 'xip:Type')
    access_type.text = rep_type
    content_objects = SubElement(representation, 'xip:ContentObjects')
    refs_dict = {}
    for f in rep_files:
        content_object = SubElement(content_objects, 'xip:ContentObject')
        content_object_ref = str(uuid.uuid4())
        content_object.text = content_object_ref
        refs_dict[content_object_ref] = f
    return refs_dict


def cvs_to_cmis_xslt(csv_file, xml_namespace, root_element, title="Metadata Title", export_folder=None,
                     additional_namespaces=None):
    """

    Deprecated function, use csv_to_cmis_xslt instead.

    """
    return csv_to_cmis_xslt(csv_file, xml_namespace, root_element, title, export_folder, additional_namespaces)

[docs] def csv_to_cmis_xslt(csv_file, xml_namespace, root_element, title="Metadata Title", export_folder=None, additional_namespaces=None): """ Create a custom CMIS XSLT transform to display metadata within Universal Access (UA). Reads the column headers from the first row of ``csv_file`` and generates an XSL stylesheet that maps the XML elements produced by :func:`csv_to_xml` to the Tessella CMIS metadata display format used by Preservica Universal Access. :param csv_file: Path to the CSV file whose headers define the metadata fields. :type csv_file: str :param xml_namespace: The XML namespace URI used in the metadata documents. :type xml_namespace: str :param root_element: The root element name used in the metadata documents. :type root_element: str :param title: Human-readable title shown in UA for this metadata group. Defaults to ``"Metadata Title"``. :type title: str :param export_folder: Directory where the generated ``.xslt`` file is written. If ``None``, the file is written to the current working directory. :type export_folder: str or None :param additional_namespaces: Mapping of namespace prefix to URI for any extra namespaces referenced in the CSV headers (e.g. ``{"dc": "http://..."}``. :type additional_namespaces: dict or None :returns: Path to the generated XSLT file. :rtype: str """ headers = set() with open(csv_file, encoding='utf-8-sig', newline='') as csvfile: reader = csv.reader(csvfile) for row in reader: for header in row: xml_tag = header.strip() xml_tag = xml_tag.replace(" ", "") xml_tag = xml_tag.replace("-", "") headers.add(xml_tag) break namespaces = {"version": "2.0", "xmlns:xsl": "http://www.w3.org/1999/XSL/Transform", "xmlns:fn": "http://www.w3.org/2005/xpath-functions", "xmlns:xs": "http://www.w3.org/2001/XMLSchema", "xmlns:csv": xml_namespace, "xmlns": "http://www.tessella.com/sdb/cmis/metadata", "exclude-result-prefixes": "csv"} if additional_namespaces is not None: for prefix, uri in additional_namespaces.items(): namespaces["xmlns:" + prefix] = uri xml_stylesheet = xml.etree.ElementTree.Element("xsl:stylesheet", namespaces) xml.etree.ElementTree.SubElement(xml_stylesheet, "xsl:output", {"method": "xml", "indent": "yes"}) xml_template = xml.etree.ElementTree.SubElement(xml_stylesheet, "xsl:template", {"match": "csv:" + root_element}) xml_group = xml.etree.ElementTree.SubElement(xml_template, "group") xml_title = xml.etree.ElementTree.SubElement(xml_group, "title") xml_title.text = title xml_templates = xml.etree.ElementTree.SubElement(xml_group, "xsl:apply-templates") elements = "" for header in headers: if ":" in header: elements = elements + "|" + header else: elements = elements + "|csv:" + header elements = elements[1:] xml_matches = xml.etree.ElementTree.SubElement(xml_stylesheet, "xsl:template", {"match": elements}) xml_item = xml.etree.ElementTree.SubElement(xml_matches, "item") xml_name = xml.etree.ElementTree.SubElement(xml_item, "name") xml_name_value = xml.etree.ElementTree.SubElement(xml_name, "xsl:value-of", { "select": "fn:replace(translate(local-name(), '_', ' '), '([a-z])([A-Z])', '$1 $2')"}) xml_value = xml.etree.ElementTree.SubElement(xml_item, "value") xml_value_value = xml.etree.ElementTree.SubElement(xml_value, "xsl:value-of", {"select": "."}) xml_type = xml.etree.ElementTree.SubElement(xml_item, "type") xml_type_value = xml.etree.ElementTree.SubElement(xml_type, "xsl:value-of", { "select": "fn:replace(translate(local-name(), '_', ' '), '([a-z])([A-Z])', '$1 $2')"}) xml_request = xml.etree.ElementTree.tostring(xml_stylesheet, encoding='utf-8', xml_declaration=True) cmis_xslt = root_element + "-cmis.xslt" if export_folder is not None: cmis_xslt = os.path.join(export_folder, cmis_xslt) file = open(cmis_xslt, "wt", encoding="utf-8") file.write(xml_request.decode("utf-8")) file.close() return cmis_xslt
def cvs_to_xsd(csv_file, xml_namespace, root_element, export_folder=None, additional_namespaces=None): """ Deprecated function, use csv_to_xsd instead. """ return csv_to_xsd(csv_file, xml_namespace, root_element, export_folder, additional_namespaces)
[docs] def csv_to_xsd(csv_file, xml_namespace, root_element, export_folder=None, additional_namespaces=None): """ Create an XSD schema definition derived from the column headers of a CSV file. Reads the column headers from the first row of ``csv_file`` and generates an XML Schema Definition (``.xsd``) file that validates the XML documents produced by :func:`csv_to_xml`. :param csv_file: Path to the CSV file whose headers define the element names. :type csv_file: str :param xml_namespace: The target namespace URI for the generated schema. :type xml_namespace: str :param root_element: The root element name that the schema will validate. :type root_element: str :param export_folder: Directory where the generated ``.xsd`` file is written. If ``None``, the file is written to the current working directory. :type export_folder: str or None :param additional_namespaces: Mapping of namespace prefix to URI for any extra namespaces referenced in the CSV headers. :type additional_namespaces: dict or None :returns: Path to the generated XSD file. :rtype: str """ headers = set() with open(csv_file, encoding='utf-8-sig', newline='') as csvfile: reader = csv.reader(csvfile) for row in reader: for header in row: xml_tag = header.strip() xml_tag = xml_tag.replace(" ", "") xml_tag = xml_tag.replace("-", "") headers.add(xml_tag) break namespaces = {"xmlns:xs": "http://www.w3.org/2001/XMLSchema", "attributeFormDefault": "unqualified", "elementFormDefault": "qualified", "targetNamespace": xml_namespace} if additional_namespaces is not None: for prefix, uri in additional_namespaces.items(): namespaces["xmlns:" + prefix.strip()] = uri.strip() xml_schema = xml.etree.ElementTree.Element("xs:schema", namespaces) if additional_namespaces is not None: for prefix, namespace in additional_namespaces.items(): xml_import = xml.etree.ElementTree.SubElement(xml_schema, "xs:import", {"namespace": namespace}) xml_element = xml.etree.ElementTree.SubElement(xml_schema, "xs:element", {"name": root_element}) xml_complex_type = xml.etree.ElementTree.SubElement(xml_element, "xs:complexType") xml_sequence = xml.etree.ElementTree.SubElement(xml_complex_type, "xs:sequence") for header in headers: if ":" in header: prefix, sep, tag = header.partition(":") try: namespace = additional_namespaces[prefix] xml.etree.ElementTree.SubElement(xml_sequence, "xs:element", {"ref": header, "xmlns:" + prefix: namespace}) except KeyError: xml.etree.ElementTree.SubElement(xml_sequence, "xs:element", {"type": "xs:string", "name": header}) else: xml.etree.ElementTree.SubElement(xml_sequence, "xs:element", {"type": "xs:string", "name": header}) xml_request = xml.etree.ElementTree.tostring(xml_schema, encoding='utf-8', xml_declaration=True) xsd_file = root_element + ".xsd" if export_folder is not None: xsd_file = os.path.join(export_folder, xsd_file) file = open(xsd_file, "wt", encoding="utf-8") file.write(xml_request.decode("utf-8")) file.close() return xsd_file
def csv_to_search_xml(csv_file, xml_namespace, root_element, title="Metadata Title", export_folder=None, additional_namespaces=None): """ Create a custom Preservica search index configuration derived from a CSV file. Reads the column headers from the first row of ``csv_file`` and generates a Preservica custom search index XML file (``*-index.xml``) that registers each column as a searchable ``STRING_DEFAULT`` field. :param csv_file: Path to the CSV file whose headers define the indexed fields. :type csv_file: str :param xml_namespace: The XML namespace URI of the metadata documents to be indexed. :type xml_namespace: str :param root_element: The root element name used in the metadata documents. :type root_element: str :param title: Human-readable name for the search schema group. Defaults to ``"Metadata Title"``. :type title: str :param export_folder: Directory where the generated index XML file is written. If ``None``, the file is written to the current working directory. :type export_folder: str or None :param additional_namespaces: Mapping of namespace prefix to URI for any extra namespaces referenced in the CSV headers. :type additional_namespaces: dict or None :returns: Path to the generated search index XML file. :rtype: str """ headers = set() with open(csv_file, encoding='utf-8-sig', newline='') as csvfile: reader = csv.reader(csvfile) for row in reader: for header in row: xml_tag = header.strip() xml_tag = xml_tag.replace(" ", "") xml_tag = xml_tag.replace("-", "") headers.add(xml_tag) break xml_index = xml.etree.ElementTree.Element("index", {"xmlns": "http://www.preservica.com/customindex/v1"}) short_name = "csv" xml_schema_name = xml.etree.ElementTree.SubElement(xml_index, "schemaName") xml_schema_name.text = title xml_schema_uri = xml.etree.ElementTree.SubElement(xml_index, "schemaUri") xml_schema_uri.text = xml_namespace xml_short_name = xml.etree.ElementTree.SubElement(xml_index, "shortName") xml_short_name.text = short_name for header in headers: if ":" in header: xpath_expression = f"//{short_name}:{root_element}/{header}" else: xpath_expression = f"//{short_name}:{root_element}/{short_name}:{header}" attr = {"indexName": header, "displayName": header, "xpath": xpath_expression, "indexType": "STRING_DEFAULT"} xml_term = xml.etree.ElementTree.SubElement(xml_index, "term", attr) if additional_namespaces is not None: for prefix, uri in additional_namespaces.items(): xml.etree.ElementTree.SubElement(xml_index, "namespaceMapping", {"key": prefix, "value": uri}) xml_request = xml.etree.ElementTree.tostring(xml_index, encoding='utf-8', xml_declaration=True) search_xml = root_element + "-index.xml" if export_folder is not None: search_xml = os.path.join(export_folder, search_xml) file = open(search_xml, "wt", encoding="utf-8") file.write(xml_request.decode("utf-8")) file.close() return search_xml def cvs_to_xml(csv_file, xml_namespace, root_element, file_name_column="filename", export_folder=None, additional_namespaces=None): """ Deprecated function, use csv_to_xml instead. """ return csv_to_xml(csv_file, xml_namespace, root_element, file_name_column, export_folder, additional_namespaces)
[docs] def csv_to_xml(csv_file, xml_namespace, root_element, file_name_column="filename", export_folder=None, additional_namespaces=None): """ Export the rows of a CSV file as individual XML metadata documents for Preservica assets. Each data row in ``csv_file`` is converted to an XML document whose root element is ``root_element`` and whose child elements correspond to the CSV column headers. One XML file is written per data row; the file is named after the value in the column identified by ``file_name_column``. :param csv_file: Path to the CSV file to process. :type csv_file: str :param xml_namespace: The XML namespace URI applied to the generated documents. :type xml_namespace: str :param root_element: The root element name for each generated XML document. :type root_element: str :param file_name_column: The CSV column header whose value is used to name each output XML file. Defaults to ``"filename"``. :type file_name_column: str :param export_folder: Directory where the generated XML files are written. If ``None``, files are written to the current working directory. :type export_folder: str or None :param additional_namespaces: Mapping of namespace prefix to URI for any extra namespaces to declare on the root element. :type additional_namespaces: dict or None :returns: Generator that yields the path to each written XML file. :rtype: Generator[str, None, None] """ headers = list() link_column_id = 0 with open(csv_file, encoding='utf-8-sig', newline='') as csvfile: reader = csv.reader(csvfile) for row in reader: col_id = 0 for header in row: col_id += 1 if header == file_name_column: link_column_id = col_id xml_tag = header.strip() xml_tag = xml_tag.replace(" ", "") xml_tag = xml_tag.replace("-", "") headers.append(xml_tag) break if link_column_id > 0: namespaces = {"xmlns": xml_namespace} if additional_namespaces is not None: for prefix, uri in additional_namespaces.items(): namespaces["xmlns:" + prefix] = uri for row in reader: col_id = 0 xml_object = xml.etree.ElementTree.Element(root_element, namespaces) for value, header in zip(row, headers): col_id += 1 xml.etree.ElementTree.SubElement(xml_object, header).text = value if col_id == link_column_id: file_name = value xml_request = xml.etree.ElementTree.tostring(xml_object, encoding='utf-8', xml_declaration=True) name = file_name + ".xml" name = sanitize(name) if export_folder is not None: name = os.path.join(export_folder, name) file = open(name, "wt", encoding="utf-8") file.write(xml_request.decode("utf-8")) file.close() yield name
def generic_asset_package(preservation_files_dict: dict|None=None, access_files_dict: dict|None=None, export_folder=None, parent_folder=None, compress=True, **kwargs): """ Create a Preservica ingest package containing an asset with an arbitrary number of representations. Both ``preservation_files_dict`` and ``access_files_dict`` map representation names to lists of file paths. Each key becomes a named representation; each file path in the associated list becomes a separate content object within that representation. :param preservation_files_dict: Mapping of preservation representation name to a list of file paths. Pass ``None`` to omit preservation representations. :type preservation_files_dict: dict[str, list[str]] or None :param access_files_dict: Mapping of access representation name to a list of file paths. Pass ``None`` to omit access representations. :type access_files_dict: dict[str, list[str]] or None :param export_folder: Directory where the package ZIP file is written. Defaults to the system temporary directory when ``None``. :type export_folder: str or None :param parent_folder: The Preservica folder (or its UUID string) under which the asset will be ingested. Required. :type parent_folder: Folder or str :param compress: Whether to compress the package ZIP file. Defaults to ``True``. :type compress: bool :param kwargs: Optional keyword arguments: - ``Title`` *(str)* – Asset title. - ``Description`` *(str)* – Asset description. - ``SecurityTag`` *(str)* – Asset security tag (default ``"open"``). - ``CustomType`` *(str)* – Asset custom type. - ``Preservation_Content_Title`` *(str or dict)* – Content object title for preservation representations. - ``Preservation_Content_Description`` *(str or dict)* – Content object description for preservation representations. - ``Access_Content_Title`` *(str or dict)* – Content object title for access representations. - ``Access_Content_Description`` *(str or dict)* – Content object description for access representations. - ``Preservation_Generation_Label`` *(str)* – Generation label for preservation representations. - ``Access_Generation_Label`` *(str)* – Generation label for access representations. - ``Preservation_files_fixity_callback`` *(callable)* – Fixity callback for preservation files. - ``Access_files_fixity_callback`` *(callable)* – Fixity callback for access files. - ``IO_Identifier_callback`` *(callable)* – Callback to supply the asset UUID. - ``Identifiers`` *(dict)* – Mapping of identifier type to identifier value. - ``Asset_Metadata`` *(dict)* – Mapping of schema URI to metadata file path or XML string. :returns: Path to the generated package ZIP file. :rtype: str :raises RuntimeError: If ``export_folder`` does not exist or ``parent_folder`` is ``None``. """ # some basic validation if export_folder is None: export_folder = tempfile.gettempdir() if not os.path.isdir(export_folder): logger.error("Export Folder Does Not Exist") raise RuntimeError(export_folder, "Export Folder Does Not Exist") if parent_folder is None: logger.error("You must specify a parent folder for the package asset") raise RuntimeError("You must specify a parent folder for the package asset") io_ref = None xip = None default_asset_title = None preservation_representation_refs_dict = {} access_representation_refs_dict = {} security_tag = kwargs.get('SecurityTag', "open") content_type = kwargs.get('CustomType', "") if not compress: shutil.register_archive_format(name="szip", function=_make_stored_zipfile, extra_args=None, description="UnCompressed ZIP file") has_preservation_files = bool((preservation_files_dict is not None) and (len(preservation_files_dict) > 0)) has_access_files = bool((access_files_dict is not None) and (len(access_files_dict) > 0)) if has_preservation_files: if default_asset_title is None: key = list(preservation_files_dict.keys())[0] preservation_files_list = preservation_files_dict[key] default_asset_title = os.path.splitext(os.path.basename(preservation_files_list[0]))[0] # create the asset xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs) if has_access_files: if default_asset_title is None: key = list(access_files_dict.keys())[0] access_files_list = access_files_dict[key] default_asset_title = os.path.splitext(os.path.basename(access_files_list[0]))[0] if io_ref is None: xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs) # loop over preservation_files_map if has_preservation_files: for representation_name in preservation_files_dict.keys(): preservation_files_list = preservation_files_dict[representation_name] preservation_refs_dict = __make_representation_multiple_co__(xip, rep_name=representation_name, rep_type="Preservation", rep_files=preservation_files_list, io_ref=io_ref) preservation_representation_refs_dict[representation_name] = preservation_refs_dict if has_access_files: for representation_name in access_files_dict.keys(): access_files_list = access_files_dict[representation_name] access_refs_dict = __make_representation_multiple_co__(xip, rep_name=representation_name, rep_type="Access", rep_files=access_files_list, io_ref=io_ref) access_representation_refs_dict[representation_name] = access_refs_dict if has_preservation_files: for representation_name in preservation_representation_refs_dict.keys(): preservation_refs_dict = preservation_representation_refs_dict[representation_name] for content_ref, filename in preservation_refs_dict.items(): default_content_objects_title = os.path.splitext(os.path.basename(filename))[0] preservation_content_title = kwargs.get('Preservation_Content_Title', default_content_objects_title) preservation_content_description = kwargs.get('Preservation_Content_Description', default_content_objects_title) if isinstance(preservation_content_title, dict): preservation_content_title = preservation_content_title.get("filename", default_content_objects_title) if isinstance(preservation_content_description, dict): preservation_content_description = preservation_content_description.get("filename", default_content_objects_title) __make_content_objects__(xip, preservation_content_title, content_ref, io_ref, security_tag, preservation_content_description, content_type) if has_access_files: for representation_name in access_representation_refs_dict.keys(): access_refs_dict = access_representation_refs_dict[representation_name] for content_ref, filename in access_refs_dict.items(): default_content_objects_title = os.path.splitext(os.path.basename(filename))[0] access_content_title = kwargs.get('Access_Content_Title', default_content_objects_title) access_content_description = kwargs.get('Access_Content_Description', default_content_objects_title) if isinstance(access_content_title, dict): access_content_title = access_content_title.get("filename", default_content_objects_title) if isinstance(access_content_description, dict): access_content_description = access_content_title.get("filename", default_content_objects_title) __make_content_objects__(xip, access_content_title, content_ref, io_ref, security_tag, access_content_description, content_type) if has_preservation_files: for representation_name in preservation_representation_refs_dict.keys(): location = sanitize(representation_name) preservation_refs_dict = preservation_representation_refs_dict[representation_name] preservation_generation_label = kwargs.get('Preservation_Generation_Label', "") for content_ref, filename in preservation_refs_dict.items(): preservation_file_name = os.path.basename(filename) __make_generation__(xip, preservation_file_name, content_ref, preservation_generation_label, location) if has_access_files: for representation_name in access_representation_refs_dict.keys(): location = sanitize(representation_name) access_refs_dict = access_representation_refs_dict[representation_name] access_generation_label = kwargs.get('Access_Generation_Label', "") for content_ref, filename in access_refs_dict.items(): access_file_name = os.path.basename(filename) __make_generation__(xip, access_file_name, content_ref, access_generation_label, location) if has_preservation_files: if 'Preservation_files_fixity_callback' in kwargs: callback = kwargs.get('Preservation_files_fixity_callback') else: callback = Sha1FixityCallBack() for representation_name in preservation_representation_refs_dict.keys(): location = sanitize(representation_name) preservation_refs_dict = preservation_representation_refs_dict[representation_name] for content_ref, filename in preservation_refs_dict.items(): preservation_file_name = os.path.basename(filename) __make_bitstream__(xip, preservation_file_name, filename, callback, location) if has_access_files: if 'Access_files_fixity_callback' in kwargs: callback = kwargs.get('Access_files_fixity_callback') else: callback = Sha1FixityCallBack() for representation_name in access_representation_refs_dict.keys(): location = sanitize(representation_name) access_refs_dict = access_representation_refs_dict[representation_name] for content_ref, filename in access_refs_dict.items(): access_file_name = os.path.basename(filename) __make_bitstream__(xip, access_file_name, filename, callback, location) if 'Identifiers' in kwargs: identifier_map = kwargs.get('Identifiers') for identifier_key, identifier_value in identifier_map.items(): if identifier_key: if identifier_value: identifier = SubElement(xip, 'xip:Identifier') id_type = SubElement(identifier, "xip:Type") id_type.text = identifier_key id_value = SubElement(identifier, "xip:Value") id_value.text = identifier_value id_io = SubElement(identifier, "xip:Entity") id_io.text = io_ref if 'Asset_Metadata' in kwargs: metadata_map = kwargs.get('Asset_Metadata') for metadata_ns, metadata_path in metadata_map.items(): if metadata_ns: if metadata_path: if os.path.exists(metadata_path) and os.path.isfile(metadata_path): descriptive_metadata = xml.etree.ElementTree.parse(source=metadata_path) metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns}) metadata_ref = SubElement(metadata, 'xip:Ref') metadata_ref.text = str(uuid.uuid4()) entity = SubElement(metadata, 'xip:Entity') entity.text = io_ref content = SubElement(metadata, 'xip:Content') content.append(descriptive_metadata.getroot()) elif isinstance(metadata_path, str): try: descriptive_metadata = xml.etree.ElementTree.fromstring(metadata_path) metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns}) metadata_ref = SubElement(metadata, 'xip:Ref') metadata_ref.text = str(uuid.uuid4()) entity = SubElement(metadata, 'xip:Entity') entity.text = io_ref content = SubElement(metadata, 'xip:Content') content.append(descriptive_metadata) except RuntimeError: logging.info(f"Could not parse asset metadata in namespace {metadata_ns}") if xip is not None: export_folder = export_folder top_level_folder = os.path.join(export_folder, io_ref) os.mkdir(top_level_folder) inner_folder = os.path.join(top_level_folder, io_ref) os.mkdir(inner_folder) content_folder = os.path.join(inner_folder, CONTENT_FOLDER) os.mkdir(content_folder) metadata_path = os.path.join(inner_folder, "metadata.xml") metadata = open(metadata_path, "wt", encoding='utf-8') metadata.write(prettify(xip)) metadata.close() for representation_name in preservation_representation_refs_dict.keys(): location = sanitize(representation_name) Path(os.path.join(content_folder, location)).mkdir(parents=True, exist_ok=True) preservation_refs_dict = preservation_representation_refs_dict[representation_name] for content_ref, filename in preservation_refs_dict.items(): src_file = filename dst_file = os.path.join(os.path.join(content_folder, location), os.path.basename(filename)) shutil.copyfile(src_file, dst_file) for representation_name in access_representation_refs_dict.keys(): location = sanitize(representation_name) Path(os.path.join(content_folder, location)).mkdir(parents=True, exist_ok=True) access_refs_dict = access_representation_refs_dict[representation_name] for content_ref, filename in access_refs_dict.items(): src_file = filename dst_file = os.path.join(os.path.join(content_folder, location), os.path.basename(filename)) shutil.copyfile(src_file, dst_file) if compress: shutil.make_archive(top_level_folder, 'zip', top_level_folder) else: shutil.make_archive(top_level_folder, 'szip', top_level_folder) shutil.rmtree(top_level_folder) return top_level_folder + ".zip" def multi_asset_package(asset_file_list:list|None=None, export_folder=None, parent_folder=None, compress=True, **kwargs): """ Create a single ingest package containing multiple assets, one per file. Each file in ``asset_file_list`` becomes a separate asset (Information Object) with a single preservation representation. All assets share the same ``parent_folder``. :param asset_file_list: List of file paths; one asset is created per file. :type asset_file_list: list[str] or None :param export_folder: Directory where the package ZIP file is written. Defaults to the system temporary directory when ``None``. :type export_folder: str or None :param parent_folder: The Preservica folder (or its UUID string) under which all assets will be ingested. Required. :type parent_folder: Folder or str :param compress: Whether to compress the package ZIP file. Defaults to ``True``. :type compress: bool :param kwargs: Optional keyword arguments: - ``SecurityTag`` *(str)* – Security tag applied to all assets (default ``"open"``). - ``CustomType`` *(str)* – Custom type applied to all assets. - ``Preservation_files_fixity_callback`` *(callable)* – Fixity callback used when computing checksums for the content files. - ``Identifiers`` *(dict)* – Mapping of file path to a dict of identifier type/value pairs that should be attached to the corresponding asset. :returns: Path to the generated package ZIP file. :rtype: str :raises RuntimeError: If ``export_folder`` does not exist or ``parent_folder`` is ``None``. """ # some basic validation if export_folder is None: export_folder = tempfile.gettempdir() if not os.path.isdir(export_folder): logger.error("Export Folder Does Not Exist") raise RuntimeError(export_folder, "Export Folder Does Not Exist") if parent_folder is None: logger.error("You must specify a parent folder for the package asset") raise RuntimeError("You must specify a parent folder for the package asset") security_tag = kwargs.get('SecurityTag', "open") content_type = kwargs.get('CustomType', "") if not compress: shutil.register_archive_format("szip", _make_stored_zipfile, None, "UnCompressed ZIP file") if 'Preservation_files_fixity_callback' in kwargs: fixity_callback = kwargs.get('Preservation_files_fixity_callback') else: fixity_callback = Sha1FixityCallBack() package_id = str(uuid.uuid4()) export_folder = export_folder top_level_folder = os.path.join(export_folder, package_id) os.mkdir(top_level_folder) inner_folder = os.path.join(top_level_folder, package_id) os.mkdir(inner_folder) os.mkdir(os.path.join(inner_folder, CONTENT_FOLDER)) asset_map = dict() xip = Element('xip:XIP') xip.set('xmlns:xip', 'http://preservica.com/XIP/v6.0') for file in asset_file_list: default_asset_title = os.path.splitext(os.path.basename(file))[0] xip, io_ref = __create_io__(xip, file_name=default_asset_title, parent_folder=parent_folder, **kwargs) asset_map[file] = io_ref representation = SubElement(xip, 'xip:Representation') io_link = SubElement(representation, 'xip:InformationObject') io_link.text = io_ref access_name = SubElement(representation, 'xip:Name') access_name.text = "Preservation" access_type = SubElement(representation, 'xip:Type') access_type.text = "Preservation" content_objects = SubElement(representation, 'xip:ContentObjects') content_object = SubElement(content_objects, 'xip:ContentObject') content_object_ref = str(uuid.uuid4()) content_object.text = content_object_ref default_content_objects_title = os.path.splitext(os.path.basename(file))[0] content_object = SubElement(xip, 'xip:ContentObject') ref_element = SubElement(content_object, "xip:Ref") ref_element.text = content_object_ref title = SubElement(content_object, "xip:Title") title.text = default_content_objects_title description = SubElement(content_object, "xip:Description") description.text = default_content_objects_title security_tag_element = SubElement(content_object, "xip:SecurityTag") security_tag_element.text = security_tag custom_type = SubElement(content_object, "xip:CustomType") custom_type.text = content_type parent = SubElement(content_object, "xip:Parent") parent.text = io_ref generation = SubElement(xip, 'xip:Generation', {"original": "true", "active": "true"}) content_object = SubElement(generation, "xip:ContentObject") content_object.text = content_object_ref label = SubElement(generation, "xip:Label") label.text = os.path.splitext(os.path.basename(file))[0] effective_date = SubElement(generation, "xip:EffectiveDate") effective_date.text = datetime.now().isoformat() bitstreams = SubElement(generation, "xip:Bitstreams") bitstream = SubElement(bitstreams, "xip:Bitstream") bitstream.text = os.path.basename(file) SubElement(generation, "xip:Formats") SubElement(generation, "xip:Properties") bitstream = SubElement(xip, 'xip:Bitstream') filename_element = SubElement(bitstream, "xip:Filename") filename_element.text = os.path.basename(file) filesize = SubElement(bitstream, "xip:FileSize") file_stats = os.stat(file) filesize.text = str(file_stats.st_size) physical_location = SubElement(bitstream, "xip:PhysicalLocation") fixities = SubElement(bitstream, "xip:Fixities") fixity_result = fixity_callback(filename_element.text, file) if type(fixity_result) == tuple: fixity = SubElement(fixities, "xip:Fixity") fixity_algorithm_ref = SubElement(fixity, "xip:FixityAlgorithmRef") fixity_value = SubElement(fixity, "xip:FixityValue") fixity_algorithm_ref.text = fixity_result[0] fixity_value.text = fixity_result[1] elif type(fixity_result) == dict: for key, val in fixity_result.items(): fixity = SubElement(fixities, "xip:Fixity") fixity_algorithm_ref = SubElement(fixity, "xip:FixityAlgorithmRef") fixity_value = SubElement(fixity, "xip:FixityValue") fixity_algorithm_ref.text = key fixity_value.text = val else: logger.error("Could Not Find Fixity Value") raise RuntimeError("Could Not Find Fixity Value") if 'Identifiers' in kwargs: identifier_map = kwargs.get('Identifiers') if str(file) in identifier_map: identifier_map_values = identifier_map[str(file)] for identifier_key, identifier_value in identifier_map_values.items(): if identifier_key: if identifier_value: identifier = SubElement(xip, 'xip:Identifier') id_type = SubElement(identifier, "xip:Type") id_type.text = identifier_key id_value = SubElement(identifier, "xip:Value") id_value.text = identifier_value id_io = SubElement(identifier, "xip:Entity") id_io.text = io_ref src_file = file dst_file = os.path.join(os.path.join(inner_folder, CONTENT_FOLDER), os.path.basename(file)) shutil.copyfile(src_file, dst_file) if xip is not None: metadata_path = os.path.join(inner_folder, "metadata.xml") metadata = open(metadata_path, "wt", encoding='utf-8') metadata.write(prettify(xip)) metadata.close() if compress: shutil.make_archive(top_level_folder, 'zip', top_level_folder) else: shutil.make_archive(top_level_folder, 'szip', top_level_folder) shutil.rmtree(top_level_folder) return top_level_folder + ".zip"
[docs] def complex_asset_package(preservation_files_list: list|None=None, access_files_list: list|None=None, export_folder=None, parent_folder=None, compress=True, **kwargs): """ Create a Preservica package containing a single Asset from a multiple preservation files and optional access files. The Asset contains multiple Content Objects within each representation. If only the preservation files are provided the asset has one representation :param list preservation_files_list: Paths to the preservation files :param list access_files_list: Paths to the access files :param str export_folder: The package location folder :param Folder parent_folder: The folder to ingest the asset into :param bool compress: Compress the ZIP file :param str Title: Asset Title :param str Description: Asset Description :param str SecurityTag: Asset SecurityTag :param str CustomType: Asset CustomType :param str Preservation_Content_Title: Title of the Preservation Representation Content Object :param str Preservation_Content_Description: Description of the Preservation Representation Content Object :param str Access_Content_Title: Title of the Access Representation Content Object :param str Access_Content_Description: Description of the Access Representation Content Object :param dict Asset_Metadata: Dictionary of Asset metadata documents :param dict Identifiers: Dictionary of Asset rd party identifiers optional kwargs map 'Title' Asset Title 'Description' Asset Description 'SecurityTag' Asset Security Tag 'CustomType' Asset Type 'Preservation_Content_Title' Content Object Title of the Preservation Object 'Preservation_Content_Description' Content Object Description of the Preservation Object 'Access_Content_Title' Content Object Title of the Access Object 'Access_Content_Description' Content Object Description of the Access Object 'Preservation_Generation_Label' Generation Label for the Preservation Object 'Access_Generation_Label' Generation Label for the Access Object 'Asset_Metadata' Map of metadata schema/documents to add to asset 'Identifiers' Map of asset identifiers 'Preservation_files_fixity_callback' Callback to allow external generated fixity values 'Access_files_fixity_callback' Callback to allow external generated fixity values 'IO_Identifier_callback' Callback to allow external generated Asset identifier 'Preservation_Representation_Name' Name of the Preservation Representation 'Access_Representation_Name' Name of the Access Representation """ xml.etree.ElementTree.register_namespace("xip", "http://preservica.com/XIP/v6.0") # some basic validation if export_folder is None: export_folder = tempfile.gettempdir() if not os.path.isdir(export_folder): logger.error("Export Folder Does Not Exist") raise RuntimeError(export_folder, "Export Folder Does Not Exist") if parent_folder is None: logger.error("You must specify a parent folder for the package asset") raise RuntimeError("You must specify a parent folder for the package asset") io_ref = None xip = None default_asset_title = None preservation_refs_dict = {} access_refs_dict = {} security_tag = kwargs.get('SecurityTag', "open") content_type = kwargs.get('CustomType', "") if not compress: shutil.register_archive_format("szip", _make_stored_zipfile, None, "UnCompressed ZIP file") has_preservation_files = bool((preservation_files_list is not None) and (len(preservation_files_list) > 0)) has_access_files = bool((access_files_list is not None) and (len(access_files_list) > 0)) if has_preservation_files: if default_asset_title is None: default_asset_title = os.path.splitext(os.path.basename(preservation_files_list[0]))[0] # create the asset if io_ref is None: xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs) if has_access_files: if default_asset_title is None: default_asset_title = os.path.splitext(os.path.basename(access_files_list[0]))[0] if io_ref is None: xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs) if io_ref is None: default_asset_title = kwargs.get('Title', None) if default_asset_title is None: default_asset_title = "New Asset" xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs) if has_preservation_files: # add the content objects representation_name = kwargs.get('Preservation_Representation_Name', "Preservation") preservation_refs_dict = __make_representation_multiple_co__(xip, rep_name=representation_name, rep_type="Preservation", rep_files=preservation_files_list, io_ref=io_ref) if has_access_files: # add the content objects access_name = kwargs.get('Access_Representation_Name', "Access") access_refs_dict = __make_representation_multiple_co__(xip, rep_name=access_name, rep_type="Access", rep_files=access_files_list, io_ref=io_ref) if has_preservation_files: for content_ref, filename in preservation_refs_dict.items(): default_content_objects_title = os.path.splitext(os.path.basename(filename))[0] preservation_content_title = kwargs.get('Preservation_Content_Title', default_content_objects_title) preservation_content_description = kwargs.get('Preservation_Content_Description', default_content_objects_title) if isinstance(preservation_content_title, dict): preservation_content_title = preservation_content_title[filename] if isinstance(preservation_content_description, dict): preservation_content_description = preservation_content_description[filename] __make_content_objects__(xip, preservation_content_title, content_ref, io_ref, security_tag, preservation_content_description, content_type) if has_access_files: for content_ref, filename in access_refs_dict.items(): default_content_objects_title = os.path.splitext(os.path.basename(filename))[0] access_content_title = kwargs.get('Access_Content_Title', default_content_objects_title) access_content_description = kwargs.get('Access_Content_Description', default_content_objects_title) if isinstance(access_content_title, dict): access_content_title = access_content_title[filename] if isinstance(access_content_description, dict): access_content_title = access_content_title[filename] __make_content_objects__(xip, access_content_title, content_ref, io_ref, security_tag, access_content_description, content_type) if has_preservation_files: preservation_generation_label = kwargs.get('Preservation_Generation_Label', "") for content_ref, filename in preservation_refs_dict.items(): preservation_file_name = os.path.basename(filename) __make_generation__(xip, preservation_file_name, content_ref, preservation_generation_label, PRESERVATION_CONTENT_FOLDER) if has_access_files: access_generation_label = kwargs.get('Access_Generation_Label', "") for content_ref, filename in access_refs_dict.items(): access_file_name = os.path.basename(filename) __make_generation__(xip, access_file_name, content_ref, access_generation_label, ACCESS_CONTENT_FOLDER) if has_preservation_files: if 'Preservation_files_fixity_callback' in kwargs: callback = kwargs.get('Preservation_files_fixity_callback') else: callback = Sha1FixityCallBack() for content_ref, filename in preservation_refs_dict.items(): preservation_file_name = os.path.basename(filename) __make_bitstream__(xip, preservation_file_name, filename, callback, PRESERVATION_CONTENT_FOLDER) if has_access_files: if 'Access_files_fixity_callback' in kwargs: callback = kwargs.get('Access_files_fixity_callback') else: callback = Sha1FixityCallBack() for content_ref, filename in access_refs_dict.items(): access_file_name = os.path.basename(filename) __make_bitstream__(xip, access_file_name, filename, callback, ACCESS_CONTENT_FOLDER) if 'Identifiers' in kwargs: identifier_map = kwargs.get('Identifiers') for identifier_key, identifier_value in identifier_map.items(): if identifier_key: if identifier_value: identifier = SubElement(xip, 'xip:Identifier') id_type = SubElement(identifier, "xip:Type") id_type.text = identifier_key id_value = SubElement(identifier, "xip:Value") id_value.text = identifier_value id_io = SubElement(identifier, "xip:Entity") id_io.text = io_ref if 'Asset_Metadata' in kwargs: metadata_map = kwargs.get('Asset_Metadata') for metadata_ns, metadata_path in metadata_map.items(): if metadata_ns: if metadata_path and isinstance(metadata_path, str): if os.path.exists(metadata_path) and os.path.isfile(metadata_path): descriptive_metadata = xml.etree.ElementTree.parse(source=metadata_path) metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns}) metadata_ref = SubElement(metadata, 'xip:Ref') metadata_ref.text = str(uuid.uuid4()) entity = SubElement(metadata, 'xip:Entity') entity.text = io_ref content = SubElement(metadata, 'xip:Content') content.append(descriptive_metadata.getroot()) elif isinstance(metadata_path, str): try: descriptive_metadata = xml.etree.ElementTree.fromstring(metadata_path) metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns}) metadata_ref = SubElement(metadata, 'xip:Ref') metadata_ref.text = str(uuid.uuid4()) entity = SubElement(metadata, 'xip:Entity') entity.text = io_ref content = SubElement(metadata, 'xip:Content') content.append(descriptive_metadata) except RuntimeError: logging.info(f"Could not parse asset metadata in namespace {metadata_ns}") if metadata_path and isinstance(metadata_path, list): for path in metadata_path: if os.path.exists(path) and os.path.isfile(path): descriptive_metadata = xml.etree.ElementTree.parse(source=path) metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns}) metadata_ref = SubElement(metadata, 'xip:Ref') metadata_ref.text = str(uuid.uuid4()) entity = SubElement(metadata, 'xip:Entity') entity.text = io_ref content = SubElement(metadata, 'xip:Content') content.append(descriptive_metadata.getroot()) if xip is not None: export_folder = export_folder top_level_folder = os.path.join(export_folder, io_ref) os.mkdir(top_level_folder) inner_folder = os.path.join(top_level_folder, io_ref) os.mkdir(inner_folder) content_folder = os.path.join(inner_folder, CONTENT_FOLDER) os.mkdir(content_folder) preservation_content_folder = os.path.join(content_folder, PRESERVATION_CONTENT_FOLDER) os.mkdir(preservation_content_folder) access_content_folder = os.path.join(content_folder, ACCESS_CONTENT_FOLDER) os.mkdir(access_content_folder) metadata_path = os.path.join(inner_folder, "metadata.xml") metadata = open(metadata_path, "wt", encoding='utf-8') metadata.write(prettify(xip)) metadata.close() for content_ref, filename in preservation_refs_dict.items(): src_file = filename dst_file = os.path.join(preservation_content_folder, os.path.basename(filename)) shutil.copyfile(src_file, dst_file) for content_ref, filename in access_refs_dict.items(): src_file = filename dst_file = os.path.join(access_content_folder, os.path.basename(filename)) shutil.copyfile(src_file, dst_file) if compress: shutil.make_archive(top_level_folder, 'zip', top_level_folder) else: shutil.make_archive(top_level_folder, 'szip', top_level_folder) shutil.rmtree(top_level_folder) return top_level_folder + ".zip"
[docs] def simple_asset_package(preservation_file: str|None=None, access_file: str|None=None, export_folder:str|None=None, parent_folder=None, compress=True, **kwargs): """ Create a Preservica package containing a single Asset from a single preservation file and an optional access file. The Asset contains one Content Object for each representation. If only the preservation file is provided the asset has one representation :param str preservation_file: Path to the preservation file :param str access_file: Path to the access file :param str export_folder: The package location folder :param Folder parent_folder: The folder to ingest the asset into :param bool compress: Compress the ZIP file :param str Title: Asset Title :param str Description: Asset Description :param str SecurityTag: Asset SecurityTag :param str CustomType: Asset CustomType :param str Preservation_Content_Title: Title of the Preservation Representation Content Object :param str Preservation_Content_Description: Description of the Preservation Representation Content Object :param str Access_Content_Title: Title of the Access Representation Content Object :param str Access_Content_Description: Description of the Access Representation Content Object :param dict Asset_Metadata: Dictionary of Asset metadata documents :param dict Identifiers: Dictionary of Asset rd party identifiers """ # some basic validation if export_folder is None: export_folder = tempfile.gettempdir() if not os.path.isdir(export_folder): logger.error("Export Folder Does Not Exist") raise RuntimeError(export_folder, "Export Folder Does Not Exist") if parent_folder is None: logger.error("You must specify a parent folder for the package asset") raise RuntimeError("You must specify a parent folder for the package asset") preservation_file_list = list() access_file_list = list() if preservation_file is not None: preservation_file_list.append(preservation_file) if access_file is not None: access_file_list.append(access_file) return complex_asset_package(preservation_files_list=preservation_file_list, access_files_list=access_file_list, export_folder=export_folder, parent_folder=parent_folder, compress=compress, **kwargs)
def upload_config(): return transfer_config def _unpad(s): return s[:-ord(s[len(s) - 1:])]
[docs] class UploadAPI(AuthenticatedAPI): def __init__(self, username: str|None = None, password: str|None = None, tenant: str|None = None, server: str|None = None, use_shared_secret: bool = False, two_fa_secret_key: str|None = None, protocol: str = "https", request_hook: Callable|None = None, credentials_path: str = 'credentials.properties'): super().__init__(username, password, tenant, server, use_shared_secret, two_fa_secret_key, protocol, request_hook, credentials_path) self.uploads_per_minute: int = 180 self.limiter: Limiter = None
[docs] def ingest_web_video(self, url=None, parent_folder=None, **kwargs): """ Download a web video (e.g. YouTube) and ingest it into Preservica as an asset. Requires the optional ``youtube_dl`` package (``pip install --upgrade youtube-dl``). The video is downloaded to the current working directory as ``<video_id>.mp4``, wrapped in a simple preservation package, and then uploaded via :meth:`upload_zip_package`. :param url: URL of the web video to download and ingest. :type url: str or None :param parent_folder: The Preservica folder into which the video asset is ingested. :type parent_folder: Folder or str or None :param Title: Optional title for the asset. Defaults to the video title from the metadata. :type Title: str :param Description: Optional description for the asset. Defaults to the video description from the metadata. :type Description: str :param SecurityTag: Optional security tag for the asset. Defaults to ``"open"``. :type SecurityTag: str :param Identifiers: Optional mapping of identifier type to value; the video id is automatically added under the key ``"Video-ID"``. :type Identifiers: dict :param Asset_Metadata: Optional mapping of schema URI to metadata document for the asset. :type Asset_Metadata: dict :param callback: Optional upload progress callback. :type callback: callable or None :raises RuntimeError: If the ``youtube_dl`` package is not installed. """ try: import youtube_dl except ImportError: logger.error("Package youtube_dl is required for this method. pip install --upgrade youtube-dl") raise RuntimeError("Package youtube_dl is required for this method. pip install --upgrade youtube-dl") ydl_opts = {} def my_hook(d): if d['status'] == 'finished': logger.info('Download Complete. Uploading to Preservica ...') ydl_opts = {'outtmpl': '%(id)s.mp4', 'progress_hooks': [my_hook], } # if True: # ydl_opts['writesubtitles'] = True # ydl_opts['writeautomaticsub'] = True # ydl_opts['subtitleslangs'] = ['en'] with youtube_dl.YoutubeDL(ydl_opts) as ydl: meta = ydl.extract_info(url, download=True) vid_id = meta.get('id') if 'Title' in kwargs: title = kwargs.get("Title") else: title = meta.get('title') if 'Description' in kwargs: description = kwargs.get("Description") else: description = meta.get('description') if 'SecurityTag' in kwargs: security_tag = kwargs.get("SecurityTag") else: security_tag = "open" if 'Identifiers' in kwargs: identifier_map = kwargs.get('Identifiers') identifier_map["Video-ID"] = vid_id else: identifier_map = {"Video-ID": vid_id} if 'Asset_Metadata' in kwargs: descriptive_metadata = kwargs.get('Asset_Metadata') else: descriptive_metadata = {} if 'callback' in kwargs: callback = kwargs.get("callback") else: callback = None upload_date = meta.get('upload_date') duration = meta.get('duration') package = simple_asset_package(preservation_file=f"{vid_id}.mp4", parent_folder=parent_folder, Title=title, Description=description, Identifiers=identifier_map, Asset_Metadata=descriptive_metadata, Preservation_Content_Title=title, SecurityTag=security_tag) self.upload_zip_package(path_to_zip_package=package, folder=parent_folder, callback=callback)
[docs] def upload_credentials(self, location_id: str) -> dict: """ Retrieve temporary upload credentials for a specific upload location. For AWS deployments, the credentials are Amazon STS temporary credentials (``key``, ``secret``, ``sessionToken``, ``endpoint``). For Azure deployments they are Azure SAS credentials (``key``, ``sessionToken``). :param location_id: The API id of the upload location (from :meth:`upload_locations`). :type location_id: str :returns: Dictionary of credential fields appropriate for the deployment type. :rtype: dict :raises HTTPException: If the server returns a non-200 response. """ headers = {HEADER_TOKEN: self.token} endpoint = f"/upload/{location_id}/upload-credentials" request = self.session.get(f'https://{self.server}/api/location{endpoint}', headers=headers) if request.status_code == requests.codes.ok: json_response = str(request.content.decode('utf-8')) return json.loads(json_response) else: exception = HTTPException(location_id, request.status_code, request.url, "upload_credentials", request.content.decode('utf-8')) logger.error(exception) raise exception
[docs] def clean_upload_bucket(self, bucket_name: str, older_than_days: int = 90): """ Delete objects in an upload bucket that are older than a given number of days. Supports both AWS S3 buckets and Azure Blob Storage containers. The appropriate cloud credentials are obtained automatically via :meth:`upload_credentials`. :param bucket_name: The name of the S3 bucket or Azure container to clean. :type bucket_name: str :param older_than_days: Objects whose last-modified timestamp is more than this many days in the past are deleted. Defaults to ``90``. :type older_than_days: int :returns: None :rtype: None """ from azure.storage.blob import ContainerClient for location in self.upload_locations(): if location['containerName'] == bucket_name: if location['type'] != 'AWS': credentials = self.upload_credentials(location['apiId']) account_key = credentials['key'] session_token = credentials['sessionToken'] sas_url = f"https://{account_key}.blob.core.windows.net/{bucket_name}" container = ContainerClient.from_container_url(container_url=sas_url, credential=session_token) now = datetime.now(timezone.utc) for blob in container.list_blobs(): if abs((blob.last_modified - now).days) > older_than_days: logger.debug(f"Deleting expired object {blob.name}") container.delete_blob(blob.name) if location['type'] == 'AWS': credentials = self.upload_credentials(location['apiId']) access_key = credentials['key'] secret_key = credentials['secret'] session_token = credentials['sessionToken'] session = boto3.Session(aws_access_key_id=access_key, aws_secret_access_key=secret_key, aws_session_token=session_token) s3_client = session.client("s3") paginator = s3_client.get_paginator('list_objects_v2') now = datetime.now(timezone.utc) for page in paginator.paginate(Bucket=bucket_name): if 'Contents' in page: for key in page['Contents']: last_modified = key['LastModified'] if abs((last_modified - now).days) > older_than_days: logger.debug(f"Deleting expired object {key['Key']}") s3_client.delete_object(Bucket=bucket_name, Key=key['Key'])
[docs] def upload_locations(self) -> dict: """ Return all configured upload locations for this Preservica system. Upload locations are configured on the Preservica Sources page as *SIP Upload* entries and represent S3 buckets or Azure containers connected to ingest workflows. :returns: List of upload location dictionaries, each containing at least ``containerName``, ``type`` (``"AWS"`` or ``"Azure"``), and ``apiId``. :rtype: list[dict] :raises HTTPException: If the server returns a non-200 response. """ headers = {HEADER_TOKEN: self.token} endpoint = "/api/location/upload" request = self.session.get(f'https://{self.server}{endpoint}', headers=headers) if request.status_code == requests.codes.ok: json_response = str(request.content.decode('utf-8')) return json.loads(json_response)['locations'] else: exception = HTTPException("", request.status_code, request.url, "upload_locations", request.content.decode('utf-8')) logger.error(exception) raise exception
[docs] def upload_buckets(self) -> dict: """ Get a list of available upload buckets. Alias for :meth:`upload_locations`; returns all upload locations configured on the Preservica system (both S3 buckets and Azure containers). :returns: List of upload location dictionaries (see :meth:`upload_locations`). :rtype: list[dict] :raises HTTPException: If the server returns a non-200 response. """ return self.upload_locations()
def crawl_filesystem(self, filesystem_path, bucket_name, preservica_parent, callback: bool = False, security_tag: str = "open", delete_after_upload: bool = True, max_MB_ingested: int = -1): from pyPreservica import EntityAPI def entity_value(client: EntityAPI, identifier: str) -> Entity: back_off: int = 5 while True: try: entities = client.identifier("code", identifier) if bool(len(entities) > 0): return entities.pop() else: return None except HTTPException as e: sleep(back_off) back_off = back_off * 2 def entity_exists(client: EntityAPI, identifier: str) -> bool: back_off: int = 5 while True: try: entities = client.identifier("code", identifier) return bool(len(entities) > 0) except HTTPException as e: sleep(back_off) back_off = back_off * 2 def get_parent(client, identifier, parent_reference): dirname_id: str = str(os.path.dirname(identifier)) if not dirname_id: dirname_id = identifier folder = entity_value(client, dirname_id) if folder is not None: folder = client.folder(folder.reference) return folder.reference else: return parent_reference def get_folder(client, name, tag, parent_reference, identifier): folder = entity_value(client, identifier) if folder is None: logger.info(f"Creating new folder with name {name}") folder = client.create_folder(name, name, tag, parent_reference) client.add_identifier(folder, "code", identifier) else: logger.info(f"Found existing folder with name {name}") return folder entity_client = EntityAPI(username=self.username, password=self.password, server=self.server, tenant=self.tenant, two_fa_secret_key=self.two_fa_secret_key, use_shared_secret=self.shared_secret, protocol=self.protocol) if preservica_parent: parent = entity_client.folder(preservica_parent) logger.info(f"Folders will be created inside Preservica collection {parent.title}") parent_ref = parent.reference else: parent_ref = None bytes_ingested = 0 folder_path = os.path.normpath(filesystem_path) for dirname, subdirs, files in os.walk(folder_path): base = os.path.basename(dirname) code = os.path.relpath(dirname, Path(folder_path).parent) p = get_parent(entity_client, code, parent_ref) f = get_folder(entity_client, base, security_tag, p, code) identifiers = dict() for file in list(files): full_path = os.path.join(dirname, file) if os.path.islink(full_path): logger.info(f"Skipping link {file}") files.remove(file) continue asset_code = os.path.join(code, file) if not entity_exists(entity_client, asset_code): bytes_ingested = bytes_ingested + os.stat(full_path).st_size logger.info(f"Adding new file: {file} to package ready for upload") file_identifiers = {"code": asset_code} identifiers[full_path] = file_identifiers else: logger.info(f"Skipping file {file} already exists in repository") files.remove(file) if len(files) > 0: full_path_list = [os.path.join(dirname, file) for file in files] package = multi_asset_package(asset_file_list=full_path_list, parent_folder=f, SecurityTag=security_tag, Identifiers=identifiers) if callback: progress_display = UploadProgressConsoleCallback(package) else: progress_display = None if bucket_name is None: self.upload_zip_package(path_to_zip_package=package, callback=progress_display, delete_after_upload=delete_after_upload) else: self.upload_zip_to_Source(path_to_zip_package=package, container_name=bucket_name, show_progress=bool(progress_display is not None), delete_after_upload=delete_after_upload) logger.info(f"Uploaded " + "{:.1f}".format(bytes_ingested / (1024 * 1024)) + " MB") if max_MB_ingested > 0: if bytes_ingested > (1024 * 1024 * max_MB_ingested): logger.info(f"Reached Max Upload Limit") break
[docs] def upload_zip_to_Source(self, path_to_zip_package: str, container_name, folder=None, delete_after_upload=False, show_progress=False): """ Upload a ZIP package to an upload source (Azure container or S3 bucket). Automatically dispatches to :meth:`upload_zip_package_to_S3` or :meth:`upload_zip_package_to_Azure` based on the deployment type reported by :meth:`upload_locations`. :param path_to_zip_package: Local path to the ZIP package file. :type path_to_zip_package: str :param container_name: Name of the S3 bucket or Azure container connected to an ingest workflow. :type container_name: str :param folder: Optional Preservica folder (or UUID string) that the ingested package should be placed into. :type folder: Folder or str or None :param delete_after_upload: When ``True``, the local ZIP file is deleted after a successful upload. Defaults to ``False``. :type delete_after_upload: bool :param show_progress: When ``True``, an upload progress bar is displayed on the console. Defaults to ``False``. :type show_progress: bool :returns: None :rtype: None """ locations = self.upload_locations() for location in locations: if location['containerName'] == container_name: if location['type'] == 'AWS': callback = None if show_progress: callback = UploadProgressConsoleCallback(path_to_zip_package) self.upload_zip_package_to_S3(path_to_zip_package=path_to_zip_package, bucket_name=container_name, folder=folder, callback=callback, delete_after_upload=delete_after_upload) else: self.upload_zip_package_to_Azure(path_to_zip_package=path_to_zip_package, container_name=container_name, folder=folder, delete_after_upload=delete_after_upload, show_progress=show_progress)
[docs] def upload_zip_package_to_Azure(self, path_to_zip_package: str, container_name: str, folder=None, delete_after_upload=False, show_progress=False): """ Uploads a zip file package to an Azure container connected to a Preservica Cloud System :param str path_to_zip_package: Path to the package :param str container_name: container connected to the ingest workflow :param Folder folder: The folder to ingest the package into :param bool delete_after_upload: Delete the local copy of the package after the upload has completed """ if (self.major_version < 7) and (self.minor_version < 5): raise RuntimeError( "This call [upload_zip_package_to_Azure] is only available against v6.5 systems and above") from azure.storage.blob import ContainerClient locations = self.upload_locations() for location in locations: if location['containerName'] == container_name: credentials = self.upload_credentials(location['apiId']) account_key = credentials['key'] session_token = credentials['sessionToken'] sas_url = f"https://{account_key}.blob.core.windows.net/{container_name}" container = ContainerClient.from_container_url(container_url=sas_url, credential=session_token) upload_key = str(uuid.uuid4()) metadata = {'key': upload_key, 'name': upload_key + ".zip", 'bucket': container_name, 'status': 'ready'} if hasattr(folder, "reference"): metadata['collectionreference'] = folder.reference elif isinstance(folder, str): metadata['collectionreference'] = folder properties = None len_bytes = Path(path_to_zip_package).stat().st_size if show_progress: with tqdm.wrapattr(open(path_to_zip_package, 'rb'), "read", total=len_bytes) as data: blob_client = container.upload_blob(name=upload_key, data=data, metadata=metadata, length=len_bytes) properties = blob_client.get_blob_properties() else: with open(path_to_zip_package, "rb") as data: blob_client = container.upload_blob(name=upload_key, data=data, metadata=metadata, length=len_bytes) properties = blob_client.get_blob_properties() if delete_after_upload: os.remove(path_to_zip_package) return properties
[docs] def upload_zip_package_to_S3(self, path_to_zip_package: str, bucket_name: str, folder=None, callback=None, delete_after_upload=False): """ Uploads a zip file package to an S3 bucket connected to a Preservica Cloud System :param str path_to_zip_package: Path to the package :param str bucket_name: Bucket connected to an ingest workflow :param Folder folder: The folder to ingest the package into :param Callable callback: Optional callback to allow the callee to monitor the upload progress :param bool delete_after_upload: Delete the local copy of the package after the upload has completed """ if (self.major_version < 7) and (self.minor_version < 5): raise RuntimeError("This call [upload_zip_package_to_S3] is only available against v6.5 systems and above") logger.debug("Finding Upload Locations") self.token = self.__token__() locations = self.upload_locations() for location in locations: if location['containerName'] == bucket_name: logger.debug(f"Found Upload Location {location['containerName']}") logger.debug(f"Fetching Upload Credentials for {location['containerName']}") credentials = self.upload_credentials(location['apiId']) access_key = credentials['key'] secret_key = credentials['secret'] session_token = credentials['sessionToken'] endpoint = credentials['endpoint'] session = boto3.Session(aws_access_key_id=access_key, aws_secret_access_key=secret_key, aws_session_token=session_token) s3 = session.resource(service_name="s3") logger.debug(f"S3 Session: {s3}") upload_key = str(uuid.uuid4()) s3_object = s3.Object(bucket_name, upload_key) metadata = {'key': upload_key, 'name': upload_key + ".zip", 'bucket': bucket_name, 'status': 'ready'} if hasattr(folder, "reference"): metadata['collectionreference'] = folder.reference elif isinstance(folder, str): metadata['collectionreference'] = folder metadata['size'] = str(Path(path_to_zip_package).stat().st_size) metadata['createdby'] = self.username metadata_map = {'Metadata': metadata} s3_object.upload_file(path_to_zip_package, Callback=callback, ExtraArgs=metadata_map, Config=transfer_config) if delete_after_upload: os.remove(path_to_zip_package)
[docs] def upload_zip_package(self, path_to_zip_package, folder=None, callback=None, delete_after_upload=False, limit_per_minute= 180): """ Upload a ZIP package directly to Preservica via the built-in S3 endpoint and start an ingest workflow. Uses auto-refreshing Preservica token credentials so that long-running uploads do not fail due to token expiry. Multipart chunk sizes are automatically adjusted based on the package size. This function can be rate limited to restrict the number of uploads per minute, it will default to a max of 180 / min. :param path_to_zip_package: Local path to the ZIP package file. :type path_to_zip_package: str :param folder: Optional Preservica folder (or UUID string) that the ingested package should be placed into. :type folder: Folder or str or None :param callback: Optional progress callback invoked during the upload. Must be compatible with the boto3 ``Callback`` interface (i.e. a callable that accepts the number of bytes transferred). :type callback: callable or None :param delete_after_upload: When ``True``, the local ZIP file is deleted after a successful upload. Defaults to ``False``. :type delete_after_upload: bool :param limit_per_minute: Restrict the number of uploads per min, default is 180 (3 per second) :type limit_per_minute: int :returns: The ``preservica-progress-token`` header value returned by the server, which can be used to monitor the ingest workflow progress. :rtype: str :raises NoCredentialsError: If AWS credentials cannot be resolved. :raises ClientError: If the S3 upload fails. """ if self.limiter is None: self.limiter = Limiter(Rate(limit_per_minute, Duration.MINUTE)) self.uploads_per_minute = limit_per_minute if limit_per_minute != self.uploads_per_minute: self.limiter = Limiter(Rate(limit_per_minute, Duration.MINUTE)) self.uploads_per_minute = limit_per_minute self.limiter.try_acquire("upload_zip_package") bucket = f'{self.tenant.lower()}.package.upload' endpoint = f'{self.protocol}://{self.server}/api/s3/buckets' self.token = self.__token__() retries= { 'max_attempts': 5, 'mode': 'adaptive' } def new_credentials(): cred_metadata: dict = {} cred_metadata['access_key'] = self.__token__() cred_metadata['secret_key'] = "NOT_USED" cred_metadata['token'] = "" cred_metadata["expiry_time"] = (datetime.now(tzlocal()) + timedelta(minutes=12)).isoformat() logger.info("Refreshing credentials at: " + str(datetime.now(tzlocal()))) return cred_metadata session = get_session() session_credentials = RefreshableCredentials.create_from_metadata( metadata=new_credentials(), refresh_using=new_credentials, advisory_timeout = 4 * 60, mandatory_timeout = 12 * 60, method = 'Preservica' ) autorefresh_session = boto3.Session(botocore_session=session) session._credentials = session_credentials config = Config(s3={'addressing_style': 'path'}, read_timeout=120, connect_timeout=120, request_checksum_calculation="WHEN_REQUIRED", response_checksum_validation="WHEN_REQUIRED", retries=retries, tcp_keepalive=True) s3_client = autorefresh_session.client('s3', endpoint_url=endpoint, config=config) metadata = {} if folder is not None: if hasattr(folder, "reference"): metadata = {'Metadata': {'structuralobjectreference': folder.reference}} elif isinstance(folder, str): metadata = {'Metadata': {'structuralobjectreference': folder}} if os.path.exists(path_to_zip_package) and os.path.isfile(path_to_zip_package): try: key_id = str(uuid.uuid4()) + ".zip" # how big is the package package_size = os.path.getsize(path_to_zip_package) if package_size > 1 * GB: transfer_config.multipart_chunksize = 16 * MB ## Min 64 Chunks if package_size > 8 * GB: transfer_config.multipart_chunksize = 32 * MB ## Min 256 Chunks if package_size > 24 * GB: transfer_config.multipart_chunksize = 48 * MB ## Min 512 Chunks if package_size > 48 * GB: transfer_config.multipart_chunksize = 64 * MB logger.info("Using Multipart Chunk Size: " + str(transfer_config.multipart_chunksize)) transfer = S3Transfer(client=s3_client, config=transfer_config) transfer.PutObjectTask = PutObjectTask transfer.CompleteMultipartUploadTask = CompleteMultipartUploadTask transfer.upload_file = upload_file response = transfer.upload_file(self=transfer, filename=path_to_zip_package, bucket=bucket, key=key_id, extra_args=metadata, callback=callback) if delete_after_upload: os.remove(path_to_zip_package) return response['ResponseMetadata']['HTTPHeaders']['preservica-progress-token'] except (NoCredentialsError, PartialCredentialsError) as ex: logger.error(ex) raise ex except ClientError as ex: logger.error(ex) raise ex