senaite.core.exportimport.genericsetup.structure.SenaiteSiteXMLAdapter._extractObjects() - Code Metrics - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

SenaiteSiteXMLAdapter._extractObjects() B
last analyzed 2025-12-16 10:03 UTC

↳ Parent: senaite.core.exportimport.genericsetup.structure

Complexity

Conditions

Size

Total Lines	17
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	16
dl	0
loc	17
rs	8
c	0
b	0
f	0
cc	7
nop	1

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE.
#
# SENAITE.CORE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2.
#
# 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 General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright 2018-2025 by it's authors.
# Some rights reserved, see README and LICENSE.

import json
from bika.lims import api
from bika.lims import logger
from bika.lims.interfaces import IAuditable
from bika.lims.interfaces import ISenaiteSiteRoot
from DateTime import DateTime
from OFS.interfaces import IOrderedContainer
from plone.dexterity.interfaces import IDexterityContainer
from plone.dexterity.interfaces import IDexterityItem
from plone.dexterity.utils import addContentToContainer
from plone.memoize.request import cache
from Products.Archetypes.interfaces import IBaseObject
from Products.CMFPlone.utils import _createObjectByType
from Products.CMFPlone.utils import safe_unicode
from Products.GenericSetup.interfaces import IBody
from Products.GenericSetup.interfaces import INode
from Products.GenericSetup.interfaces import ISetupEnviron
from Products.GenericSetup.utils import I18NURI
from Products.GenericSetup.utils import ObjectManagerHelpers
from Products.GenericSetup.utils import XMLAdapterBase
from senaite.core.p3compat import cmp
from senaite.core.registry import get_registry_record
from xml.dom.minidom import parseString
from zope.component import adapts
from zope.component import getUtility
from zope.component import queryMultiAdapter
from zope.component.interfaces import IFactory
from zope.interface import alsoProvides

from .config import SITE_ID

ID_MAP = {}


class SenaiteSiteXMLAdapter(XMLAdapterBase, ObjectManagerHelpers):
    """Adapter for the SENAITE root object (portal)
    """
    adapts(ISenaiteSiteRoot, ISetupEnviron)

    def __init__(self, context, environ):
        super(SenaiteSiteXMLAdapter, self).__init__(context, environ)

    def _exportNode(self):
        """Export the object as a DOM node.
        """
        node = self._getObjectNode("object")

        # remember the UID of the item for reference fields
        node.setAttribute("uid", "0")

        # Extract all contained objects
        node.appendChild(self._extractObjects())

        # Extract Groups
        node.appendChild(self._extractGroups(self.context))

        # Extract Users
        node.appendChild(self._extractUsers(self.context))

        return node

    def _importNode(self, node):
        """Import the object from the DOM node.
        """
        obj_id = str(node.getAttribute("name"))

        if "acl_users" not in self.context:
            return

        # Add groups and users
        self._initGroups(self.context, node)
        self._initUsers(self.context, node)

        self._logger.info("Imported '%r'" % obj_id)

    def _initGroups(self, context, node):
        group_tool = api.get_tool("portal_groups")
        for child in node.childNodes:
            if child.nodeName != "groups":
                continue
            for cn in child.childNodes:
                if cn.nodeName != "group":
                    continue
                group_id = cn.firstChild.nodeValue
                group = api.user.get_group(group_id)
                if not group:
                    self._logger.info("Adding group {}".format(group_id))
                    roles = cn.getAttribute("roles").split(",")
                    group_tool.addGroup(group_id, roles=roles)
                    group = group_tool.getGroupById(group_id)

                # set the group properties
                group.setProperties(properties={
                    "title": cn.getAttribute("name"),
                    "email": cn.getAttribute("email"),
                })

    def _initUsers(self, context, node):
        reg_tool = api.get_tool("portal_registration")
        for child in node.childNodes:
            if child.nodeName != "users":
                continue
            for cn in child.childNodes:
                if cn.nodeName != "user":
                    continue
                user_id = cn.firstChild.nodeValue
                user = api.user.get_user(user_id)

                if not user: # add new user with password
                    self._logger.info("Adding user {}".format(user_id))
                    user = reg_tool.addMember(user_id, '12345') 

                # set the user properties
                user.setProperties(properties={
                    "fullname": cn.getAttribute("name"),
                    "email": cn.getAttribute("email"),
                })

                # add the user to the groups
                groups = cn.getAttribute("groups")
                if groups:
                    group_ids = groups.split(",")
                    api.user.add_group(group_ids, user_id)

    def _get_users(self):
        acl_users = api.get_tool("acl_users")
        return acl_users.getUsers()

    def _get_groups(self):
        acl_users = api.get_tool("acl_users")
        return acl_users.getGroups()

    def _get_roles_for_principal(self, principal):
        """Returs a list of roles for the user/group
        """
        ignored_roles = ["Authenticated"]
        roles = filter(lambda r: r not in ignored_roles,
                       principal.getRoles())
        return roles

    def _get_groups_for_principal(self, principal):
        """Returs a list of groups for the user/group
        """
        ignored_groups = ["AuthenticatedUsers"]
        groups = filter(lambda r: r not in ignored_groups,
                        principal.getGroupIds())
        return groups

    def _extractGroups(self, context):
        node = self._doc.createElement("groups")
        for group in self._get_groups():
            name = group.getGroupName()
            roles = self._get_roles_for_principal(group)
            child = self._doc.createElement("group")
            child.setAttribute("name", safe_unicode(name))
            child.setAttribute("roles", ",".join(roles))
            text = self._doc.createTextNode(group.getGroupId())
            child.appendChild(text)
            node.appendChild(child)
        return node

    def _extractUsers(self, context):
        node = self._doc.createElement("users")
        for user in self._get_users():
            name = user.getProperty("fullname")
            groups = self._get_groups_for_principal(user)
            child = self._doc.createElement("user")
            child.setAttribute("name", safe_unicode(name))
            child.setAttribute("email", user.getProperty("email"))
            child.setAttribute("groups", ",".join(groups))
            text = self._doc.createTextNode(user.getId())
            child.appendChild(text)
            node.appendChild(child)
        return node

    def _extractObjects(self):
        fragment = self._doc.createDocumentFragment()
        objects = self.context.objectValues()
        if not IOrderedContainer.providedBy(self.context):
            objects = list(objects)
            objects.sort(lambda x, y: cmp(x.getId(), y.getId()))
        for obj in objects:
            # Check if the object can be exported
            if not can_export(obj):
                logger.info("Skipping {}".format(repr(obj)))
                continue
            exporter = queryMultiAdapter((obj, self.environ), INode)
            if exporter:
                node = exporter.node
                if node is not None:
                    fragment.appendChild(exporter.node)
        return fragment


