profile_generation.py

# copyright 2016-2021 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr -- mailto:contact@logilab.fr
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""cubicweb-seda adapter classes for profile (schema) generation"""

from collections import defaultdict, namedtuple
from functools import partial
from itertools import chain

from lxml import etree
from pyxst.xml_struct import graph_nodes

from logilab.common import attrdict

from yams import BASE_TYPES

from cubicweb.predicates import is_instance
from cubicweb.view import EntityAdapter

from ..xsd import XSDM_MAPPING, JUMP_ELEMENTS
from ..xsd2yams import SKIP_ATTRS
from . import simplified_profile, wrap_dataobjects


JUMPED_OPTIONAL_ELEMENTS = set(
    ("DataObjectPackage", "FileInfo", "PhysicalDimensions", "Coverage")
)


def substitute_xml_prefix(prefix_name, namespaces):
    """Given an XML prefixed name in the form `'ns:name'`, return the string `'{<ns_uri>}name'`
    where `<ns_uri>` is the URI for the namespace prefix found in `namespaces`.

    This new string is then suitable to build an LXML etree.Element object.

    Example::

      >>> substitude_xml_prefix('xlink:href', {'xlink': 'http://wwww.w3.org/1999/xlink'})
      '{http://www.w3.org/1999/xlink}href'

    """
    try:
        prefix, name = prefix_name.split(":", 1)
    except ValueError:
        return prefix_name
    assert prefix in namespaces, "Unknown namespace prefix: {0}".format(prefix)
    return "{{{0}}}".format(namespaces[prefix]) + name


def content_types(content_type):
    """Return an ordered tuple of content types from pyxst `textual_content_type` that may be None, a
    set or a string value.
    """
    if content_type:
        if isinstance(content_type, set):
            content_types = sorted(content_type)
        else:
            if content_type == "IDREF":
                content_type = "NCName"
            content_types = (content_type,)
    else:
        content_types = ()
    return content_types


def _internal_reference(value):
    """Return True if the given value is a reference to an entity within the profile."""
    return getattr(value, "cw_etype", None) in (
        "SEDAArchiveUnit",
        "SEDABinaryDataObject",
        "SEDAPhysicalDataObject",
    )


def _concept_value(concept, language):
    """Return string value to be inserted in a SEDA export for the given concept.

    * `concept` may be None, in which case None will be returned

    * `language` is the language matching the exported format (one of 'seda-2', 'seda-1' or
      'seda-02')
    """
    assert language in ("seda-2", "seda-1", "seda-02")
    if concept is None:
        return None
    for code in (language, "seda", "en", "fr"):
        try:
            return concept.labels[code]
        except KeyError:
            continue
    return concept.label()


def xmlid(entity):
    """Return a value usable as ID/IDREF for the given entity."""
    return entity.cw_adapt_to("IXmlId").id()


def serialize(value, build_url):
    """Return typed `value` as an XSD string."""
    if value is None:
        return None
    if hasattr(value, "eid"):
        if value.cw_etype == "ConceptScheme":
            return build_url(value)
        if value.cw_etype == "Concept":
            return _concept_value(value, "seda-2")
        if _internal_reference(value):
            return xmlid(value)
        return None  # intermediary entity
    if isinstance(value, bool):
        return "true" if value else "false"
    assert isinstance(value, str), repr(value)
    return value


def integrity_cardinality(data_object):
    minvalue, maxvalue = minmax_cardinality(data_object.user_cardinality)
    itree = data_object.cw_adapt_to("ITreeBase")
    for parent in itree.iterancestors():
        try:
            parent_cardinality = parent.user_cardinality
        except AttributeError:
            continue
        minc, maxc = minmax_cardinality(parent_cardinality)
        minvalue = min(minc, minvalue)
        if maxc == graph_nodes.INFINITY or maxvalue == graph_nodes.INFINITY:
            maxvalue = graph_nodes.INFINITY
        else:
            maxvalue = max(maxc, maxvalue)
    if maxvalue == graph_nodes.INFINITY:
        maxvalue = "n"
    if minvalue == maxvalue == 1:
        return "1"
    return "{}..{}".format(minvalue, maxvalue)


def minmax_cardinality(string_cardinality, _allowed=("0..1", "0..n", "1", "1..n")):
    """Return (minimum, maximum) cardinality for the cardinality as string (one of '0..1', '0..n',
    '1' or '1..n').
    """
    assert string_cardinality in _allowed, "%s not allowed %s" % (
        string_cardinality,
        _allowed,
    )
    if string_cardinality[0] == "0":
        minimum = 0
    else:
        minimum = 1
    if string_cardinality[-1] == "n":
        maximum = graph_nodes.INFINITY
    else:
        maximum = 1
    return minimum, maximum


def element_minmax_cardinality(occ, card_entity):
    """Return (minimum, maximum) cardinality for the given pyxst Occurence and entity.

    Occurence 's cardinality may be overriden by the entity's user_cardinality value.
    """
    cardinality = getattr(card_entity, "user_cardinality", None)
    if cardinality is None:
        return occ.minimum, occ.maximum
    else:
        return minmax_cardinality(cardinality)


def attribute_minimum_cardinality(occ, card_entity):
    """Return 0 or 1 for the given pyxst attribute's Occurence. Cardinality may be overriden by
    the data model's user_cardinality value.
    """
    cardinality = getattr(card_entity, "user_cardinality", None)
    if cardinality is None:
        return occ.minimum
    else:
        return minmax_cardinality(cardinality, ("0..1", "1"))[0]


def iter_path_children(xselement, entity):
    """Return an iterator on `entity` children entities according to `xselement` definition.

    (`path`, `target`) is returned with `path` the path definition leading to the target, and
    `target` either a final value in case of attributes or a list of entities.
    """
    for rtype, role, _path in XSDM_MAPPING.iter_rtype_role(xselement.local_name):
        if _path[0][2] in BASE_TYPES:
            # entity attribute
            if getattr(entity, rtype) is not None:
                yield _path, getattr(entity, rtype)
        else:
            related = entity.related(rtype, role, entities=True)
            if related:
                yield _path, related


class RNGMixin(object):
    """Mixin class providing some Relax NG schema generation helper methods."""

    def rng_element_parent(self, parent, minimum, maximum=1):
        """Given a etree node and minimum/maximum cardinalities of a desired child element,
        return suitable parent node for it.

        This will be one of rng:optional, rng:zeroOrMore or rng:oneOrMore that will be created by
        this method or the given parent itself if minimum == maximum == 1.
        """
        if minimum == 1 and maximum == 1:
            return parent
        elif minimum == 0 and maximum == 1:
            return self.element("rng:optional", parent)
        elif minimum == 0 and maximum == graph_nodes.INFINITY:
            return self.element("rng:zeroOrMore", parent)
        elif minimum == 1 and maximum == graph_nodes.INFINITY:
            return self.element("rng:oneOrMore", parent)
        else:
            assert False, ("unexpected min/max cardinality:", minimum, maximum)

    def rng_attribute_parent(self, parent, minimum):
        """Given a etree node and minimum cardinality of a desired attribute,
        return suitable parent node for it.

        This will be rng:optional that will be created by this method or the given parent itself if
        minimum == 1.
        """
        if minimum == 1:
            return parent
        else:
            return self.element("rng:optional", parent)

    def rng_value(
        self, element, qualified_datatype, fixed_value=None, default_value=None
    ):
        """Given a (etree) schema element, a data type (e.g. 'xsd:token') and an optional fixed
        value, add RNG declaration to the element to declare the datatype and fix the value if
        necessary.
        """
        prefix, datatype = qualified_datatype.split(":")
        if prefix != "xsd":
            # XXX RelaxNG compatible version of custom types? this would allow
            # `type_attrs['datatypeLibrary'] = self.namespaces[prefix]`. In the mean time, turn
            # every custom type to string, supposing transfer are also checked against the original
            # schema (as agape v1 was doing).
            datatype = "string"
        type_attrs = {"type": datatype}
        if fixed_value is not None:
            if isinstance(fixed_value, (tuple, list)):
                choice = self.element("rng:choice", element)
                for value in fixed_value:
                    self.element("rng:value", choice, type_attrs, text=value)
            else:
                self.element("rng:value", element, type_attrs, text=fixed_value)
        elif default_value is not None:
            element.attrib[self.qname("a:defaultValue")] = default_value
            self.element("rng:data", element, type_attrs)
        else:
            self.element("rng:data", element, type_attrs)


class SEDA2ExportAdapter(EntityAdapter):
    """Abstract base class for export of SEDA profile."""

    __abstract__ = True
    __select__ = is_instance("SEDAArchiveTransfer")
    encoding = "utf-8"
    content_type = "application/xml"
    # to be defined in concret implementations
    namespaces = {}
    _root_attributes = {}

    @property
    def root_attributes(self):
        if self.entity.compat_list is None:
            # uncommited transfer may occurs during tests
            return self._root_attributes
        diag = "" if "RNG" in self.entity.compat_list else "rng-ambiguous"
        attributes = {"seda:warnings": diag}
        attributes.update(self._root_attributes)
        return attributes

    def dump(self, _encoding=None):
        """Return an schema string for the adapted SEDA profile

        _encoding will be used as "encoding" argument of lxml's tostring, in
        order to retrieve a unicode string. This is useful for tests.
        """
        root = self.dump_etree()
        kwargs = {}
        if _encoding is None:
            # We only want the XML declaration at all if _encoding is not specified.
            kwargs["standalone"] = False
        kwargs["encoding"] = _encoding or self.encoding
        return etree.tostring(root, pretty_print=True, **kwargs)

    def dump_etree(self):
        """Return an XSD etree for the adapted SEDA profile."""
        raise NotImplementedError()

    def qname(self, tag):
        return substitute_xml_prefix(tag, self.namespaces)

    def element(self, tag, parent=None, attributes=None, text=None):
        """Generic method to build a XSD element tag.

        Params:

        * `tag`, tag name of the element

        * `parent`, the parent etree node

        * `attributes`, dictionary of attributes - may contain a special 'documentation' attribute
          that will be added in a xsd:annotation node

        * `text`, textual content of the tag if any
        """
        attributes = attributes or {}
        tag = self.qname(tag)
        documentation = attributes.pop("documentation", None)
        for attr, value in list(attributes.items()):
            newattr = substitute_xml_prefix(attr, self.namespaces)
            attributes[newattr] = value
            if newattr != attr:
                attributes.pop(attr)
        if parent is None:
            elt = etree.Element(tag, attributes, nsmap=self.namespaces)
        else:
            elt = etree.SubElement(parent, tag, attributes)
        if text is not None:
            elt.text = text
        if documentation:
            annot = self.element("xsd:annotation", elt)
            self.element("xsd:documentation", annot).text = documentation
        return elt

    def dispatch_occ(self, profile_element, occ, target_value, to_process, card_entity):
        callback = getattr(self, "element_" + occ.target.__class__.__name__.lower())
        callback(occ, profile_element, target_value, to_process, card_entity)

    def _dump(self, root):
        entity = self.entity
        xselement = XSDM_MAPPING.root_xselement
        transfer_element = self.init_transfer_element(xselement, root, entity)
        to_process = defaultdict(list)
        to_process[xselement].append((entity, transfer_element))
        # first round to ensure we have necessary basic structure
        for xselement, etype, child_defs in XSDM_MAPPING:
            # print 'PROCESS', getattr(xselement, 'local_name', xselement.__class__.__name__), etype
            for entity, profile_element in to_process.pop(xselement, ()):
                assert etype == entity.cw_etype
                self._process(entity, profile_element, child_defs, to_process)
        # then process remaining elements
        # print 'STARTING ROUND 2'
        while to_process:
            xselement = next(iter(to_process))
            entities_profiles = to_process.pop(xselement, ())
            if entities_profiles:
                try:
                    etype, child_defs = XSDM_MAPPING[xselement]
                except KeyError:
                    # element has no children
                    continue
                for entity, profile_element in entities_profiles:
                    assert etype == entity.cw_etype
                    self._process(entity, profile_element, child_defs, to_process)

        assert not to_process, to_process

    def _process(self, entity, profile_element, child_defs, to_process):
        for occ, path in child_defs:
            # print '  child', getattr(occ.target, 'local_name', occ.target.__class__.__name__), \
            #    [x[:-1] for x in path]
            if not path:
                assert not isinstance(occ.target, graph_nodes.XMLAttribute)
                assert occ.target.local_name in JUMP_ELEMENTS, occ.target
                if occ.minimum == 0 and not any(iter_path_children(occ.target, entity)):
                    # element has no children, skip it
                    continue
                if occ.target.local_name in JUMPED_OPTIONAL_ELEMENTS:
                    # elements in JUMPED_OPTIONAL_ELEMENTS are jumped but have optional cardinality,
                    # so search in all it's child element, and mark it as mandatory if one of them
                    # is mandatory, else keep it optional
                    cardinality = "0..1"
                    for _path, target in iter_path_children(occ.target, entity):
                        if _path[0][2] in BASE_TYPES:
                            # special case of a mandatory attribute: parent element will be
                            # mandatory if some value is specified, else that's fine
                            if target is not None:
                                cardinality = "1"
                                break
                        elif any(te.user_cardinality == "1" for te in target):
                            cardinality = "1"
                            break
                else:
                    cardinality = None
                # jumped element: give None as target_value but register the generated element for
                # later processing
                self.dispatch_occ(
                    profile_element,
                    occ,
                    None,
                    to_process,
                    card_entity=attrdict({"user_cardinality": cardinality}),
                )
                to_process[occ.target].append(
                    (entity, self.jumped_element(profile_element))
                )
            else:
                # print '  values', _path_target_values(entity, path)
                for card_entity, target_value in _path_target_values(entity, path):
                    self.dispatch_occ(
                        profile_element,
                        occ,
                        target_value,
                        to_process,
                        card_entity=card_entity,
                    )

    def init_transfer_element(self, xselement, root, entity):
        """Initialize and return the XML element holding the ArchiveTransfer definition, as well as
        any other necessary global definitions.
        """
        raise NotImplementedError()

    def jumped_element(self, profile_element):
        """Return the last generated element, for insertion of its content."""
        raise NotImplementedError()

    @staticmethod
    def cwuri_url(entity):
        """Return "public" URI for the given entity.

        In a staticmethod to ease overriding in subclasses (eg saem).
        """
        return entity.cwuri


class SEDA2RelaxNGExport(RNGMixin, SEDA2ExportAdapter):
    """Abstract Adapter to build a Relax NG representation of a SEDA profile, using SEDA 2.1 specification."""

    __regid__ = "SEDA-2.1.rng"
    __abstract__ = True

    namespaces = {
        None: "fr:gouv:culture:archivesdefrance:seda:v2.1",
        "seda": "fr:gouv:culture:archivesdefrance:seda:v2.1",
        "xml": "http://www.w3.org/XML/1998/namespace",
        "xsd": "http://www.w3.org/2001/XMLSchema",
        "xlink": "http://www.w3.org/1999/xlink",
        "rng": "http://relaxng.org/ns/structure/1.0",
        "a": "http://relaxng.org/ns/compatibility/annotations/1.0",
    }

    _root_attributes = {
        "ns": "fr:gouv:culture:archivesdefrance:seda:v2.1",
        "datatypeLibrary": "http://www.w3.org/2001/XMLSchema-datatypes",
    }

    def dump_etree(self):
        """Return a RelaxNG etree for the adapted SEDA profile."""
        root = self.element("rng:grammar", attributes=self.root_attributes)
        start = self.element("rng:start", root)
        # XXX http://lists.xml.org/archives/xml-dev/200206/msg01074.html ?
        # self.element('xsd:import', parent=root,
        #              attributes={'namespace': 'http://www.w3.org/1999/xlink',
        #                          'schemaLocation': 'http://www.w3.org/1999/xlink.xsd'})
        self._dump(start)

        open_type = self.element("rng:define", root, {"name": "OpenType"})
        open_elt = self._create_hierarchy(open_type, ["rng:zeroOrMore", "rng:element"])
        self.element("rng:anyName", open_elt)
        self._create_hierarchy(
            open_elt, ["rng:zeroOrMore", "rng:attribute", "rng:anyName"]
        )

        # add a 'text' node to empty rng:element to satisfy the RNG grammar
        namespaces = self.namespaces.copy()
        del namespaces[None]  # xpath engine don't want None prefix
        for element in root.xpath("//rng:element[not(*)]", namespaces=namespaces):
            self.element("rng:text", element)

        self.postprocess_dataobjects(root, namespaces)
        return root

    def postprocess_dataobjects(self, root, namespaces):
        """Insert rng:group node as parent of [Binary|Physical]DataObject node
        to avoid forcing an order among them
        """
        raise NotImplementedError()

    def init_transfer_element(self, xselement, root, entity):
        transfer_element = self.element(
            "rng:element",
            root,
            {"name": xselement.local_name, "documentation": entity.user_annotation},
        )
        exc = self._create_hierarchy(
            transfer_element,
            ["rng:zeroOrMore", "rng:attribute", "rng:anyName", "rng:except"],
        )
        self.element("rng:nsName", exc)
        self.element("rng:nsName", exc, {"ns": ""})
        return transfer_element

    def jumped_element(self, profile_element):
        element = profile_element[-1]
        if element.tag != "{http://relaxng.org/ns/structure/1.0}element":
            # optional, zeroOrMore, etc.: should pick their child element
            element = element[-1]
            assert (
                element.tag == "{http://relaxng.org/ns/structure/1.0}element"
            ), element
        return element

    def element_alternative(
        self, occ, profile_element, target_value, to_process, card_entity
    ):
        parent_element = self._rng_element_parent(occ, card_entity, profile_element)
        target_element = self.element("rng:choice", parent_element)
        to_process[occ.target].append((target_value, target_element))

    def element_sequence(
        self, occ, profile_element, target_value, to_process, card_entity
    ):
        parent_element = self._rng_element_parent(occ, card_entity, profile_element)
        target_element = self.element("rng:group", parent_element)  # XXX sequence
        to_process[occ.target].append((target_value, target_element))

    def element_xmlattribute(
        self, occ, profile_element, target_value, to_process, card_entity
    ):
        parent_element = self._rng_attribute_parent(occ, card_entity, profile_element)
        self._rng_attribute(
            occ.target, parent_element, serialize(target_value, self.cwuri_url)
        )

    def element_xmlelement(
        self, occ, profile_element, target_value, to_process, card_entity
    ):  # noqa
        parent_element = self._rng_element_parent(occ, card_entity, profile_element)
        xselement = occ.target
        attrs = {
            "documentation": getattr(card_entity, "user_annotation", None),
            "name": xselement.local_name,
        }
        if xselement.local_name == "Signature":
            element = self.element("rng:element", parent_element, attrs)
            self.element("rng:ref", element, {"name": "OpenType"})
        elif isinstance(
            occ, dict
        ):  # fake occurence introduced for some elements'content
            # target element has already been introduced: it is now given as profile_element
            self.fill_element(xselement, profile_element, target_value, card_entity)
        else:
            target_element = self.element("rng:element", parent_element, attrs)
            xstypes = content_types(xselement.textual_content_type)
            if xstypes:
                if len(xstypes) == 1:
                    parent_element = target_element
                else:
                    parent_element = self.element("rng:choice", target_element)
                for xstype in xstypes:
                    self.fill_element(
                        xselement,
                        parent_element,
                        target_value,
                        card_entity,
                        xstype=xstype,
                        copy_attributes=True,
                    )
            else:
                # target is a complex element
                if getattr(target_value, "eid", None):  # value is an entity
                    if target_value.cw_etype == "AuthorityRecord":
                        self.fill_organization_element(target_element, target_value)
                elif xselement.local_name in ("ArchivalAgency", "TransferringAgency"):
                    self.fill_organization_element(target_element, None)
                elif target_value is not None:
                    assert False, (xselement, target_value)
            if getattr(target_value, "eid", None):  # value is an entity
                to_process[xselement].append((target_value, target_element))

    def fill_element(
        self,
        xselement,
        profile_element,
        value,
        card_entity,  # noqa
        copy_attributes=False,
        xstype=None,
    ):
        if xselement.local_name == "KeywordType":
            attr = self.element(
                "rng:attribute",
                attributes={"name": "listVersionID"},
                parent=self.element("rng:optional", profile_element),
            )
            if value:
                list_value = value.scheme.description or value.scheme.dc_title()
                attrs = {"type": xstype} if xstype else {}
                self.element("rng:value", attr, attrs, text=list_value)
            else:
                attr.attrib[self.qname("a:defaultValue")] = "edition 2009"

        elif xselement.local_name == "KeywordReference" and card_entity.scheme:
            self.concept_scheme_attribute(
                xselement, profile_element, card_entity.scheme
            )

        elif getattr(value, "cw_etype", None) == "Concept":
            self.concept_scheme_attribute(xselement, profile_element, value.scheme)

        elif copy_attributes:
            for attrname, occ in xselement.attributes.items():
                if attrname in ("id", "href") or attrname.startswith(
                    ("list", "scheme")
                ):
                    parent_element = self._rng_attribute_parent(
                        occ, None, profile_element
                    )
                    self._rng_attribute(occ.target, parent_element)
        # special case for KeywordReference content, the only known case where we want URL instead
        # of label of its concept value
        if value is not None and xselement.local_name == "KeywordReference":
            fixed_value = self.cwuri_url(value)
        elif isinstance(value, (tuple, list)):
            fixed_value = [serialize(val, self.cwuri_url) for val in value]
        else:
            fixed_value = serialize(value, self.cwuri_url)
        if fixed_value is not None:
            if _internal_reference(value):
                profile_element.attrib[self.qname("a:defaultValue")] = fixed_value
                self.element("rng:data", profile_element, {"type": xstype})
            else:
                if len(profile_element):
                    # As there is a fixed value search for potential element
                    # tag data and extract its type
                    for elem in profile_element:
                        if elem.tag == "{http://relaxng.org/ns/structure/1.0}data":
                            xstype = elem.attrib.get("type")
                            profile_element.remove(elem)
                            break
                attrs = {"type": xstype} if xstype else {}
                if isinstance(fixed_value, (tuple, list)):
                    choice = self.element("rng:choice", profile_element)
                    for val in fixed_value:
                        self.element("rng:value", choice, attrs, text=val)
                else:
                    self.element("rng:value", profile_element, attrs, text=fixed_value)
        elif xstype is not None:
            self.element("rng:data", profile_element, {"type": xstype})

    def concept_scheme_attribute(self, xselement, type_element, scheme):
        try:
            scheme_attr = xselement_scheme_attribute(xselement)
        except KeyError:
            return
        scheme_attr = self.element(
            "rng:attribute", type_element, attributes={"name": scheme_attr}
        )
        self.element("rng:value", scheme_attr, text=self.cwuri_url(scheme))

    def fill_organization_element(self, parent_element, value):
        target_element = self.element(
            "rng:element", parent_element, {"name": "Identifier"}
        )
        if value:
            self.element("rng:value", target_element, text=self.cwuri_url(value))

    def _rng_element_parent(self, occ, card_entity, profile_element):
        minimum, maximum = element_minmax_cardinality(occ, card_entity)
        return self.rng_element_parent(profile_element, minimum, maximum)

    def _rng_attribute_parent(self, occ, card_entity, profile_element):
        minimum = attribute_minimum_cardinality(occ, card_entity)
        return self.rng_element_parent(profile_element, minimum)

    def _rng_attribute(self, xselement, parent_element, value=None):
        xstypes = content_types(xselement.textual_content_type)
        if len(xstypes) > 1:
            parent_element = self.element("rng:choice", parent_element)
        for xstype in xstypes:
            attr_element = self.element(
                "rng:attribute", parent_element, {"name": xselement.local_name}
            )
            if value is not None:
                if xselement.local_name == "id":
                    attr_element.attrib[self.qname("xml:id")] = value
                    self.element("rng:data", attr_element, {"type": "ID"})
                else:
                    self.element(
                        "rng:value", attr_element, {"type": xstype}, text=value
                    )
            else:
                self.element("rng:data", attr_element, {"type": xstype})

    def _create_hierarchy(self, parent, tags):
        for tag in tags:
            parent = self.element(tag, parent)
        return parent


def _safe_cardinality(entity):
    """Return entity's cardinality if some entity is given, else None."""
    if entity is None:
        return None
    return entity.user_cardinality


