# # dumpDCWorkflow.py # (c) inQual 2001 # written by Sebastien Bigaret # """ Through this script you can either: - dumps a (e.g. TTW defined) DCWorkflow into a python module so that the latter can be integrated in a Zope Product, making it available as a workflow type in CMF --just like default DCWorkflows: Web-configurable workflow [Classic] and Web-configurable workflow [Revision 2]. - exports the 'scripts' folder within a given DCWorkflow - imports a previously 'scripts' folder into a given DCWorkflow. Please note that this operation is destructive: all previously defined scripts within the DCWorkflow will be deleted. See http://cmf.zope.org/Members/Big/dumpDCWorkflowHowto/view for details """ __version__='$Revision: 1.2 $'[11:-2] from Products.DCWorkflow.DCWorkflow import DCWorkflowDefinition from Acquisition import aq_base from StringIO import StringIO from string import capitalize from DateTime import DateTime from Globals import MessageDialog triple_quotes='"""' def DCWorkflow_importExport(self, aDCWorkflowId, importFilename, set_owner, REQUEST): """ Export/import facilities for DCWorkflows. If REQUEST has key 'exportScripts', it exports the 'scripts' Folder. If REQUEST has key 'importScripts', it imports the 'scripts' Folder. Otherwise, it dumps the workflow 'aDCWorkflowId' into a python module. Returns the python file corresponding to the workflow identified by the supplied id """ wf = self.portal_workflow.getWorkflowById(aDCWorkflowId) if wf is None: msg='Workflow id:'+aDCWorkflowId+' not found!' return MessageDialog(title=msg, message=msg, action='./manage_main',) if type(aq_base(wf))!=type(DCWorkflowDefinition('test')): msg='This script only handles DCWorkflow workflows', return MessageDialog(title=msg, message=msg, action='./manage_main',) if REQUEST and hasattr(REQUEST, 'exportScripts'): export=exportWorkflowScripts(wf, REQUEST) return export if REQUEST and hasattr(REQUEST, 'importScripts'): importWorkflowScripts(self, wf, REQUEST) return MessageDialog(title='Scripts successfully imported!', message='Scripts successfully imported!', action ='./portal_workflow/%s/scripts/manage_main'%wf.id,) _result= StringIO() _result.write('# \n') dumpDCwf(wf, _result) _result.flush() _ret = _result.getvalue() _result.close() return _ret def importWorkflowScripts(self, wf, REQUEST): "Imports the 'scripts' folder within the supplied workflow" importFile=REQUEST['importFilename'] # First delete the 'scripts' ContainerTab wf._delObject('scripts') wf.manage_importObject(importFile, REQUEST, REQUEST['set_owner']) def exportWorkflowScripts(wf, REQUEST): "Exports the scripts within the supplied wf" export=wf.scripts.manage_exportObject(id='', download='Yes!', RESPONSE=REQUEST.RESPONSE) REQUEST.RESPONSE.setHeader('Content-Disposition', 'inline;filename=%s_scripts.zexp' % (wf.id,)) return export def dumpDCwf(wf, aStringIO): "Dumps the supplied workflow into the stringIO" writeHeader(wf, aStringIO) dumpMainVariables(wf, aStringIO) dumpStates(wf, aStringIO) dumpTransitions(wf, aStringIO) dumpVariables(wf, aStringIO) dumpWorklists(wf, aStringIO) writeFooter(wf, aStringIO) def dumpMainVariables(wf, aStrIO): aStrIO.write(""" from Products.CMFCore.WorkflowTool import addWorkflowFactory from Products.DCWorkflow.DCWorkflow import DCWorkflowDefinition def setup%s(wf): "..." wf.setProperties(title='%s') for s in %s: wf.states.addState(s) for t in %s: wf.transitions.addTransition(t) for v in %s: wf.variables.addVariable(v) for l in %s: wf.worklists.addWorklist(l) for p in %s: wf.addManagedPermission(p) """ % (capitalize(wf.id), wf.title, wf.states.objectIds(), wf.transitions.objectIds(), wf.variables.objectIds(), wf.worklists.objectIds(), wf.permissions ) ) # States def dumpStates(wf, aStrIO): aStrIO.write(""" ## Initial State wf.states.setInitialState('%s') ## States initialization""" % wf.initial_state) for state in wf.states.objectValues(): dumpState(state, aStrIO) def dumpState(aState, aStrIO): aStrIO.write(""" sdef = wf.states['%s'] sdef.setProperties(title=\"\"\"%s\"\"\", transitions=%s) """ % ( aState.getId(), aState.title, aState.getTransitions()) ) # Permissions for _perm in aState.getManagedPermissions(): _perm_info = aState.getPermissionInfo(_perm) aStrIO.write(""" sdef.setPermission(%s, %i, %s)\n""" % ( repr(_perm), _perm_info['acquired'], _perm_info['roles']) ) # Assignments to variables vars = aState.var_values if vars: for _id in vars.keys(): aStrIO.write(" sdef.addVariable(id='%s', value=\"\"\"%s\"\"\")\n" % ( _id, vars.get(_id)) ) # Transitions def dumpTransitions(wf, aStrIO): aStrIO.write(""" ## Transitions initialization""") for transition in wf.transitions.objectValues(): dumpTransition(transition, aStrIO) def dumpTransition(aTransition, aStrIO): aStrIO.write(""" tdef = wf.transitions['%s'] tdef.setProperties(title=\"\"\"%s\"\"\", new_state_id=\"\"\"%s\"\"\", trigger_type=%i, script_name=\"\"\"%s\"\"\", after_script_name=\"\"\"%s\"\"\", actbox_name=\"\"\"%s\"\"\", actbox_url=\"\"\"%s\"\"\", actbox_category=\"\"\"%s\"\"\", props=%s, ) """ % ( aTransition.getId(), aTransition.title, aTransition.new_state_id, aTransition.trigger_type, aTransition.script_name, aTransition.after_script_name, aTransition.actbox_name, aTransition.actbox_url, aTransition.actbox_category, getGuardProps(aTransition.getGuard()) ) ) # Var. assignments vars = aTransition.var_exprs if vars: for _id in vars.keys(): aStrIO.write(" tdef.addVariable(id='%s', text=%s)\n" % ( _id, repr(aTransition.getVarExprText(_id))) ) # Variables def dumpVariables(wf, aStrIO): aStrIO.write(""" ## State Variable wf.variables.setStateVar('%s') ## Variables initialization""" % wf.state_var) for variable in wf.variables.objectValues(): dumpVariable(variable, aStrIO) def dumpVariable(aVariable, aStrIO): aStrIO.write(""" vdef = wf.variables['%s'] vdef.setProperties(description=\"\"\"%s\"\"\", default_value=\"\"\"%s\"\"\", default_expr=\"\"\"%s\"\"\", for_catalog=%i, for_status=%i, update_always=%i, props=%s) """ % ( aVariable.getId(), aVariable.description, aVariable.default_value, aVariable.getDefaultExprText(), aVariable.for_catalog, aVariable.for_status, aVariable.update_always, getGuardProps(aVariable.getInfoGuard()) ) ) # Worklists def dumpWorklists(wf, aStrIO): aStrIO.write(""" ## Worklists Initialization""") for worklist in wf.worklists.objectValues(): dumpWorklist(worklist, aStrIO) def dumpWorklist(aWorklist, aStrIO): "..." aStrIO.write(""" ldef = wf.worklists['%s'] ldef.setProperties(description=\"\"\"%s\"\"\", actbox_name=\"\"\"%s\"\"\", actbox_url=\"\"\"%s\"\"\", actbox_category=\"\"\"%s\"\"\", props=%s) """ % ( aWorklist.getId(), aWorklist.description, aWorklist.actbox_name, aWorklist.actbox_url, aWorklist.actbox_category, worklistProps(aWorklist, getGuardProps(aWorklist.getGuard())) ) ) def worklistProps(aWorklist, props): """Updates the props dictionary with key/value pairs stored in the worklist's var_match """ if props is None: props={} for key in aWorklist.var_matches.keys(): props['var_match_'+key]=aWorklist.var_matches.get(key) return props # Guard def getGuardProps(aGuard): "Inspects aGuard's properties and returns them in a dictionary" props={} if (aGuard.getPermissionsText()): props['guard_permissions']=aGuard.getPermissionsText() if (aGuard.getRolesText()): props['guard_roles']=aGuard.getRolesText() if (aGuard.getExprText()): props['guard_expr']=aGuard.getExprText() return props or None # Enrobage def writeHeader(wf, aStrIO): """Writes the beginning of of python script, and indicates which scripts should be backed up, if applicable""" aStrIO.write("""# # Generated by dumpDCWorkflow.py written by Sebastien Bigaret # Original workflow id/title: %s/%s # Date: %s # # WARNING: this dumps does NOT contain any scripts you might have added to # the workflow, IT IS YOUR RESPONSABILITY TO MAKE BACKUPS FOR THESE SCRIPTS. # """ % (wf.id, wf.title, DateTime()) ) _scripts = wf.scripts.objectValues() if len(_scripts): aStrIO.write("# The following scripts have been detected and should be backed up:\n") for _script in _scripts: _id = _script.id if callable(_id): _id=_id() aStrIO.write("# - %s (%s)\n" % (_id, _script.meta_type)) else: aStrIO.write("# No script detected in this workflow\n") aStrIO.write("# \n") aStrIO.write(triple_quotes+""" Programmatically creates a workflow type """+triple_quotes+'\n') aStrIO.write("""__version__ = "$Revision: 1.2 $"[11:-2]\n""") def writeFooter(wf, aStrIO): "..." aStrIO.write(""" def create%s(id): "..." ob = DCWorkflowDefinition(id) setup%s(ob) return ob addWorkflowFactory(create%s, id=%s, title=%s) """ % ( capitalize(wf.id), capitalize(wf.id), capitalize(wf.id), repr(wf.id), repr(wf.title) ) )