import logging
import os
from itertools import chain
from uuid import uuid4
from lxml import etree
from . import exceptions
from . import utils
from .di import Dependency
from .di import DependencyPossessor
from .di import has_class_methods
from .di import has_methods
from .di import is_class
from .metadata import AMDSec
from .metadata import MDRef
from .metadata import MDWrap
from .metadata import SubSection
LOGGER = logging.getLogger(__name__)
[docs]class FSEntry(DependencyPossessor):
"""A class representing a filesystem entry - either a file or a directory.
When passed to a :class:`metsrw.mets.METSDocument` instance, the tree of
FSEntry objects will be used to construct the <fileSec> and <structMap>
elements of a METS document.
Unless otherwise specified, an FSEntry object is assumed to be a file;
pass the `type` value as 'Directory' to specify that the object is
instead a directory.
An FSEntry object must be instantiated with a path as the first
argument to the constructor, which represents its path on disk.
An FSEntry object which is a Directory may have one or more children,
representing files or directories contained within itself. Directory
trees are designed for top-to-bottom traversal. Files cannot have
children, and attempting to instantiate a file FSEntry object with
children will raise a ValueError.
Any FSEntry object may have one or more metadata entries associated
with it; these can take the form of either references to other XML
files on disk, which should be wrapped in MDRef objects, or
wrapped copies of those XML files, which should be wrapped in
MDWrap objects.
:param str path: Path to the file on disk, as a bytestring. This will
populate FLocat @xlink:href
:param str label: Label in the structMap. If not provided, will be populated
with the basename of path
:param str fileid: Provides a mechanism to assign a FILEID to an FSENTRY
when a pointer file is being created, so when a METS file is being
written for an package-file-type, i.e. an AIP. The FILE ID is an XML
NC (Non-colonized) name and so callers must understand the restricted
character-set of that type to use it properly. There is currently no
validation on this attribute on generation.
:param str use: Use for the fileGrp. Items with identical uses will be
grouped together.
:param str type: Type of FSEntry this is. This will appear in the structMap.
:param list children: List of :class:`metsrw.fsentry.FSEntry` that are
direct children of this element in the structMap. Only allowed if type
is 'Directory'
:param str file_uuid: UUID of this entry. Will be used to construct the
FILEID used in the fileSec and structMap, and GROUPID. Only required if
type is 'Item'.
:param metsrw.fsentry.FSEntry derived_from: FSEntry that this FSEntry is
derived_from. This is used to set the GROUPID in the fileSec.
:param str checksum: Value of the file's checksum. Required if checksumtype passed.
:param str checksumtype: Type of the checksum. Must be one of
:const:`FSEntry.ALLOWED_CHECKSUMS`. Required if checksum passed.
:param list transform_files: a list of dicts representing METS transform
file elements, which provide "a means to access any subsidiary files
listed below a <file> element by indicating the steps required to
'unpack' or transform the subsidiary files."
:raises ValueError: if children passed when type is not 'Directory'
:raises ValueError: if only one of checksum or checksumtype passed
:raises ValueError: if checksumtype is not in :const:`FSEntry.ALLOWED_CHECKSUMS`
"""
ALLOWED_CHECKSUMS = (
"Adler-32",
"CRC32",
"HAVAL",
"MD5",
"MNP",
"SHA-1",
"SHA-256",
"SHA-384",
"SHA-512",
"TIGER WHIRLPOOL",
)
# Dependencies that must be injected. This means that an
# ``FSEntry`` instance must be able to call ``self.premis_object_class`` and
# get a class with methods ``fromtree`` and ``serialize``.
premis_object_class = Dependency(
has_methods("serialize"), has_class_methods("fromtree"), is_class
)
premis_event_class = Dependency(
has_methods("serialize"), has_class_methods("fromtree"), is_class
)
premis_agent_class = Dependency(
has_methods("serialize"), has_class_methods("fromtree"), is_class
)
premis_rights_class = Dependency(
has_methods("serialize"), has_class_methods("fromtree"), is_class
)
PREMIS_OBJECT = "PREMIS:OBJECT"
PREMIS_EVENT = "PREMIS:EVENT"
PREMIS_AGENT = "PREMIS:AGENT"
PREMIS_RIGHTS = "PREMIS:RIGHTS"
def __init__(
self,
path=None,
fileid=None,
label=None,
use="original",
type="Item",
children=None,
file_uuid=None,
derived_from=None,
checksum=None,
checksumtype=None,
transform_files=None,
mets_div_type=None,
):
if isinstance(path, bytes):
self.path = path.decode("utf-8", errors="strict")
else:
self.path = path
if label is None and path is not None:
label = os.path.basename(path)
self._fileid = fileid
self.label = label
self.use = use
self.type = str(type)
self.parent = None
self._children = []
if not transform_files:
transform_files = []
self.transform_files = transform_files
self.mets_div_type = mets_div_type or self.type
children = children or []
for child in children:
self.add_child(child)
self.file_uuid = file_uuid
self.derived_from = derived_from
if bool(checksum) != bool(checksumtype):
raise ValueError(
"Must provide both checksum and checksumtype, or neither. Provided values: %s and %s"
% (checksum, checksumtype)
)
if checksumtype and checksumtype not in self.ALLOWED_CHECKSUMS:
raise ValueError(f"{checksumtype} must be one of {self.ALLOWED_CHECKSUMS}")
self.checksum = checksum
self.checksumtype = checksumtype
self.amdsecs = []
self.dmdsecs = []
self.dmdsecs_by_mdtype = {}
[docs] @classmethod
def dir(cls, label, children):
"""Return ``FSEntry`` directory object."""
return FSEntry(label=label, children=children, type="Directory", use=None)
[docs] @classmethod
def from_fptr(cls, label, type_, fptr):
"""Return ``FSEntry`` object."""
return FSEntry(
fileid=fptr.fileid,
label=label,
type=type_,
path=fptr.path,
use=fptr.use,
file_uuid=fptr.file_uuid,
derived_from=fptr.derived_from,
checksum=fptr.checksum,
checksumtype=fptr.checksumtype,
transform_files=fptr.transform_files,
)
def __str__(self):
return "{s.type}: {s.path}".format(s=self)
def __repr__(self):
return "FSEntry(type={s.type!r}, path={s.path!r}, use={s.use!r}, label={s.label!r}, file_uuid={s.file_uuid!r}, checksum={s.checksum!r}, checksumtype={s.checksumtype!r}, fileid={s._fileid!r})".format(
s=self
)
# PROPERTIES
[docs] def file_id(self):
"""Returns the fptr @FILEID if this is not a Directory."""
if self.type.lower() == "directory":
return None
if self.file_uuid is None:
raise exceptions.MetsError(
"No FILEID: File %s does not have file_uuid set" % self.path
)
if self.is_aip:
if self._fileid:
return self._fileid
return (
utils.FILE_ID_PREFIX + os.path.splitext(os.path.basename(self.path))[0]
)
return utils.FILE_ID_PREFIX + self.file_uuid
[docs] def group_id(self):
"""
Returns the @GROUPID.
If derived_from is set, returns that group_id.
"""
if self.derived_from is not None:
return self.derived_from.group_id()
if self.file_uuid is None:
return None
return utils.GROUP_ID_PREFIX + self.file_uuid
@property
def admids(self):
"""Returns a list of ADMIDs for this entry."""
return [a.id_string for a in self.amdsecs]
@property
def dmdids(self):
"""Returns a list of DMDIDs for this entry."""
return [d.id_string for d in self.dmdsecs]
@property
def children(self):
# Read-only
return self._children
@property
def is_aip(self):
return self.type.lower() == "archival information package"
# ADD ATTRIBUTES
def _add_metadata_element(self, md, subsection, mdtype, mode="mdwrap", **kwargs):
"""
:param md: Value to pass to the MDWrap/MDRef
:param str subsection: Metadata tag to create. See :const:`SubSection.ALLOWED_SUBSECTIONS`
:param str mdtype: Value for mdWrap/mdRef @MDTYPE
:param str mode: 'mdwrap' or 'mdref'
:param str loctype: Required if mode is 'mdref'. LOCTYPE of a mdRef
:param str label: Optional. Label of a mdRef
:param str otherloctype: Optional. OTHERLOCTYPE of a mdRef.
:param str othermdtype: Optional. OTHERMDTYPE of a mdWrap.
"""
# HELP how handle multiple amdSecs?
# When adding *MD which amdSec to add to?
if mode.lower() == "mdwrap":
othermdtype = kwargs.get("othermdtype")
mdsec = MDWrap(md, mdtype, othermdtype)
elif mode.lower() == "mdref":
loctype = kwargs.get("loctype")
label = kwargs.get("label")
otherloctype = kwargs.get("otherloctype")
xptr = kwargs.get("xptr")
othermdtype = kwargs.get("othermdtype")
mdsec = MDRef(md, mdtype, loctype, label, otherloctype, xptr, othermdtype)
subsection = SubSection(subsection, mdsec)
if subsection.subsection == "dmdSec":
self.dmdsecs.append(subsection)
else:
try:
amdsec = self.amdsecs[0]
except IndexError:
amdsec = AMDSec()
self.amdsecs.append(amdsec)
amdsec.subsections.append(subsection)
return subsection
[docs] def add_techmd(self, md, mdtype, mode="mdwrap", **kwargs):
return self._add_metadata_element(md, "techMD", mdtype, mode, **kwargs)
[docs] def add_digiprovmd(self, md, mdtype, mode="mdwrap", **kwargs):
return self._add_metadata_element(md, "digiprovMD", mdtype, mode, **kwargs)
[docs] def add_rightsmd(self, md, mdtype, mode="mdwrap", **kwargs):
return self._add_metadata_element(md, "rightsMD", mdtype, mode, **kwargs)
[docs] def add_dmdsec(self, md, mdtype, mode="mdwrap", **kwargs):
"""Add dmdsec.
Extension of _add_metadata_element that adds a dmdSec and updates the
previous dmdSecs with the same MDTYPE and OTHERMDTYPE attribute values,
marking them as "superseded" and using the same group_id for all of them.
"""
dmdsec = self._add_metadata_element(md, "dmdSec", mdtype, mode, **kwargs)
dmdsec.status = kwargs.get("status") or "original"
mdtype_key = utils.generate_mdtype_key(mdtype, kwargs.get("othermdtype", ""))
if mdtype_key in self.dmdsecs_by_mdtype:
group_id = getattr(
self.dmdsecs_by_mdtype[mdtype_key][0], "group_id", str(uuid4())
)
dmdsec.group_id = group_id
for previous_dmdsec in self.dmdsecs_by_mdtype[mdtype_key]:
previous_dmdsec.group_id = group_id
if not previous_dmdsec.status:
previous_dmdsec.status = "original"
if not previous_dmdsec.status.endswith("-superseded"):
previous_dmdsec.status += "-superseded"
self.dmdsecs_by_mdtype.setdefault(mdtype_key, []).append(dmdsec)
return dmdsec
[docs] def delete_dmdsec(self, mdtype, othermdtype=""):
"""Mark latest dmdsec of mdtype_othermdtype as deleted.
It doesn't delete the dmdsec from the METS. It only sets its status
attribute to "deleted".
"""
mdtype_key = utils.generate_mdtype_key(mdtype, othermdtype)
if mdtype_key in self.dmdsecs_by_mdtype:
self.dmdsecs_by_mdtype[mdtype_key][-1].status = "deleted"
[docs] def has_dmdsec(self, mdtype, othermdtype=""):
"""Check if a dmdsec of mdtype_othermdtype exists for this entry."""
mdtype_key = utils.generate_mdtype_key(mdtype, othermdtype)
return mdtype_key in self.dmdsecs_by_mdtype
[docs] def serialize_md_inst(self, md_inst, md_class):
"""Serialize object ``md_inst`` by transforming it into an
``lxml.etree._ElementTree``. If it already is such, return it. If not,
make sure it is the correct type and return the output of calling
``seriaize()`` on it.
"""
valid_insts = tuple(chain((etree._ElementTree, etree._Element), (str,)))
if isinstance(md_inst, valid_insts):
return md_inst
if not isinstance(md_inst, md_class):
raise TypeError(f"Instance {md_inst!r} must be instance of {md_class!r}")
return md_inst.serialize()
[docs] def add_premis_object(self, md, mode="mdwrap"):
meth = self.add_techmd
if self.is_empty_dir:
meth = self.add_dmdsec
return meth(
self.serialize_md_inst(md, self.premis_object_class),
self.PREMIS_OBJECT,
mode,
)
[docs] def add_premis_event(self, md, mode="mdwrap"):
return self.add_digiprovmd(
self.serialize_md_inst(md, self.premis_event_class), self.PREMIS_EVENT, mode
)
[docs] def add_premis_agent(self, md, mode="mdwrap"):
return self.add_digiprovmd(
self.serialize_md_inst(md, self.premis_agent_class), self.PREMIS_AGENT, mode
)
[docs] def add_premis_rights(self, md, mode="mdwrap"):
return self.add_rightsmd(
self.serialize_md_inst(md, self.premis_rights_class),
self.PREMIS_RIGHTS,
mode,
)
[docs] def add_dublin_core(self, md, mode="mdwrap", **kwargs):
# TODO create DC object here
return self.add_dmdsec(md, "DC", mode, **kwargs)
[docs] def add_child(self, child):
"""Add a child FSEntry to this FSEntry.
Only FSEntrys with a type of 'directory' can have children.
This does not detect cyclic parent/child relationships, but that will
cause problems.
:param metsrw.fsentry.FSEntry child: FSEntry to add as a child
:return: The newly added child
:raises ValueError: If this FSEntry cannot have children.
:raises ValueError: If the child and the parent are the same
"""
if self.type.lower() != "directory":
raise ValueError("Only directory objects can have children")
if child is self:
raise ValueError("Cannot be a child of itself!")
if child not in self._children:
self._children.append(child)
child.parent = self
return child
[docs] def remove_child(self, child):
"""
Remove a child from this FSEntry
If `child` is not actually a child of this entry, nothing happens.
:param child: Child to remove
"""
try:
self._children.remove(child)
except ValueError: # Child may not be in list
pass
else:
child.parent = None
# SERIALIZE
[docs] def serialize_filesec(self):
"""
Return the file Element for this file, appropriate for use in a fileSec.
If this is not an Item or has no use, return None.
:return: fileSec element for this FSEntry
"""
if (
self.type.lower() not in ("item", "archival information package")
or self.use is None
):
return None
el = etree.Element(utils.lxmlns("mets") + "file", ID=self.file_id())
if self.group_id():
el.attrib["GROUPID"] = self.group_id()
if self.admids:
el.set("ADMID", " ".join(self.admids))
if self.dmdids and self.use == "original":
el.set("DMDID", " ".join(self.dmdids))
if self.checksum and self.checksumtype:
el.attrib["CHECKSUM"] = self.checksum
el.attrib["CHECKSUMTYPE"] = self.checksumtype
if self.path:
flocat = etree.SubElement(el, utils.lxmlns("mets") + "FLocat")
# Setting manually so order is correct
try:
flocat.set(utils.lxmlns("xlink") + "href", utils.urlencode(self.path))
except ValueError:
raise exceptions.SerializeError(
'Value "{}" (for attribute xlink:href) is not a valid'
" URL.".format(self.path)
)
flocat.set("LOCTYPE", "OTHER")
flocat.set("OTHERLOCTYPE", "SYSTEM")
for transform_file in self.transform_files:
transform_file_el = etree.SubElement(
el, utils.lxmlns("mets") + "transformFile"
)
for key, val in transform_file.items():
attribute = f"transform{key}".upper()
transform_file_el.attrib[attribute] = str(val)
return el
@property
def is_empty_dir(self):
"""Returns ``True`` if this fs item is a directory with no children or
a directory with only other empty directories as children.
"""
if self.mets_div_type == "Directory":
children = self._children
if children:
if all(child.is_empty_dir for child in children):
return True
else:
return False
else:
return True
else:
return False
[docs] def serialize_structmap(self, recurse=True, normative=False):
"""Return the div Element for this file, appropriate for use in a
structMap.
If this FSEntry represents a directory, its children will be
recursively appended to itself. If this FSEntry represents a file, it
will contain a <fptr> element.
:param bool recurse: If true, serialize and apppend all children.
Otherwise, only serialize this element but not any children.
:param bool normative: If true, we are creating a "Normative Directory
Structure" logical structmap, in which case we add div elements for
empty directories and do not add fptr elements for files.
:return: structMap element for this FSEntry
"""
if not self.label:
return None
# Empty directories are not included in the physical structmap.
if self.is_empty_dir and not normative:
return None
el = etree.Element(utils.lxmlns("mets") + "div", TYPE=self.mets_div_type)
el.attrib["LABEL"] = self.label
if (not normative) and self.file_id():
etree.SubElement(el, utils.lxmlns("mets") + "fptr", FILEID=self.file_id())
if self.dmdids:
el.set("DMDID", " ".join(self.dmdids))
if self.mets_div_type.lower() == "directory" and self.admids:
el.set("ADMID", " ".join(self.admids))
if recurse and self._children:
for child in self._children:
child_el = child.serialize_structmap(
recurse=recurse, normative=normative
)
if child_el is not None:
el.append(child_el)
return el
[docs] def get_subsections_of_type(self, mdtype, md_class):
try:
return [
md_class.fromtree(ss.contents.document)
for ss in self.amdsecs[0].subsections
if ss.contents.mdtype == mdtype
]
except IndexError:
return []
[docs] def get_premis_objects(self):
return self.get_subsections_of_type(
self.PREMIS_OBJECT, self.premis_object_class
)
[docs] def get_premis_agents(self):
return self.get_subsections_of_type(self.PREMIS_AGENT, self.premis_agent_class)
[docs] def get_premis_events(self):
return self.get_subsections_of_type(self.PREMIS_EVENT, self.premis_event_class)
[docs] def get_premis_event(self, event_uuid):
for evt in self.get_premis_events():
if evt.identifier_value == event_uuid:
return evt
return None
[docs] def get_premis_rights(self):
return self.get_subsections_of_type(
self.PREMIS_RIGHTS, self.premis_rights_class
)
[docs] def get_premis_rights_statement(self, rights_statement_uuid):
for rights in self.get_premis_rights():
for statement in rights.rights_statement:
if statement.rights_statement_identifier_value == rights_statement_uuid:
return statement
return None
[docs] def get_path(self):
"""Return the relative path to this FSEntry.
If the path is not set, it's generated from the ancestor labels. Raises an
AttributeError if the path cannot be generated. Returns None for the top
level FSEntry.
"""
if self.path:
return self.path
if not self.label:
raise AttributeError("The path cannot be generated")
if self.parent:
if self.parent.parent:
return self.parent.get_path() + os.sep + self.label
return self.label