class ATContentXMLAdapter(SenaiteSiteXMLAdapter):
    """AT Content XML Importer/Exporter
    """
    adapts(IBaseObject, ISetupEnviron)

    def __init__(self, context, environ):
        super(ATContentXMLAdapter, self).__init__(context, environ)

    def _getObjectNode(self, name, i18n=True):
        node = self._doc.createElement(name)
        # Attach the UID of the object as well
        node.setAttribute("id", api.get_id(self.context))
        node.setAttribute("uid", api.get_uid(self.context))
        node.setAttribute("name", api.get_id(self.context))
        node.setAttribute("meta_type", self.context.meta_type)
        node.setAttribute("portal_type", api.get_portal_type(self.context))
        i18n_domain = getattr(self.context, "i18n_domain", None)
        if i18n and i18n_domain:
            node.setAttributeNS(I18NURI, "i18n:domain", i18n_domain)
            self._i18n_props = ("title", "description")
        return node

    def _exportNode(self):
        """Export the object as a DOM node.
        """
        node = self._getObjectNode("object")

        # remember the UID of the item for reference fields
        node.setAttribute("uid", api.get_uid(self.context))

        # remember the WF Status
        # TODO: Export the complete Review History
        state = api.get_workflow_status_of(self.context)
        node.setAttribute("state", state)

        # Extract AuditLog
        node.appendChild(self._extractAuditLog(self.context))

        # Extract all fields of the current context
        node.appendChild(self._extractFields(self.context))

        # Extract all contained objects
        node.appendChild(self._extractObjects())

        return node

    def _importNode(self, node):
        """Import the object from the DOM node.
        """

        # set workflow state
        self._initAuditLog(self.context, node)
        self._initWorkflow(self.context, node)
        self._initFields(self.context, node)

        # reindex the object
        self.context.reindexObject()

        # set a new snapshot
        # api.snapshot.take_snapshot(self.context)

        obj_id = str(node.getAttribute("name"))
        self._logger.info("Imported '%r'" % obj_id)

    def _initAuditLog(self, context, node):
        for child in node.childNodes:
            if child.nodeName == "auditlog":
                snapshots = json.loads(child.firstChild.nodeValue)
                storage = api.snapshot.get_storage(context)
                storage[:] = map(json.dumps, snapshots)[:]
                # make sure the object provides `IAuditable`
                alsoProvides(context, IAuditable)
                return

    def _initWorkflow(self, context, node):
        state = node.getAttribute("state")

        if not state:
            return

        if state == api.get_workflow_status_of(context):
            return

        wf_state = {
            "action": None,
            "actor": None,
            "comments": "Generic Setup Import",
            "review_state": state,
            "time": DateTime(),
        }

        wf = api.get_tool("portal_workflow")
        wf_id = wf.getChainFor(context)[0]
        wf.setStatusOf(wf_id, context, wf_state)

    def _initFields(self, context, node):
        fields = api.get_fields(context)

        for child in node.childNodes:
            # we only handle filed nodes
            if child.nodeName != "field":
                continue

            name = child.getAttribute("name")
            field = fields.get(name)
            if field is None:
                self._logger.warning("Unrecognized field '{}'".format(name))
                continue

            importer = queryMultiAdapter((context, field, self.environ), INode)
            if importer:
                importer.node = child

    def _extractAuditLog(self, context):
        snapshots = api.snapshot.get_snapshots(self.context)
        node = self._doc.createElement("auditlog")
        child = self._doc.createTextNode(json.dumps(snapshots))
        node.appendChild(child)
        return node

    def _extractFields(self, context):
        fragment = self._doc.createDocumentFragment()

        fields = api.get_fields(context)
        for name, field in fields.items():
            # query the field adapter
            exporter = queryMultiAdapter((context, field, self.environ), INode)
            if not exporter:
                continue
            node = exporter.node
            if node is not None:
                fragment.appendChild(node)
        return fragment


