collective.exportimport 1.12

Python 2兼容性

此包与 Python 3 和 Python 2 兼容。根据 Python 版本，将安装其依赖项的不同版本。如果您遇到问题，请在：https://github.com/collective/collective.exportimport/issues 上提交问题

导出

使用 URL /@@export_content 的表单，并选择您想要导出的内容

您可以导出一种或多种类型，整个站点或仅站点中的特定路径。由于项目是按路径顺序导出的，因此导入它们将创建与最初相同的结构。

下载的 json 文件将具有您导出的路径的名称，例如 Plone.json。

成员、关系、本地角色和关系的导出链接已在此表单中列出，但也可以单独调用：/@@export_members、/@@export_relations、/@@export_localroles、/@@export_translations、/@@export_ordering、/@@export_discussion。

导入

使用URL为/@@import_content的表单，上传您想要导入的json文件

成员、关系、本地角色和关系的导入链接已在此表单中列出，但也可以单独调用：/@@import_members、/@@import_relations、/@@import_localroles、/@@import_translations、/@@import_ordering、/@@import_discussion。

在迁移的最后一步，还有一个名为@@reset_dates的视图，该视图将导入内容的修改日期重置为导入的json文件中最初包含的日期。这是必要的，因为在迁移过程中，各种更改可能会导致修改日期的更新。在导入过程中，原始日期作为每个新对象的obj.modification_date_migrated存储，此视图设置此日期。

导出和导入位置

如果您选择“保存到服务器上的文件”，导出视图将保存json文件到您的Plone实例的<var>目录中，位于/var/instance。导入视图将在/var/instance/import下查找文件。这些目录通常在不同Plone实例中，可能在不同的服务器上。

您可以将环境变量‘COLLECTIVE_EXPORTIMPORT_CENTRAL_DIRECTORY’设置为在服务器上或可能网络上添加一个“共享”目录。设置此变量后，collective.exportimport将同时保存到和从同一服务器目录加载.json文件。这样可以节省时间，不需要在导出和导入位置之间移动.json文件。请注意，导出视图将覆盖任何具有相同名称的现有以前的.json文件导出。

迁移

如果不需要就地迁移，可以选择此插件将网站最重要的部分迁移到json，然后将其导入到您目标版本的新Plone实例中

• 从Plone站点导出内容（它支持Plone 4和5，Archetypes和Dexterity，Python 2和3）。

• 将导出的内容导入到新站点（Plone 5.2+，Dexterity，Python 3）

• 导出和导入关系、用户和组及其角色、翻译、本地角色、排序、默认页面、评论、小部件和重定向。

有关如何迁移附加功能（如注释或标记接口）的讨论，请参阅FAQ部分。

其他

您可以使用此插件来

• 将内容存档为JSON。

• 导出数据以准备迁移到另一个系统。

• 将多个plone站点的内容合并到一个。

• 将plone站点作为子站点导入到另一个。

• 导入其他系统的内容，只要它符合所需格式。

• 更新或替换现有数据。

导出内容

导出内容基本上是plone.restapi序列化器的包装器

from plone.restapi.interfaces import ISerializeToJson
from zope.component import getMultiAdapter

serializer = getMultiAdapter((obj, request), ISerializeToJson)
data = serializer(include_items=False)

导入内容

导入内容是plone.restapi反序列化器的复杂包装器

from plone.restapi.interfaces import IDeserializeFromJson
from zope.component import getMultiAdapter

container.invokeFactory(item['@type'], item['id'])
deserializer = getMultiAdapter((new, self.request), IDeserializeFromJson)
new = deserializer(validate_all=False, data=item)

用于迁移

此包的主要用例是从一个Plone版本迁移到另一个版本。

将Archetypes内容导出并作为Dexterity内容导入可以正常工作，但由于字段名的更改，一些设置会丢失。例如，从导航中排除内容的设置已从excludeFromNav重命名为exclude_from_nav。

要修复此问题，您可以检查“修改导出数据以进行迁移”复选框。这将修改导出过程中的数据

• 删除未使用的数据（例如，next_item和components）

• 删除所有关系字段

• 更改Archetypes和Dexterity之间更改的一些字段名

  • excludeFromNav → exclude_from_nav

  • allowDiscussion → allow_discussion

  • subject → subjects

  • expirationDate → expires

  • effectiveDate → effective

  • creation_date → created

  • modification_date → modified

  • startDate → start

  • endDate → end

  • openEnd → open_end

  • wholeDay → whole_day

  • contactEmail → contact_email

  • contactName → contact_name

  • contactPhone → contact_phone

• 更新自Plone 4以来更改的文件夹和集合的视图名称。

• 将ATTopic及其标准导出到带有查询字符串的集合中。

• 更新集合标准。

• 内容和小部件的Richtext字段中的链接和图片自Plone 4以来已更改。视图/@@fix_html允许您修复这些问题。

控制导入内容的创建

您可以选择四种方式处理已存在的内容：

• 跳过：完全不导入

• 替换：删除项目并创建新的

• 更新：重用并仅覆盖导入的数据

• 忽略：使用新ID创建

导入的内容最初使用invokeFactory创建，使用导出项的portal_type和id，在反序列化其余数据之前。您可以通过指定一个字典factory_kwargs来设置附加值，该字典将被传递给工厂。这样，您可以设置导入对象上的值，这些值是IObjectAddedEvent订阅者预期的。

导出版本化内容

由于plone.restapi中的错误（https://github.com/plone/plone.restapi/issues/1335），导出Archetypes内容的版本将不起作用。要使导出工作，您需要使用7.7.0和8.0.0之间的版本（如果已发布）或7.x.x分支的源签出。

导出示例

from collective.exportimport.export_content import ExportContent

class CustomExportContent(ExportContent):

    QUERY = {
        'Document': {'review_state': ['published', 'pending']},
    }

    DROP_PATHS = [
        '/Plone/userportal',
        '/Plone/en/obsolete_content',
    ]

    DROP_UIDS = [
        '71e3e0a6f06942fea36536fbed0f6c42',
    ]

    def update(self):
        """Use this to override stuff before the export starts
        (e.g. force a specific language in the request)."""

    def start(self):
        """Hook to do something before export."""

    def finish(self):
        """Hook to do something after export."""

    def global_obj_hook(self, obj):
        """Inspect the content item before serialisation data.
        Bad: Changing the content-item is a horrible idea.
        Good: Return None if you want to skip this particular object.
        """
        return obj

    def global_dict_hook(self, item, obj):
        """Use this to modify or skip the serialized data.
        Return None if you want to skip this particular object.
        """
        return item

    def dict_hook_document(self, item, obj):
        """Use this to modify or skip the serialized data by type.
        Return the modified dict (item) or None if you want to skip this particular object.
        """
        return item

