Source code for MyCapytain.common.metadata

# -*- coding: utf-8 -*-
"""
.. module:: MyCapytain.common.metadata
   :synopsis: Metadata related objects

.. moduleauthor:: Thibault Clérice <leponteineptique@gmail.com>


"""
from MyCapytain.common.utils.xml import make_xml_node
from MyCapytain.common.constants import Mimetypes, get_graph
from MyCapytain.common.base import Exportable
from rdflib import BNode, Literal, Graph, URIRef, term
from typing import Union, Optional


__all__ = [
    "Metadata"
]


[docs]class Metadata(Exportable): """ A metadatum aggregation object provided to centralize metadata :param keys: A metadata field names list :type keys: [text_type] :cvar EXPORT_TO: List of exportable supported formats :cvar DEFAULT_EXPORT: Default export (CTS XML Inventory) :cvar STORE: RDF Store """ EXPORT_TO = [Mimetypes.JSON.Std, Mimetypes.XML.RDF, Mimetypes.XML.RDFa, Mimetypes.JSON.LD, Mimetypes.XML.CapiTainS.CTS] DEFAULT_EXPORT = Mimetypes.JSON.Std def __init__(self, node=None, *args, **kwargs): super(Metadata, self).__init__(*args, **kwargs) self.__graph__ = get_graph() if node is not None: self.__node__ = node else: self.__node__ = BNode() def asNode(self): return self.__node__ @property def graph(self): """ Quick access to the graph this node is connected to :rtype: Graph """ return self.__graph__
[docs] def set(self, key: URIRef, value: Union[Literal, BNode, URIRef, str, int], lang: Optional[str]=None): """ Set the VALUE for KEY predicate in the Metadata Graph :param key: Predicate to be set (eg. DCT.creator) :param value: Value to be stored (eg. "Cicero") :param lang: [Optional] Language of the value (eg. "la") """ if not isinstance(value, Literal) and lang is not None: value = Literal(value, lang=lang) elif not isinstance(value, (BNode, URIRef)): value = Literal(value) self.graph.set((self.asNode(), key, value))
[docs] def add(self, key, value, lang=None): """ Add a triple to the graph related to this node :param key: Predicate of the triple :param value: Object of the triple :param lang: Language of the triple if applicable """ if not isinstance(value, Literal) and lang is not None: value = Literal(value, lang=lang) elif not isinstance(value, (BNode, URIRef)): value = Literal(value) self.graph.add((self.asNode(), key, value))
[docs] def get(self, key, lang=None): """ Returns triple related to this node. Can filter on lang :param key: Predicate of the triple :param lang: Language of the triple if applicable :rtype: Literal or BNode or URIRef """ if lang is not None: for o in self.graph.objects(self.asNode(), key): if o.language == lang: yield o else: for o in self.graph.objects(self.asNode(), key): yield o
[docs] def get_single(self, key, lang=None): """ Returns a single triple related to this node. :param key: Predicate of the triple :param lang: Language of the triple if applicable :rtype: Literal or BNode or URIRef """ if not isinstance(key, URIRef): key = URIRef(key) if lang is not None: default = None for o in self.graph.objects(self.asNode(), key): default = o if o.language == lang: return o return default else: for o in self.graph.objects(self.asNode(), key): return o
def __getitem__(self, item): """ Quick access method. If :param item: Identifier :return: List or Literal or BNode or URIRef .. example:: Metadata[Title, lang] == Metadata.get(Title, lang=lang) Metadata[Title] == [Metadata.get(Title, lang=lang1), Metadata.get(Title, lang=lang2)] """ if isinstance(item, tuple): return self.get_single(item[0], item[1]) return list(self.get(item))
[docs] def remove(self, predicate=None, obj=None): """ Remove triple matching the predicate or the object :param predicate: Predicate to match, None to match all :param obj: Object to match, None to match all """ self.graph.remove((self.asNode(), predicate, obj))
[docs] def predicate_object(self, predicate=None, obj=None): """ Retrieve predicate and object around this object :param predicate: Predicate to match, None to match all :param obj: Object to match, None to match all :return: List of resources """
def __export__(self, output=Mimetypes.JSON.Std, only=None, exclude=None, **kwargs): """ Export a set of Metadata :param output: Mimetype to export to :param only: Includes only term from given namespaces :param exclude: Includes only term from given namespaces :return: Formatted Export .. warning:: exclude and warning cannot be used together """ graph = Graph() graph.namespace_manager = get_graph().namespace_manager if only is not None: _only = only only = [str(s) for s in only] for predicate in set(self.graph.predicates(subject=self.asNode())): if str(predicate) not in only: prefix, namespace, name = self.graph.compute_qname(predicate) if str(namespace) in only: _only.append(predicate) for predicate, obj in self.graph[self.asNode()]: if predicate in _only: graph.add((self.asNode(), predicate, obj)) elif exclude is not None: _only = [] exclude = [str(s) for s in exclude] for predicate in set(self.graph.predicates(subject=self.asNode())): prefix, namespace, name = self.graph.compute_qname(predicate) if str(predicate) not in exclude and not str(namespace) in exclude: _only.append(predicate) for predicate, obj in self.graph[self.asNode()]: if predicate in _only: graph.add((self.asNode(), predicate, obj)) else: for predicate, object in self.graph[self.asNode()]: graph.add((self.asNode(), predicate, object)) if output == Mimetypes.JSON.Std: out = {} for _, predicate, object in graph: predicate = str(predicate) if predicate not in out: out[predicate] = {} if isinstance(object, Literal): if object.language in out[predicate]: if isinstance(out[predicate][object.language], str): out[predicate][object.language] = [out[predicate][object.language]] out[predicate][object.language].append(object.title()) else: out[predicate][object.language] = object.title() del graph return out elif output == Mimetypes.JSON.LD: out = graph.serialize(format="json-ld", context={}) del graph return out elif output == Mimetypes.XML.RDF: out = graph.serialize(format="xml") del graph return out elif output == Mimetypes.XML.CapiTainS.CTS: strings = [] for pred, obj in graph.predicate_objects(self.asNode()): kwargs = {} if hasattr(obj, "language") and obj.language is not None: kwargs["xml:lang"] = obj.language if hasattr(obj, "datatype") and obj.datatype is not None: kwargs["rdf:type"] = obj.datatype strings.append(make_xml_node(graph, pred, text=obj, attributes=kwargs, complete=True)) del graph return "\n".join(strings)
[docs] @staticmethod def getOr(subject, predicate, *args, **kwargs): """ Retrieve a metadata node or generate a new one :param subject: Subject to which the metadata node should be connected :param predicate: Predicate by which the metadata node should be connected :return: Metadata for given node :rtype: Metadata """ if (subject, predicate, None) in get_graph(): return Metadata(node=get_graph().objects(subject, predicate).__next__()) return Metadata(*args, **kwargs)