class DXContainerXMLAdapter(ATContentXMLAdapter):
    """DX Container XML Importer/Exporter
    """
    adapts(IDexterityContainer, ISetupEnviron)

    def __init__(self, context, environ):
        super(DXContainerXMLAdapter, self).__init__(context, environ)


class DXItemXMLAdapter(ATContentXMLAdapter):
    """DX Item XML Importer/Exporter
    """
    adapts(IDexterityItem, ISetupEnviron)

    def __init__(self, context, environ):
        super(DXItemXMLAdapter, self).__init__(context, environ)


def create_content_slugs(parent, parent_path, context):
    """Helper function to create initial content slugs
    """
    logger.info("create_content_slugs: parent={} parent_path={}".format(
        repr(parent), parent_path))

    path = "%s%s" % (parent_path, get_id(parent))
    filename = "%s.xml" % (path)
    xml = context.readDataFile(filename)

    if xml is None:
        logger.error("File not found: '{}'".format(filename))
        return

    # parse the XML data
    node = parseString(xml)

    if node.nodeName == "#document":
        node = node.firstChild

    # read the node attributes
    name = node.getAttribute("name")
    uid = node.getAttribute("uid")
    logger.info("::: Processing '{}' (UID {}) in path '{}' :::"
                .format(name, uid, path))

    def is_object_node(n):
        return getattr(n, "nodeName", "") == "object"

    def get_child_nodes(n):
        return getattr(n, "childNodes", [])

    for child in get_child_nodes(node):
        # only process `<object ../>` nodes
        if not is_object_node(child):
            continue
        # extract node attributes (see `_exportNode` method)
        child_id = child.getAttribute("name")
        child_uid = child.getAttribute("uid")
        portal_type = child.getAttribute("portal_type")
        # get or create object
        obj = create_or_get(parent, child_id, child_uid, portal_type)
        # handle vanished objects
        if obj is None:
            logger.warn("Skipping object creation for '{}'".format(path))
            continue
        # get the id of the new object
        obj_id = api.get_id(obj)
        # track new ID -> old ID
        ID_MAP[obj_id] = child_id
        # recursively create contents
        create_content_slugs(obj, path + "/", context)


def create_or_get(parent, id, uid, portal_type):
    """Create or get the object
    """
    # return first level objects directly
    if api.is_portal(parent):
        return parent.get(id)
    # Check both bika_setup (AT, has subfolders) and senaite_setup (DX)
    bika_setup = api.get_portal().get("bika_setup")
    if bika_setup and bika_setup == parent:
        return parent.get(id)
    elif api.get_senaite_setup() == parent:
        return parent.get(id)

    # query object by UID
    query = {
        "UID": uid,
        "portal_type": portal_type,
        "path": {
            "query": api.get_path(parent),
        }
    }
    results = api.search(query, "uid_catalog")
    if results:
        return api.get_object(results[0])

    # create object slug
    obj = None
    # get the fti
    types_tool = api.get_tool("portal_types")
    fti = types_tool.getTypeInfo(portal_type)
    # removed
    if not fti:
        return None
    # old style factory
    if fti.product:
        # Create AT Content Slug (we take the UID as ID to avoid clashes)
        obj = _createObjectByType(portal_type, parent, uid)
        # set the old UID to maintain references
        obj._setUID(uid)
        # IMPORTANT: this will generate a new ID by the ID Server config
        obj.processForm()
    else:
        # Create DX Content Slug
        factory = getUtility(IFactory, fti.factory)
        tmp_id = str(uid)
        obj = factory(tmp_id)
        if hasattr(obj, "_setPortalTypeName"):
            obj._setPortalTypeName(fti.getId())
        # set the old UID to maintain references
        setattr(obj, "_plone.uuid", uid)
        # IMPORTANT: this will generate a new ID by the ID Server config
        obj = addContentToContainer(parent, obj, checkConstraints=False)

    return obj