将其与自己的browserlayer注册以覆盖默认值。

<browser:page
    name="export_content"
    for="zope.interface.Interface"
    class=".custom_export.CustomExportContent"
    layer="My.Custom.IBrowserlayer"
    permission="cmf.ManagePortal"
    />

导入示例

from collective.exportimport.import_content import ImportContent

class CustomImportContent(ImportContent):

    CONTAINER = {'Event': '/imported-events'}

    # These fields will be ignored
    DROP_FIELDS = ['relatedItems']

    # Items with these uid will be ignored
    DROP_UIDS = ['04d1477583c74552a7fcd81a9085c620']

    # These paths will be ignored
    DROP_PATHS = ['/Plone/doormat/', '/Plone/import_files/']

    # Default values for some fields
    DEFAULTS = {'which_price': 'normal'}

    def start(self):
        """Hook to do something before importing one file."""

    def finish(self):
        """Hook to do something after importing one file."""

    def global_dict_hook(self, item):
        if isinstance(item.get('description', None), dict):
            item['description'] = item['description']['data']
        if isinstance(item.get('rights', None), dict):
            item['rights'] = item['rights']['data']
        return item

    def dict_hook_customtype(self, item):
        # change the type
        item['@type'] = 'anothertype'
        # drop a field
        item.pop('experiences', None)
        return item

    def handle_file_container(self, item):
        """Use this to specify the container in which to create the item in.
        Return the container for this particular object.
        """
        return self.portal['imported_files']

注册它

<browser:page
    name="import_content"
    for="zope.interface.Interface"
    class=".custom_import.CustomImportContent"
    layer="My.Custom.IBrowserlayer"
    permission="cmf.ManagePortal"
    />

自动化导出和导入

运行所有导出并将所有数据保存到var/instance/

from plone import api
from Products.Five import BrowserView

class ExportAll(BrowserView):

    def __call__(self):
        export_content = api.content.get_view("export_content", self.context, self.request)
        self.request.form["form.submitted"] = True
        export_content(
            portal_type=["Folder", "Document", "News Item", "File", "Image"],  # only export these
            include_blobs=2,  # Export files and images as blob paths
            download_to_server=True)

        other_exports = [
            "export_relations",
            "export_members",
            "export_translations",
            "export_localroles",
            "export_ordering",
            "export_defaultpages",
            "export_discussion",
            "export_portlets",
            "export_redirects",
        ]
        for name in other_exports:
            view = api.content.get_view(name, portal, request)
            # This saves each export in var/instance/export_xxx.json
            view(download_to_server=True)

        # Important! Redirect to prevent infinite export loop :)
        return self.request.response.redirect(self.context.absolute_url())

使用上面示例中的导出数据进行所有导入

from collective.exportimport.fix_html import fix_html_in_content_fields
from collective.exportimport.fix_html import fix_html_in_portlets
from pathlib import Path
from plone import api
from Products.Five import BrowserView


class ImportAll(BrowserView):

    def __call__(self):
        portal = api.portal.get()

        # Import content
        view = api.content.get_view("import_content", portal, request)
        request.form["form.submitted"] = True
        request.form["commit"] = 500
        view(server_file="Plone.json", return_json=True)
        transaction.commit()

        # Run all other imports
        other_imports = [
            "relations",
            "members",
            "translations",
            "localroles",
            "ordering",
            "defaultpages",
            "discussion",
            "portlets",
            "redirects",
        ]
        cfg = getConfiguration()
        directory = Path(cfg.clienthome) / "import"
        for name in other_imports:
            view = api.content.get_view(f"import_{name}", portal, request)
            path = Path(directory) / f"export_{name}.json"
            results = view(jsonfile=path.read_text(), return_json=True)
            logger.info(results)
            transaction.commit()

        # Run cleanup steps
        results = fix_html_in_content_fields()
        logger.info("Fixed html for %s content items", results)
        transaction.commit()

        results = fix_html_in_portlets()
        logger.info("Fixed html for %s portlets", results)
        transaction.commit()

        reset_dates = api.content.get_view("reset_dates", portal, request)
        reset_dates()
        transaction.commit()

在导出期间使用 global_obj_hook

在导出期间使用global_obj_hook来检查内容并决定跳过它。

def global_obj_hook(self, obj):
    # Drop subtopics
    if obj.portal_type == "Topic" and obj.__parent__.portal_type == "Topic":
        return

    # Drop files and images from PFG formfolders
    if obj.__parent__.portal_type == "FormFolder":
        return
    return obj

在导出期间使用 dict-hooks

在导出期间使用global_dict_hook来检查内容并修改序列化的json。您还可以使用dict_hook_<somecontenttype>来更好地结构您的代码以提高可读性。

有时您需要在导出期间的global_dict_hook中处理数据，并在导入期间对应的global_object_hook中的相应代码中处理这些数据。

以下关于地方性工作流策略的示例是这种模式的完美示例。

导出/导入 placeful 工作流程策略

导出

def global_dict_hook(self, item, obj):
    if obj.isPrincipiaFolderish and ".wf_policy_config" in obj.keys():
        wf_policy = obj[".wf_policy_config"]
        item["exportimport.workflow_policy"] = {
            "workflow_policy_below": wf_policy.workflow_policy_below,
            "workflow_policy_in": wf_policy.workflow_policy_in,
        }
    return item

导入

def global_obj_hook(self, obj, item):
    wf_policy = item.get("exportimport.workflow_policy")
    if wf_policy:
        obj.manage_addProduct["CMFPlacefulWorkflow"].manage_addWorkflowPolicyConfig()
        wf_policy_config = obj[".wf_policy_config"]
        wf_policy_config.setPolicyIn(wf_policy["workflow_policy_in"], update_security=True)
        wf_policy_config.setPolicyBelow(wf_policy["workflow_policy_below"], update_security=True)

在导入期间使用 dict-hooks

您可以使用global_dict_hook或dict_hook_<contenttype>在导入期间执行许多修复。

我们在此处防止到期日期在有效日期之前，因为这会导致反序列化时的验证错误。

