mirror of
https://github.com/sartography/cr-connect-workflow.git
synced 2025-02-22 20:58:28 +00:00
There was a problem with the python script engine as well that wasn't handling the de-serialize properly and didn't correctly pick back up on the script engine, and the renaming of methods in PythonScriptEngine created some conflicts with the way we override functions. We were not handling ldap looks up efficiently, and this was also breaking in Study Info. Finally we had a bug in SpiffWorkflow that did not allow us to reset back to the previous task in some cases where nested call activities happen far later in the process and are currently active when the reset is created.
107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
import hashlib
|
|
import io
|
|
import json
|
|
import inspect
|
|
import os
|
|
|
|
import connexion
|
|
from flask import send_file
|
|
from jinja2 import Template, UndefinedError
|
|
|
|
from crc.api.common import ApiError
|
|
from crc.scripts.complete_template import CompleteTemplate
|
|
from crc.scripts.script import Script
|
|
|
|
from crc.services.email_service import EmailService
|
|
from config.default import DEFAULT_SENDER
|
|
from crc.services.workflow_processor import CustomBpmnScriptEngine
|
|
|
|
|
|
def render_markdown(data, template):
|
|
"""
|
|
Provides a quick way to very that a Jinja markdown template will work properly on a given json
|
|
data structure. Useful for folks that are building these markdown templates.
|
|
"""
|
|
try:
|
|
template = Template(template)
|
|
data = json.loads(data)
|
|
return template.render(**data)
|
|
except UndefinedError as ue:
|
|
raise ApiError(code="undefined_field", message=ue.message)
|
|
except Exception as e:
|
|
raise ApiError(code="invalid_render", message=str(e))
|
|
|
|
|
|
def render_docx():
|
|
"""
|
|
Provides a quick way to verify that a Jinja docx template will work properly on a given json
|
|
data structure. Useful for folks that are building these templates.
|
|
"""
|
|
try:
|
|
file = connexion.request.files['file']
|
|
data = connexion.request.form['data']
|
|
target_stream = CompleteTemplate().make_template(file, json.loads(data))
|
|
return send_file(
|
|
io.BytesIO(target_stream.read()),
|
|
as_attachment=True,
|
|
attachment_filename="output.docx",
|
|
mimetype="application/octet-stream",
|
|
cache_timeout=-1 # Don't cache these files on the browser.
|
|
)
|
|
except ValueError as e:
|
|
raise ApiError(code="undefined_field", message=str(e))
|
|
except Exception as e:
|
|
raise ApiError(code="invalid_render", message=str(e))
|
|
|
|
|
|
def list_scripts():
|
|
"""Provides a list of scripts that can be called by a script task."""
|
|
scripts = Script.get_all_subclasses()
|
|
script_meta = []
|
|
for script_class in scripts:
|
|
file_path = inspect.getfile(script_class)
|
|
file_name = os.path.basename(file_path)
|
|
# This grabs the filename without the suffix,
|
|
# which is what configurators actually use in a script task.
|
|
handle = os.path.splitext(file_name)[0]
|
|
meta_data = {
|
|
"name": script_class.__name__,
|
|
"handle": handle,
|
|
"description": script_class().get_description()
|
|
}
|
|
script_meta.append(meta_data)
|
|
return script_meta
|
|
|
|
|
|
def send_email(subject, address, body, data=None):
|
|
"""Just sends a quick test email to assure the system is working."""
|
|
if address and body:
|
|
body = body.decode('UTF-8')
|
|
return send_test_email(subject, address, body, json.loads(data))
|
|
else:
|
|
raise ApiError(code='missing_parameter',
|
|
message='You must provide an email address and a message.')
|
|
|
|
|
|
def evaluate_python_expression(body):
|
|
"""Evaluate the given python expression, returning its result. This is useful if the
|
|
front end application needs to do real-time processing on task data. If for instance
|
|
there is a hide expression that is based on a previous value in the same form.
|
|
The response includes both the result, and a hash of the original query, subsequent calls
|
|
of the same hash are unnecessary. """
|
|
try:
|
|
script_engine = CustomBpmnScriptEngine()
|
|
result = script_engine._evaluate(body['expression'], **body['data'])
|
|
return {"result": result, "expression": body['expression'], "key": body['key']}
|
|
except Exception as e:
|
|
return {"result": False, "expression": body['expression'], "key": body['key'], "error": str(e)}
|
|
|
|
def send_test_email(subject, address, message, data=None):
|
|
rendered, wrapped = EmailService().get_rendered_content(message, data)
|
|
EmailService.add_email(
|
|
subject=subject,
|
|
sender=DEFAULT_SENDER,
|
|
recipients=[address],
|
|
content=rendered,
|
|
content_html=wrapped)
|