@cache(get_key=lambda *args: 'can-export-%s' % args[1])

def can_export_type(portal_type, request):
    """Returns whether objects from this type can be exported
    """
    name = "generic_setup_skip_export_types"
    skip = get_registry_record(name, default=[])
    return portal_type not in skip


def can_export(obj):
    """Decides if the object can be exported or not
    """
    if not api.is_object(obj):
        return False
    portal_type = api.get_portal_type(obj)
    request = api.get_request()
    return can_export_type(portal_type, request)


def can_import(obj):
    """Decides if the object can be imported or not
    """
    if not api.is_object(obj):
        return False
    return True


def get_id(obj):
    if api.is_portal(obj):
        return SITE_ID
    oid = api.get_id(obj)
    # resolve id from mapping
    rid = ID_MAP.get(oid, oid)
    return rid.replace(" ", "_")


def exportObjects(obj, parent_path, context):
    """ Export subobjects recursively.
    """

    if not can_export(obj):
        logger.info("Skipping export of {}".format(repr(obj)))
        return

    if api.is_portal(obj):
        # explicitly instantiate the exporter to avoid adapter clash of
        # Products.CMFCore.exportimport.properties.PropertiesXMLAdapter
        exporter = SenaiteSiteXMLAdapter(obj, context)
    else:
        exporter = queryMultiAdapter((obj, context), IBody)

    path = "%s%s" % (parent_path, get_id(obj))
    if exporter:
        if exporter.name:
            path = "%s%s" % (parent_path, exporter.name)
        filename = "%s%s" % (path, exporter.suffix)
        body = exporter.body
        if body is not None:
            context.writeDataFile(filename, body, exporter.mime_type)
    else:
        raise ValueError("No exporter found for object: %r" % obj)

    if getattr(obj, "objectValues", False):
        for sub in obj.objectValues():
            exportObjects(sub, path + "/", context)


def importObjects(obj, parent_path, context):
    """ Import subobjects recursively.
    """

    if not can_import(obj):
        logger.info("Skipping import of {}".format(repr(obj)))
        return

    if api.is_portal(obj):
        # explicitly instantiate the importer to avoid adapter clash of
        # Products.CMFCore.exportimport.properties.PropertiesXMLAdapter
        importer = SenaiteSiteXMLAdapter(obj, context)
    else:
        importer = queryMultiAdapter((obj, context), IBody)

    path = "%s%s" % (parent_path, get_id(obj))
    __traceback_info__ = path
    if importer:
        if importer.name:
            path = "%s%s" % (parent_path, importer.name)
        filename = "%s%s" % (path, importer.suffix)
        body = context.readDataFile(filename)
        if body is not None:
            importer.filename = filename  # for error reporting
            importer.body = body

    if getattr(obj, "objectValues", False):
        for sub in obj.objectValues():
            importObjects(sub, path + "/", context)


def export_xml(context):
    portal = context.getSite()
    exportObjects(portal, "", context)