def global_dict_hook(self, item):
    effective = item.get('effective', None)
    expires = item.get('expires', None)
    if effective and expires and expires <= effective:
        item.pop('expires')
    return item

我们在此处删除创建者中的空行。

def global_dict_hook(self, item):
    item["creators"] = [i for i in item.get("creators", []) if i]
    return item

此示例在导入期间将PloneHelpCenter迁移到简单的文件夹/文档结构。还有其他一些类型要处理（作为文件夹或文档），但您应该明白这个概念，对吗？

def dict_hook_helpcenter(self, item):
    item["@type"] = "Folder"
    item["layout"] = "listing_view"
    return item

def dict_hook_helpcenterglossary(self, item):
    item["@type"] = "Folder"
    item["layout"] = "listing_view"
    return item

def dict_hook_helpcenterinstructionalvideo(self, item):
    item["@type"] = "File"
    if item.get("video_file"):
        item["file"] = item["video_file"]
    return item

def dict_hook_helpcenterlink(self, item):
    item["@type"] = "Link"
    item["remoteUrl"] = item.get("url", None)
    return item

def dict_hook_helpcenterreferencemanualpage(self, item):
    item["@type"] = "Document"
    return item

如果您在导入期间更改类型，您需要注意其他引用类型的情况。例如，集合查询（请参阅“修复无效的集合查询”）或约束（请参阅此处）。

PORTAL_TYPE_MAPPING = {
    "Topic": "Collection",
    "FormFolder": "EasyForm",
    "HelpCenter": "Folder",
}

def global_dict_hook(self, item):
    if item.get("exportimport.constrains"):
        types_fixed = []
        for portal_type in item["exportimport.constrains"]["locally_allowed_types"]:
            if portal_type in PORTAL_TYPE_MAPPING:
                types_fixed.append(PORTAL_TYPE_MAPPING[portal_type])
            elif portal_type in ALLOWED_TYPES:
                types_fixed.append(portal_type)
        item["exportimport.constrains"]["locally_allowed_types"] = list(set(types_fixed))

        types_fixed = []
        for portal_type in item["exportimport.constrains"]["immediately_addable_types"]:
            if portal_type in PORTAL_TYPE_MAPPING:
                types_fixed.append(PORTAL_TYPE_MAPPING[portal_type])
            elif portal_type in ALLOWED_TYPES:
                types_fixed.append(portal_type)
        item["exportimport.constrains"]["immediately_addable_types"] = list(set(types_fixed))
    return item

更改工作流程

REVIEW_STATE_MAPPING = {
    "internal": "published",
    "internally_published": "published",
    "obsolete": "private",
    "hidden": "private",
}

def global_dict_hook(self, item):
    if item.get("review_state") in REVIEW_STATE_MAPPING:
        item["review_state"] = REVIEW_STATE_MAPPING[item["review_state"]]
    return item

导出/导入注释

Plone的一些核心功能（例如评论）使用注解来存储数据。核心功能已经覆盖，但您的自定义代码或社区插件也可能使用注解。以下是迁移它们的方法。

导出：仅导出您真正需要的注解。

from zope.annotation.interfaces import IAnnotations
ANNOTATIONS_TO_EXPORT = [
    "syndication_settings",
]
ANNOTATIONS_KEY = 'exportimport.annotations'

class CustomExportContent(ExportContent):

    def global_dict_hook(self, item, obj):
        item = self.export_annotations(item, obj)
        return item

    def export_annotations(self, item, obj):
        results = {}
        annotations = IAnnotations(obj)
        for key in ANNOTATIONS_TO_EXPORT:
            data = annotations.get(key)
            if data:
                results[key] = IJsonCompatible(data, None)
        if results:
            item[ANNOTATIONS_KEY] = results
        return item

导入:

from zope.annotation.interfaces import IAnnotations
ANNOTATIONS_KEY = "exportimport.annotations"

class CustomImportContent(ImportContent):

    def global_obj_hook(self, obj, item):
        item = self.import_annotations(obj, item)
        return item

    def import_annotations(self, obj, item):
        annotations = IAnnotations(obj)
        for key in item.get(ANNOTATIONS_KEY, []):
            annotations[key] = item[ANNOTATIONS_KEY][key]
        return item

一些功能也在门户的注解中存储数据，例如 plone.contentrules.localassignments，plone.portlets.categoryblackliststatus，plone.portlets.contextassignments，syndication_settings。根据您的需求，您可能还想导出和导入这些内容。

导出/导入标记接口

导出：您可能只想导出所需的标记接口。在决定要迁移什么之前，检查门户中所有使用的标记接口列表是个好主意。

from zope.interface import directlyProvidedBy

MARKER_INTERFACES_TO_EXPORT = [
    "collective.easyslider.interfaces.ISliderPage",
    "plone.app.layout.navigation.interfaces.INavigationRoot",
]
MARKER_INTERFACES_KEY = "exportimport.marker_interfaces"

class CustomExportContent(ExportContent):

    def global_dict_hook(self, item, obj):
        item = self.export_marker_interfaces(item, obj)
        return item

    def export_marker_interfaces(self, item, obj):
        interfaces = [i.__identifier__ for i in directlyProvidedBy(obj)]
        interfaces = [i for i in interfaces if i in MARKER_INTERFACES_TO_EXPORT]
        if interfaces:
            item[MARKER_INTERFACES_KEY] = interfaces
        return item

导入:

from plone.dexterity.utils import resolveDottedName
from zope.interface import alsoProvides

MARKER_INTERFACES_KEY = "exportimport.marker_interfaces"

class CustomImportContent(ImportContent):

    def global_obj_hook_before_deserializing(self, obj, item):
        """Apply marker interfaces before deserializing."""
        for iface_name in item.pop(MARKER_INTERFACES_KEY, []):
            try:
                iface = resolveDottedName(iface_name)
                if not iface.providedBy(obj):
                    alsoProvides(obj, iface)
                    logger.info("Applied marker interface %s to %s", iface_name, obj.absolute_url())
            except ModuleNotFoundError:
                pass
        return obj, item

导入时跳过版本控制

版本控制的事件处理器可能会严重影响您的导入速度。在导入之前跳过它是明智的。

VERSIONED_TYPES = [
    "Document",
    "News Item",
    "Event",
    "Link",
]

def start(self):
    self.items_without_parent = []
    portal_types = api.portal.get_tool("portal_types")
    for portal_type in VERSIONED_TYPES:
        fti = portal_types.get(portal_type)
        behaviors = list(fti.behaviors)
        if 'plone.versioning' in behaviors:
            logger.info(f"Disable versioning for {portal_type}")
            behaviors.remove('plone.versioning')
        fti.behaviors = behaviors