def _safe_concept_value(entity, concepts_language):
    """Return entity's targetted concept if some entity is given, else None."""
    if entity is None:
        return None
    return _concept_value(entity.concept, concepts_language)


class SEDA22RelaxNGExport(SEDA2RelaxNGExport):
    __select__ = SEDA2RelaxNGExport.__select__ & wrap_dataobjects()

    def postprocess_dataobjects(self, root, namespaces):
        """Insert rng:group node as parent of [Binary|Physical]DataObject node
        to avoid forcing an order among them
        """
        # start by looking for the rng:element node for DataObjectPackage or its parent rng:optional
        dops = root.xpath(
            "/rng:grammar/rng:start/rng:element/"
            'rng:element[@name="DataObjectPackage"]',
            namespaces=namespaces,
        )
        if not dops:
            dops = root.xpath(
                "/rng:grammar/rng:start/rng:element/"
                'rng:optional/rng:element[@name="DataObjectPackage"]',
                namespaces=namespaces,
            )
        if dops:
            assert len(dops) == 1
            dop = dops[0]
            nodes = dop.xpath(
                'rng:element[@name="BinaryDataObject" or @name="PhysicalDataObject"]',
                namespaces=namespaces,
            )
            opt_nodes = dop.xpath(
                'rng:optional[rng:element[@name="BinaryDataObject" or @name="PhysicalDataObject"]]',
                namespaces=namespaces,
            )
            if nodes or opt_nodes:
                # insert after definition of dop's id attribute
                for node in chain(nodes, opt_nodes):
                    zeroormore = self.element("rng:zeroOrMore")
                    choice = self.element("rng:choice", zeroormore)
                    dop[0].addnext(zeroormore)
                    # insert DataObjectGroup
                    group = self.element(
                        "rng:element", choice, attributes={"name": "DataObjectGroup"}
                    )
                    group.append(node)