def import_xml(context):
    portal = context.getSite()

    qi = api.get_tool("portal_quickinstaller")
    installed = qi.isProductInstalled("senaite.core")
    request = api.get_request()

    # tests call it w/o request
    if request and not installed:
        logger.debug("Nothing to import.")
        return

    if not context.readDataFile("senaite.xml"):
        logger.debug("Nothing to import.")
        return

    # create content slugs for UID references
    create_content_slugs(portal, "", context)

    # import objects
    importObjects(portal, "", context)


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE.
4			#
5			# SENAITE.CORE is free software: you can redistribute it and/or modify it under
6			# the terms of the GNU General Public License as published by the Free Software
7			# Foundation, version 2.
8			#
9			# This program is distributed in the hope that it will be useful, but WITHOUT
10			# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11			# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12			# details.
13			#
14			# You should have received a copy of the GNU General Public License along with
15			# this program; if not, write to the Free Software Foundation, Inc., 51
16			# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17			#
18			# Copyright 2018-2025 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			import json
22			from bika.lims import api
23			from bika.lims import logger
24			from bika.lims.interfaces import IAuditable
25			from bika.lims.interfaces import ISenaiteSiteRoot
26			from DateTime import DateTime
27			from OFS.interfaces import IOrderedContainer
28			from plone.dexterity.interfaces import IDexterityContainer
29			from plone.dexterity.interfaces import IDexterityItem
30			from plone.dexterity.utils import addContentToContainer
31			from plone.memoize.request import cache
32			from Products.Archetypes.interfaces import IBaseObject
33			from Products.CMFPlone.utils import _createObjectByType
34			from Products.CMFPlone.utils import safe_unicode
35			from Products.GenericSetup.interfaces import IBody
36			from Products.GenericSetup.interfaces import INode
37			from Products.GenericSetup.interfaces import ISetupEnviron
38			from Products.GenericSetup.utils import I18NURI
39			from Products.GenericSetup.utils import ObjectManagerHelpers
40			from Products.GenericSetup.utils import XMLAdapterBase
41			from senaite.core.p3compat import cmp
42			from senaite.core.registry import get_registry_record
43			from xml.dom.minidom import parseString
44			from zope.component import adapts
45			from zope.component import getUtility
46			from zope.component import queryMultiAdapter
47			from zope.component.interfaces import IFactory
48			from zope.interface import alsoProvides
49
50			from .config import SITE_ID
51
52			ID_MAP = {}
53
54
55			class SenaiteSiteXMLAdapter(XMLAdapterBase, ObjectManagerHelpers):
56			"""Adapter for the SENAITE root object (portal)
57			"""
58			adapts(ISenaiteSiteRoot, ISetupEnviron)
59
60			def __init__(self, context, environ):
61			super(SenaiteSiteXMLAdapter, self).__init__(context, environ)
62
63			def _exportNode(self):
64			"""Export the object as a DOM node.
65			"""
66			node = self._getObjectNode("object")
67
68			# remember the UID of the item for reference fields
69			node.setAttribute("uid", "0")
70
71			# Extract all contained objects
72			node.appendChild(self._extractObjects())
73
74			# Extract Groups
75			node.appendChild(self._extractGroups(self.context))
76
77			# Extract Users
78			node.appendChild(self._extractUsers(self.context))
79
80			return node
81
82			def _importNode(self, node):
83			"""Import the object from the DOM node.
84			"""
85			obj_id = str(node.getAttribute("name"))
86
87			if "acl_users" not in self.context:
88			return
89
90			# Add groups and users
91			self._initGroups(self.context, node)
92			self._initUsers(self.context, node)
93
94			self._logger.info("Imported '%r'" % obj_id)
95
96			def _initGroups(self, context, node):
97			group_tool = api.get_tool("portal_groups")
98			for child in node.childNodes:
99			if child.nodeName != "groups":
100			continue
101			for cn in child.childNodes:
102			if cn.nodeName != "group":
103			continue
104			group_id = cn.firstChild.nodeValue
105			group = api.user.get_group(group_id)
106			if not group:
107			self._logger.info("Adding group {}".format(group_id))
108			roles = cn.getAttribute("roles").split(",")
109			group_tool.addGroup(group_id, roles=roles)
110			group = group_tool.getGroupById(group_id)
111
112			# set the group properties
113			group.setProperties(properties={
114			"title": cn.getAttribute("name"),
115			"email": cn.getAttribute("email"),
116			})
117
118			def _initUsers(self, context, node):
119			reg_tool = api.get_tool("portal_registration")
120			for child in node.childNodes:
121			if child.nodeName != "users":
122			continue
123			for cn in child.childNodes:
124			if cn.nodeName != "user":
125			continue
126			user_id = cn.firstChild.nodeValue
127			user = api.user.get_user(user_id)
128
129			if not user: # add new user with password
130			self._logger.info("Adding user {}".format(user_id))
131			user = reg_tool.addMember(user_id, '12345')
132
133			# set the user properties
134			user.setProperties(properties={
135			"fullname": cn.getAttribute("name"),
136			"email": cn.getAttribute("email"),
137			})
138
139			# add the user to the groups
140			groups = cn.getAttribute("groups")
141			if groups:
142			group_ids = groups.split(",")
143			api.user.add_group(group_ids, user_id)
144
145			def _get_users(self):
146			acl_users = api.get_tool("acl_users")
147			return acl_users.getUsers()
148
149			def _get_groups(self):
150			acl_users = api.get_tool("acl_users")
151			return acl_users.getGroups()
152
153			def _get_roles_for_principal(self, principal):
154			"""Returs a list of roles for the user/group
155			"""
156			ignored_roles = ["Authenticated"]
157			roles = filter(lambda r: r not in ignored_roles,
158			principal.getRoles())
159			return roles
160
161			def _get_groups_for_principal(self, principal):
162			"""Returs a list of groups for the user/group
163			"""
164			ignored_groups = ["AuthenticatedUsers"]
165			groups = filter(lambda r: r not in ignored_groups,
166			principal.getGroupIds())
167			return groups
168
169			def _extractGroups(self, context):
170			node = self._doc.createElement("groups")
171			for group in self._get_groups():
172			name = group.getGroupName()
173			roles = self._get_roles_for_principal(group)
174			child = self._doc.createElement("group")
175			child.setAttribute("name", safe_unicode(name))
176			child.setAttribute("roles", ",".join(roles))
177			text = self._doc.createTextNode(group.getGroupId())
178			child.appendChild(text)
179			node.appendChild(child)
180			return node
181
182			def _extractUsers(self, context):
183			node = self._doc.createElement("users")
184			for user in self._get_users():
185			name = user.getProperty("fullname")
186			groups = self._get_groups_for_principal(user)
187			child = self._doc.createElement("user")
188			child.setAttribute("name", safe_unicode(name))
189			child.setAttribute("email", user.getProperty("email"))
190			child.setAttribute("groups", ",".join(groups))
191			text = self._doc.createTextNode(user.getId())
192			child.appendChild(text)
193			node.appendChild(child)
194			return node
195
196			def _extractObjects(self):
197			fragment = self._doc.createDocumentFragment()
198			objects = self.context.objectValues()
199			if not IOrderedContainer.providedBy(self.context):
200			objects = list(objects)
201			objects.sort(lambda x, y: cmp(x.getId(), y.getId()))
202			for obj in objects:
203			# Check if the object can be exported
204			if not can_export(obj):
205			logger.info("Skipping {}".format(repr(obj)))
206			continue
207			exporter = queryMultiAdapter((obj, self.environ), INode)
208			if exporter:
209			node = exporter.node
210			if node is not None:
211			fragment.appendChild(exporter.node)
212			return fragment
213
214
215			class ATContentXMLAdapter(SenaiteSiteXMLAdapter):
216			"""AT Content XML Importer/Exporter
217			"""
218			adapts(IBaseObject, ISetupEnviron)
219
220			def __init__(self, context, environ):
221			super(ATContentXMLAdapter, self).__init__(context, environ)
222
223			def _getObjectNode(self, name, i18n=True):
224			node = self._doc.createElement(name)
225			# Attach the UID of the object as well
226			node.setAttribute("id", api.get_id(self.context))
227			node.setAttribute("uid", api.get_uid(self.context))
228			node.setAttribute("name", api.get_id(self.context))
229			node.setAttribute("meta_type", self.context.meta_type)
230			node.setAttribute("portal_type", api.get_portal_type(self.context))
231			i18n_domain = getattr(self.context, "i18n_domain", None)
232			if i18n and i18n_domain:
233			node.setAttributeNS(I18NURI, "i18n:domain", i18n_domain)
234			self._i18n_props = ("title", "description")
235			return node
236
237			def _exportNode(self):
238			"""Export the object as a DOM node.
239			"""
240			node = self._getObjectNode("object")
241
242			# remember the UID of the item for reference fields
243			node.setAttribute("uid", api.get_uid(self.context))
244
245			# remember the WF Status
246			# TODO: Export the complete Review History
247			state = api.get_workflow_status_of(self.context)
248			node.setAttribute("state", state)
249
250			# Extract AuditLog
251			node.appendChild(self._extractAuditLog(self.context))
252
253			# Extract all fields of the current context
254			node.appendChild(self._extractFields(self.context))
255
256			# Extract all contained objects
257			node.appendChild(self._extractObjects())
258
259			return node
260
261			def _importNode(self, node):
262			"""Import the object from the DOM node.
263			"""
264
265			# set workflow state
266			self._initAuditLog(self.context, node)
267			self._initWorkflow(self.context, node)
268			self._initFields(self.context, node)
269
270			# reindex the object
271			self.context.reindexObject()
272
273			# set a new snapshot
274			# api.snapshot.take_snapshot(self.context)
275
276			obj_id = str(node.getAttribute("name"))
277			self._logger.info("Imported '%r'" % obj_id)
278
279			def _initAuditLog(self, context, node):
280			for child in node.childNodes:
281			if child.nodeName == "auditlog":
282			snapshots = json.loads(child.firstChild.nodeValue)
283			storage = api.snapshot.get_storage(context)
284			storage[:] = map(json.dumps, snapshots)[:]
285			# make sure the object provides `IAuditable`
286			alsoProvides(context, IAuditable)
287			return
288
289			def _initWorkflow(self, context, node):
290			state = node.getAttribute("state")
291
292			if not state:
293			return
294
295			if state == api.get_workflow_status_of(context):
296			return
297
298			wf_state = {
299			"action": None,
300			"actor": None,
301			"comments": "Generic Setup Import",
302			"review_state": state,
303			"time": DateTime(),
304			}
305
306			wf = api.get_tool("portal_workflow")
307			wf_id = wf.getChainFor(context)[0]
308			wf.setStatusOf(wf_id, context, wf_state)
309
310			def _initFields(self, context, node):
311			fields = api.get_fields(context)
312
313			for child in node.childNodes:
314			# we only handle filed nodes
315			if child.nodeName != "field":
316			continue
317
318			name = child.getAttribute("name")
319			field = fields.get(name)
320			if field is None:
321			self._logger.warning("Unrecognized field '{}'".format(name))
322			continue
323
324			importer = queryMultiAdapter((context, field, self.environ), INode)
325			if importer:
326			importer.node = child
327
328			def _extractAuditLog(self, context):
329			snapshots = api.snapshot.get_snapshots(self.context)
330			node = self._doc.createElement("auditlog")
331			child = self._doc.createTextNode(json.dumps(snapshots))
332			node.appendChild(child)
333			return node
334
335			def _extractFields(self, context):
336			fragment = self._doc.createDocumentFragment()
337
338			fields = api.get_fields(context)
339			for name, field in fields.items():
340			# query the field adapter
341			exporter = queryMultiAdapter((context, field, self.environ), INode)
342			if not exporter:
343			continue
344			node = exporter.node
345			if node is not None:
346			fragment.appendChild(node)
347			return fragment
348
349
350			class DXContainerXMLAdapter(ATContentXMLAdapter):
351			"""DX Container XML Importer/Exporter
352			"""
353			adapts(IDexterityContainer, ISetupEnviron)
354
355			def __init__(self, context, environ):
356			super(DXContainerXMLAdapter, self).__init__(context, environ)
357
358
359			class DXItemXMLAdapter(ATContentXMLAdapter):
360			"""DX Item XML Importer/Exporter
361			"""
362			adapts(IDexterityItem, ISetupEnviron)
363
364			def __init__(self, context, environ):
365			super(DXItemXMLAdapter, self).__init__(context, environ)
366
367
368			def create_content_slugs(parent, parent_path, context):
369			"""Helper function to create initial content slugs
370			"""
371			logger.info("create_content_slugs: parent={} parent_path={}".format(
372			repr(parent), parent_path))
373
374			path = "%s%s" % (parent_path, get_id(parent))
375			filename = "%s.xml" % (path)
376			xml = context.readDataFile(filename)
377
378			if xml is None:
379			logger.error("File not found: '{}'".format(filename))
380			return
381
382			# parse the XML data
383			node = parseString(xml)
384
385			if node.nodeName == "#document":
386			node = node.firstChild
387
388			# read the node attributes
389			name = node.getAttribute("name")
390			uid = node.getAttribute("uid")
391			logger.info("::: Processing '{}' (UID {}) in path '{}' :::"
392			.format(name, uid, path))
393
394			def is_object_node(n):
395			return getattr(n, "nodeName", "") == "object"
396
397			def get_child_nodes(n):
398			return getattr(n, "childNodes", [])
399
400			for child in get_child_nodes(node):
401			# only process `<object ../>` nodes
402			if not is_object_node(child):
403			continue
404			# extract node attributes (see `_exportNode` method)
405			child_id = child.getAttribute("name")
406			child_uid = child.getAttribute("uid")
407			portal_type = child.getAttribute("portal_type")
408			# get or create object
409			obj = create_or_get(parent, child_id, child_uid, portal_type)
410			# handle vanished objects
411			if obj is None:
412			logger.warn("Skipping object creation for '{}'".format(path))
413			continue
414			# get the id of the new object
415			obj_id = api.get_id(obj)
416			# track new ID -> old ID
417			ID_MAP[obj_id] = child_id
418			# recursively create contents
419			create_content_slugs(obj, path + "/", context)
420
421
422			def create_or_get(parent, id, uid, portal_type):
423			"""Create or get the object
424			"""
425			# return first level objects directly
426			if api.is_portal(parent):
427			return parent.get(id)
428			# Check both bika_setup (AT, has subfolders) and senaite_setup (DX)
429			bika_setup = api.get_portal().get("bika_setup")
430			if bika_setup and bika_setup == parent:
431			return parent.get(id)
432			elif api.get_senaite_setup() == parent:
433			return parent.get(id)
434
435			# query object by UID
436			query = {
437			"UID": uid,
438			"portal_type": portal_type,
439			"path": {
440			"query": api.get_path(parent),
441			}
442			}
443			results = api.search(query, "uid_catalog")
444			if results:
445			return api.get_object(results[0])
446
447			# create object slug
448			obj = None
449			# get the fti
450			types_tool = api.get_tool("portal_types")
451			fti = types_tool.getTypeInfo(portal_type)
452			# removed
453			if not fti:
454			return None
455			# old style factory
456			if fti.product:
457			# Create AT Content Slug (we take the UID as ID to avoid clashes)
458			obj = _createObjectByType(portal_type, parent, uid)
459			# set the old UID to maintain references
460			obj._setUID(uid)
461			# IMPORTANT: this will generate a new ID by the ID Server config
462			obj.processForm()
463			else:
464			# Create DX Content Slug
465			factory = getUtility(IFactory, fti.factory)
466			tmp_id = str(uid)
467			obj = factory(tmp_id)
468			if hasattr(obj, "_setPortalTypeName"):
469			obj._setPortalTypeName(fti.getId())
470			# set the old UID to maintain references
471			setattr(obj, "_plone.uuid", uid)
472			# IMPORTANT: this will generate a new ID by the ID Server config
473			obj = addContentToContainer(parent, obj, checkConstraints=False)
474
475			return obj
476
477
478			@cache(get_key=lambda *args: 'can-export-%s' % args[1])
			0 ignored issues – show Comprehensibility Best Practice introduced 2024-03-24 10:39 UTC by Report Bug Copy Issue Report The variable `args` does not seem to be defined. Loading history...