完成所有导入和修复后，重新启用版本控制并创建初始版本，例如在 @@import_all 视图中。

from Products.CMFEditions.interfaces.IModifier import FileTooLargeToVersionError

VERSIONED_TYPES = [
    "Document",
    "News Item",
    "Event",
    "Link",
]

class ImportAll(BrowserView):

    # re-enable versioning
    portal_types = api.portal.get_tool("portal_types")
    for portal_type in VERSIONED_TYPES:
        fti = portal_types.get(portal_type)
        behaviors = list(fti.behaviors)
        if "plone.versioning" not in behaviors:
            behaviors.append("plone.versioning")
            logger.info(f"Enable versioning for {portal_type}")
        if "plone.locking" not in behaviors:
            behaviors.append("plone.locking")
            logger.info(f"Enable locking for {portal_type}")
        fti.behaviors = behaviors
    transaction.get().note("Re-enabled versioning")
    transaction.commit()

    # create initial version for all versioned types
    logger.info("Creating initial versions")
    portal_repository = api.portal.get_tool("portal_repository")
    brains = api.content.find(portal_type=VERSIONED_TYPES)
    total = len(brains)
    for index, brain in enumerate(brains):
        obj = brain.getObject()
        try:
            portal_repository.save(obj=obj, comment="Imported Version")
        except FileTooLargeToVersionError:
            pass
        if not index % 1000:
            msg = f"Created versions for {index} of {total} items."
            logger.info(msg)
            transaction.get().note(msg)
            transaction.commit()
    msg = "Created initial versions"
    transaction.get().note(msg)
    transaction.commit()

处理验证错误

有时在导入过程中会得到验证错误，因为数据无法验证。这可能发生在字段中的选项由网站中的内容生成时。在这些情况下，在导入内容时，您不能确定所有选项都已在门户中存在。

这也可能发生在您有依赖于导入时不存在的内容或配置的验证器时。

有两种处理这些问题的方法

• 使用简单的setter绕过restapi使用的验证

• 将导入推迟到所有其他导入都运行完毕

SIMPLE_SETTER_FIELDS = {
    "ALL": ["some_shared_field"],
    "CollaborationFolder": ["allowedPartnerDocTypes"],
    "DocType": ["automaticTransferTargets"],
    "DPDocument": ["scenarios"],
    "DPEvent" : ["Status"],
}

class CustomImportContent(ImportContent):

    def global_dict_hook(self, item):
        simple = {}
        for fieldname in SIMPLE_SETTER_FIELDS.get("ALL", []):
            if fieldname in item:
                value = item.pop(fieldname)
                if value:
                    simple[fieldname] = value
        for fieldname in SIMPLE_SETTER_FIELDS.get(item["@type"], []):
            if fieldname in item:
                value = item.pop(fieldname)
                if value:
                    simple[fieldname] = value
        if simple:
            item["exportimport.simplesetter"] = simple

    def global_obj_hook_before_deserializing(self, obj, item):
        """Hook to modify the created obj before deserializing the data.
        """
        # import simplesetter data before the rest
        for fieldname, value in item.get("exportimport.simplesetter", {}).items():
            setattr(obj, fieldname, value)

from plone.restapi.interfaces import IDeserializeFromJson
from zope.annotation.interfaces import IAnnotations
from zope.component import getMultiAdapter

DEFERRED_KEY = "exportimport.deferred"
DEFERRED_FIELD_MAPPING = {
    "talk": ["somefield"],
    "speaker": [
        "custom_field",
        "another_field",
    ]
}
SIMPLE_SETTER_FIELDS = {"custom_type": ["another_field"]}

class CustomImportContent(ImportContent):

    def global_dict_hook(self, item):
        # Move deferred values to a different key to not deserialize.
        # This could also be done during export.
        item[DEFERRED_KEY] = {}
        for fieldname in DEFERRED_FIELD_MAPPING.get(item["@type"], []):
            if item.get(fieldname):
                item[DEFERRED_KEY][fieldname] = item.pop(fieldname)
        return item

    def global_obj_hook(self, obj, item):
        # Store deferred data in an annotation.
        deferred = item.get(DEFERRED_KEY, {})
        if deferred:
            annotations = IAnnotations(obj)
            annotations[DEFERRED_KEY] = {}
            for key, value in deferred.items():
                annotations[DEFERRED_KEY][key] = value

class ImportDeferred(BrowserView):

    def __call__(self):
        # This example reuses the form export_other.pt from collective.exportimport
        self.title = "Import deferred data"
        if not self.request.form.get("form.submitted", False):
            return self.index()
        portal = api.portal.get()
        self.results = []
        for brain in api.content.find(DEFERRED_FIELD_MAPPING.keys()):
            obj = brain.getObject()
            self.import_deferred(obj)
        api.portal.show_message(f"Imported deferred data for {len(self.results)} items!", self.request)

    def import_deferred(self, obj):
        annotations = IAnnotations(obj, {})
        deferred = annotations.get(DEFERRED_KEY, None)
        if not deferred:
            return
        # Shortcut for simple fields (e.g. storing strings, uuids etc.)
        for fieldname in SIMPLE_SETTER_FIELDS.get(obj.portal_type, []):
            value = deferred.pop(fieldname, None)
            if value:
                setattr(obj, fieldname, value)
        if not deferred:
            return
        # This approach validates the values and converts more complex data
        deserializer = getMultiAdapter((obj, self.request), IDeserializeFromJson)
        try:
            obj = deserializer(validate_all=False, data=deferred)
        except Exception as e:
            logger.info("Error while importing deferred data for %s", obj.absolute_url(), exc_info=True)
            logger.info("Data: %s", deferred)
        else:
            self.results.append(obj.absolute_url())
        # cleanup
        del annotations[DEFERRED_KEY]

<browser:page
    name="import_deferred"
    for="zope.interface.Interface"
    class=".import_content.ImportDeferred"
    template="export_other.pt"
    permission="cmf.ManagePortal"
    />

处理 LinguaPlone 内容

导出

