--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/src/pyams_content/workflow/basic.py Fri Jan 26 16:43:52 2018 +0100
@@ -0,0 +1,343 @@
+#
+# Copyright (c) 2008-2015 Thierry Florac <tflorac AT ulthar.net>
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+
+__docformat__ = 'restructuredtext'
+
+
+# import standard library
+from datetime import datetime
+
+# import interfaces
+from pyams_content.interfaces import PUBLISH_CONTENT_PERMISSION, MANAGE_SITE_ROOT_PERMISSION, WEBMASTER_ROLE, \
+ PILOT_ROLE, MANAGER_ROLE, OWNER_ROLE, READER_ROLE, MANAGE_CONTENT_PERMISSION, CREATE_CONTENT_PERMISSION
+from pyams_content.shared.common.interfaces import IWfSharedContentRoles
+from pyams_content.workflow.interfaces import IBasicWorkflow
+from pyams_content.shared.common.interfaces import IManagerRestrictions
+from pyams_security.interfaces import IRoleProtectedObject
+from pyams_workflow.interfaces import IWorkflowStateLabel, IWorkflowState, IWorkflow, IWorkflowPublicationInfo, \
+ IWorkflowVersions, IWorkflowInfo, ObjectClonedEvent
+
+# import packages
+from pyams_utils.adapter import adapter_config, ContextAdapter
+from pyams_utils.date import format_datetime
+from pyams_utils.registry import utility_config, get_current_registry
+from pyams_utils.request import check_request
+from pyams_workflow.workflow import Transition, Workflow
+from zope.copy import copy
+from zope.interface import implementer
+from zope.location import locate
+from zope.schema.vocabulary import SimpleVocabulary, SimpleTerm
+
+from pyams_content import _
+
+
+DRAFT = 'draft'
+PUBLISHED = 'published'
+ARCHIVED = 'archived'
+DELETED = 'deleted'
+
+STATES_IDS = (DRAFT,
+ PUBLISHED,
+ ARCHIVED,
+ DELETED)
+
+STATES_LABELS = (_("Draft"),
+ _("Published"),
+ _("Archived"),
+ _("Deleted"))
+
+STATES_VOCABULARY = SimpleVocabulary([SimpleTerm(STATES_IDS[i], title=t)
+ for i, t in enumerate(STATES_LABELS)])
+
+STATES_HEADERS = {DRAFT: _("draft created"),
+ PUBLISHED: _("published"),
+ ARCHIVED: _("archived")}
+
+UPDATE_STATES = (DRAFT, )
+'''Default state available to contributors in update mode'''
+
+READONLY_STATES = (ARCHIVED, DELETED)
+'''Retired and archived contents can't be modified'''
+
+PROTECTED_STATES = (PUBLISHED, )
+'''Protected states are available to webmasters in update mode'''
+
+MANAGER_STATES = ()
+'''No custom state available to managers!'''
+
+VISIBLE_STATES = PUBLISHED_STATES = (PUBLISHED, )
+
+WAITING_STATES = ()
+
+RETIRED_STATES = ()
+
+ARCHIVED_STATES = (ARCHIVED, )
+
+
+#
+# Workflow conditions
+#
+
+def can_manage_content(wf, context):
+ """Check if a manager can handle content"""
+ request = check_request()
+ # grant access to webmaster
+ if request.has_permission(MANAGE_SITE_ROOT_PERMISSION, context):
+ return True
+ # local content managers can manage content
+ principal_id = request.principal.id
+ if principal_id in context.managers:
+ return True
+ # shared tool managers can manage content if restrictions apply
+ restrictions = IManagerRestrictions(context).get_restrictions(principal_id)
+ return restrictions and restrictions.check_access(context, permission=PUBLISH_CONTENT_PERMISSION, request=request)
+
+
+def can_create_new_version(wf, context):
+ """Check if we can create a new version"""
+ # can't create new version when previous draft already exists
+ versions = IWorkflowVersions(context)
+ if versions.has_version(DRAFT):
+ return False
+ request = check_request()
+ # grant access to webmaster
+ if request.has_permission(MANAGE_SITE_ROOT_PERMISSION, context):
+ return True
+ # grant access to owner, creator and local contributors
+ principal_id = request.principal.id
+ if principal_id in context.owner | {context.creator} | context.contributors:
+ return True
+ # grant access to local content managers
+ if principal_id in context.managers:
+ return True
+ # grant access to shared tool managers if restrictions apply
+ restrictions = IManagerRestrictions(context).get_restrictions(principal_id)
+ return restrictions and restrictions.check_access(context, permission=CREATE_CONTENT_PERMISSION, request=request)
+
+
+def can_delete_version(wf, context):
+ """Check if we can delete a draft version"""
+ request = check_request()
+ # grant access to webmaster
+ if request.has_permission(MANAGE_SITE_ROOT_PERMISSION, context):
+ return True
+ # grant access to owner, creator and local contributors
+ principal_id = request.principal.id
+ if principal_id in context.owner | {context.creator} | context.contributors:
+ return True
+ # grant access to local content managers
+ if principal_id in context.managers:
+ return True
+ # grant access to shared tool managers if restrictions apply
+ restrictions = IManagerRestrictions(context).get_restrictions(principal_id)
+ return restrictions and restrictions.check_access(context, permission=MANAGE_CONTENT_PERMISSION, request=request)
+
+
+#
+# Workflow actions
+#
+
+def publish_action(wf, context):
+ """Publish version"""
+ request = check_request()
+ translate = request.localizer.translate
+ publication_info = IWorkflowPublicationInfo(context)
+ publication_info.publication_date = datetime.utcnow()
+ publication_info.publisher = request.principal.id
+ version_id = IWorkflowState(context).version_id
+ for version in IWorkflowVersions(context).get_versions((PUBLISHED, )):
+ if version is not context:
+ IWorkflowInfo(version).fire_transition_toward(ARCHIVED,
+ comment=translate(_("Published version {0}")).format(version_id))
+
+
+def archive_action(wf, context):
+ """Remove readers when a content is archived"""
+ roles = IWfSharedContentRoles(context, None)
+ if roles is not None:
+ IRoleProtectedObject(context).revoke_role(READER_ROLE, roles.readers)
+
+
+def clone_action(wf, context):
+ """Create new version"""
+ result = copy(context)
+ locate(result, context.__parent__)
+ registry = get_current_registry()
+ registry.notify(ObjectClonedEvent(result, context))
+ return result
+
+
+def delete_action(wf, context):
+ """Delete draft version, and parent if single version"""
+ versions = IWorkflowVersions(context)
+ versions.remove_version(IWorkflowState(context).version_id)
+
+
+#
+# Workflow transitions
+#
+
+init = Transition(transition_id='init',
+ title=_("Initialize"),
+ source=None,
+ destination=DRAFT,
+ history_label=_("Draft creation"))
+
+draft_to_published = Transition(transition_id='draft_to_published',
+ title=_("Publish"),
+ source=DRAFT,
+ destination=PUBLISHED,
+ permission=PUBLISH_CONTENT_PERMISSION,
+ condition=can_manage_content,
+ action=publish_action,
+ menu_css_class='fa fa-fw fa-thumbs-o-up',
+ view_name='wf-publish.html',
+ history_label=_("Content published"),
+ notify_roles={WEBMASTER_ROLE, PILOT_ROLE, MANAGER_ROLE, OWNER_ROLE},
+ notify_message=_("published the content « {0} »"),
+ order=1)
+
+published_to_archived = Transition(transition_id='published_to_archived',
+ title=_("Archive content"),
+ source=PUBLISHED,
+ destination=ARCHIVED,
+ permission=PUBLISH_CONTENT_PERMISSION,
+ condition=can_manage_content,
+ action=archive_action,
+ menu_css_class='fa fa-fw fa-archive',
+ view_name='wf-archive.html',
+ history_label=_("Content archived"),
+ notify_roles={WEBMASTER_ROLE, PILOT_ROLE, MANAGER_ROLE, OWNER_ROLE},
+ notify_message=_("archived content « {0} »"),
+ order=2)
+
+published_to_draft = Transition(transition_id='published_to_draft',
+ title=_("Create new version"),
+ source=PUBLISHED,
+ destination=DRAFT,
+ permission=CREATE_CONTENT_PERMISSION,
+ condition=can_create_new_version,
+ action=clone_action,
+ menu_css_class='fa fa-fw fa-file-o',
+ view_name='wf-clone.html',
+ history_label=_("New version created"),
+ order=3)
+
+archived_to_draft = Transition(transition_id='archived_to_draft',
+ title=_("Create new version"),
+ source=ARCHIVED,
+ destination=DRAFT,
+ permission=CREATE_CONTENT_PERMISSION,
+ condition=can_create_new_version,
+ action=clone_action,
+ menu_css_class='fa fa-fw fa-file-o',
+ view_name='wf-clone.html',
+ history_label=_("New version created"),
+ order=4)
+
+delete = Transition(transition_id='delete',
+ title=_("Delete version"),
+ source=DRAFT,
+ destination=DELETED,
+ permission=MANAGE_CONTENT_PERMISSION,
+ condition=can_delete_version,
+ action=delete_action,
+ menu_css_class='fa fa-fw fa-trash',
+ view_name='wf-delete.html',
+ history_label=_("Version deleted"),
+ order=99)
+
+wf_transitions = {init,
+ draft_to_published,
+ published_to_archived,
+ published_to_draft,
+ archived_to_draft,
+ delete}
+
+
+@implementer(IBasicWorkflow)
+class BasicWorkflow(Workflow):
+ """PyAMS basic workflow"""
+
+
+@adapter_config(context=IBasicWorkflow, provides=IWorkflowStateLabel)
+class WorkflowStateLabelAdapter(ContextAdapter):
+ """Generic state label adapter"""
+
+ @staticmethod
+ def get_label(content, request=None, format=True):
+ if request is None:
+ request = check_request()
+ translate = request.localizer.translate
+ state = IWorkflowState(content)
+ header = STATES_HEADERS.get(state.state)
+ if header is not None:
+ state_label = translate(header)
+ if format:
+ state_label = translate(_('{state} {date}')).format(state=state_label,
+ date=format_datetime(state.state_date))
+ else:
+ state_label = translate(_("Unknown state"))
+ return state_label
+
+
+@adapter_config(name=DRAFT, context=IBasicWorkflow, provides=IWorkflowStateLabel)
+class DraftWorkflowStateLabelAdapter(ContextAdapter):
+ """Draft state label adapter"""
+
+ @staticmethod
+ def get_label(content, request=None, format=True):
+ if request is None:
+ request = check_request()
+ translate = request.localizer.translate
+ state = IWorkflowState(content)
+ if len(state.history) <= 2:
+ header = STATES_HEADERS.get(state.state)
+ if header is not None:
+ if state.version_id == 1:
+ state_label = translate(header)
+ else:
+ state_label = translate(_("new version created"))
+ else:
+ state_label = translate(_("Unknown state"))
+ else:
+ state_label = translate(_('publication refused'))
+ if format:
+ state_label = translate(_('{state} {date}')).format(state=state_label,
+ date=format_datetime(state.state_date))
+ return state_label
+
+
+wf = BasicWorkflow(wf_transitions,
+ states=STATES_VOCABULARY,
+ initial_state=DRAFT,
+ update_states=UPDATE_STATES,
+ readonly_states=READONLY_STATES,
+ protected_states=PROTECTED_STATES,
+ manager_states=MANAGER_STATES,
+ published_states=VISIBLE_STATES,
+ waiting_states=WAITING_STATES,
+ retired_states=RETIRED_STATES,
+ archived_states=ARCHIVED_STATES,
+ auto_retired_state=ARCHIVED)
+
+
+@utility_config(name='PyAMS basic workflow', provides=IWorkflow)
+class WorkflowUtility(object):
+ """PyAMS basic workflow utility
+
+ This is a basic workflow implementation for PyAMS contents.
+ It only implements three states which are *draft*, *published* and *archived*.
+ """
+
+ def __new__(cls):
+ return wf