text: from __future__ import (absolute_import, division, print_function,
                          unicode_literals)
  
  __author__ = 'Lion Krischer'
  __email__ = 'krischer@geophysik.uni-muenchen.de'
  
  import datetime
  import logging
  from lxml import etree
  import io
  import warnings
  import six
  
  logger = logging.getLogger(__name__)
  
  import prov
  import prov.identifier
  from prov.model import DEFAULT_NAMESPACES, sorted_attributes
  from prov.constants import *  # NOQA
  
  
  # Create a dictionary containing all top-level PROV XML elements for an easy
  # mapping.
  FULL_NAMES_MAP = dict(PROV_N_MAP)
  FULL_NAMES_MAP.update(ADDITIONAL_N_MAP)
  # Inverse mapping.
  FULL_PROV_RECORD_IDS_MAP = dict((FULL_NAMES_MAP[rec_type_id], rec_type_id) for
                                  rec_type_id in FULL_NAMES_MAP)
  
  XML_XSD_URI = 'http://www.w3.org/2001/XMLSchema'
  
  
  class ProvXMLException(prov.Error):
      pass
  
  
  class ProvXMLSerializer(prov.serializers.Serializer):
      """PROV-XML serializer for :class:`~prov.model.ProvDocument`
      """
      def serialize(self, stream, force_types=False, **kwargs):
          """
          Serializes a :class:`~prov.model.ProvDocument` instance to `PROV-XML
          <http://www.w3.org/TR/prov-xml/>`_.
  
          :param stream: Where to save the output.
          :type force_types: boolean, optional
          :param force_types: Will force xsd:types to be written for most
              attributes mainly PROV-"attributes", e.g. tags not in the
              PROV namespace. Off by default meaning xsd:type attributes will
              only be set for prov:type, prov:location, and prov:value as is
              done in the official PROV-XML specification. Furthermore the
              types will always be set if the Python type requires it. False
              is a good default and it should rarely require changing.
          """
          xml_root = self.serialize_bundle(bundle=self.document,
                                           force_types=force_types)
          for bundle in self.document.bundles:
              self.serialize_bundle(bundle=bundle, element=xml_root,
                                    force_types=force_types)
          # No encoding must be specified when writing to String object which
          # does not have the concept of an encoding as it should already
          # represent unicode code points.
          et = etree.ElementTree(xml_root)
          if isinstance(stream, io.TextIOBase):
              stream.write(etree.tostring(et, xml_declaration=True,
                                          pretty_print=True).decode('utf-8'))
          else:
              et.write(stream, pretty_print=True, xml_declaration=True,
                       encoding="UTF-8")
  
      def serialize_bundle(self, bundle, element=None, force_types=False):
          """
          Serializes a bundle or document to PROV XML.
  
          :param bundle: The bundle or document.
          :param element: The XML element to write to. Will be created if None.
          :type force_types: boolean, optional
          :param force_types: Will force xsd:types to be written for most
              attributes mainly PROV-"attributes", e.g. tags not in the
              PROV namespace. Off by default meaning xsd:type attributes will
              only be set for prov:type, prov:location, and prov:value as is
              done in the official PROV-XML specification. Furthermore the
              types will always be set if the Python type requires it. False
              is a good default and it should rarely require changing.
          """
          # Build the namespace map for lxml and attach it to the root XML
          # element. No dictionary comprehension in Python 2.6!
          nsmap = dict((ns.prefix, ns.uri) for ns in
                       self.document._namespaces.get_registered_namespaces())
          if self.document._namespaces._default:
              nsmap[None] = self.document._namespaces._default.uri
          for namespace in bundle.namespaces:
              if namespace not in nsmap:
                  nsmap[namespace.prefix] = namespace.uri
  
          for key, value in DEFAULT_NAMESPACES.items():
              uri = value.uri
              if value.prefix == "xsd":
                  # The XSD namespace for some reason has no hash at the end
                  # for PROV XML, but for all other serializations it does.
                  uri = uri.rstrip("#")
              nsmap[value.prefix] = uri
  
          if element is not None:
              xml_bundle_root = etree.SubElement(
                  element, _ns_prov("bundleContent"), nsmap=nsmap)
          else:
              xml_bundle_root = etree.Element(_ns_prov("document"), nsmap=nsmap)
  
          if bundle.identifier:
              xml_bundle_root.attrib[_ns_prov("id")] = \
                  six.text_type(bundle.identifier)
  
          for record in bundle._records:
              rec_type = record.get_type()
              identifier = six.text_type(record._identifier) \
                  if record._identifier else None
  
              if identifier:
                  attrs = {_ns_prov("id"): identifier}
              else:
                  attrs = None
  
              # Derive the record label from its attributes which is sometimes
              # needed.
              attributes = list(record.attributes)
              rec_label = self._derive_record_label(rec_type, attributes)
  
              elem = etree.SubElement(xml_bundle_root,
                                      _ns_prov(rec_label), attrs)
  
              for attr, value in sorted_attributes(rec_type, attributes):
                  subelem = etree.SubElement(
                      elem, _ns(attr.namespace.uri, attr.localpart))
                  if isinstance(value, prov.model.Literal):
                      if value.datatype not in \
                              [None, PROV["InternationalizedString"]]:
                          subelem.attrib[_ns_xsi("type")] = "%s:%s" % (
                              value.datatype.namespace.prefix,
                              value.datatype.localpart)
                      if value.langtag is not None:
                          subelem.attrib[_ns_xml("lang")] = value.langtag
                      v = value.value
                  elif isinstance(value, prov.model.QualifiedName):
                      if attr not in PROV_ATTRIBUTE_QNAMES:
                          subelem.attrib[_ns_xsi("type")] = "xsd:QName"
                      v = six.text_type(value)
                  elif isinstance(value, datetime.datetime):
                      v = value.isoformat()
                  else:
                      v = six.text_type(value)
  
                  # xsd type inference.
                  #
                  # This is a bit messy and there are all kinds of special
                  # rules but it appears to get the job done.
                  #
                  # If it is a type element and does not yet have an
                  # associated xsi type, try to infer it from the value.
                  # The not startswith("prov:") check is a little bit hacky to
                  # avoid type interference when the type is a standard prov
                  # type.
                  #
                  # To enable a mapping of Python types to XML and back,
                  # the XSD type must be written for these types.
                  ALWAYS_CHECK = [bool, datetime.datetime, float,
                                  prov.identifier.Identifier]
                  # Add long and int on Python 2, only int on Python 3.
                  ALWAYS_CHECK.extend(six.integer_types)
                  ALWAYS_CHECK = tuple(ALWAYS_CHECK)
                  if (force_types or
                          type(value) in ALWAYS_CHECK or
                          attr in [PROV_TYPE, PROV_LOCATION, PROV_VALUE]) and \
                          _ns_xsi("type") not in subelem.attrib and \
                          not six.text_type(value).startswith("prov:") and \
                          not (attr in PROV_ATTRIBUTE_QNAMES and v) and \
                          attr not in [PROV_ATTR_TIME, PROV_LABEL]:
                      xsd_type = None
                      if isinstance(value, bool):
                          xsd_type = XSD_BOOLEAN
                          v = v.lower()
                      elif isinstance(value, six.string_types):
                          xsd_type = XSD_STRING
                      elif isinstance(value, float):
                          xsd_type = XSD_DOUBLE
                      elif isinstance(value, six.integer_types):
                          xsd_type = XSD_INT
                      elif isinstance(value, datetime.datetime):
                          # Exception of the exception, while technically
                          # still correct, do not write XSD dateTime type for
                          # attributes in the PROV namespaces as the type is
                          # already declared in the XSD and PROV XML also does
                          # not specify it in the docs.
                          if attr.namespace.prefix != "prov" \
                                  or "time" not in attr.localpart.lower():
                              xsd_type = XSD_DATETIME
                      elif isinstance(value, prov.identifier.Identifier):
                          xsd_type = XSD_ANYURI
  
                      if xsd_type is not None:
                          subelem.attrib[_ns_xsi("type")] = \
                              six.text_type(xsd_type)
  
                  if attr in PROV_ATTRIBUTE_QNAMES and v:
                      subelem.attrib[_ns_prov("ref")] = v
                  else:
                      subelem.text = v
          return xml_bundle_root
  
      def deserialize(self, stream, **kwargs):
          """
          Deserialize from `PROV-XML <http://www.w3.org/TR/prov-xml/>`_
          representation to a :class:`~prov.model.ProvDocument` instance.
  
          :param stream: Input data.
          """
          if isinstance(stream, io.TextIOBase):
              with io.BytesIO() as buf:
                  buf.write(stream.read().encode('utf-8'))
                  buf.seek(0, 0)
                  xml_doc = etree.parse(buf).getroot()
          else:
              xml_doc = etree.parse(stream).getroot()
  
          # Remove all comments.
          for c in xml_doc.xpath("//comment()"):
              p = c.getparent()
              p.remove(c)
  
          document = prov.model.ProvDocument()
          self.deserialize_subtree(xml_doc, document)
          return document
  
      def deserialize_subtree(self, xml_doc, bundle):
          """
          Deserialize an etree element containing a PROV document or a bundle
          and write it to the provided internal object.
  
          :param xml_doc: An etree element containing the information to read.
          :param bundle: The bundle object to write to.
          """
  
          for element in xml_doc:
              qname = etree.QName(element)
              if qname.namespace != DEFAULT_NAMESPACES["prov"].uri:
                  raise ProvXMLException("Non PROV element discovered in "
                                         "document or bundle.")
              # Ignore the <prov:other> element storing non-PROV information.
              if qname.localname == "other":
                  warnings.warn(
                      "Document contains non-PROV information in "
                      "<prov:other>. It will be ignored in this package.",
                      UserWarning)
                  continue
  
              id_tag = _ns_prov("id")
              rec_id = element.attrib[id_tag] if id_tag in element.attrib \
                  else None
  
              if rec_id is not None:
                  # Try to make a qualified name out of it!
                  rec_id = xml_qname_to_QualifiedName(element, rec_id)
  
              # Recursively read bundles.
              if qname.localname == "bundleContent":
                  b = bundle.bundle(identifier=rec_id)
                  self.deserialize_subtree(element, b)
                  continue
  
              attributes = _extract_attributes(element)
  
              # Map the record type to its base type.
              q_prov_name = FULL_PROV_RECORD_IDS_MAP[qname.localname]
              rec_type = PROV_BASE_CLS[q_prov_name]
  
              if _ns_xsi("type") in element.attrib:
                  value = xml_qname_to_QualifiedName(
                      element, element.attrib[_ns_xsi("type")]
                  )
                  attributes.append((PROV["type"], value))
  
              rec = bundle.new_record(rec_type, rec_id, attributes)
  
              # Add the actual type in case a base type has been used.
              if rec_type != q_prov_name:
                  rec.add_asserted_type(q_prov_name)
          return bundle
  
      def _derive_record_label(self, rec_type, attributes):
          """
          Helper function trying to derive the record label taking care of
          subtypes and what not. It will also remove the type declaration for
          the attributes if it was used to specialize the type.
  
          :param rec_type: The type of records.
          :param attributes: The attributes of the record.
          """
          rec_label = FULL_NAMES_MAP[rec_type]
  
          for key, value in list(attributes):
              if key != PROV_TYPE:
                  continue
              if isinstance(value, prov.model.Literal):
                  value = value.value
              if value in PROV_BASE_CLS and PROV_BASE_CLS[value] != value:
                  attributes.remove((key, value))
                  rec_label = FULL_NAMES_MAP[value]
                  break
          return rec_label
  
  
  def _extract_attributes(element):
      """
      Extract the PROV attributes from an etree element.
  
      :param element: The lxml.etree.Element instance.
      """
      attributes = []
      for subel in element:
          sqname = etree.QName(subel)
          _t = xml_qname_to_QualifiedName(
              subel, "%s:%s" % (subel.prefix, sqname.localname)
          )
  
          for key, value in subel.attrib.items():
              if key == _ns_xsi("type"):
                  datatype = xml_qname_to_QualifiedName(subel, value)
                  if datatype == XSD_QNAME:
                      _v = xml_qname_to_QualifiedName(subel, subel.text)
                  else:
                      _v = prov.model.Literal(subel.text, datatype)
              elif key == _ns_prov("ref"):
                  _v = xml_qname_to_QualifiedName(subel, value)
              elif key == _ns_xml("lang"):
                  _v = prov.model.Literal(subel.text, langtag=value)
              else:
                  warnings.warn(
                      "The element '%s' contains an attribute %s='%s' "
                      "which is not representable in the prov module's "
                      "internal data model and will thus be ignored." %
                      (_t, six.text_type(key), six.text_type(value)),
                      UserWarning)
  
          if not subel.attrib:
              _v = subel.text
  
          attributes.append((_t, _v))
  
      return attributes
  
  
  def xml_qname_to_QualifiedName(element, qname_str):
      if ':' in qname_str:
          prefix, localpart = qname_str.split(':', 1)
          if prefix in element.nsmap:
              ns_uri = element.nsmap[prefix]
              if ns_uri == XML_XSD_URI:
                  ns = XSD  # use the standard xsd namespace (i.e. with #)
              elif ns_uri == PROV.uri:
                  ns = PROV
              else:
                  ns = Namespace(prefix, ns_uri)
              return ns[localpart]
      # case 1: no colon
      # case 2: unknown prefix
      if None in element.nsmap:
          ns_uri = element.nsmap[None]
          ns = Namespace('', ns_uri)
          return ns[qname_str]
      # no default namespace
      raise ProvXMLException(
          'Could not create a valid QualifiedName for "%s"' % qname_str
      )
  
  def _ns(ns, tag):
      return "{%s}%s" % (ns, tag)
  
  
  def _ns_prov(tag):
      return _ns(DEFAULT_NAMESPACES['prov'].uri, tag)
  
  
  def _ns_xsi(tag):
      return _ns(DEFAULT_NAMESPACES['xsi'].uri, tag)
  
  
  def _ns_xml(tag):
      NS_XML = "http://www.w3.org/XML/1998/namespace"
      return _ns(NS_XML, tag)
  