def global_dict_hook(self, item, obj):
    # Find language of the nearest parent with a language
    # Usefull for LinguaPlone sites where some content is languageindependent
    parent = obj.__parent__
    for ancestor in parent.aq_chain:
        if IPloneSiteRoot.providedBy(ancestor):
            # keep language for root content
            nearest_ancestor_lang = item["language"]
            break
        if getattr(ancestor, "getLanguage", None) and ancestor.getLanguage():
            nearest_ancestor_lang = ancestor.getLanguage()
            item["parent"]["language"] = nearest_ancestor_lang
            break

    # This forces "wrong" languages to the nearest parents language
    if "language" in item and item["language"] != nearest_ancestor_lang:
        logger.info(u"Forcing %s (was %s) for %s %s ", nearest_ancestor_lang, item["language"], item["@type"], item["@id"])
        item["language"] = nearest_ancestor_lang

    # set missing language
    if not item.get("language"):
        item["language"] = nearest_ancestor_lang

    # add info on translations to help find the right container
    # usually this idone by export_translations
    # but when migrating from LP to pam you sometimes want to check the
    # tranlation info during import
    if getattr(obj.aq_base, "getTranslations", None) is not None:
        translations = obj.getTranslations()
        if translations:
            item["translation"] = {}
            for lang in translations:
                uuid = IUUID(translations[lang][0], None)
                if uuid == item["UID"]:
                    continue
                translation = translations[lang][0]
                if not lang:
                    lang = "no_language"
                item["translation"][lang] = translation.absolute_url()

导入

def global_dict_hook(self, item):

    # Adapt this to your site
    languages = ["en", "fr", "de"]
    default_language = "en"
    portal_id = "Plone"

    # No language => lang of parent or default
    if item.get("language") not in languages:
        if item["parent"].get("language"):
            item["language"] = item["parent"]["language"]
        else:
            item["language"] = default_language

    lang = item["language"]

    if item["parent"].get("language") != item["language"]:
        logger.debug(f"Inconsistent lang: item is {lang}, parent is {item['parent'].get('language')} for {item['@id']}")

    # Move item to the correct language-root-folder
    # This is only relevant for items in the site-root.
    # Most items containers are usually looked up by the uuid of the old parent
    url = item["@id"]
    parent_url = item["parent"]["@id"]

    url = url.replace(f"/{portal_id}/", f"/{portal_id}/{lang}/", 1)
    parent_url = parent_url.replace(f"/{portal_id}", f"/{portal_id}/{lang}", 1)

    item["@id"] = url
    item["parent"]["@id"] = parent_url

    return item

处理无父项项的替代方法

通常，最好导出和记录无法找到容器的项目，而不是重新创建原始结构。

def update(self):
    self.items_without_parent = []

def create_container(self, item):
    # Override create_container to never create parents
    self.items_without_parent.append(item)

def finish(self):
    # export content without parents
    if self.items_without_parent:
        data = json.dumps(self.items_without_parent, sort_keys=True, indent=4)
        number = len(self.items_without_parent)
        cfg = getConfiguration()
        filename = 'content_without_parent.json'
        filepath = os.path.join(cfg.clienthome, filename)
        with open(filepath, 'w') as f:
            f.write(data)
        msg = u"Saved {} items without parent to {}".format(number, filepath)
        logger.info(msg)
        api.portal.show_message(msg, self.request)

导出/导入 Zope 用户

默认情况下，只有存储在Plone中的用户和组会被导出/导入。您可以使用这种方法导出/导入Zope用户。

导出

from collective.exportimport.export_other import BaseExport
from plone import api

import six

class ExportZopeUsers(BaseExport):

    AUTO_ROLES = ["Authenticated"]

    def __call__(self, download_to_server=False):
        self.title = "Export Zope users"
        self.download_to_server = download_to_server
        portal = api.portal.get()
        app = portal.__parent__
        self.acl = app.acl_users
        self.pms = api.portal.get_tool("portal_membership")
        data = self.all_zope_users()
        self.download(data)

    def all_zope_users(self):
        results = []
        for user in self.acl.searchUsers():
            data = self._getUserData(user["userid"])
            data['title'] = user['title']
            results.append(data)
        return results

    def _getUserData(self, userId):
        member = self.pms.getMemberById(userId)
        roles = [
            role
            for role in member.getRoles()
            if role not in self.AUTO_ROLES
        ]
        # userid, password, roles
        props = {
            "username": userId,
            "password": json_compatible(self._getUserPassword(userId)),
            "roles": json_compatible(roles),
        }
        return props

    def _getUserPassword(self, userId):
        users = self.acl.users
        passwords = users._user_passwords
        password = passwords.get(userId, "")
        return password

导入:

class ImportZopeUsers(BrowserView):

    def __call__(self, jsonfile=None, return_json=False):
        if jsonfile:
            self.portal = api.portal.get()
            status = "success"
            try:
                if isinstance(jsonfile, str):
                    return_json = True
                    data = json.loads(jsonfile)
                elif isinstance(jsonfile, FileUpload):
                    data = json.loads(jsonfile.read())
                else:
                    raise ("Data is neither text nor upload.")
            except Exception as e:
                status = "error"
                logger.error(e)
                api.portal.show_message(
                    u"Failure while uploading: {}".format(e),
                    request=self.request,
                )
            else:
                members = self.import_members(data)
                msg = u"Imported {} members".format(members)
                api.portal.show_message(msg, self.request)
            if return_json:
                msg = {"state": status, "msg": msg}
                return json.dumps(msg)

        return self.index()

    def import_members(self, data):
        app = self.portal.__parent__
        acl = app.acl_users
        counter = 0
        for item in data:
            username = item["username"]
            password = item.pop("password")
            roles = item.pop("roles", [])
            if not username or not password or not roles:
                continue
            title = item.pop("title", None)
            acl.users.addUser(username, title, password)
            for role in roles:
                acl.roles.assignRoleToPrincipal(role, username)
            counter += 1
        return counter

导出/导入属性，注册设置和已安装的附加组件

当您迁移多个手动配置的类似站点时，导出和导入手动设置的配置可能很有用。

from collective.exportimport.export_other import BaseExport
from logging import getLogger
from plone import api
from plone.restapi.serializer.converters import json_compatible

logger = getLogger(__name__)