class SEDA21RelaxNGExport(SEDA2RelaxNGExport):
    __select__ = SEDA2RelaxNGExport.__select__ & ~wrap_dataobjects()

    def postprocess_dataobjects(self, root, namespaces):
        """Insert rng:group node as parent of [Binary|Physical]DataObject node
        to avoid forcing an order among them
        """
        # start by looking for the rng:element node for DataObjectPackage or its parent rng:optional
        dops = root.xpath(
            "/rng:grammar/rng:start/rng:element/"
            'rng:element[@name="DataObjectPackage"]',
            namespaces=namespaces,
        )
        if not dops:
            dops = root.xpath(
                "/rng:grammar/rng:start/rng:element/"
                'rng:optional/rng:element[@name="DataObjectPackage"]',
                namespaces=namespaces,
            )
        if dops:
            assert len(dops) == 1
            dop = dops[0]
            nodes = dop.xpath(
                'rng:element[@name="BinaryDataObject" or @name="PhysicalDataObject"]',
                namespaces=namespaces,
            )
            opt_nodes = dop.xpath(
                'rng:optional[rng:element[@name="BinaryDataObject" or @name="PhysicalDataObject"]]',
                namespaces=namespaces,
            )
            if nodes or opt_nodes:
                group = self.element("rng:group")
                # insert after definition of dop's id attribute
                dop[0].addnext(group)
                for node in chain(nodes, opt_nodes):
                    group.append(node)