479			def can_export_type(portal_type, request):
480			"""Returns whether objects from this type can be exported
481			"""
482			name = "generic_setup_skip_export_types"
483			skip = get_registry_record(name, default=[])
484			return portal_type not in skip
485
486
487			def can_export(obj):
488			"""Decides if the object can be exported or not
489			"""
490			if not api.is_object(obj):
491			return False
492			portal_type = api.get_portal_type(obj)
493			request = api.get_request()
494			return can_export_type(portal_type, request)
495
496
497			def can_import(obj):
498			"""Decides if the object can be imported or not
499			"""
500			if not api.is_object(obj):
501			return False
502			return True
503
504
505			def get_id(obj):
506			if api.is_portal(obj):
507			return SITE_ID
508			oid = api.get_id(obj)
509			# resolve id from mapping
510			rid = ID_MAP.get(oid, oid)
511			return rid.replace(" ", "_")
512
513
514			def exportObjects(obj, parent_path, context):
515			""" Export subobjects recursively.
516			"""
517
518			if not can_export(obj):
519			logger.info("Skipping export of {}".format(repr(obj)))
520			return
521
522			if api.is_portal(obj):
523			# explicitly instantiate the exporter to avoid adapter clash of
524			# Products.CMFCore.exportimport.properties.PropertiesXMLAdapter
525			exporter = SenaiteSiteXMLAdapter(obj, context)
526			else:
527			exporter = queryMultiAdapter((obj, context), IBody)
528
529			path = "%s%s" % (parent_path, get_id(obj))
530			if exporter:
531			if exporter.name:
532			path = "%s%s" % (parent_path, exporter.name)
533			filename = "%s%s" % (path, exporter.suffix)
534			body = exporter.body
535			if body is not None:
536			context.writeDataFile(filename, body, exporter.mime_type)
537			else:
538			raise ValueError("No exporter found for object: %r" % obj)
539
540			if getattr(obj, "objectValues", False):
541			for sub in obj.objectValues():
542			exportObjects(sub, path + "/", context)
543
544
545			def importObjects(obj, parent_path, context):
546			""" Import subobjects recursively.
547			"""
548
549			if not can_import(obj):
550			logger.info("Skipping import of {}".format(repr(obj)))
551			return
552
553			if api.is_portal(obj):
554			# explicitly instantiate the importer to avoid adapter clash of
555			# Products.CMFCore.exportimport.properties.PropertiesXMLAdapter
556			importer = SenaiteSiteXMLAdapter(obj, context)
557			else:
558			importer = queryMultiAdapter((obj, context), IBody)
559
560			path = "%s%s" % (parent_path, get_id(obj))
561			__traceback_info__ = path
562			if importer:
563			if importer.name:
564			path = "%s%s" % (parent_path, importer.name)
565			filename = "%s%s" % (path, importer.suffix)
566			body = context.readDataFile(filename)
567			if body is not None:
568			importer.filename = filename # for error reporting
569			importer.body = body
570
571			if getattr(obj, "objectValues", False):
572			for sub in obj.objectValues():
573			importObjects(sub, path + "/", context)
574
575
576			def export_xml(context):
577			portal = context.getSite()
578			exportObjects(portal, "", context)
579
580
581			def import_xml(context):
582			portal = context.getSite()
583
584			qi = api.get_tool("portal_quickinstaller")
585			installed = qi.isProductInstalled("senaite.core")
586			request = api.get_request()
587
588			# tests call it w/o request
589			if request and not installed:
590			logger.debug("Nothing to import.")
591			return
592
593			if not context.readDataFile("senaite.xml"):
594			logger.debug("Nothing to import.")
595			return
596
597			# create content slugs for UID references
598			create_content_slugs(portal, "", context)
599
600			# import objects
601			importObjects(portal, "", context)
602

senaite / senaite.core

SenaiteSiteXMLAdapter._extractObjects() B last analyzed 2025-12-16 10:03 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

SenaiteSiteXMLAdapter._extractObjects() B
last analyzed 2025-12-16 10:03 UTC