class ExportSettings(BaseExport):
    """Export various settings for haiku sites
    """

    def __call__(self, download_to_server=False):
        self.title = "Export installed add-ons various settings"
        self.download_to_server = download_to_server
        if not self.request.form.get("form.submitted", False):
            return self.index()

        data = self.export_settings()
        self.download(data)

    def export_settings(self):
        results = {}
        addons = []
        qi = api.portal.get_tool("portal_quickinstaller")
        for product in qi.listInstalledProducts():
            if product["id"].startswith("myproject."):
                addons.append(product["id"])
        results["addons"] = addons

        portal = api.portal.get()
        registry = {}
        registry["plone.email_from_name"] = portal.getProperty('email_from_name', '')
        registry["plone.email_from_address"] = portal.getProperty('email_from_address', '')
        registry["plone.smtp_host"] = getattr(portal.MailHost, 'smtp_host', '')
        registry["plone.smtp_port"] = int(getattr(portal.MailHost, 'smtp_port', 25))
        registry["plone.smtp_userid"] = portal.MailHost.get('smtp_user_id')
        registry["plone.smtp_pass"] = portal.MailHost.get('smtp_pass')
        registry["plone.site_title"] = portal.title

        portal_properties = api.portal.get_tool("portal_properties")
        iprops = portal_properties.imaging_properties
        registry["plone.allowed_sizes"] = iprops.getProperty('allowed_sizes')
        registry["plone.quality"] = iprops.getProperty('quality')
        site_props = portal_properties.site_properties
        if site_props.hasProperty("webstats_js"):
            registry["plone.webstats_js"] = site_props.webstats_js
        results["registry"] = json_compatible(registry)
        return results

from logging import getLogger
from plone import api
from plone.registry.interfaces import IRegistry
from Products.CMFPlone.utils import get_installer
from Products.Five import BrowserView
from zope.component import getUtility
from ZPublisher.HTTPRequest import FileUpload

import json

logger = getLogger(__name__)

class ImportSettings(BrowserView):
    """Import various settings"""

    def __call__(self, jsonfile=None, return_json=False):
        if jsonfile:
            self.portal = api.portal.get()
            status = "success"
            try:
                if isinstance(jsonfile, str):
                    return_json = True
                    data = json.loads(jsonfile)
                elif isinstance(jsonfile, FileUpload):
                    data = json.loads(jsonfile.read())
                else:
                    raise ("Data is neither text nor upload.")
            except Exception as e:
                status = "error"
                logger.error(e)
                api.portal.show_message(
                    "Failure while uploading: {}".format(e),
                    request=self.request,
                )
            else:
                self.import_settings(data)
                msg = "Imported addons and settings"
                api.portal.show_message(msg, self.request)
            if return_json:
                msg = {"state": status, "msg": msg}
                return json.dumps(msg)

        return self.index()

    def import_settings(self, data):
        installer = get_installer(self.context)
        for addon in data["addons"]:
            if not installer.is_product_installed(addon) and installer.is_product_installable(addon):
                installer.install_product(addon)
                logger.info(f"Installed addon {addon}")
        registry = getUtility(IRegistry)
        for key, value in data["registry"].items():
            registry[key] = value
            logger.info(f"Imported record {key}: {value}")

将 PloneFormGen 迁移到 Easyform

要能够将 PFG 导出为 easyform，您应该在您的旧站点的 collective.easyform 中使用 migration_features_1.x 分支。Easyform 不需要安装，我们只需要 fields_model 和 actions_model 这两个方法。

导出

def dict_hook_formfolder(self, item, obj):
    item["@type"] = "EasyForm"
    item["is_folderish"] = False

    from collective.easyform.migration.fields import fields_model
    from collective.easyform.migration.actions import actions_model

    # this does most of the heavy lifting...
    item["fields_model"] = fields_model(obj)
    item["actions_model"] = actions_model(obj)

    # handle thankspage
    pfg_thankspage = obj.get(obj.getThanksPage(), None)
    if pfg_thankspage:
        item["thankstitle"] = pfg_thankspage.title
        item["thanksdescription"] = pfg_thankspage.Description()
        item["showAll"] = pfg_thankspage.showAll
        item["showFields"] = pfg_thankspage.showFields
        item["includeEmpties"] = pfg_thankspage.includeEmpties
        item["thanksPrologue"] = json_compatible(pfg_thankspage.thanksPrologue.raw)
        item["thanksEpilogue"] = json_compatible(pfg_thankspage.thanksEpilogue.raw)

    # optional
    item["exportimport._inputStorage"] = self.export_saved_data(obj)

    # Drop some PFG fields no longer needed
    obsolete_fields = [
        "layout",
        "actionAdapter",
        "checkAuthenticator",
        "constrainTypesMode",
        "location",
        "thanksPage",
    ]
    for key in obsolete_fields:
        item.pop(key, None)

    # optional: disable tabs for imported forms
    item["form_tabbing"] = False

    # fix some custom validators
    replace_mapping = {
        "request.form['": "request.form['form.widgets.",
        "request.form.get('": "request.form.get('form.widgets.",
        "member and member.id or ''": "member and member.getProperty('id', '') or ''",
    }

    # fix overrides in actions and fields to use form.widgets.xyz instead of xyz
    for schema in ["actions_model", "fields_model"]:
        for old, new in replace_mapping.items():
            if old in item[schema]:
                item[schema] = item[schema].replace(old, new)

        # add your own fields if you have these issues...
        for fieldname in [
            "email",
            "replyto",
        ]:
            if "request/form/{}".format(fieldname) in item[schema]:
                item[schema] = item[schema].replace("request/form/{}".format(fieldname), "python: request.form.get('form.widgets.{}')".format(fieldname))

    return item

def export_saved_data(self, obj):
    actions = {}
    for data_adapter in obj.objectValues("FormSaveDataAdapter"):
        data_adapter_name = data_adapter.getId()
        actions[data_adapter_name] = {}
        cols = data_adapter.getColumnNames()
        column_count_mismatch = False
        for idx, row in enumerate(data_adapter.getSavedFormInput()):
            if len(row) != len(cols):
                column_count_mismatch = True
                logger.debug("Column count mismatch at row %s", idx)
                continue
            data = {}
            for key, value in zip(cols, row):
                data[key] = json_compatible(value)
            id_ = int(time() * 1000)
            while id_ in actions[data_adapter_name]:  # avoid collisions during export
                id_ += 1
            data["id"] = id_
            actions[data_adapter_name][id_] = data
        if column_count_mismatch:
            logger.info(
                "Number of columns does not match for all rows. Some data were skipped in "
                "data adapter %s/%s",
                "/".join(obj.getPhysicalPath()),
                data_adapter_name,
            )
    return actions

将导出的 PloneFormGen 数据导入到 Easyform