class XAttr(
    namedtuple("_XAttr", ["name", "qualified_type", "cardinality", "fixed_value"])
):
    """Simple representation of an attribute element in a schema (RNG or XSD).

    Parameters:

    * `name`, the attribute's name,

    * `qualified_type`, its qualified type (e.g. 'xsd:string'),

    * `cardinality`, optional cardinality as string (None, '1' or '0..1') - default to '1' if some
      fixed value is provided, else to None (i.e. attribute is prohibited),

    * `fixed_value`, optional fixed value for the attribute.

    """

    def __new__(cls, name, qualified_type, cardinality="0..1", fixed_value=None):
        assert cardinality in (None, "1", "0..1"), cardinality
        if fixed_value:
            cardinality = "1"
            if isinstance(fixed_value, (tuple, list)) and len(fixed_value) == 1:
                fixed_value = fixed_value[0]
        else:
            fixed_value = None
        return super(XAttr, cls).__new__(
            cls, name, qualified_type, cardinality, fixed_value
        )


LIST_VERSION_ID_2009 = XAttr("listVersionID", "xsd:token", "1", "edition 2009")
LIST_VERSION_ID_2011 = XAttr("listVersionID", "xsd:token", "1", "edition 2011")


class SEDA1XSDExport(SEDA2ExportAdapter):
    """Adapter to build an XSD representation of a simplified SEDA profile, using SEDA 1.0
    specification.

    The SEDA2XSDExport implementation may be driven by the SEDA 2.1 XSD model because it's used as
    the basis for the Yams model generation. We can't do the same thing with lower version of SEDA,
    hence the limitation to simplified profile, and a direct implementation of the export.
    """

    __regid__ = "SEDA-1.0.xsd"
    __select__ = SEDA2ExportAdapter.__select__ & simplified_profile()

    namespaces = {
        None: "fr:gouv:culture:archivesdefrance:seda:v1.0",
        "seda": "fr:gouv:culture:archivesdefrance:seda:v2.1",
        "xml": "http://www.w3.org/XML/1998/namespace",
        "xsd": "http://www.w3.org/2001/XMLSchema",
        "qdt": "fr:gouv:culture:archivesdefrance:seda:v1.0:QualifiedDataType:1",
        "udt": "urn:un:unece:uncefact:data:standard:UnqualifiedDataType:10",
        "clmDAFFileTypeCode": "urn:un:unece:uncefact:codelist:draft:DAF:fileTypeCode:2009-08-18",
        "clmIANACharacterSetCode": "urn:un:unece:uncefact:codelist:standard:IANA:CharacterSetCode:2007-05-14",
        "clmIANAMIMEMediaType": "urn:un:unece:uncefact:codelist:standard:IANA:MIMEMediaType:2008-11-12",
        "clm60133": "urn:un:unece:uncefact:codelist:standard:6:0133:40106",
    }
    _root_attributes = {
        "targetNamespace": "fr:gouv:culture:archivesdefrance:seda:v1.0",
        "attributeFormDefault": "unqualified",
        "elementFormDefault": "qualified",
        "version": "1.0",
    }

    concepts_language = "seda-1"

    def element_schema(
        self,
        parent,
        name,
        xsd_type=None,
        fixed_value=None,
        default_value=None,
        cardinality="1",
        documentation=None,
        xsd_attributes=(),
        extra_attributes=None,
    ):
        assert not (
            fixed_value and default_value
        ), "only one of fixed_value or default_value may be specified"
        attributes = {"name": name}
        if extra_attributes is not None:
            attributes.update(extra_attributes)
        if fixed_value is not None:
            attributes["fixed"] = str(fixed_value)
        elif default_value is not None:
            attributes["default"] = str(default_value)
        if xsd_type is not None and not xsd_attributes:
            attributes["type"] = xsd_type
        assert cardinality in ("0..1", "0..n", "1", "1..n")
        if cardinality != "1":
            if cardinality[0] == "0":
                attributes["minOccurs"] = "0"
            if cardinality[-1] == "n":
                attributes["maxOccurs"] = "unbounded"
        if documentation:
            attributes["documentation"] = documentation
        element = self.element("xsd:element", parent, attributes)
        children_parent = None
        if xsd_type is None:
            attributes_parent = self.element("xsd:complexType", element)
            children_parent = self.element("xsd:sequence", attributes_parent)
        elif xsd_attributes:
            ct = self.element("xsd:complexType", element)
            scontent = self.element("xsd:simpleContent", ct)
            attributes_parent = self.element(
                "xsd:extension", scontent, {"base": xsd_type}
            )
        for xattr in xsd_attributes:
            self.attribute_schema(attributes_parent, xattr)
        return children_parent

    def attribute_schema(self, parent, xattr):
        attrs = {"name": xattr.name}
        if xattr.cardinality is None:
            attrs["use"] = "prohibited"
        elif xattr.cardinality == "1":
            attrs["use"] = "required"
        else:
            attrs["use"] = "optional"
        if not isinstance(xattr.fixed_value, (tuple, list)):
            attrs["type"] = xattr.qualified_type
            if isinstance(xattr.fixed_value, str):
                attrs["fixed"] = xattr.fixed_value
        attribute_element = self.element("xsd:attribute", parent, attrs)
        if isinstance(xattr.fixed_value, (tuple, list)):
            type_element = self.element("xsd:simpleType", attribute_element)
            restriction_element = self.element(
                "xsd:restriction", type_element, {"base": "xsd:token"}
            )
            for value in xattr.fixed_value:
                self.element("xsd:enumeration", restriction_element, {"value": value})

    # business visit methods #######################################################################

    def dump_etree(self):
        """Return an XSD etree for the adapted SEDA profile."""
        root = self.element("xsd:schema", attributes=self.root_attributes)
        # self.element('xsd:import', parent=root,
        #              attributes={'namespace': 'http://www.w3.org/XML/1998/namespace',
        #                          'schemaLocation': 'http://www.w3.org/2001/xml.xsd'})
        self.xsd_transfer(root, self.entity)
        return root

    def xsd_transfer(self, parent, archive_transfer):
        """Append XSD elements for the archive transfer to the given parent node."""
        transfer_node = self.xsd_transfer_base(parent, archive_transfer)
        for archive_unit in archive_transfer.archive_units:
            self.xsd_archive(transfer_node, archive_unit)

    def xsd_archive(self, parent, archive_unit):
        """Append XSD elements for an archive to the given parent node."""
        archive_node = self.element_schema(
            parent,
            "Archive",
            cardinality=archive_unit.user_cardinality,
            documentation=archive_unit.user_annotation,
            xsd_attributes=[XAttr("Id", "xsd:ID")],
            extra_attributes={"xml:id": xmlid(archive_unit)},
        )
        transfer = archive_unit.cw_adapt_to("ITreeBase").parent()
        self.xsd_archival_agreement(archive_node, transfer)
        # hard-coded description's language XXX fine, content language may be specified
        self.element_schema(
            archive_node,
            "DescriptionLanguage",
            "qdt:CodeLanguageType",
            fixed_value="fra",
            xsd_attributes=[LIST_VERSION_ID_2011],
        )
        name_entity = self.archive_unit_name(archive_unit)
        self.element_schema(
            archive_node,
            "Name",
            "udt:TextType",
            fixed_value=name_entity.title,
            documentation=name_entity.user_annotation,
            xsd_attributes=[XAttr("languageID", "xsd:language")],
        )
        content_entity = self.archive_unit_content(archive_unit)
        self.xsd_transferring_agency_archive_identifier(
            archive_node, content_entity, "TransferringAgencyArchiveIdentifier"
        )
        self.xsd_content_description(archive_node, content_entity)
        self.xsd_rules(archive_node, archive_unit)
        self.xsd_children(archive_node, archive_unit)

    archive_object_tag_name = "ArchiveObject"

    def xsd_archive_object(self, parent, archive_unit):
        """Append XSD elements for the archive object to the given parent node."""
        ao_node = self.element_schema(
            parent,
            self.archive_object_tag_name,
            cardinality=archive_unit.user_cardinality,
            documentation=archive_unit.user_annotation,
            xsd_attributes=[XAttr("Id", "xsd:ID")],
            extra_attributes={"xml:id": xmlid(archive_unit)},
        )
        content_entity = self.archive_unit_content(archive_unit)
        self.element_schema(
            ao_node,
            "Name",
            "udt:TextType",
            fixed_value=content_entity.title.title,
            documentation=content_entity.title.user_annotation,
            xsd_attributes=[XAttr("languageID", "xsd:language")],
        )
        self.xsd_transferring_agency_archive_identifier(
            ao_node, content_entity, "TransferringAgencyObjectIdentifier"
        )
        if (
            self.__regid__.startswith("SEDA-1.0")
            or content_entity.start_date
            or content_entity.end_date
            or content_entity.description
            or content_entity.keywords
        ):