""" Part of the bsie module. A copy of the license is provided with the project. Author: Matthias Baumgartner, 2022 """ # standard imports import os import typing # bsie imports from bsie.extractor import base from bsie.utils import bsfs, node, ns # exports __all__: typing.Sequence[str] = ( 'Path', ) ## code ## class Path(base.Extractor): """Extract information from file's path.""" CONTENT_READER = 'bsie.reader.path.Path' # mapping from predicate to handler function. _callmap: typing.Dict[bsfs.schema.Predicate, typing.Callable[[str], typing.Any]] def __init__(self): super().__init__(bsfs.schema.from_string(base.SCHEMA_PREAMBLE + ''' bse:filename rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:File ; rdfs:range xsd:string ; rdfs:label "File name"^^xsd:string ; schema:description "Filename of entity in some filesystem."^^xsd:string ; bsfs:unique "false"^^xsd:boolean . ''')) self._callmap = { self.schema.predicate(ns.bse.filename): self.__filename, } def extract( self, subject: node.Node, content: str, principals: typing.Iterable[bsfs.schema.Predicate], ) -> typing.Iterator[typing.Tuple[node.Node, bsfs.schema.Predicate, typing.Any]]: for pred in principals: # find callback clbk = self._callmap.get(pred) if clbk is None: continue # get value value = clbk(content) if value is None: continue # produce triple yield subject, pred, value def __filename(self, path: str) -> typing.Optional[str]: try: return os.path.basename(path) except Exception: # pylint: disable=broad-except # we explicitly want to catch everything # some error, skip # FIXME: some kind of error reporting (e.g. logging)? # Options: (a) Fail silently (current); (b) Skip and report to log; # (c) Raise ExtractorError (aborts extraction); (d) separate content type # checks from basename errors (report content type errors, skip basename # errors) return None ## EOF ##