def obj_hook_easyform(self, obj, item):
    if not item.get("exportimport._inputStorage"):
        return
    from collective.easyform.actions import SavedDataBTree
    from persistent.mapping import PersistentMapping
    if not hasattr(obj, '_inputStorage'):
        obj._inputStorage = PersistentMapping()
    for name, data in item["exportimport._inputStorage"].items():
        obj._inputStorage[name] = SavedDataBTree()
        for key, row in data.items():
             obj._inputStorage[name][int(key)] = row

导出和导入 collective.cover 内容

导出

from collective.exportimport.serializer import get_dx_blob_path
from plone.app.textfield.value import RichTextValue
from plone.namedfile.file import NamedBlobImage
from plone.restapi.interfaces import IJsonCompatible
from z3c.relationfield import RelationValue
from zope.annotation.interfaces import IAnnotations

def global_dict_hook(self, item, obj):
    item = self.handle_cover(item, obj)
    return item

def handle_cover(self, item, obj):
    if ICover.providedBy(obj):
        item['tiles'] = {}
        annotations = IAnnotations(obj)
        for tile in obj.get_tiles():
            annotation_key = 'plone.tiles.data.{}'.format(tile['id'])
            annotation = annotations.get(annotation_key, None)
            if annotation is None:
                continue
            tile_data = self.serialize_tile(annotation)
            tile_data['type'] = tile['type']
            item['tiles'][tile['id']] = tile_data
    return item

def serialize_tile(self, annotation):
    data = {}
    for key, value in annotation.items():
        if isinstance(value, RichTextValue):
            value = value.raw
        elif isinstance(value, RelationValue):
            value = value.to_object.UID()
        elif isinstance(value, NamedBlobImage):
            blobfilepath = get_dx_blob_path(value)
            if not blobfilepath:
                continue
            value = {
                "filename": value.filename,
                "content-type": value.contentType,
                "size": value.getSize(),
                "blob_path": blobfilepath,
            }
        data[key] = IJsonCompatible(value, None)
    return data

导入

from collections import defaultdict
from collective.exportimport.import_content import get_absolute_blob_path
from plone.app.textfield.interfaces import IRichText
from plone.app.textfield.interfaces import IRichTextValue
from plone.namedfile.file import NamedBlobImage
from plone.namedfile.interfaces import INamedBlobImageField
from plone.tiles.interfaces import ITileType
from zope.annotation.interfaces import IAnnotations
from zope.component import getUtilitiesFor
from zope.schema import getFieldsInOrder

COVER_CONTENT = [
    "collective.cover.content",
]

def global_obj_hook(self, obj, item):
    if item["@type"] in COVER_CONTENT and "tiles" in item:
        item = self.import_tiles(obj, item)

def import_tiles(self, obj, item):
    RICHTEXT_TILES = defaultdict(list)
    IMAGE_TILES = defaultdict(list)
    for tile_name, tile_type in getUtilitiesFor(ITileType):
        for fieldname, field in getFieldsInOrder(tile_type.schema):
            if IRichText.providedBy(field):
                RICHTEXT_TILES[tile_name].append(fieldname)
            if INamedBlobImageField.providedBy(field):
                IMAGE_TILES[tile_name].append(fieldname)

    annotations = IAnnotations(obj)
    prefix = "plone.tiles.data."
    for uid, tile in item["tiles"].items():
        # TODO: Maybe create all tiles that do not need to be defferred?
        key = prefix + uid
        tile_name = tile.pop("type", None)
        # first set raw data
        annotations[key] = item["tiles"][uid]
        for fieldname in RICHTEXT_TILES.get(tile_name, []):
            raw = annotations[key][fieldname]
            if raw is not None and not IRichTextValue.providedBy(raw):
                annotations[key][fieldname] = RichTextValue(raw, "text/html", "text/x-html-safe")
        for fieldname in IMAGE_TILES.get(tile_name, []):
            data = annotations[key][fieldname]
            if data is not None:
                blob_path = data.get("blob_path")
                if not blob_path:
                    continue

                abs_blob_path = get_absolute_blob_path(obj, blob_path)
                if not abs_blob_path:
                    logger.info("Blob path %s for tile %s of %s %s does not exist!", blob_path, tile, obj.portal_type, obj.absolute_url())
                    continue
                # Determine the class to use: file or image.
                filename = data["filename"]
                content_type = data["content-type"]

                # Write the field.
                with open(abs_blob_path, "rb") as myfile:
                    blobdata = myfile.read()
                image = NamedBlobImage(
                    data=blobdata,
                    contentType=content_type,
                    filename=filename,
                )
                annotations[key][fieldname] = image
    return item

修复无效的集合查询

Plone 4 和 5 之间存在一些查询变化。这修复了这些问题。

在 collective.exportimport.serializer.SerializeTopicToJson 中，将主题迁移到集合的实际迁移（目前）还没有注意这一点。

class CustomImportContent(ImportContent):

    def global_dict_hook(self, item):
        if item["@type"] in ["Collection", "Topic"]:
            item = self.fix_query(item)

    def fix_query(self, item):
        item["@type"] = "Collection"
        query = item.pop("query", [])
        if not query:
            logger.info("Drop item without query: %s", item["@id"])
            return

        fixed_query = []
        indexes_to_fix = [
            "portal_type",
            "review_state",
            "Creator",
            "Subject",
        ]
        operator_mapping = {
            # old -> new
            "plone.app.querystring.operation.selection.is":
                "plone.app.querystring.operation.selection.any",
            "plone.app.querystring.operation.string.is":
                "plone.app.querystring.operation.selection.any",
        }

        for crit in query:
            if crit["i"] == "portal_type" and len(crit["v"]) > 30:
                # Criterion is all types
                continue

            if crit["o"].endswith("relativePath") and crit["v"] == "..":
                # relativePath no longer accepts ..
                crit["v"] = "..::1"

            if crit["i"] in indexes_to_fix:
                for old_operator, new_operator in operator_mapping.items():
                    if crit["o"] == old_operator:
                        crit["o"] = new_operator

            if crit["i"] == "portal_type":
                # Some types may have changed their names
                fixed_types = []
                for portal_type in crit["v"]:
                    fixed_type = PORTAL_TYPE_MAPPING.get(portal_type, portal_type)
                    fixed_types.append(fixed_type)
                crit["v"] = list(set(fixed_types))

            if crit["i"] == "review_state":
                # Review states may have changed their names
                fixed_states = []
                for review_state in crit["v"]:
                    fixed_state = REVIEW_STATE_MAPPING.get(review_state, review_state)
                    fixed_states.append(fixed_state)
                crit["v"] = list(set(fixed_states))

            if crit["o"] == "plone.app.querystring.operation.string.currentUser":
                crit["v"] = ""

            fixed_query.append(crit)
        item["query"] = fixed_query

        if not item["query"]:
            logger.info("Drop collection without query: %s", item["@id"])
            return
        return item

