wcs/wcs/admin/forms.py

1965 lines
77 KiB
Python

# -*- coding: utf-8 -*-
#
# w.c.s. - web application for online forms
# Copyright (C) 2005-2010 Entr'ouvert
#
# This program 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; either version 2 of the License, or
# (at your option) any later version.
#
# 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, see <http://www.gnu.org/licenses/>.
import datetime
import difflib
import io
import tarfile
import time
import xml.etree.ElementTree as ET
from quixote import get_publisher
from quixote import get_request
from quixote import get_response
from quixote import get_session
from quixote import redirect
from quixote.directory import AccessControlled
from quixote.directory import Directory
from quixote.html import TemplateIO
from quixote.html import htmltext
from wcs.backoffice.snapshots import SnapshotsDirectory
from wcs.carddef import CardDef
from wcs.categories import Category
from wcs.formdef import DRAFTS_DEFAULT_LIFESPAN
from wcs.formdef import FormDef
from wcs.formdef import FormdefImportError
from wcs.formdef import FormdefImportRecoverableError
from wcs.forms.root import qrcode
from wcs.qommon import N_
from wcs.qommon import _
from wcs.qommon import force_str
from wcs.qommon import get_logger
from wcs.qommon import misc
from wcs.qommon import template
from wcs.qommon.afterjobs import AfterJob
from wcs.qommon.backoffice.menu import html_top
from wcs.qommon.errors import TraversalError
from wcs.qommon.form import CheckboxesWidget
from wcs.qommon.form import CheckboxWidget
from wcs.qommon.form import DateTimeWidget
from wcs.qommon.form import DateWidget
from wcs.qommon.form import FileWidget
from wcs.qommon.form import Form
from wcs.qommon.form import HtmlWidget
from wcs.qommon.form import SingleSelectWidget
from wcs.qommon.form import StringWidget
from wcs.qommon.form import UrlWidget
from wcs.qommon.form import ValidatedStringWidget
from wcs.qommon.form import WcsExtraStringWidget
from wcs.qommon.form import WidgetList
from wcs.qommon.form import WysiwygTextWidget
from wcs.qommon.misc import C_
from wcs.qommon.storage import Equal
from wcs.qommon.storage import NotEqual
from wcs.qommon.storage import Null
from wcs.roles import get_user_roles
from wcs.roles import logged_users_role
from wcs.workflows import Workflow
from . import utils
from .blocks import BlocksDirectory
from .categories import CategoriesDirectory
from .data_sources import NamedDataSourcesDirectory
from .fields import FieldDefPage
from .fields import FieldsDirectory
from .logged_errors import LoggedErrorsDirectory
def get_categories(category_class):
t = sorted([(misc.simplify(x.name), x.id, x.name, x.id) for x in category_class.select()])
return [x[1:] for x in t]
class FormDefUI:
formdef_class = FormDef
category_class = Category
def __init__(self, formdef):
self.formdef = formdef
def get_categories(self):
return get_categories(self.category_class)
@classmethod
def get_workflows(cls, condition=lambda x: True):
default_workflow = cls.formdef_class.get_default_workflow()
t = sorted([(misc.simplify(x.name), x.id, x.name, x.id) for x in Workflow.select() if condition(x)])
return [(None, default_workflow.name, '')] + [x[1:] for x in t]
def new_form_ui(self):
form = Form(enctype='multipart/form-data')
if self.formdef:
formdef = self.formdef
else:
formdef = self.formdef_class()
form.add(StringWidget, 'name', title=_('Name'), required=True, size=40, value=formdef.name)
categories = self.get_categories()
if categories:
form.add(
SingleSelectWidget,
'category_id',
title=_('Category'),
value=formdef.category_id,
options=[(None, '---', '')] + categories,
)
workflows = self.get_workflows()
if len(workflows) > 1:
form.add(
SingleSelectWidget,
'workflow_id',
title=_('Workflow'),
value=formdef.workflow_id,
options=workflows,
)
if not formdef.is_readonly():
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
return form
def submit_form(self, form):
if self.formdef:
formdef = self.formdef
else:
formdef = self.formdef_class()
name = form.get_widget('name').parse()
formdefs_name = [
x.name
for x in self.formdef_class.select(ignore_errors=True, lightweight=True)
if x.id != formdef.id
]
if name in formdefs_name:
form.get_widget('name').set_error(_('This name is already used'))
raise ValueError()
for f in (
'name',
'confirmation',
'only_allow_one',
'category_id',
'disabled',
'enable_tracking_codes',
'workflow_id',
'disabled_redirection',
'always_advertise',
'publication_date',
'expiration_date',
):
widget = form.get_widget(f)
if widget:
setattr(formdef, f, widget.parse())
if not formdef.fields:
formdef.fields = []
formdef.store()
return formdef
class FieldDefPage(FieldDefPage):
section = 'forms'
def get_deletion_extra_warning(self):
if not self.objectdef.data_class().count():
return None
return _('Warning: this field data will be permanently deleted from existing forms.')
class FieldsDirectory(FieldsDirectory):
field_def_page_class = FieldDefPage
field_var_prefix = 'form_var_'
readonly_message = N_('This form is readonly.')
def index_bottom(self):
if self.objectdef.is_readonly():
return
if hasattr(self.objectdef, str('disabled')) and self.objectdef.disabled:
r = TemplateIO(html=True)
r += htmltext('<div class="warningnotice">')
r += _('This form is currently disabled.')
if hasattr(self.objectdef, str('disabled_redirection')) and self.objectdef.disabled_redirection:
r += htmltext(' (<a href="%s">') % self.objectdef.disabled_redirection
r += _('redirection')
r += htmltext('</a>)')
r += htmltext(' <a href="../enable?back=fields">%s</a>') % _('Enable')
r += htmltext('</div>')
return r.getvalue()
class OptionsDirectory(Directory):
category_class = Category
category_empty_choice = N_('Select a category for this form')
_q_exports = [
'confirmation',
'only_allow_one',
'always_advertise',
'tracking_code',
'online_status',
'captcha',
'description',
'keywords',
'category',
'management',
'geolocations',
'appearance',
'templates',
'user_support',
]
def __init__(self, formdef):
self.formdef = formdef
self.changed = False
def confirmation(self):
form = Form(enctype='multipart/form-data')
form.add(
CheckboxWidget,
'confirmation',
title=_('Include confirmation page'),
value=self.formdef.confirmation,
)
return self.handle(form, _('Confirmation Page'))
def only_allow_one(self):
form = Form(enctype='multipart/form-data')
form.add(
CheckboxWidget,
'only_allow_one',
title=_('Only allow one form per user'),
value=self.formdef.only_allow_one,
)
return self.handle(form, _('Limit to one form'))
def always_advertise(self):
form = Form(enctype='multipart/form-data')
form.add(
CheckboxWidget,
'always_advertise',
title=_('Advertise to unlogged users'),
value=self.formdef.always_advertise,
)
return self.handle(form, _('Display to unlogged users'))
def tracking_code(self):
form = Form(enctype='multipart/form-data')
form.add(
CheckboxWidget,
'enable_tracking_codes',
title=_('Enable support for tracking codes'),
value=self.formdef.enable_tracking_codes,
)
widget = form.add(
WcsExtraStringWidget,
'drafts_lifespan',
title=_('Lifespan of drafts (in days)'),
value=self.formdef.drafts_lifespan,
hint=_('By default drafts are removed after %s days.') % DRAFTS_DEFAULT_LIFESPAN,
)
def check_lifespan(value):
if not value:
return True
try:
return bool(int(value) >= 2 and int(value) <= 100)
except (ValueError, TypeError):
return False
widget.validation_function = check_lifespan
widget.validation_function_error_message = _('Lifespan must be between 2 and 100 days.')
return self.handle(form, _('Tracking Code'))
def captcha(self):
form = Form(enctype='multipart/form-data')
form.add(
CheckboxWidget,
'has_captcha',
title=_('Prepend a CAPTCHA page for anonymous users'),
value=self.formdef.has_captcha,
)
return self.handle(form, _('CAPTCHA'))
def management(self):
form = Form(enctype='multipart/form-data')
form.add(
CheckboxWidget,
'include_download_all_button',
title=_('Include button to download all files'),
value=self.formdef.include_download_all_button,
)
form.add(
CheckboxWidget,
'skip_from_360_view',
title=_('Skip from per user view'),
value=self.formdef.skip_from_360_view,
)
return self.handle(form, _('Management'))
def online_status(self):
form = Form(enctype='multipart/form-data')
form.add(CheckboxWidget, 'disabled', title=_('Disable access to form'), value=self.formdef.disabled)
form.add(
StringWidget,
'disabled_redirection',
title=_('If disabled, redirect to this URL'),
size=40,
hint=_(
'Redirection will only be performed if the form is disabled and a URL is given. '
'Common variables are available with the {{variable}} syntax.'
),
value=self.formdef.disabled_redirection,
)
form.add(
DateTimeWidget,
'publication_date',
title=_('Publication Date'),
value=self.formdef.publication_date,
)
form.add(
DateTimeWidget, 'expiration_date', title=_('Expiration Date'), value=self.formdef.expiration_date
)
return self.handle(form, _('Online Status'))
def description(self):
form = Form(enctype='multipart/form-data')
form.add(WysiwygTextWidget, 'description', title=_('Description'), value=self.formdef.description)
return self.handle(form, _('Description'))
def keywords(self):
form = Form(enctype='multipart/form-data')
form.add(
StringWidget,
'keywords',
title=_('Keywords'),
value=self.formdef.keywords,
size=50,
hint=_('Keywords need to be separated with commas.'),
)
return self.handle(form, _('Keywords'))
def category(self):
categories = get_categories(self.category_class)
form = Form(enctype='multipart/form-data')
form.widgets.append(HtmlWidget('<p>%s</p>' % _(self.category_empty_choice)))
form.add(
SingleSelectWidget,
'category_id',
title=_('Category'),
value=self.formdef.category_id,
options=[(None, '---', '')] + categories,
)
return self.handle(form, _('Category'))
def geolocations(self):
form = Form(enctype='multipart/form-data')
geoloc_label = (self.formdef.geolocations or {}).get('base')
form.add(
StringWidget,
'geoloc_label',
title=_('Geolocation Label'),
value=geoloc_label,
size=50,
hint=_('Location label (empty to disable geolocation)'),
)
return self.handle(form, _('Geolocation'))
def appearance(self):
form = Form(enctype='multipart/form-data')
form.add(
StringWidget,
'appearance_keywords',
title=_('Appearance keywords'),
value=self.formdef.appearance_keywords,
size=50,
hint=_(
'Serie of keywords to alter form appearance using CSS or '
'custom templates, separated by spaces.'
),
)
return self.handle(form, _('Appearance'))
def templates(self):
form = Form(enctype='multipart/form-data')
form.add(
StringWidget, 'digest_template', title=_('Digest'), value=self.formdef.digest_template, size=50
)
form.add(
WysiwygTextWidget,
'lateral_template',
title=_('Lateral Block'),
value=self.formdef.lateral_template,
)
form.add(
WysiwygTextWidget,
'submission_lateral_template',
title=_('Submission Lateral Block'),
value=self.formdef.submission_lateral_template,
)
result = self.handle(form, _('Templates'))
if self.changed and self.formdef.data_class().count():
get_response().add_after_job(UpdateDigestAfterJob(formdef=self.formdef))
if isinstance(self.formdef, CardDef):
get_session().message = ('info', _('Existing cards will be updated in the background.'))
else:
get_session().message = ('info', _('Existing forms will be updated in the background.'))
return result
def user_support(self):
form = Form(enctype='multipart/form-data')
form.add(
SingleSelectWidget,
'user_support',
title=_('User support'),
value=self.formdef.user_support,
options=[
(None, C_('user_support|No'), ''),
('optional', C_('user_support|Optional'), 'optional'),
],
)
return self.handle(form, _('User support'))
def handle(self, form, title):
if not self.formdef.is_readonly():
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('..')
if form.is_submitted() and not form.has_errors():
attrs = [
'confirmation',
'only_allow_one',
'disabled',
'enable_tracking_codes',
'always_advertise',
'disabled_redirection',
'publication_date',
'expiration_date',
'has_captcha',
'description',
'keywords',
'category_id',
'skip_from_360_view',
'geoloc_label',
'appearance_keywords',
'include_download_all_button',
'digest_template',
'lateral_template',
'submission_lateral_template',
'drafts_lifespan',
'user_support',
]
for attr in attrs:
widget = form.get_widget(attr)
if widget:
if hasattr(self, 'clean_%s' % attr):
has_error = getattr(self, 'clean_%s' % attr)(form)
if has_error:
continue
if attr == 'geoloc_label':
if widget.parse():
self.formdef.geolocations = {'base': widget.parse()}
else:
self.formdef.geolocations = None
else:
new_value = widget.parse()
if getattr(self.formdef, attr, None) != new_value:
if attr == 'digest_template':
self.changed = True
setattr(self.formdef, attr, new_value)
if not form.has_errors():
self.formdef.store(comment=_('Changed "%s" parameters') % title)
return redirect('..')
html_top('forms', title=self.formdef.name)
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % title
r += form.render()
return r.getvalue()
def clean_digest_template(self, form):
if not isinstance(self.formdef, CardDef):
return False
widget = form.get_widget('digest_template')
new_value = widget.parse()
if new_value:
return False
if any(self.formdef.usage_in_formdefs()):
widget.set_error(
_('Can not empty digest template: this card model is used as data source in some forms.')
)
return True
return False
def _q_traverse(self, path):
get_response().breadcrumb.append((path[0] + '/', self.formdef.name))
return super()._q_traverse(path)
class WorkflowRoleDirectory(Directory):
def __init__(self, formdef):
self.formdef = formdef
def _q_lookup(self, component):
if not component in self.formdef.workflow.roles:
raise TraversalError()
if not self.formdef.workflow_roles:
self.formdef.workflow_roles = {}
role_id = self.formdef.workflow_roles.get(component)
options = [(None, '---', None)]
options.extend(get_user_roles())
form = Form(enctype='multipart/form-data')
form.add(SingleSelectWidget, 'role_id', value=role_id, options=options)
if not self.formdef.is_readonly():
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if not form.is_submitted() or form.has_errors():
get_response().breadcrumb.append(('role/%s' % component, _('Workflow Role')))
self.html_top(title=self.formdef.name)
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Role')
r += htmltext('<p>%s</p>') % self.formdef.workflow.roles.get(component)
r += form.render()
return r.getvalue()
else:
self.formdef.workflow_roles[component] = form.get_widget('role_id').parse()
self.formdef.store(
comment=_('Change in function "%s"') % self.formdef.workflow.roles.get(component)
)
# instruct formdef to update its security rules
self.formdef.data_class().rebuild_security()
return redirect('..')
class FormDefPage(Directory):
_q_exports = [
'',
'fields',
'delete',
'duplicate',
'export',
'anonymise',
'archive',
'enable',
'workflow',
'role',
('workflow-options', 'workflow_options'),
('workflow-variables', 'workflow_variables'),
('workflow-status-remapping', 'workflow_status_remapping'),
'roles',
'title',
'options',
'overwrite',
'qrcode',
'information',
('public-url', 'public_url'),
('backoffice-submission-roles', 'backoffice_submission_roles'),
('logged-errors', 'logged_errors_dir'),
('history', 'snapshots_dir'),
]
formdef_class = FormDef
formdef_export_prefix = 'form'
formdef_ui_class = FormDefUI
formdef_default_workflow = '_default'
options_directory_class = OptionsDirectory
delete_message = N_('You are about to irrevocably delete this form.')
delete_title = N_('Deleting Form:')
overwrite_message = N_('You can replace this form by uploading a file ' 'or by pointing to a form URL.')
overwrite_success_message = N_(
'The form has been successfully overwritten. '
'Do note it kept its existing address and role and workflow parameters.'
)
def __init__(self, component, instance=None):
try:
self.formdef = instance or self.formdef_class.get(component)
except KeyError:
raise TraversalError()
self.formdefui = self.formdef_ui_class(self.formdef)
get_response().breadcrumb.append((component + '/', self.formdef.name))
self.fields = FieldsDirectory(self.formdef)
self.fields.html_top = self.html_top
self.role = WorkflowRoleDirectory(self.formdef)
self.role.html_top = self.html_top
self.options = self.options_directory_class(self.formdef)
self.logged_errors_dir = LoggedErrorsDirectory(
parent_dir=self, formdef_class=self.formdef_class, formdef_id=self.formdef.id
)
self.snapshots_dir = SnapshotsDirectory(self.formdef)
def html_top(self, title):
return html_top('forms', title)
def add_option_line(self, link, label, current_value, popup=True):
return htmltext(
'<li><a rel="%(popup)s" href="%(link)s">'
'<span class="label">%(label)s</span> '
'<span class="value">%(current_value)s</span>'
'</a></li>'
% {
'popup': 'popup' if popup else '',
'link': link,
'label': label,
'current_value': current_value,
}
)
def _q_index(self):
self.html_top(title=self.formdef.name)
r = TemplateIO(html=True)
get_response().filter['sidebar'] = self.get_sidebar()
get_response().add_javascript(['jquery.js', 'widget_list.js', 'qommon.wysiwyg.js'])
DateWidget.prepare_javascript()
r += htmltext('<div id="appbar">')
r += htmltext('<h2>%s</h2>') % self.formdef.name
r += htmltext('<span class="actions">')
if not self.formdef.is_readonly():
r += htmltext('<a rel="popup" href="title">%s</a>') % _('change title')
r += htmltext('</span>')
r += htmltext('</div>')
r += utils.last_modification_block(obj=self.formdef)
r += get_session().display_message()
r += htmltext('<div class="bo-block">')
r += htmltext('<h3>%s</h3>') % _('Information')
r += htmltext('<ul class="biglist optionslist">')
r += self.add_option_line(
'options/description',
_('Description'),
self.formdef.description and C_('description|On') or C_('description|None'),
)
r += self.add_option_line(
'options/keywords',
_('Keywords'),
self.formdef.keywords and self.formdef.keywords or C_('keywords|None'),
)
r += self.add_option_line(
'options/category',
_('Category'),
self.formdef.category_id
and self.formdef.category
and self.formdef.category.name
or C_('category|None'),
)
r += htmltext('</ul>')
r += htmltext('</div>')
r += htmltext('<div class="splitcontent-left">')
r += htmltext('<div class="bo-block">')
r += htmltext('<h3>%s</h3>') % _('Workflow')
r += htmltext('<ul class="biglist optionslist">')
if get_publisher().get_backoffice_root().is_accessible('workflows'):
# custom option line to also include a link to the workflow itself.
r += htmltext(
'<li><a rel="popup" href="%(link)s">'
'<span class="label">%(label)s</span> '
'<span class="value offset">%(current_value)s</span>'
'</a>'
'<a class="extra-link" title="%(title)s" href="%(workflow_url)s">↗</a>'
'</li>'
% {
'link': 'workflow',
'label': _('Workflow'),
'title': _('Open workflow page'),
'workflow_url': self.formdef.workflow.get_admin_url(),
'current_value': self.formdef.workflow.name or '-',
}
)
else:
r += self.add_option_line(
'workflow', _('Workflow'), self.formdef.workflow and self.formdef.workflow.name or '-'
)
if self.formdef.workflow_id:
pristine_workflow = Workflow.get(self.formdef.workflow_id, ignore_errors=True)
if pristine_workflow and pristine_workflow.variables_formdef:
r += self.add_option_line('workflow-variables', _('Options'), '')
elif self.formdef.workflow_options:
# there are no variables defined but there are some values
# in workflow_options, this is probably the legacy stuff.
if any((x for x in self.formdef.workflow_options if '*' in x)):
r += self.add_option_line('workflow-options', _('Options'), '')
if self.formdef.workflow.roles:
if not self.formdef.workflow_roles:
self.formdef.workflow_roles = {}
workflow_roles = list((self.formdef.workflow.roles or {}).items())
workflow_roles.sort(key=lambda x: '' if x[0] == '_receiver' else misc.simplify(x[1]))
for (wf_role_id, wf_role_label) in workflow_roles:
role_id = self.formdef.workflow_roles.get(wf_role_id)
if role_id:
try:
role = get_publisher().role_class.get(role_id)
role_label = role.name
except KeyError:
# removed role ?
role_label = _('Unknown role (%s)') % role_id
else:
role_label = '-'
r += self.add_option_line('role/%s' % wf_role_id, wf_role_label, role_label)
r += self.add_option_line('roles', _('User Roles'), self._get_roles_label_and_auth_context('roles'))
r += self.add_option_line(
'backoffice-submission-roles',
_('Backoffice Submission Role'),
self._get_roles_label('backoffice_submission_roles'),
)
r += htmltext('</ul>')
r += htmltext('</div>')
r += htmltext('</div>')
r += htmltext('<div class="splitcontent-right">')
r += htmltext('<div class="bo-block">')
r += htmltext('<h3>%s</h3>') % _('Options')
r += htmltext('<ul class="biglist optionslist">')
r += self.add_option_line(
'options/confirmation',
_('Confirmation Page'),
self.formdef.confirmation and C_('confirmation page|Enabled') or C_('confirmation page|Disabled'),
)
r += self.add_option_line(
'options/only_allow_one',
_('Limit to one form'),
self.formdef.only_allow_one and C_('limit to one|Enabled') or C_('limit to one|Disabled'),
)
if self.formdef.roles:
r += self.add_option_line(
'options/always_advertise',
_('Display to unlogged users'),
self.formdef.always_advertise
and C_('display to unlogged|Enabled')
or C_('display to unlogged|Disabled'),
)
r += self.add_option_line(
'options/management',
_('Management'),
_('Custom')
if (self.formdef.skip_from_360_view or self.formdef.include_download_all_button)
else _('Default'),
)
r += self.add_option_line(
'options/tracking_code',
_('Tracking Code'),
self.formdef.enable_tracking_codes
and C_('tracking code|Enabled')
or C_('tracking code|Disabled'),
)
r += self.add_option_line(
'options/geolocations',
_('Geolocation'),
self.formdef.geolocations and C_('geolocation|Enabled') or C_('geolocation|Disabled'),
)
if get_publisher().has_site_option('formdef-captcha-option'):
r += self.add_option_line(
'options/captcha',
_('CAPTCHA for anonymous users'),
self.formdef.has_captcha and C_('captcha|Enabled') or C_('captcha|Disabled'),
)
if get_publisher().has_site_option('formdef-appearance-keywords'):
r += self.add_option_line(
'options/appearance',
_('Appearance'),
self.formdef.appearance_keywords
and self.formdef.appearance_keywords
or C_('appearance|Standard'),
)
if (
self.formdef.digest_template
or self.formdef.lateral_template
or self.formdef.submission_lateral_template
):
template_status = C_('template|Custom')
else:
template_status = C_('template|None')
r += self.add_option_line('options/templates', _('Templates'), template_status, popup=False)
online_status = C_('online status|Active')
if self.formdef.disabled:
# manually disabled
online_status = C_('online status|Disabled')
if self.formdef.disabled_redirection:
online_status = _('Redirected')
elif self.formdef.is_disabled():
# disabled by date
online_status = C_('online status|Inactive by date')
r += self.add_option_line('options/online_status', _('Online Status'), online_status)
r += htmltext('</ul>')
r += htmltext('</div>')
r += htmltext('</div>')
r += htmltext('<div class="bo-block clear">')
r += htmltext('<h3 class="clear">%s <span class="change">(<a href="fields/">%s</a>)</span></h3>') % (
_('Fields'),
_('edit'),
)
r += self.get_preview()
r += htmltext('</div>')
return r.getvalue()
def _get_roles_label(self, attribute):
if getattr(self.formdef, attribute):
roles = []
for x in getattr(self.formdef, attribute):
if x == logged_users_role().id:
roles.append(logged_users_role().name)
else:
try:
roles.append(get_publisher().role_class.get(x).name)
except KeyError:
# removed role ?
roles.append(_('Unknown role (%s)') % x)
value = htmltext(', ').join(roles)
else:
value = C_('roles|None')
return value
def _get_roles_label_and_auth_context(self, attribute):
value = self._get_roles_label(attribute)
if self.formdef.required_authentication_contexts:
auth_contexts = get_publisher().get_supported_authentication_contexts()
value += ' (%s)' % ', '.join(
[
auth_contexts.get(x)
for x in self.formdef.required_authentication_contexts
if auth_contexts.get(x)
]
)
return value
def get_sidebar(self):
r = TemplateIO(html=True)
if self.formdef.is_readonly():
r += htmltext('<div class="infonotice"><p>%s</p></div>') % _('This form is readonly.')
r += utils.snapshot_info_block(snapshot=self.formdef.snapshot_object)
return r.getvalue()
r += htmltext('<ul id="sidebar-actions">')
r += htmltext('<li><a href="delete" rel="popup">%s</a></li>') % _('Delete')
r += htmltext('<li><a href="duplicate">%s</a></li>') % _('Duplicate')
r += htmltext('<li><a rel="popup" href="overwrite">%s</a></li>') % _('Overwrite with new import')
r += htmltext('<li><a href="export">%s</a></li>') % _('Export')
if get_publisher().snapshot_class:
r += htmltext('<li><a rel="popup" href="history/save">%s</a></li>') % _('Save snapshot')
r += htmltext('<li><a href="history/">%s</a></li>') % _('History')
r += htmltext('<li><a href="anonymise">%s</a></li>') % _('Anonymise forms')
if not get_publisher().is_using_postgresql():
r += htmltext('<li><a href="archive">%s</a></li>') % _('Archive')
r += htmltext('</ul>')
r += htmltext('<ul>')
if self.formdef.is_disabled():
r += htmltext('<li><a href="%s">%s</a></li>') % (
self.formdef.get_url(preview=True),
_('Preview Online'),
)
else:
r += htmltext('<li><a href="%s">%s</a></li>') % (self.formdef.get_url(), _('Display Online'))
r += htmltext('<li><a href="public-url" rel="popup">%s</a></li>') % _('Display public URL')
if qrcode is not None:
r += htmltext('<li><a href="qrcode" rel="popup">%s</a></li>') % _('Display QR Code')
r += htmltext('</ul>')
r += LoggedErrorsDirectory.errors_block(formdef_class=self.formdef_class, formdef_id=self.formdef.id)
return r.getvalue()
def public_url(self):
self.html_top(title=self.formdef.name)
get_response().breadcrumb.append(('public-url', _('Public URL')))
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>' % _('Public URL'))
r += htmltext('<div>')
r += htmltext('<p>%s</p>') % _('The public URL of this form is:')
url = self.formdef.get_url()
r += htmltext('<a href="%s">%s</a>') % (url, url)
r += htmltext('</div>')
return r.getvalue()
def qrcode(self):
self.html_top(title=self.formdef.name)
get_response().breadcrumb.append(('qrcode', _('QR Code')))
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>' % _('QR Code'))
r += htmltext('<div id="qrcode">')
r += htmltext('<img width="410px" height="410px" src="%sqrcode" alt=""/>' % self.formdef.get_url())
r += htmltext('<a href="%sqrcode?download">%s</a></p>') % (self.formdef.get_url(), _('Download'))
r += htmltext('</div>')
return r.getvalue()
def _roles_selection(self, title, attribute, description=None, include_logged_users_role=True):
form = Form(enctype='multipart/form-data')
options = [(None, '---', None)]
if include_logged_users_role:
options.append((logged_users_role().id, logged_users_role().name, logged_users_role().id))
options += get_user_roles()
form.add(
WidgetList,
'roles',
element_type=SingleSelectWidget,
value=getattr(self.formdef, attribute),
add_element_label=_('Add Role'),
element_kwargs={'render_br': False, 'options': options},
)
auth_contexts = get_publisher().get_supported_authentication_contexts()
if attribute == 'roles' and auth_contexts:
form.add(
CheckboxesWidget,
'required_authentication_contexts',
title=_('Required authentication contexts'),
value=self.formdef.required_authentication_contexts,
options=list(auth_contexts.items()),
)
if not self.formdef.is_readonly():
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if not form.is_submitted() or form.has_errors():
get_response().breadcrumb.append(('roles', title))
self.html_top(title=self.formdef.name)
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % title
if description:
r += htmltext('<p>%s</p>') % description
r += form.render()
return r.getvalue()
else:
roles = form.get_widget('roles').parse() or []
setattr(self.formdef, attribute, [x for x in roles if x])
if form.get_widget('required_authentication_contexts'):
self.formdef.required_authentication_contexts = form.get_widget(
'required_authentication_contexts'
).parse()
self.formdef.store(comment=_('Change of %s') % title)
return redirect('.')
def roles(self):
return self._roles_selection(
title=_('User Roles'),
attribute='roles',
description=_('Select the roles that can access this form.'),
)
def backoffice_submission_roles(self):
return self._roles_selection(
title=_('Backoffice Submission Roles'),
attribute='backoffice_submission_roles',
include_logged_users_role=False,
description=_(
'Select the roles that will be allowed to ' 'fill out forms of this kind in the backoffice.'
),
)
def title(self):
form = Form(enctype='multipart/form-data')
kwargs = {}
if self.formdef.url_name == misc.simplify(self.formdef.name):
# if name and url name are in sync, keep them that way
kwargs['data-slug-sync'] = 'url_name'
form.add(
StringWidget, 'name', title=_('Name'), required=True, size=40, value=self.formdef.name, **kwargs
)
disabled_url_name = bool(self.formdef.data_class().count())
kwargs = {}
if disabled_url_name:
kwargs['readonly'] = 'readonly'
form.add(
ValidatedStringWidget,
'url_name',
title=_('Identifier in URLs'),
size=40,
required=True,
value=self.formdef.url_name,
regex=r'^[a-zA-Z0-9_-]+',
**kwargs,
)
if not self.formdef.is_readonly():
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if form.is_submitted() and not form.has_errors():
new_name = form.get_widget('name').parse()
new_url_name = form.get_widget('url_name').parse()
formdefs = [
x
for x in self.formdef_class.select(ignore_errors=True, lightweight=True)
if x.id != self.formdef.id
]
if new_name in [x.name for x in formdefs]:
form.get_widget('name').set_error(_('This name is already used.'))
if new_url_name in [x.url_name for x in formdefs]:
form.get_widget('url_name').set_error(_('This identifier is already used.'))
if not form.has_errors():
self.formdef.name = new_name
self.formdef.url_name = new_url_name
self.formdef.store(comment=_('Change of title / URL'))
return redirect('.')
if disabled_url_name:
form.widgets.append(
HtmlWidget(
'<p>%s<br>'
% _('The form identifier should not be modified as there is already some data.')
)
)
form.widgets.append(
HtmlWidget(
'<a href="" class="change-nevertheless">%s</a></p>'
% _('I understand the danger, make it editable nevertheless.')
)
)
get_response().breadcrumb.append(('title', _('Title')))
self.html_top(title=self.formdef.name)
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Title')
r += form.render()
return r.getvalue()
def workflow(self):
form = Form(enctype='multipart/form-data')
workflows = self.formdef_ui_class.get_workflows(condition=lambda x: x.possible_status)
form.add(SingleSelectWidget, 'workflow_id', value=self.formdef.workflow_id, options=workflows)
if not self.formdef.is_readonly():
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if not form.is_submitted() or form.has_errors():
get_response().breadcrumb.append(('workflow', _('Workflow')))
self.html_top(title=self.formdef.name)
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Workflow')
r += htmltext('<p>%s</p>') % _('Select the workflow that will handle those forms.')
r += form.render()
return r.getvalue()
else:
workflow_id = form.get_widget('workflow_id').parse()
if self.formdef.data_class().keys():
# there are existing formdata, status will have to be mapped
if workflow_id is None:
workflow_id = self.formdef_default_workflow
return redirect('workflow-status-remapping?new=%s' % workflow_id)
self.formdef.workflow_id = workflow_id
self.formdef.store(comment=_('Workflow change'))
return redirect('.')
def workflow_status_remapping(self):
new_workflow = Workflow.get(get_request().form.get('new'))
if get_request().get_method() == 'GET':
get_request().form = None # do not be considered submitted already
new_workflow_status = [(x.id, x.name) for x in new_workflow.possible_status]
form = Form(enctype='multipart/form-data')
for status in self.formdef.workflow.possible_status:
default = status.id
if not default in [x.id for x in new_workflow.possible_status]:
default = new_workflow_status[0]
form.add(
SingleSelectWidget,
'mapping-%s' % status.id,
title=status.name,
value=default,
options=new_workflow_status,
)
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if not form.is_submitted() or form.has_errors():
get_response().breadcrumb.append(('workflow-status-remapping', _('Workflow Status Remapping')))
self.html_top(title=self.formdef.name)
r = TemplateIO(html=True)
r += htmltext('<p>')
r += _('From %(here)s to %(there)s') % {
'here': self.formdef.workflow.name,
'there': new_workflow.name,
}
r += htmltext('</p>')
r += form.render()
return r.getvalue()
else:
get_logger().info(
'admin - form "%s", workflow is now "%s" (was "%s")'
% (self.formdef.name, new_workflow.name, self.formdef.workflow.name)
)
self.workflow_status_remapping_submit(form)
if new_workflow.id == self.formdef_default_workflow:
self.formdef.workflow_id = None
else:
self.formdef.workflow_id = new_workflow.id
self.formdef.store(comment=_('Workflow change'))
# instruct formdef to update its security rules
self.formdef.data_class().rebuild_security()
return redirect('.')
def workflow_status_remapping_submit(self, form):
status_mapping = {}
for status in self.formdef.workflow.possible_status:
status_mapping['wf-%s' % status.id] = 'wf-%s' % form.get_widget('mapping-%s' % status.id).parse()
if any([x[0] != x[1] for x in status_mapping.items()]):
# if there are status changes, update all formdatas (except drafts)
status_mapping.update({'draft': 'draft'})
for item in self.formdef.data_class().select([NotEqual('status', 'draft')]):
item.status = status_mapping.get(item.status)
if item.evolution:
for evo in item.evolution:
evo.status = status_mapping.get(evo.status)
item.store()
def get_preview(self):
form = Form(action='#', use_tokens=False)
on_page = 0
for i, field in enumerate(self.formdef.fields):
field.id = i
if hasattr(field, str('add_to_form')):
try:
field.add_to_form(form)
except Exception as e:
form.widgets.append(
HtmlWidget(
htmltext('<div class="errornotice"><p>%s (%s)</p></div>')
% (_('Error previewing field.'), e)
)
)
else:
if field.key == 'page':
if on_page:
form.widgets.append(HtmlWidget('</fieldset>'))
form.widgets.append(HtmlWidget('<fieldset class="formpage">'))
on_page += 1
form.widgets.append(
HtmlWidget('<legend>%s %s</legend>' % (_('Page #%s:') % on_page, field.label))
)
elif field.key == 'title':
form.widgets.append(HtmlWidget('<h3>%s</h3>' % field.label))
elif field.key == 'subtitle':
form.widgets.append(HtmlWidget('<h4>%s</h4>' % field.label))
elif field.key == 'comment':
form.widgets.append(HtmlWidget('<p>%s</p>' % field.label))
if on_page:
form.widgets.append(HtmlWidget('</fieldset>'))
r = TemplateIO(html=True)
r += htmltext('<div class="form-preview">')
r += form.render()
r += htmltext('</div>')
return r.getvalue()
def duplicate(self):
self.formdefui.formdef.id = None
original_name = self.formdefui.formdef.name
self.formdefui.formdef.name = self.formdefui.formdef.name + _(' (copy)')
formdef_names = [x.name for x in self.formdef_class.select(lightweight=True)]
no = 2
while self.formdefui.formdef.name in formdef_names:
self.formdefui.formdef.name = _('%(name)s (copy %(no)d)') % {'name': original_name, 'no': no}
no += 1
self.formdefui.formdef.url_name = None
self.formdefui.formdef.table_name = None
self.formdefui.formdef.disabled = True
self.formdefui.formdef.store()
return redirect('../%s/' % self.formdefui.formdef.id)
def get_check_count_before_deletion_message(self):
if not get_publisher().is_using_postgresql():
return None
from wcs import sql
criterias = [
Equal('formdef_id', self.formdefui.formdef.id),
NotEqual('status', 'draft'),
Equal('is_at_endpoint', False),
Null('anonymised'),
]
if sql.AnyFormData.count(criterias):
return _('Deletion is not possible as there are open forms.')
def delete(self):
form = Form(enctype='multipart/form-data')
check_count_message = self.get_check_count_before_deletion_message()
if check_count_message:
form.widgets.append(HtmlWidget('<p>%s</p>' % check_count_message))
else:
form.widgets.append(HtmlWidget('<p>%s</p>' % _(self.delete_message)))
form.add_submit('delete', _('Delete'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse() or (form.is_submitted() and check_count_message):
return redirect('..')
if not form.is_submitted() or form.has_errors():
get_response().breadcrumb.append(('delete', _('Delete')))
self.html_top(title=_(self.delete_title))
r = TemplateIO(html=True)
r += htmltext('<h2>%s %s</h2>') % (_(self.delete_title), self.formdef.name)
r += form.render()
return r.getvalue()
else:
self.formdef.remove_self()
return redirect('..')
def overwrite(self):
form = Form(enctype='multipart/form-data', use_tokens=False)
form.add(FileWidget, 'file', title=_('File'), required=False)
form.add(UrlWidget, 'url', title=_('Address'), required=False, size=50)
form.add_hidden('new_formdef', required=False)
form.add_hidden('force', required=False)
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if form.is_submitted() and not form.has_errors():
try:
return self.overwrite_submit(form)
except ValueError:
pass
get_response().breadcrumb.append(('overwrite', _('Overwrite')))
self.html_top(title=_('Overwrite'))
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Overwrite')
r += htmltext('<p>%s</p>') % _(self.overwrite_message)
r += form.render()
return r.getvalue()
def overwrite_submit(self, form):
if form.get_widget('file').parse():
fp = form.get_widget('file').parse().fp
elif form.get_widget('new_formdef').parse():
fp = io.StringIO(form.get_widget('new_formdef').parse())
elif form.get_widget('url').parse():
url = form.get_widget('url').parse()
try:
fp = misc.urlopen(url)
except misc.ConnectionError as e:
form.set_error('url', _('Error loading form (%s).') % str(e))
raise ValueError()
else:
form.set_error('file', _('You have to enter a file or a URL.'))
raise ValueError()
error, reason = False, None
try:
new_formdef = self.formdef_class.import_from_xml(fp, include_id=True)
except FormdefImportError as e:
error = True
reason = _(e.msg) % e.msg_args
if e.details:
reason += ' [%s]' % e.details
except ValueError:
error = True
if error:
if reason:
msg = _('Invalid File (%s)') % reason
else:
msg = _('Invalid File')
if form.get_widget('url').parse():
form.set_error('url', msg)
else:
form.set_error('file', msg)
raise ValueError()
# it's been through the summary page, or there is no data yet
if not self.formdef.data_class().count() or form.get_widget('force').parse():
# doing it!
return self.overwrite_by_formdef(new_formdef)
return self.overwrite_warning_summary(new_formdef)
def overwrite_by_formdef(self, new_formdef):
incompatible_field_ids = self.get_incompatible_field_ids(new_formdef)
if incompatible_field_ids:
# if there are incompatible field ids, remove them first
self.formdef.fields = [x for x in self.formdef.fields if x.id not in incompatible_field_ids]
self.formdef.store(comment=_('Overwritten (removal of incompatible fields)'))
# keep current formdef id, url_name, internal identifier and sql table name
new_formdef.id = self.formdef.id
new_formdef.internal_identifier = self.formdef.internal_identifier
new_formdef.url_name = self.formdef.url_name
new_formdef.table_name = self.formdef.table_name
# keep currently assigned category and workflow
new_formdef.category_id = self.formdef.category_id
new_formdef.workflow_id = self.formdef.workflow_id
new_formdef.workflow_options = self.formdef.workflow_options
# keep currently assigned roles
new_formdef.workflow_roles = self.formdef.workflow_roles
new_formdef.backoffice_submission_roles = self.formdef.backoffice_submission_roles
new_formdef.roles = self.formdef.roles
self.formdef = new_formdef
self.formdef.store(comment=_('Overwritten'))
get_session().message = ('info', _(self.overwrite_success_message))
return redirect('.')
def get_incompatible_field_ids(self, new_formdef):
incompatible_field_ids = []
current_fields = {}
for field in self.formdef.fields:
current_fields[field.id] = field
for field in new_formdef.fields:
current_field = current_fields.get(field.id)
if current_field and current_field.type != field.type:
incompatible_field_ids.append(field.id)
return incompatible_field_ids
def overwrite_warning_summary(self, new_formdef):
self.html_top(title=_('Overwrite'))
get_response().breadcrumb.append(('overwrite', _('Overwrite')))
r = TemplateIO(html=True)
r += htmltext('<h2>%s - %s</h2>') % (_('Overwrite'), _('Summary of changes'))
current_fields_list = [str(x.id) for x in self.formdef.fields]
new_fields_list = [str(x.id) for x in new_formdef.fields]
current_fields = {}
new_fields = {}
for field in self.formdef.fields:
current_fields[field.id] = field
for field in new_formdef.fields:
new_fields[field.id] = field
table = TemplateIO(html=True)
table += htmltext('<table id="table-diff">')
def ellipsize_html(field):
return misc.ellipsize(field.unhtmled_label, 60)
nodata_types = ('page', 'title', 'subtitle', 'comment')
display_warning = False
for diffinfo in difflib.ndiff(current_fields_list, new_fields_list):
if diffinfo[0] == '?':
# detail line, ignored
continue
field_id = diffinfo[2:].split()[0]
current_field = current_fields.get(field_id)
new_field = new_fields.get(field_id)
current_label = (
htmltext('%s - %s') % (ellipsize_html(current_field), current_field.get_type_label())
if current_field
else ''
)
new_label = (
htmltext('%s - %s') % (ellipsize_html(new_field), new_field.get_type_label())
if new_field
else ''
)
if diffinfo[0] == ' ':
# unchanged line
if current_field and new_field and current_field.type != new_field.type:
# different datatypes
if current_field.type in nodata_types:
# but current field doesn't hold data, not a problem
table += htmltext('<tr class="added-field"><td class="indicator">+</td>')
current_label = ''
elif new_field.type in nodata_types:
# new field won't hold data, but old data will be removed
table += htmltext('<tr class="removed-field"><td class="indicator">-</td>')
new_label = ''
display_warning = True
else:
# and real incompatibility, data will need to be wiped out.
table += htmltext('<tr class="type-change"><td class="indicator">!</td>')
display_warning = True
elif (
current_field
and new_field
and ET.tostring(current_field.export_to_xml('utf-8'))
!= ET.tostring(new_field.export_to_xml('utf-8'))
):
# same type, but changes within field
table += htmltext('<tr class="modified-field"><td class="indicator">~</td>')
else:
table += htmltext('<tr><td class="indicator"></td>')
elif diffinfo[0] == '-':
# removed field
table += htmltext('<tr class="removed-field"><td class="indicator">-</td>')
display_warning = True
elif diffinfo[0] == '+':
# added field
table += htmltext('<tr class="added-field"><td class="indicator">+</td>')
table += htmltext('<td>%s</td> <td>%s</td></tr>') % (current_label, new_label)
table += htmltext('</table>')
if display_warning:
r += htmltext('<div class="errornotice"><p>%s</p></div>') % _(
'The form removes or changes fields, you should review the '
'changes carefully as some data will be lost.'
)
r += htmltext('<div class="section">')
r += htmltext('<div id="form-diff">')
r += table.getvalue()
r += htmltext('<div id="legend">')
r += htmltext('<table>')
r += htmltext('<tr class="added-field"><td class="indicator">+</td><td>%s</td></tr>') % (
_('Added field')
)
r += htmltext('</table>')
r += htmltext('<table>')
r += htmltext('<tr class="removed-field"><td class="indicator">-</td><td>%s</td></tr>') % (
_('Removed field')
)
r += htmltext('</table>')
r += htmltext('<table>')
r += htmltext('<tr class="modified-field"><td class="indicator">~</td><td>%s</td></tr>') % (
_('Modified field')
)
r += htmltext('<table>')
r += htmltext('<tr class="type-change"><td class="indicator">!</td><td>%s</td></tr>') % (
_('Incompatible field')
)
r += htmltext('</table>')
r += htmltext('</div>') # .legend
get_request().method = 'GET'
get_request().form = {}
form = Form(enctype='multipart/form-data', use_tokens=False)
if display_warning:
form.add(CheckboxWidget, 'force', title=_('Overwrite despite data loss'))
else:
form.add_hidden('force', 'ok')
form.add_hidden('new_formdef', force_str(ET.tostring(new_formdef.export_to_xml(include_id=True))))
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
r += form.render()
r += htmltext('</div>') # #form-diff
r += htmltext('</div>') # .section
return r.getvalue()
def export(self):
x = self.formdef.export_to_xml(include_id=True)
misc.indent_xml(x)
response = get_response()
response.set_content_type('application/x-wcs-form')
response.set_header(
'content-disposition',
'attachment; filename=%s-%s.wcs' % (self.formdef_export_prefix, self.formdef.url_name),
)
return '<?xml version="1.0"?>\n' + force_str(ET.tostring(x))
def archive(self):
if get_publisher().is_using_postgresql():
raise TraversalError()
if get_request().form.get('download'):
return self.archive_download()
form = Form(enctype='multipart/form-data')
form.add(DateWidget, 'date', title=_('Archive forms handled before'))
form.add(CheckboxWidget, 'not-done', title=_('Include forms that have not been handled'), value=False)
form.add(CheckboxWidget, 'keep', title=_('Do not remove forms'), value=False)
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if not form.is_submitted() or form.has_errors():
get_response().breadcrumb.append(('archive', _('Archive')))
self.html_top(title=_('Archive Forms'))
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Archive Forms')
r += form.render()
return r.getvalue()
else:
return self.archive_submit(form)
def archive_submit(self, form):
class Archiver:
def __init__(self, formdef):
self.formdef = formdef
def archive(self, job=None):
all_forms = self.formdef.data_class().select()
if form.get_widget('not-done').parse() is False:
not_endpoint_status = self.formdef.workflow.get_not_endpoint_status()
not_endpoint_status_ids = ['wf-%s' % x.id for x in not_endpoint_status]
all_forms = [x for x in all_forms if x.status not in not_endpoint_status_ids]
if form.get_widget('date').parse():
date = form.get_widget('date').parse()
date = time.strptime(date, misc.date_format())
all_forms = [x for x in all_forms if x.last_update_time < date]
self.fd = io.BytesIO()
t = tarfile.open('wcs.tar.gz', 'w:gz', fileobj=self.fd)
t.add(self.formdef.get_object_filename(), 'formdef')
for formdata in all_forms:
t.add(formdata.get_object_filename(), '%s/%s' % (self.formdef.url_name, str(formdata.id)))
t.close()
if form.get_widget('keep').parse() is False:
for f in all_forms:
f.remove_self()
if job:
job.file_content = self.fd.getvalue()
job.store()
count = self.formdef.data_class().count()
archiver = Archiver(self.formdef)
if count > 100: # Arbitrary threshold
job = get_response().add_after_job(str(N_('Archiving forms')), archiver.archive)
job.done_action_url = self.formdef.get_admin_url() + 'archive?job=%s' % job.id
job.done_action_label = _('Download Archive')
job.store()
return redirect(job.get_processing_url())
else:
archiver.archive()
response = get_response()
response.set_content_type('application/x-wcs-archive')
response.set_header(
'content-disposition', 'attachment; filename=%s-archive.wcs' % self.formdef.url_name
)
return archiver.fd.getvalue()
def archive_download(self):
try:
job = AfterJob.get(get_request().form.get('download'))
except KeyError:
return redirect('.')
if not job.status == 'completed':
raise TraversalError()
response = get_response()
response.set_content_type('application/x-wcs-archive')
response.set_header(
'content-disposition', 'attachment; filename=%s-archive.wcs' % self.formdef.url_name
)
return job.file_content
def anonymise(self):
endpoints = []
for status in self.formdef.workflow.get_endpoint_status():
endpoints.append((str(status.id), status.name, str(status.id)))
form = Form(enctype='multipart/form-data')
form.add(
DateWidget,
'before_request_date',
title=_('Forms ended before'),
value=datetime.date.today() - datetime.timedelta(30),
required=True,
)
form.add(
CheckboxesWidget,
'endpoints',
title=_('Status of the forms to anonymise'),
value=[endpoint[0] for endpoint in endpoints],
options=endpoints,
inline=False,
required=True,
)
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if not form.is_submitted() or form.has_errors():
get_response().breadcrumb.append(('anonymise', _('Anonymise')))
html_top('forms', title=_('Anonymise Forms'))
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Anonymise Forms')
r += htmltext('<p>%s</p>' % _("You are about to irrevocably anonymise forms."))
r += form.render()
return r.getvalue()
else:
return self.anonymise_submit(form)
def anonymise_submit(self, form):
class Anonymiser:
def __init__(self, formdef, status_ids, before_date):
self.formdef = formdef
self.status_ids = ["wf-%s" % id for id in status_ids]
self.before_date = before_date
def anonymise(self, job=None):
for formdata in self.formdef.data_class().select():
if formdata.anonymised:
continue
if formdata.status not in self.status_ids:
continue
if (formdata.evolution and formdata.evolution[-1].time >= self.before_date) or (
formdata.receipt_time >= self.before_date
):
continue
formdata.anonymise()
before_date = form.get_widget('before_request_date').parse()
before_date = time.strptime(before_date, misc.date_format())
status_ids = form.get_widget('endpoints').parse()
count = self.formdef.data_class().count()
anonymiser = Anonymiser(self.formdef, status_ids, before_date)
if count > 100: # Arbitrary threshold
job = get_response().add_after_job(str(N_('Anonymising forms')), anonymiser.anonymise)
job.done_action_url = self.formdef.get_admin_url()
job.done_action_label = _('Back')
job.store()
return redirect(job.get_processing_url())
else:
anonymiser.anonymise()
return redirect('.')
def enable(self):
self.formdef.disabled = False
self.formdef.store(comment=_('Enable'))
if get_request().form.get('back') == 'fields':
return redirect('fields/')
return redirect('.')
def workflow_variables(self):
if not self.formdef.workflow.variables_formdef:
raise TraversalError()
self.html_top(title=_('Options'))
form = Form(enctype='multipart/form-data')
self.formdef.workflow.variables_formdef.add_fields_to_form(
form, form_data=self.formdef.get_variable_options_for_form()
)
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if form.is_submitted() and not form.has_errors():
self.formdef.set_variable_options(form)
self.formdef.store(comment=_('Change in workflow variables'))
return redirect('.')
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Options')
r += form.render()
return r.getvalue()
def workflow_options(self):
request = get_request()
if request.get_method() == 'GET' and request.form.get('file'):
value = self.formdef.workflow_options.get(request.form.get('file'))
if value:
return value.build_response()
self.html_top(title=_('Workflow Options'))
form = Form(enctype='multipart/form-data')
pristine_workflow = Workflow.get(self.formdef.workflow_id)
for status in self.formdef.workflow.possible_status:
had_options = False
for item in status.items:
prefix = '%s*%s*' % (status.id, item.id)
pristine_item = pristine_workflow.get_status(status.id).get_item(item.id)
parameters = [x for x in item.get_parameters() if not getattr(pristine_item, x)]
if not parameters:
continue
if not had_options:
form.widgets.append(HtmlWidget('<h3>%s</h3>' % status.name))
had_options = True
label = getattr(item, str('label'), None) or _(item.description)
form.widgets.append(HtmlWidget('<h4>%s</h4>' % label))
item.add_parameters_widgets(form, parameters, prefix=prefix, formdef=self.formdef)
form.add_submit('submit', _('Submit'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if form.is_submitted() and not form.has_errors():
self.workflow_options_submit(form)
return redirect('.')
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Workflow Options')
r += form.render()
return r.getvalue()
def workflow_options_submit(self, form):
self.formdef.workflow_options = {}
for widget in form.get_all_widgets():
if widget in form.get_submit_widgets():
continue
if widget.name.startswith('_'):
continue
self.formdef.workflow_options[widget.name] = widget.parse()
self.formdef.store(comment=_('Change in workflow options'))
class NamedDataSourcesDirectoryInForms(NamedDataSourcesDirectory):
pass
class FormsDirectory(AccessControlled, Directory):
_q_exports = ['', 'new', ('import', 'p_import'), 'blocks', 'categories', ('data-sources', 'data_sources')]
category_class = Category
categories = CategoriesDirectory()
blocks = BlocksDirectory(section='forms')
data_sources = NamedDataSourcesDirectoryInForms()
formdef_class = FormDef
formdef_page_class = FormDefPage
formdef_ui_class = FormDefUI
top_title = N_('Forms')
import_title = N_('Import Form')
import_submit_label = N_('Import Form')
import_paragraph = N_('You can install a new form by uploading a file ' 'or by pointing to the form URL.')
import_loading_error_message = N_('Error loading form (%s).')
import_success_message = N_(
'This form has been successfully imported. ' 'Do note it is disabled by default.'
)
import_error_message = N_(
'Imported form contained errors and has been automatically fixed, '
'you should nevertheless check everything is ok. '
'Do note it is disabled by default.'
)
def html_top(self, title):
return html_top('forms', title)
def _q_traverse(self, path):
get_response().breadcrumb.append(('forms/', _('Forms')))
return super()._q_traverse(path)
def _q_index(self):
self.html_top(title=_(self.top_title))
r = TemplateIO(html=True)
get_response().add_javascript(['jquery.js', 'widget_list.js'])
r += self.form_actions()
cats = self.category_class.select()
self.category_class.sort_by_position(cats)
one = False
formdefs = self.formdef_class.select(order_by='name', ignore_errors=True, lightweight=True)
for c in cats:
l2 = [x for x in formdefs if str(x.category_id) == str(c.id)]
l2 = [x for x in l2 if not x.disabled or (x.disabled and x.disabled_redirection)] + [
x for x in l2 if x.disabled and not x.disabled_redirection
]
if l2:
r += self.form_list(l2, title=c.name)
one = True
l2 = [x for x in formdefs if not x.category]
if l2:
if one:
title = _('Misc')
else:
title = None
l2 = [x for x in l2 if not x.disabled or (x.disabled and x.disabled_redirection)] + [
x for x in l2 if x.disabled and not x.disabled_redirection
]
r += self.form_list(l2, title=title)
return r.getvalue()
def form_actions(self):
r = TemplateIO(html=True)
has_roles = bool(get_publisher().role_class.count())
r += htmltext('<div id="appbar">')
r += htmltext('<h2>%s</h2>') % _('Forms')
if has_roles:
r += htmltext('<span class="actions">')
r += htmltext('<a href="data-sources/">%s</a>') % _('Data sources')
if get_publisher().has_site_option('fields-blocks'):
r += htmltext('<a href="blocks/">%s</a>') % _('Fields blocks')
if get_publisher().get_backoffice_root().is_accessible('categories'):
r += htmltext('<a href="categories/">%s</a>') % _('Categories')
r += htmltext('<a href="import" rel="popup">%s</a>') % _('Import')
r += htmltext('<a class="new-item" href="new" rel="popup">%s</a>') % _('New Form')
r += htmltext('</span>')
r += htmltext('</div>')
if not has_roles:
r += htmltext('<p>%s</p>') % _('You first have to define roles.')
return r.getvalue()
def form_list(self, formdefs, title):
r = TemplateIO(html=True)
if title:
r += htmltext('<h2>%s</h2>') % title
r += htmltext('<ul class="biglist">')
for formdef in formdefs:
if formdef.disabled:
r += htmltext('<li class="disabled">')
else:
r += htmltext('<li>')
r += htmltext('<strong class="label"><a href="%s/">%s</a></strong>') % (formdef.id, formdef.name)
if formdef.disabled and formdef.disabled_redirection:
r += htmltext('<a class="redirection" href="%s">(') % formdef.disabled_redirection
r += _('redirection')
r += htmltext(')</a>')
r += htmltext('</li>')
r += htmltext('</ul>')
return r.getvalue()
def new(self):
get_response().breadcrumb.append(('new', _('New')))
if get_publisher().role_class.count() == 0:
return template.error_page('forms', _('You first have to define roles.'))
formdefui = self.formdef_ui_class(None)
form = formdefui.new_form_ui()
if form.get_widget('cancel').parse():
return redirect('.')
if form.is_submitted() and not form.has_errors():
try:
formdef = formdefui.submit_form(form)
formdef.disabled = True
formdef.store()
except ValueError:
pass
else:
return redirect(str(formdef.id) + '/')
self.html_top(title=_('New Form'))
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('New Form')
r += form.render()
return r.getvalue()
def _q_lookup(self, component):
return self.formdef_page_class(component)
def p_import(self):
form = Form(enctype='multipart/form-data')
form.add(FileWidget, 'file', title=_('File'), required=False)
form.add(UrlWidget, 'url', title=_('Address'), required=False, size=50)
form.add_submit('submit', _(self.import_submit_label))
form.add_submit('cancel', _('Cancel'))
if form.get_submit() == 'cancel':
return redirect('.')
if form.is_submitted() and not form.has_errors():
try:
return self.import_submit(form)
except ValueError:
pass
get_response().breadcrumb.append(('import', _('Import')))
self.html_top(title=_(self.import_title))
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _(self.import_title)
r += htmltext('<p>%s</p>') % _(self.import_paragraph)
r += form.render()
return r.getvalue()
def import_submit(self, form):
self.imported_formdef = None
if form.get_widget('file').parse():
fp = form.get_widget('file').parse().fp
elif form.get_widget('url').parse():
url = form.get_widget('url').parse()
try:
fp = misc.urlopen(url)
except misc.ConnectionError as e:
form.set_error('url', _(self.import_loading_error_message) % str(e))
raise ValueError()
else:
form.set_error('file', _('You have to enter a file or a URL.'))
raise ValueError()
error, reason = False, None
try:
try:
formdef = self.formdef_class.import_from_xml(fp)
get_session().message = ('info', _(self.import_success_message))
except FormdefImportRecoverableError as e:
fp.seek(0)
formdef = self.formdef_class.import_from_xml(fp, fix_on_error=True)
get_session().message = ('info', _(self.import_error_message))
except FormdefImportError as e:
error = True
reason = _(e.msg) % e.msg_args
if e.details:
reason += ' [%s]' % e.details
except ValueError:
error = True
if error:
if reason:
msg = _('Invalid File (%s)') % reason
else:
msg = _('Invalid File')
if form.get_widget('url').parse():
form.set_error('url', msg)
else:
form.set_error('file', msg)
raise ValueError()
self.imported_formdef = formdef
formdef.internal_identifier = None # a new one will be set in .store()
formdef.disabled = True
formdef.store()
return redirect('%s/' % formdef.id)
class UpdateDigestAfterJob(AfterJob):
label = N_('Updating digests')
def __init__(self, formdef):
super().__init__(formdef_class=formdef.__class__, formdef_id=formdef.id)
def execute(self):
formdef = self.kwargs['formdef_class'].get(self.kwargs['formdef_id'])
for formdata in formdef.data_class().select(order_by='id'):
formdata.store()