迁移到 Volto

您可以在迁移中重用 plone.volto 中的 @@migrate_to_volto 提供的迁移代码。以下示例（用于将 https://plone.org 迁移到 Volto）可以用来将任何旧版本站点迁移到 Plone 6 并使用 Volto。

您需要运行 https://github.com/plone/blocks-conversion-tool 以处理将富文本值迁移到 Volto-blocks。

有关迁移到 Volto 的更改的更多详细信息，请参阅 https://6.docs.plone.org/backend/upgrading/version-specific-migration/migrate-to-volto.html。

from App.config import getConfiguration
from bs4 import BeautifulSoup
from collective.exportimport.fix_html import fix_html_in_content_fields
from collective.exportimport.fix_html import fix_html_in_portlets
from contentimport.interfaces import IContentimportLayer
from logging import getLogger
from pathlib import Path
from plone import api
from plone.volto.browser.migrate_to_volto import migrate_richtext_to_blocks
from plone.volto.setuphandlers import add_behavior
from plone.volto.setuphandlers import remove_behavior
from Products.CMFPlone.utils import get_installer
from Products.Five import BrowserView
from zope.interface import alsoProvides

import requests
import transaction

logger = getLogger(__name__)

DEFAULT_ADDONS = []


class ImportAll(BrowserView):

    def __call__(self):

        request = self.request

        # Check if Blocks-conversion-tool is running
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
        }
        r = requests.post(
            "https://127.0.0.1:5000/html", headers=headers, json={"html": "<p>text</p>"}
        )
        r.raise_for_status()

        # Submit a simple form template to trigger the import
        if not request.form.get("form.submitted", False):
            return self.index()

        portal = api.portal.get()
        alsoProvides(request, IContentimportLayer)

        installer = get_installer(portal)
        if not installer.is_product_installed("contentimport"):
            installer.install_product("contentimport")

        # install required add-ons
        for addon in DEFAULT_ADDONS:
            if not installer.is_product_installed(addon):
                installer.install_product(addon)

        # Fake the target being a classic site even though plone.volto is installed...
        # 1. Allow Folders and Collections (they are disabled in Volto by default)
        portal_types = api.portal.get_tool("portal_types")
        portal_types["Collection"].global_allow = True
        portal_types["Folder"].global_allow = True
        # 2. Enable richtext behavior (otherwise no text will be imported)
        for type_ in ["Document", "News Item", "Event"]:
            add_behavior(type_, "plone.richtext")

        transaction.commit()
        cfg = getConfiguration()
        directory = Path(cfg.clienthome) / "import"

        # Import content
        view = api.content.get_view("import_content", portal, request)
        request.form["form.submitted"] = True
        request.form["commit"] = 500
        view(server_file="Plone.json", return_json=True)
        transaction.commit()

        # Run all other imports
        other_imports = [
            "relations",
            "members",
            "translations",
            "localroles",
            "ordering",
            "defaultpages",
            "discussion",
            "portlets",  # not really useful in Volto
            "redirects",
        ]
        for name in other_imports:
            view = api.content.get_view(f"import_{name}", portal, request)
            path = Path(directory) / f"export_{name}.json"
            if path.exists():
                results = view(jsonfile=path.read_text(), return_json=True)
                logger.info(results)
                transaction.get().note(f"Finished import_{name}")
                transaction.commit()
            else:
                logger.info(f"Missing file: {path}")

        # Optional: Run html-fixers on richtext
        fixers = [anchor_fixer]
        results = fix_html_in_content_fields(fixers=fixers)
        msg = "Fixed html for {} content items".format(results)
        logger.info(msg)
        transaction.get().note(msg)
        transaction.commit()

        results = fix_html_in_portlets()
        msg = "Fixed html for {} portlets".format(results)
        logger.info(msg)
        transaction.get().note(msg)
        transaction.commit()

        view = api.content.get_view("updateLinkIntegrityInformation", portal, request)
        results = view.update()
        msg = f"Updated linkintegrity for {results} items"
        logger.info(msg)
        transaction.get().note(msg)
        transaction.commit()

        # Rebuilding the catalog is necessary to prevent issues later on
        catalog = api.portal.get_tool("portal_catalog")
        logger.info("Rebuilding catalog...")
        catalog.clearFindAndRebuild()
        msg = "Finished rebuilding catalog!"
        logger.info(msg)
        transaction.get().note(msg)
        transaction.commit()

        # This uses the blocks-conversion-tool to migrate to blocks
        logger.info("Start migrating richtext to blocks...")
        migrate_richtext_to_blocks()
        msg = "Finished migrating richtext to blocks"
        transaction.get().note(msg)
        transaction.commit()

        # Reuse the migration-form from plon.volto to do some more tasks
        view = api.content.get_view("migrate_to_volto", portal, request)
        # Yes, wen want to migrate default pages
        view.migrate_default_pages = True
        view.slate = True
        logger.info("Start migrating Folders to Documents...")
        view.do_migrate_folders()
        msg = "Finished migrating Folders to Documents!"
        transaction.get().note(msg)
        transaction.commit()

        logger.info("Start migrating Collections to Documents...")
        view.migrate_collections()
        msg = "Finished migrating Collections to Documents!"
        transaction.get().note(msg)
        transaction.commit()

        reset_dates = api.content.get_view("reset_dates", portal, request)
        reset_dates()
        transaction.commit()

        # Disallow folders and collections again
        portal_types["Collection"].global_allow = False
        portal_types["Folder"].global_allow = False

        # Disable richtext behavior again
        for type_ in ["Document", "News Item", "Event"]:
            remove_behavior(type_, "plone.richtext")

        return request.response.redirect(portal.absolute_url())


def anchor_fixer(text, obj=None):
    """Remove anchors since they are not supported by Volto yet"""
    soup = BeautifulSoup(text, "html.parser")
    for link in soup.find_all("a"):
        if not link.get("href") and not link.text:
            # drop empty links (e.g. anchors)
            link.decompose()
        elif not link.get("href") and link.text:
            # drop links without a href but keep the text
            link.unwrap()
    return soup.decode()