1655 lines
66 KiB
Python
1655 lines
66 KiB
Python
# 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 json
|
|
import time
|
|
from StringIO import StringIO
|
|
import urllib2
|
|
|
|
try:
|
|
from PIL import Image
|
|
except ImportError:
|
|
Image = None
|
|
|
|
try:
|
|
import qrcode
|
|
except ImportError:
|
|
qrcode = None
|
|
|
|
from quixote import (get_publisher, get_request, get_response, get_session,
|
|
get_session_manager, redirect)
|
|
from quixote.directory import Directory, AccessControlled
|
|
from quixote.util import randbytes
|
|
from quixote.form.widget import *
|
|
from quixote.html import TemplateIO, htmltext
|
|
|
|
from qommon import _
|
|
from qommon.admin.emails import EmailsDirectory
|
|
from qommon import errors, get_cfg
|
|
from qommon import misc, get_logger
|
|
from qommon import template
|
|
from qommon.form import *
|
|
from qommon.logger import BotFilter
|
|
from qommon import tokens
|
|
from qommon import emails
|
|
|
|
from wcs.categories import Category
|
|
from wcs.formdef import FormDef
|
|
from wcs.formdata import FormData
|
|
from wcs.roles import logged_users_role
|
|
from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef
|
|
|
|
from qommon.admin.texts import TextsDirectory
|
|
|
|
|
|
class SubmittedDraftException(Exception):
|
|
pass
|
|
|
|
|
|
def html_top(title = None):
|
|
template.html_top(title = title, default_org = _('Forms'))
|
|
|
|
def get_user_forms(formdef):
|
|
"""Return forms data for the current user
|
|
|
|
formdef - the formdef from which we want form datas
|
|
|
|
"""
|
|
session = get_session()
|
|
user = session.get_user()
|
|
user_forms = []
|
|
if user and not user.anonymous:
|
|
user_forms = formdef.data_class().get_with_indexed_value('user_id', user.id)
|
|
return user_forms
|
|
|
|
from wcs.forms.common import FormStatusPage
|
|
|
|
def tryauth(url):
|
|
# tries to log the user in before redirecting to the asked url; this won't
|
|
# do anything for local logins but will use a passive SAML request when
|
|
# configured to use an external identity provider.
|
|
if get_request().user:
|
|
return redirect(url)
|
|
ident_methods = get_cfg('identification', {}).get('methods', ['idp'])
|
|
if not 'idp' in ident_methods:
|
|
# when configured with local logins and not logged in, redirect to
|
|
# asked url.
|
|
return redirect(url)
|
|
login_url = '/login/?ReturnUrl=%s&IsPassive=true' % urllib2.quote(url)
|
|
return redirect(login_url)
|
|
|
|
def auth(url):
|
|
# logs the user in before redirecting to asked url.
|
|
if get_request().user:
|
|
return redirect(url)
|
|
login_url = '/login/?ReturnUrl=%s' % urllib2.quote(url)
|
|
return redirect(login_url)
|
|
|
|
|
|
class TokenDirectory(Directory):
|
|
_q_exports = ['']
|
|
|
|
def __init__(self, formdef, token):
|
|
self.formdef = formdef
|
|
self.token = token
|
|
|
|
def _q_index(self):
|
|
if self.token.type != 'form-invite':
|
|
raise errors.TraversalError()
|
|
|
|
if self.token.formdef_id != self.formdef.id:
|
|
raise errors.TraversalError()
|
|
|
|
try:
|
|
self.token.remove_self()
|
|
except OSError:
|
|
# race condition, and the token already got removed (??!)
|
|
self.token.type = None
|
|
|
|
get_session().set_user(self.token.user_id)
|
|
return redirect(self.formdef.get_url())
|
|
|
|
|
|
class TokensDirectory(Directory):
|
|
def __init__(self, formdef):
|
|
self.formdef = formdef
|
|
|
|
def _q_lookup(self, component):
|
|
try:
|
|
token = tokens.Token.get(component)
|
|
except KeyError:
|
|
raise errors.TraversalError()
|
|
return TokenDirectory(self.formdef, token)
|
|
|
|
|
|
class TrackingCodeDirectory(Directory):
|
|
_q_exports = ['', 'load']
|
|
|
|
def __init__(self, code, formdef):
|
|
self.code = code
|
|
self.formdef = formdef
|
|
|
|
def _q_index(self):
|
|
if self.formdef is None:
|
|
raise errors.TraversalError()
|
|
form = Form()
|
|
if get_request().user and get_request().user.email:
|
|
email = get_request().user.email
|
|
else:
|
|
email = None
|
|
form.add(EmailWidget, 'email', value=email, title=_('Email'), size=25, required=True)
|
|
form.add_submit('submit', _('Send email'))
|
|
form.add_submit('cancel', _('Cancel'))
|
|
|
|
if form.get_submit() == 'cancel':
|
|
return redirect('./load')
|
|
|
|
if form.is_submitted() and not form.has_errors():
|
|
email = form.get_widget('email').parse()
|
|
data = {
|
|
'form_tracking_code': self.code,
|
|
'tracking_code': self.code,
|
|
'email': email
|
|
}
|
|
data.update(self.formdef.get_substitution_variables(minimal=True))
|
|
emails.custom_ezt_email('tracking-code-reminder', data,
|
|
email, fire_and_forget=True)
|
|
return redirect('./load')
|
|
|
|
html_top()
|
|
r = TemplateIO(html=True)
|
|
r += htmltext('<h2>%s</h2>') % _('Keep your tracking code')
|
|
r += TextsDirectory.get_html_text('tracking-code-email-dialog')
|
|
r += form.render()
|
|
return r.getvalue()
|
|
|
|
def load(self):
|
|
try:
|
|
tracking_code = get_publisher().tracking_code_class.get(self.code)
|
|
if tracking_code.formdata_id is None:
|
|
# this tracking code was not associated with any data; return a 404
|
|
raise KeyError
|
|
formdata = tracking_code.formdata
|
|
except KeyError:
|
|
raise errors.TraversalError()
|
|
if formdata.formdef.enable_tracking_codes is False:
|
|
raise errors.TraversalError()
|
|
if BotFilter.is_bot():
|
|
raise errors.AccessForbiddenError()
|
|
get_session().mark_anonymous_formdata(formdata)
|
|
return redirect(formdata.get_url().rstrip('/'))
|
|
|
|
|
|
class TrackingCodesDirectory(Directory):
|
|
_q_exports = ['load']
|
|
|
|
def __init__(self, formdef=None):
|
|
self.formdef = formdef
|
|
|
|
def load(self):
|
|
code = get_request().form.get('code')
|
|
return redirect('./%s/load' % code)
|
|
|
|
def _q_lookup(self, component):
|
|
return TrackingCodeDirectory(component, self.formdef)
|
|
|
|
class FormPage(Directory):
|
|
_q_exports = ['', 'tempfile', 'tokens', 'schema', 'tryauth',
|
|
'auth', 'qrcode', 'autosave', 'code', 'removedraft']
|
|
|
|
def __init__(self, component):
|
|
try:
|
|
self.formdef = FormDef.get_by_urlname(component)
|
|
except KeyError:
|
|
raise errors.TraversalError()
|
|
|
|
self.substvars = {}
|
|
get_publisher().substitutions.feed(self)
|
|
get_publisher().substitutions.feed(self.formdef)
|
|
|
|
self.tokens = TokensDirectory(self.formdef)
|
|
self.code = TrackingCodesDirectory(self.formdef)
|
|
self.action_url = '.'
|
|
self.edit_mode = False
|
|
|
|
self.page_number = len([
|
|
x for x in self.formdef.fields[1:] if x.type == 'page']) + 1
|
|
|
|
self.user = get_request().user
|
|
get_response().breadcrumb.append( (component + '/', self.formdef.name) )
|
|
|
|
def html_top(self, *args, **kwargs):
|
|
html_top(*args, **kwargs)
|
|
|
|
def get_substitution_variables(self):
|
|
return self.substvars
|
|
|
|
def schema(self):
|
|
# backward compatibility
|
|
from wcs.api import ApiFormdefDirectory
|
|
return ApiFormdefDirectory(self.formdef).schema()
|
|
|
|
def check_role(self):
|
|
if self.formdef.roles:
|
|
if not self.user:
|
|
raise errors.AccessUnauthorizedError()
|
|
if logged_users_role().id not in self.formdef.roles and not (
|
|
self.user and self.user.is_admin):
|
|
if self.user and self.user.roles:
|
|
user_roles = set(self.user.roles)
|
|
else:
|
|
user_roles = set([])
|
|
other_roles = (self.formdef.roles or [])
|
|
if self.formdef.workflow_roles:
|
|
other_roles.extend(self.formdef.workflow_roles.values())
|
|
if not user_roles.intersection(other_roles):
|
|
raise errors.AccessForbiddenError()
|
|
|
|
def has_confirmation_page(self):
|
|
if self.formdef.confirmation:
|
|
return True
|
|
if self.formdef.has_captcha:
|
|
session = get_session()
|
|
if not (session.get_user() or session.won_captcha):
|
|
return True
|
|
return False
|
|
|
|
def step(self, step_no, page_no, log_detail=None, data=None):
|
|
if step_no == 0:
|
|
self.substvars['current_page_no'] = str(page_no + 1)
|
|
if log_detail:
|
|
get_logger().info('form %s - step %s (%s)' % (self.formdef.name, step_no, log_detail))
|
|
else:
|
|
get_logger().info('form %s - step %s' % (self.formdef.name, step_no))
|
|
|
|
page_labels = []
|
|
current_position = 1
|
|
|
|
page_index = self.formdef.get_start_page()
|
|
for field in self.formdef.fields:
|
|
if field.type != 'page':
|
|
continue
|
|
if field.is_visible(data, self.formdef):
|
|
page_labels.append(field.label)
|
|
if page_index == page_no:
|
|
current_position = len(page_labels)
|
|
page_index += 1
|
|
|
|
if not page_labels:
|
|
page_labels.append(_('Filling'))
|
|
|
|
if step_no > 0:
|
|
current_position = len(page_labels) + step_no
|
|
|
|
if self.has_confirmation_page() and not self.edit_mode:
|
|
page_labels.append(_('Validating'))
|
|
|
|
r = TemplateIO(html=True)
|
|
r += htmltext('<div id="steps"><ol>')
|
|
|
|
for i, page_label in enumerate(page_labels):
|
|
classes = []
|
|
index = i + 1
|
|
if index == 1:
|
|
classes.append('first')
|
|
if index == len(page_labels):
|
|
classes.append('last')
|
|
if index == current_position:
|
|
classes.append('current')
|
|
elif index < current_position:
|
|
classes.append('step-before')
|
|
elif index > current_position:
|
|
classes.append('step-after')
|
|
r += htmltext('<li class="%s">') % ' '.join(classes)
|
|
r += htmltext('<span class="marker">%d</span> <span class="label">%s</span>') % (
|
|
index, page_label)
|
|
r += htmltext('</li>')
|
|
|
|
r += htmltext('</ol></div>')
|
|
return r.getvalue()
|
|
|
|
def page(self, page_no, page_change=True, log_detail=None, page_error_messages=None):
|
|
r = TemplateIO(html=True)
|
|
displayed_fields = []
|
|
|
|
session = get_session()
|
|
|
|
if page_no > self.formdef.get_start_page():
|
|
magictoken = get_request().form['magictoken']
|
|
self.feed_current_data(magictoken)
|
|
|
|
form = self.create_form(page_no, displayed_fields)
|
|
if page_error_messages:
|
|
form.add_global_errors(page_error_messages)
|
|
if getattr(session, 'ajax_form_token', None):
|
|
form.add_hidden('_ajax_form_token', session.ajax_form_token)
|
|
if get_request().is_in_backoffice():
|
|
form.attrs['data-is-backoffice'] = 'true'
|
|
form.action = self.action_url
|
|
# include a data-has-draft attribute on the <form> element when a draft
|
|
# already exists for the form; this will activate the autosave.
|
|
magictoken = get_request().form.get('magictoken')
|
|
if magictoken:
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
if self.formdef.enable_tracking_codes and not self.edit_mode:
|
|
form.attrs['data-has-draft'] = 'yes'
|
|
else:
|
|
form_data = {}
|
|
|
|
if page_no == self.formdef.get_start_page() and not get_request().form.has_key('magictoken'):
|
|
magictoken = randbytes(8)
|
|
else:
|
|
magictoken = get_request().form['magictoken']
|
|
form.add_hidden('magictoken', magictoken)
|
|
data = session.get_by_magictoken(magictoken, {})
|
|
|
|
if page_no == self.formdef.get_start_page() and get_request().form.has_key('cancelurl'):
|
|
cancelurl = get_request().form['cancelurl']
|
|
session.add_magictoken(magictoken, {'__cancelurl': cancelurl})
|
|
|
|
if self.edit_mode and page_no == self.page_number - 1:
|
|
form.add_submit('submit', _('Save Changes'))
|
|
elif not self.has_confirmation_page() and page_no == self.page_number - 1:
|
|
form.add_submit('submit', _('Submit'))
|
|
else:
|
|
form.add_submit('submit', _('Next'))
|
|
|
|
if page_no > self.formdef.get_start_page():
|
|
form.add_submit('previous', _('Previous'))
|
|
|
|
if page_change:
|
|
# on page change, we fake a GET request so the form is not altered
|
|
# with errors from the previous submit; if the page was already
|
|
# visited, we restore values; otherwise we set req.form as empty.
|
|
req = get_request()
|
|
req.environ['REQUEST_METHOD'] = 'GET'
|
|
one = False
|
|
for field in displayed_fields:
|
|
k = field.id
|
|
v = None
|
|
prefilled = False
|
|
verified = False
|
|
|
|
if field.prefill:
|
|
prefill_user = get_request().user
|
|
if get_request().is_in_backoffice():
|
|
prefill_user = get_publisher().substitutions.get_context_variables(
|
|
).get('form_user')
|
|
v, verified = field.get_prefill_value(user=prefill_user)
|
|
|
|
# always set additional attributes as they will be used for
|
|
# "live prefill", regardless of existing data.
|
|
form.get_widget('f%s' % k).prefill_attributes = field.get_prefill_attributes()
|
|
|
|
if data.has_key(k):
|
|
v = data[k]
|
|
elif field.prefill:
|
|
if get_request().is_in_backoffice() and (
|
|
field.prefill and field.prefill.get('type') == 'geoloc'):
|
|
# turn off prefilling from geolocation attributes if
|
|
# the form is filled from the backoffice
|
|
v = None
|
|
if v:
|
|
prefilled = True
|
|
if field.prefill and field.prefill.get('type') != 'string' and not verified:
|
|
# unless we prefilled with a fixed string, we
|
|
# inform the user the field value has been
|
|
# prefilled.
|
|
form.get_widget('f%s' % k).set_message(
|
|
_('Value has been automatically prefilled.'))
|
|
form.get_widget('f%s' % k).prefilled = True
|
|
|
|
if not prefilled and form.get_widget('f%s' % k):
|
|
form.get_widget('f%s' % k).clear_error()
|
|
|
|
if v is not None:
|
|
if not isinstance(v, str) and field.convert_value_to_str:
|
|
v = field.convert_value_to_str(v)
|
|
form.get_widget('f%s' % k).set_value(v)
|
|
if verified:
|
|
form.get_widget('f%s' % k).readonly = 'readonly'
|
|
form.get_widget('f%s' % k).attrs['readonly'] = 'readonly'
|
|
req.form['f%s' % k] = v
|
|
one = True
|
|
|
|
if not one:
|
|
req.form = {}
|
|
|
|
for field in displayed_fields:
|
|
if field.prefill:
|
|
# always set additional attributes as they will be used for
|
|
# "live prefill", regardless of existing data.
|
|
form.get_widget('f%s' % field.id).prefill_attributes = field.get_prefill_attributes()
|
|
|
|
self.html_top(self.formdef.name)
|
|
r += self.form_side(0, page_no, log_detail=log_detail, data=data, magictoken=magictoken)
|
|
r += get_session().display_message()
|
|
|
|
form.add_hidden('step', '0')
|
|
form.add_hidden('page', page_no)
|
|
|
|
form.add_submit('cancel', _('Cancel'), css_class = 'cancel')
|
|
if self.formdef.enable_tracking_codes and not self.edit_mode:
|
|
form.add_submit('savedraft', _('Save Draft'), css_class = 'save-draft',
|
|
attrs={'style': 'display: none'})
|
|
|
|
r += form.render()
|
|
return r.getvalue()
|
|
|
|
def form_side(self, step_no, page_no, log_detail=None, data=None, magictoken=None):
|
|
'''Create the elements that typically appear aside the main form
|
|
(tracking code and steps).'''
|
|
r = TemplateIO(html=True)
|
|
r += htmltext('<div id="side">')
|
|
if self.formdef.enable_tracking_codes and data:
|
|
# display tracking code box if they are enabled and there's some
|
|
# data (e.g. the user is not on a insufficient authenticiation
|
|
# context page)
|
|
r += self.tracking_code_box(data, magictoken)
|
|
r += self.step(step_no, page_no, log_detail, data=data)
|
|
r += htmltext('</div> <!-- #side -->')
|
|
return r.getvalue()
|
|
|
|
def tracking_code_box(self, data, magictoken):
|
|
'''Create the tracking code box, it displays the current tracking code
|
|
and a 'remove draft' button if the current form is a draft that has
|
|
been recalled.'''
|
|
r = TemplateIO(html=True)
|
|
r += htmltext('<div id="tracking-code">')
|
|
r += htmltext('<h3>%s</h3>') % _('Tracking code')
|
|
|
|
tracking_code = None
|
|
draft_formdata_id = data.get('draft_formdata_id')
|
|
if draft_formdata_id:
|
|
formdata = self.formdef.data_class().get(draft_formdata_id)
|
|
tracking_code = formdata.tracking_code
|
|
else:
|
|
tracking_code = data.get('future_tracking_code')
|
|
|
|
if tracking_code:
|
|
get_response().add_javascript(['jquery.js', 'jquery-ui.js', 'popup.js'])
|
|
r += htmltext('<a rel="popup" href="%s">%s</a>') % (
|
|
'code/%s/' % tracking_code, tracking_code)
|
|
r += TextsDirectory.get_html_text('tracking-code-short-text')
|
|
|
|
if data.get('is_recalled_draft'):
|
|
r += htmltext('<form action="removedraft" method="POST">')
|
|
r += htmltext('<input type="hidden" name="magictoken" value="%s">') % magictoken
|
|
r += htmltext('<button>%s</button>') % _('Remove Draft')
|
|
r += htmltext('</form>')
|
|
r += htmltext('</div>') # <!-- #tracking-code -->
|
|
return r.getvalue()
|
|
|
|
def feed_current_data(self, magictoken):
|
|
# create a fake FormData to feed variables
|
|
formdata = FormData()
|
|
formdata._formdef = self.formdef
|
|
formdata.user = get_request().user
|
|
formdata.data = get_session().get_by_magictoken(magictoken, {})
|
|
if get_request().is_in_backoffice():
|
|
formdata.user_id = None
|
|
draft_formdata_id = formdata.data.get('draft_formdata_id')
|
|
if draft_formdata_id:
|
|
formdata = self.formdef.data_class().get(draft_formdata_id)
|
|
formdata.status = str('')
|
|
get_publisher().substitutions.feed(formdata)
|
|
|
|
def check_disabled(self):
|
|
if self.formdef.is_disabled():
|
|
if self.formdef.disabled_redirection:
|
|
return misc.get_variadic_url(self.formdef.disabled_redirection,
|
|
get_publisher().substitutions.get_context_variables())
|
|
else:
|
|
raise errors.AccessForbiddenError()
|
|
return False
|
|
|
|
def create_form(self, *args, **kwargs):
|
|
return self.formdef.create_form(*args, **kwargs)
|
|
|
|
def create_view_form(self, *args, **kwargs):
|
|
return self.formdef.create_view_form(*args, **kwargs)
|
|
|
|
def check_authentication_context(self):
|
|
if not self.formdef.required_authentication_contexts:
|
|
return
|
|
if get_session().get_authentication_context() in self.formdef.required_authentication_contexts:
|
|
return
|
|
|
|
self.html_top(self.formdef.name)
|
|
r = TemplateIO(html=True)
|
|
r += self.form_side(step_no=0, page_no=self.formdef.get_start_page())
|
|
auth_contexts = get_publisher().get_supported_authentication_contexts()
|
|
r += htmltext('<div class="errornotice">')
|
|
r += htmltext('<p>%s</p>') % _('You need a stronger authentication level to fill this form.')
|
|
r += htmltext('</div>')
|
|
root_url = get_publisher().get_root_url()
|
|
for auth_context in self.formdef.required_authentication_contexts:
|
|
r += htmltext('<p><a class="button" href="%slogin/?forceAuthn=true">%s</a></p>') % (
|
|
root_url, _('Login with %s') % auth_contexts[auth_context])
|
|
return r.getvalue()
|
|
|
|
def _q_index(self, log_detail=None):
|
|
self.check_role()
|
|
authentication_context_check_result = self.check_authentication_context()
|
|
if authentication_context_check_result:
|
|
return authentication_context_check_result
|
|
|
|
if self.check_disabled():
|
|
return redirect(self.check_disabled())
|
|
|
|
session = get_session()
|
|
if self.formdef.enable_tracking_codes:
|
|
if get_request().form.get('_ajax_form_token'):
|
|
# _ajax_form_token is immediately removed, this prevents
|
|
# late autosave() to overwrite data after the user went to a
|
|
# different page.
|
|
try:
|
|
session.remove_form_token(get_request().form.get('_ajax_form_token'))
|
|
except ValueError:
|
|
# already got removed, this may be because the form got
|
|
# submitted twice.
|
|
pass
|
|
session.ajax_form_token = session.create_form_token()
|
|
|
|
if get_request().form.get('magictoken'):
|
|
no_magic = object()
|
|
session_magic_token = session.get_by_magictoken(
|
|
get_request().form.get('magictoken'), no_magic)
|
|
if session_magic_token is no_magic:
|
|
if (get_request().form.get('page') != str(self.formdef.get_start_page()) or
|
|
get_request().form.get('step') != '0'):
|
|
# the magictoken that has been submitted is not available
|
|
# in the session and we're not on the first page of the
|
|
# first step that means we probably lost the session in
|
|
# mid-air.
|
|
get_session().message = ('error',
|
|
_('Sorry, your session have been lost.'))
|
|
return redirect(self.formdef.get_url())
|
|
|
|
else:
|
|
# first hit on first page, if tracking code are enabled and we
|
|
# are not editing an existing formdata, generate a new tracking
|
|
# code.
|
|
if not self.edit_mode and self.formdef.enable_tracking_codes and not get_request().form.has_key('mt'):
|
|
tracking_code = get_publisher().tracking_code_class()
|
|
tracking_code.store()
|
|
token = randbytes(8)
|
|
get_request().form['magictoken'] = token
|
|
session.add_magictoken(token, {'future_tracking_code': tracking_code.id})
|
|
|
|
existing_formdata = None
|
|
if self.edit_mode:
|
|
existing_formdata = self.edited_data.data
|
|
if not get_request().form:
|
|
# on the initial visit editing the form (i.e. not after
|
|
# clicking for previous or next page), we need to load the
|
|
# existing data into the session
|
|
self.edited_data.feed_session()
|
|
token = randbytes(8)
|
|
get_request().form['magictoken'] = token
|
|
session.add_magictoken(token, self.edited_data.data)
|
|
elif self.formdef.only_allow_one:
|
|
user_forms = get_user_forms(self.formdef)
|
|
if [x for x in user_forms if not x.is_draft()]:
|
|
return redirect('%s/' % user_forms[0].id)
|
|
|
|
get_response().add_javascript(['jquery.js', 'qommon.forms.js'])
|
|
form = Form()
|
|
form.add_hidden('step', '-1')
|
|
form.add_hidden('page', '-1')
|
|
form.add_hidden('magictoken', '-1')
|
|
form.add_submit('cancel')
|
|
|
|
if self.formdef.enable_tracking_codes:
|
|
form.add_submit('removedraft')
|
|
form.add_submit('savedraft')
|
|
|
|
if not form.is_submitted():
|
|
if get_request().form.has_key('mt'):
|
|
magictoken = get_request().form['mt']
|
|
data = session.get_by_magictoken(magictoken, {})
|
|
if not get_request().is_in_backoffice():
|
|
# don't remove magictoken as the backoffice agent may get
|
|
# the page reloaded.
|
|
session.remove_magictoken(magictoken)
|
|
if data:
|
|
# create a new one since the other has been exposed in a url
|
|
magictoken = randbytes(8)
|
|
session.add_magictoken(magictoken, data)
|
|
get_request().form['magictoken'] = magictoken
|
|
if data.has_key('page_no'):
|
|
page_no = int(data['page_no'])
|
|
del data['page_no']
|
|
if page_no == -1 or page_no >= self.page_number:
|
|
req = get_request()
|
|
for k, v in data.items():
|
|
req.form['f%s' % k] = v
|
|
return self.validating(data)
|
|
else:
|
|
page_no = self.formdef.get_start_page()
|
|
self.feed_current_data(magictoken)
|
|
return self.page(page_no, True)
|
|
self.feed_current_data(None)
|
|
if self.formdef.get_start_page() is None:
|
|
raise errors.TraversalError()
|
|
return self.page(self.formdef.get_start_page())
|
|
|
|
if form.get_submit() == 'cancel':
|
|
get_logger().info('form %s - cancel' % (self.formdef.name))
|
|
if self.edit_mode:
|
|
return redirect('.')
|
|
try:
|
|
magictoken = form.get_widget('magictoken').parse()
|
|
cancelurl = session.get_by_magictoken(magictoken, {}).get('__cancelurl')
|
|
if cancelurl:
|
|
return redirect(cancelurl)
|
|
except KeyError:
|
|
pass
|
|
if get_request().is_in_backoffice():
|
|
return redirect(get_publisher().get_backoffice_url() + '/submission/')
|
|
return redirect(get_publisher().get_root_url())
|
|
|
|
try:
|
|
step = int(form.get_widget('step').parse())
|
|
except TypeError:
|
|
step = 0
|
|
|
|
# reset verified fields, making sure the user cannot alter them.
|
|
prefill_user = get_request().user
|
|
if get_request().is_in_backoffice():
|
|
prefill_user = get_publisher().substitutions.get_context_variables().get('form_user')
|
|
if prefill_user:
|
|
for field in self.formdef.fields:
|
|
if not field.prefill:
|
|
continue
|
|
if not 'f%s' % field.id in get_request().form:
|
|
continue
|
|
v, verified = field.get_prefill_value(user=prefill_user)
|
|
if verified:
|
|
get_request().form['f%s' % field.id] = v
|
|
|
|
if step == 0:
|
|
try:
|
|
page_no = int(form.get_widget('page').parse())
|
|
except TypeError:
|
|
page_no = -1
|
|
try:
|
|
magictoken = form.get_widget('magictoken').parse()
|
|
except KeyError:
|
|
magictoken = randbytes(8)
|
|
|
|
self.feed_current_data(magictoken)
|
|
|
|
form = self.create_form(page_no)
|
|
form.add_submit('previous')
|
|
if self.formdef.enable_tracking_codes:
|
|
form.add_submit('removedraft')
|
|
form.add_submit('savedraft')
|
|
form.add_submit('submit')
|
|
if page_no > self.formdef.get_start_page() and form.get_submit() == 'previous':
|
|
return self.previous_page(page_no, magictoken)
|
|
|
|
if self.formdef.enable_tracking_codes and form.get_submit() == 'removedraft':
|
|
return self.removedraft()
|
|
|
|
if self.formdef.enable_tracking_codes and form.get_submit() == 'savedraft':
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
data = self.formdef.get_data(form)
|
|
form_data.update(data)
|
|
filled = self.save_draft(form_data, page_no)
|
|
return redirect(filled.get_url().rstrip('/'))
|
|
|
|
page_error_messages = []
|
|
if form.get_submit() == 'submit':
|
|
pages = [x for x in self.formdef.fields if x.type == 'page']
|
|
try:
|
|
page = pages[page_no]
|
|
post_conditions = page.post_conditions or []
|
|
except IndexError:
|
|
post_conditions = []
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
data = self.formdef.get_data(form)
|
|
form_data.update(data)
|
|
for i, post_condition in enumerate(post_conditions):
|
|
condition = post_condition.get('condition')
|
|
error_message = post_condition.get('error_message')
|
|
errored = False
|
|
try:
|
|
if not page.evaluate_condition(form_data, self.formdef, condition):
|
|
errored = True
|
|
except RuntimeError:
|
|
errored = True
|
|
if errored:
|
|
form.add(HiddenErrorWidget, 'post_condition%d' % i)
|
|
form.set_error('post_condition%d' % i, 'error')
|
|
page_error_messages.append(error_message)
|
|
|
|
# form.get_submit() returns the name of the clicked button, and
|
|
# it will return True if the form has been submitted, but not
|
|
# by clicking on a submit widget; for example if an "add row"
|
|
# button is clicked.
|
|
if form.has_errors() or form.get_submit() is True:
|
|
if page_no == -1:
|
|
# this situation shouldn't arise (that likely means the
|
|
# page hidden field had an error in its submission), in
|
|
# that case we just fall back to the first page.
|
|
page_no = self.formdef.get_start_page()
|
|
return self.page(page_no, page_change=False,
|
|
page_error_messages=page_error_messages)
|
|
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
data = self.formdef.get_data(form)
|
|
form_data.update(data)
|
|
|
|
session.add_magictoken(magictoken, form_data)
|
|
|
|
while True:
|
|
page_no = int(page_no) + 1
|
|
try:
|
|
next_page = self.formdef.get_page(page_no)
|
|
except IndexError:
|
|
break
|
|
if next_page.is_visible(form_data, self.formdef):
|
|
break
|
|
|
|
draft_id = session.get_by_magictoken(magictoken, {}).get('draft_formdata_id')
|
|
if draft_id:
|
|
# if there's a draft (be it because drafts are enabled or
|
|
# because the formdata was created as a draft via the
|
|
# submission API), update it with current data.
|
|
try:
|
|
self.autosave_draft(draft_id, page_no, form_data)
|
|
except SubmittedDraftException:
|
|
if get_request().is_in_backoffice():
|
|
get_session().message = ('error', _('This form has already been submitted.'))
|
|
return redirect(get_publisher().get_backoffice_url() + '/submission/')
|
|
return template.error_page(_('This form has already been submitted.'))
|
|
elif self.formdef.enable_tracking_codes and not self.edit_mode:
|
|
# if there's no draft yet and tracking codes are enabled, create one
|
|
filled = self.save_draft(form_data, page_no)
|
|
|
|
if page_no == self.page_number:
|
|
# last page has been submitted
|
|
req = get_request()
|
|
for field in self.formdef.fields:
|
|
k = field.id
|
|
if form_data.has_key(k):
|
|
v = form_data[k]
|
|
if field.convert_value_to_str:
|
|
v = field.convert_value_to_str(v)
|
|
req.form['f%s' % k] = v
|
|
if self.edit_mode:
|
|
form = self.create_view_form(form_data, use_tokens=False)
|
|
return self.submitted_existing(form)
|
|
if self.has_confirmation_page():
|
|
return self.validating(form_data)
|
|
else:
|
|
step = 1 # so it will flow to submit
|
|
# kind of restore state
|
|
form = Form()
|
|
form.add_hidden('step', '-1')
|
|
form.add_hidden('page', '-1')
|
|
form.add_hidden('magictoken', '-1')
|
|
form.add_submit('cancel')
|
|
if self.formdef.enable_tracking_codes:
|
|
form.add_submit('removedraft')
|
|
form.add_submit('savedraft')
|
|
|
|
else:
|
|
return self.page(page_no)
|
|
|
|
if step == 1:
|
|
form.add_submit('previous')
|
|
magictoken = form.get_widget('magictoken').parse()
|
|
|
|
if form.get_submit() == 'previous':
|
|
return self.previous_page(self.page_number, magictoken)
|
|
magictoken = form.get_widget('magictoken').parse()
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
data = self.formdef.get_data(form)
|
|
form_data.update(data)
|
|
session.add_magictoken(magictoken, form_data)
|
|
|
|
step = 2 # so it will flow to submit
|
|
form = Form()
|
|
form.add_hidden('step', '-1')
|
|
form.add_hidden('page', '-1')
|
|
form.add_hidden('magictoken', '-1')
|
|
form.add_submit('cancel')
|
|
|
|
if step == 2:
|
|
form.add_submit('previous')
|
|
magictoken = form.get_widget('magictoken').parse()
|
|
self.feed_current_data(magictoken)
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
|
|
if form.get_submit() == 'previous':
|
|
return self.previous_page(self.page_number, magictoken)
|
|
|
|
if self.formdef.enable_tracking_codes and form.get_submit() == 'removedraft':
|
|
return self.removedraft()
|
|
|
|
if self.formdef.enable_tracking_codes and form.get_submit() == 'savedraft':
|
|
filled = self.save_draft(form_data, page_no = -1)
|
|
return redirect(filled.get_url().rstrip('/'))
|
|
|
|
# so it gets FakeFileWidget in preview mode
|
|
form = self.create_view_form(form_data,
|
|
use_tokens=self.has_confirmation_page())
|
|
if self.formdef.has_captcha and not (get_session().get_user() or get_session().won_captcha):
|
|
form.add_captcha(hint='')
|
|
if form.captcha.has_error():
|
|
return self.validating(form_data)
|
|
|
|
if form.has_errors():
|
|
# the only possible error here is a token error if the form is
|
|
# submitted a second time
|
|
return template.error_page(_('This form has already been submitted.'))
|
|
|
|
return self.submitted(form, existing_formdata)
|
|
|
|
|
|
def previous_page(self, page_no, magictoken):
|
|
session = get_session()
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
|
|
while True:
|
|
page_no = int(page_no) - 1
|
|
try:
|
|
previous_page = self.formdef.get_page(page_no)
|
|
except IndexError:
|
|
break
|
|
if not previous_page.condition:
|
|
break
|
|
|
|
if previous_page.is_visible(form_data, self.formdef):
|
|
break
|
|
|
|
return self.page(page_no, page_change=True)
|
|
|
|
def removedraft(self):
|
|
magictoken = get_request().form.get('magictoken')
|
|
if magictoken:
|
|
form_data = get_session().get_by_magictoken(magictoken, {})
|
|
if form_data.get('draft_formdata_id'):
|
|
self.formdef.data_class().remove_object(form_data.get('draft_formdata_id'))
|
|
return redirect(get_publisher().get_root_url())
|
|
|
|
def autosave_draft(self, draft_id, page_no, form_data):
|
|
try:
|
|
formdata = self.formdef.data_class().get(draft_id)
|
|
except KeyError:
|
|
return
|
|
|
|
if not formdata.status == 'draft':
|
|
raise SubmittedDraftException()
|
|
|
|
formdata.page_no = page_no
|
|
formdata.data = form_data
|
|
formdata.receipt_time = time.localtime()
|
|
if not get_request().is_in_backoffice():
|
|
session = get_session()
|
|
if session and session.user and not str(session.user).startswith('anonymous-'):
|
|
formdata.user_id = session.user
|
|
formdata.store()
|
|
|
|
def autosave(self):
|
|
get_response().set_content_type('application/json')
|
|
def result_error(reason):
|
|
get_request().ignore_session = True
|
|
return json.dumps({'result': 'error', 'reason': reason})
|
|
|
|
if not get_session().has_form_token(get_request().form.get('_ajax_form_token')):
|
|
return result_error('obsolete ajax form token')
|
|
|
|
try:
|
|
page_no = int(get_request().form.get('page'))
|
|
except TypeError:
|
|
return result_error('missing page_no')
|
|
|
|
magictoken = get_request().form.get('magictoken')
|
|
if not magictoken:
|
|
return result_error('missing magictoken')
|
|
|
|
session = get_session()
|
|
if not session:
|
|
return result_error('missing session')
|
|
|
|
self.feed_current_data(magictoken)
|
|
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
if not form_data:
|
|
return result_error('missing data')
|
|
|
|
form = self.create_form(page_no)
|
|
data = self.formdef.get_data(form)
|
|
if not data:
|
|
return result_error('nothing to save')
|
|
|
|
form_data.update(data)
|
|
|
|
# reload session to make sure _ajax_form_token is still valid
|
|
session = get_session_manager().get(get_session().id)
|
|
if not session:
|
|
return result_error('cannot get ajax form token (lost session)')
|
|
if not session.has_form_token(get_request().form.get('_ajax_form_token')):
|
|
return result_error('obsolete ajax form token (late check)')
|
|
|
|
try:
|
|
draft_formdata = self.save_draft(form_data, page_no)
|
|
except SubmittedDraftException:
|
|
return result_error('form has already been submitted')
|
|
|
|
return json.dumps({'result': 'success'})
|
|
|
|
def save_draft(self, data, page_no):
|
|
filled = self.get_current_draft() or self.formdef.data_class()()
|
|
if filled.id and filled.status != 'draft':
|
|
raise SubmittedDraftException()
|
|
filled.data = data
|
|
filled.status = 'draft'
|
|
filled.page_no = page_no
|
|
filled.receipt_time = time.localtime()
|
|
session = get_session()
|
|
if get_request().is_in_backoffice():
|
|
# if submitting via backoffice store fhe formdata as is.
|
|
filled.store()
|
|
else:
|
|
# if submitting via frontoffice, attach current user, eventually
|
|
# anonymous, to the formdata
|
|
if session and session.user and not str(session.user).startswith('anonymous-'):
|
|
filled.user_id = session.user
|
|
filled.store()
|
|
|
|
if not filled.user_id:
|
|
get_session().mark_anonymous_formdata(filled)
|
|
|
|
data['draft_formdata_id'] = filled.id
|
|
self.set_tracking_code(filled, data)
|
|
|
|
get_logger().info('form %s - saving draft (id: %s)' % (self.formdef.name, filled.id))
|
|
|
|
return filled
|
|
|
|
def get_current_draft(self):
|
|
magictoken = get_request().form.get('magictoken')
|
|
if magictoken:
|
|
session = get_session()
|
|
form_data = session.get_by_magictoken(magictoken, {})
|
|
draft_formdata_id = form_data.get('draft_formdata_id')
|
|
if draft_formdata_id:
|
|
# there was a draft, use it.
|
|
return self.formdef.data_class().get(draft_formdata_id)
|
|
return None
|
|
|
|
def submitted(self, form, existing_formdata = None):
|
|
if existing_formdata: # modifying
|
|
filled = existing_formdata
|
|
filled.last_modification_time = time.localtime()
|
|
# XXX: what about status?
|
|
else:
|
|
filled = self.get_current_draft() or self.formdef.data_class()()
|
|
filled.just_created()
|
|
|
|
filled.data = self.formdef.get_data(form)
|
|
session = get_session()
|
|
if session and session.user and not str(session.user).startswith('anonymous-'):
|
|
filled.user_id = get_request().user.id
|
|
|
|
if get_request().get_path().startswith('/backoffice/'):
|
|
filled.user_id = None
|
|
|
|
if self.formdef.only_allow_one:
|
|
# this is already checked in _q_index but it's done a second time
|
|
# just before a new form is to be stored.
|
|
user_forms = get_user_forms(self.formdef)
|
|
if [x for x in user_forms if not x.is_draft()]:
|
|
return redirect('%s/' % user_forms[0].id)
|
|
|
|
filled.store()
|
|
self.set_tracking_code(filled)
|
|
session.remove_magictoken(get_request().form.get('magictoken'))
|
|
|
|
get_logger().info('form %s - done (id: %s)' % (self.formdef.name, filled.id))
|
|
url = None
|
|
if existing_formdata is None:
|
|
url = filled.perform_workflow()
|
|
|
|
if not filled.user_id:
|
|
get_session().mark_anonymous_formdata(filled)
|
|
|
|
if not url:
|
|
if get_request().get_path().startswith('/backoffice/'):
|
|
url = filled.get_url(backoffice=True)
|
|
else:
|
|
url = filled.get_url()
|
|
return redirect(url)
|
|
|
|
def set_tracking_code(self, formdata, magictoken_data=None):
|
|
if not self.formdef.enable_tracking_codes:
|
|
return
|
|
if formdata.tracking_code:
|
|
return
|
|
code = get_publisher().tracking_code_class()
|
|
if magictoken_data and 'future_tracking_code' in magictoken_data:
|
|
code.id = magictoken_data['future_tracking_code']
|
|
code.formdata = formdata # this will .store() the code
|
|
|
|
def submitted_existing(self, form):
|
|
new_data = self.formdef.get_data(form)
|
|
for k, v in self.edited_data.data.items():
|
|
if k.startswith(WorkflowBackofficeFieldsFormDef.field_prefix):
|
|
new_data[k] = v
|
|
self.edited_data.data = new_data
|
|
self.edited_data.store()
|
|
wf_status = self.edited_data.get_status()
|
|
url = None
|
|
for item in wf_status.items:
|
|
if item.id == self.edit_action_id and item.status:
|
|
self.edited_data.jump_status(item.status)
|
|
url = self.edited_data.perform_workflow()
|
|
break
|
|
return redirect(url or '.')
|
|
|
|
def tempfile(self):
|
|
self.check_role()
|
|
if self.user and not self.user.id == get_session().user:
|
|
self.check_receiver()
|
|
try:
|
|
t = get_request().form['t']
|
|
tempfile = get_session().tempfiles[t]
|
|
except KeyError:
|
|
raise errors.TraversalError()
|
|
response = get_response()
|
|
if tempfile['content_type']:
|
|
response.set_content_type(tempfile['content_type'])
|
|
else:
|
|
response.set_content_type('application/octet-stream')
|
|
if tempfile['charset']:
|
|
response.set_charset(tempfile['charset'])
|
|
|
|
file_pointer = get_session().get_tempfile_content(t).get_file_pointer()
|
|
if Image is not None and get_request().form.get('thumbnail') == '1':
|
|
try:
|
|
image = Image.open(file_pointer)
|
|
image.thumbnail((500, 300))
|
|
image_thumb_fp = StringIO()
|
|
image.save(image_thumb_fp, "PNG")
|
|
except IOError:
|
|
# too bad we couldn't load the image, return the whole file :/
|
|
return file_pointer.read()
|
|
response.set_content_type('image/png')
|
|
return image_thumb_fp.getvalue()
|
|
else:
|
|
return file_pointer.read()
|
|
|
|
def validating(self, data):
|
|
self.html_top(self.formdef.name)
|
|
r = TemplateIO(html=True)
|
|
r += htmltext('<div class="form-validation">')
|
|
r += self.form_side(step_no=1, page_no=None, data=data,
|
|
magictoken=get_request().form['magictoken'])
|
|
r += TextsDirectory.get_html_text('check-before-submit')
|
|
form = self.create_view_form(data)
|
|
token_widget = form.get_widget(form.TOKEN_NAME)
|
|
token_widget._parsed = True
|
|
if self.formdef.has_captcha and not (get_session().get_user() or get_session().won_captcha):
|
|
get_request().form['captcha$q'] = ''
|
|
captcha_text = TextsDirectory.get_html_text('captcha-page')
|
|
if captcha_text:
|
|
form.widgets.append(HtmlWidget(captcha_text))
|
|
form.add_captcha(hint='')
|
|
form.captcha.has_error = lambda request: False
|
|
form.add_submit('submit', _('Submit'))
|
|
form.add_submit('previous', _('Previous'))
|
|
form.add_submit('cancel', _('Cancel'), css_class = 'cancel')
|
|
session = get_session()
|
|
if self.formdef.enable_tracking_codes:
|
|
form.add_submit('savedraft', _('Save Draft'), css_class = 'save-draft',
|
|
attrs={'style': 'display: none'})
|
|
form.add_hidden('step', '2')
|
|
magictoken = get_request().form['magictoken']
|
|
form.add_hidden('magictoken', magictoken)
|
|
r += form.render()
|
|
r += htmltext('</div>')
|
|
return r.getvalue()
|
|
|
|
def error(self, msg):
|
|
self.html_top(self.formdef.name)
|
|
homepage = get_publisher().get_root_url()
|
|
r = TemplateIO(html=True)
|
|
r += htmltext('<div class="errornotice">%s</div>') % msg
|
|
r += htmltext('<a href="%s">%s</a>') % (homepage, _('Back Home'))
|
|
return r.getvalue()
|
|
|
|
def tryauth(self):
|
|
return tryauth(self.formdef.get_url())
|
|
|
|
def auth(self):
|
|
return auth(self.formdef.get_url())
|
|
|
|
def qrcode(self):
|
|
img = qrcode.make(self.formdef.get_url())
|
|
s = StringIO()
|
|
img.save(s)
|
|
if get_request().get_query() == 'download':
|
|
get_response().set_header('content-disposition',
|
|
'attachment; filename=qrcode-%s.png' % self.formdef.url_name)
|
|
get_response().set_content_type('image/png')
|
|
return s.getvalue()
|
|
|
|
def _q_lookup(self, component):
|
|
try:
|
|
filled = self.formdef.data_class().get(component)
|
|
except KeyError:
|
|
raise errors.TraversalError()
|
|
|
|
if not filled.is_draft():
|
|
if get_request().is_in_backoffice():
|
|
get_session().message = ('error', _('This form has already been submitted.'))
|
|
return redirect(get_publisher().get_backoffice_url() + '/submission/')
|
|
return PublicFormStatusPage(self.formdef, filled)
|
|
|
|
if not (get_request().is_in_backoffice() or filled.formdef.enable_tracking_codes):
|
|
# don't allow restoring drafts if drafts are no longer enabled for
|
|
# this form.
|
|
raise errors.AccessForbiddenError()
|
|
|
|
session = get_session()
|
|
if not (get_request().is_in_backoffice() and filled.backoffice_submission):
|
|
if session.user:
|
|
if str(session.user) != str(filled.user_id):
|
|
raise errors.AccessForbiddenError()
|
|
else:
|
|
if not session.is_anonymous_submitter(filled):
|
|
raise errors.AccessUnauthorizedError()
|
|
|
|
if get_request().get_query() == 'remove-draft':
|
|
filled.remove_self()
|
|
return redirect(get_publisher().get_root_url())
|
|
|
|
magictoken = randbytes(8)
|
|
filled.feed_session()
|
|
form_data = filled.data
|
|
# remove keys that were not set, this is required when we restore a
|
|
# draft from SQL (where all columns are always defined).
|
|
for field, value in form_data.items():
|
|
if value is None:
|
|
del form_data[field]
|
|
form_data['is_recalled_draft'] = True
|
|
form_data['draft_formdata_id'] = filled.id
|
|
form_data['page_no'] = filled.page_no
|
|
session.add_magictoken(magictoken, form_data)
|
|
|
|
if get_request().is_in_backoffice():
|
|
return redirect('./?mt=%s' % magictoken)
|
|
else:
|
|
return redirect('%s?mt=%s' % (filled.formdef.get_url(), magictoken))
|
|
|
|
|
|
class RootDirectory(AccessControlled, Directory):
|
|
_q_exports = ['', 'json', 'categories', 'code', 'tryauth', 'auth']
|
|
|
|
category = None
|
|
code = TrackingCodesDirectory()
|
|
|
|
def __init__(self, category = None):
|
|
self.category = category
|
|
get_publisher().substitutions.feed(category)
|
|
|
|
def tryauth(self):
|
|
if self.category:
|
|
base_url = self.category.get_url()
|
|
else:
|
|
base_url = get_publisher().get_root_url()
|
|
return tryauth(base_url)
|
|
|
|
def auth(self):
|
|
if self.category:
|
|
base_url = self.category.get_url()
|
|
else:
|
|
base_url = get_publisher().get_root_url()
|
|
return auth(base_url)
|
|
|
|
def _q_access(self):
|
|
if self.category:
|
|
response = get_response()
|
|
response.breadcrumb.append( ('%s/' % self.category.url_name, self.category.name ) )
|
|
|
|
def get_list_of_forms(self, formdefs, user):
|
|
list_forms = []
|
|
advertised_forms = []
|
|
|
|
for formdef in formdefs:
|
|
if formdef.roles:
|
|
if not user:
|
|
if formdef.always_advertise:
|
|
advertised_forms.append(formdef)
|
|
continue
|
|
if logged_users_role().id not in formdef.roles:
|
|
for q in user.roles or []:
|
|
if q in formdef.roles:
|
|
break
|
|
else:
|
|
if formdef.always_advertise:
|
|
advertised_forms.append(formdef)
|
|
continue
|
|
list_forms.append(formdef)
|
|
|
|
return list_forms, advertised_forms
|
|
|
|
def _q_index(self):
|
|
if get_request().get_header(str('Accept'), '') == 'application/json':
|
|
return self.json()
|
|
|
|
if not self.category:
|
|
redirect_url = get_cfg('misc', {}).get('homepage-redirect-url')
|
|
else:
|
|
redirect_url = self.category.redirect_url
|
|
|
|
if redirect_url:
|
|
return redirect(misc.get_variadic_url(redirect_url,
|
|
get_publisher().substitutions.get_context_variables()))
|
|
|
|
template.html_top(default_org = _('Forms'))
|
|
r = TemplateIO(html=True)
|
|
|
|
session = get_session()
|
|
request = get_request()
|
|
user = request.user
|
|
|
|
if not self.category:
|
|
get_logger().info('home page')
|
|
if user:
|
|
message = TextsDirectory.get_html_text('welcome-logged')
|
|
else:
|
|
message = TextsDirectory.get_html_text('welcome-unlogged')
|
|
|
|
if message:
|
|
r += htmltext('<div id="welcome-message">')
|
|
r += message
|
|
r += htmltext('</div>')
|
|
|
|
if self.category:
|
|
all_formdefs = FormDef.select(
|
|
lambda x: str(x.category_id) == str(self.category.id),
|
|
order_by='name', ignore_errors=True)
|
|
else:
|
|
all_formdefs = FormDef.select(order_by='name', ignore_errors=True)
|
|
|
|
formdefs = [x for x in all_formdefs if (not x.is_disabled() or x.disabled_redirection)]
|
|
|
|
if not self.category and any((x for x in formdefs if x.enable_tracking_codes)):
|
|
r += htmltext('<div id="side">')
|
|
r += htmltext('<div id="tracking-code">')
|
|
r += htmltext('<h3>%s</h3>') % _('Tracking code')
|
|
r += htmltext('<form action="/code/load">')
|
|
r += htmltext('<input size="12" name="code" placeholder="%s"/>') % _('ex: RPQDFVCD')
|
|
r += htmltext('<input type="submit" value="%s"/>') % _('Load')
|
|
r += htmltext('</form>')
|
|
r += htmltext('</div>')
|
|
r += htmltext('</div> <!-- #side -->')
|
|
|
|
list_forms, advertised_forms = self.get_list_of_forms(formdefs, user)
|
|
|
|
if formdefs and not list_forms and not advertised_forms:
|
|
# there is forms, but none can be displayed
|
|
raise errors.AccessUnauthorizedError()
|
|
|
|
user_forms = []
|
|
if user:
|
|
for formdef in all_formdefs:
|
|
user_forms.extend(get_user_forms(formdef))
|
|
user_forms = [x for x in user_forms if x.formdef.is_user_allowed_read(user, x)]
|
|
user_forms.sort(lambda x,y: cmp(x.receipt_time, y.receipt_time))
|
|
|
|
if self.category:
|
|
r += self.form_list(list_forms, category = self.category,
|
|
session = session, user_forms = user_forms,
|
|
advertised_forms = advertised_forms)
|
|
else:
|
|
cats = Category.select()
|
|
Category.sort_by_position(cats)
|
|
one = False
|
|
for c in cats:
|
|
l2 = [x for x in list_forms if str(x.category_id) == str(c.id)]
|
|
l2_advertise = [x for x in advertised_forms if str(x.category_id) == str(c.id)]
|
|
if l2 or l2_advertise:
|
|
r += self.form_list(l2, category = c,
|
|
session = session, user_forms = user_forms,
|
|
advertised_forms = l2_advertise)
|
|
one = True
|
|
|
|
l2 = [x for x in list_forms if not x.category]
|
|
l2_advertise = [x for x in advertised_forms if not x.category]
|
|
if l2 or l2_advertise:
|
|
if one:
|
|
title = _('Misc')
|
|
else:
|
|
title = None
|
|
r += self.form_list(l2, title = title,
|
|
session = session, user_forms = user_forms,
|
|
advertised_forms = l2_advertise)
|
|
|
|
root_url = get_publisher().get_root_url()
|
|
if user:
|
|
r += self.user_forms(user_forms)
|
|
|
|
if not self.category:
|
|
r += htmltext('<p id="logout">')
|
|
if user.can_go_in_backoffice():
|
|
r += htmltext('<a href="%sbackoffice/">%s</a> - ') % (root_url, _('Back Office'))
|
|
if user.anonymous:
|
|
if get_cfg('saml_identities', {}).get('creation', 'admin') != 'admin':
|
|
r += htmltext('<a href="%sregister">%s</a> - ') % (root_url, _('Register'))
|
|
|
|
r += htmltext('<a href="%slogout">%s</a></p>') % (root_url, _('Logout'))
|
|
|
|
elif get_cfg('sp') or get_cfg('identification', {}).get('methods'):
|
|
if not self.category:
|
|
r += htmltext('<p id="login"><a href="%slogin">%s</a>') % (root_url, _('Login'))
|
|
identities_cfg = get_cfg('identities', {})
|
|
if identities_cfg.get('creation') in ('self', 'moderated'):
|
|
r += htmltext(' - <a href="%sregister">%s</a>') % (root_url, _('Register'))
|
|
r += htmltext('</p>')
|
|
return r.getvalue()
|
|
|
|
|
|
def user_forms(self, user_forms):
|
|
r = TemplateIO(html=True)
|
|
draft = [x for x in user_forms if x.is_draft() and not x.formdef.is_disabled()]
|
|
if draft:
|
|
r += htmltext('<h2 id="drafts">%s</h2>') % _('Your Current Drafts')
|
|
r += htmltext('<ul>')
|
|
for f in draft:
|
|
r += htmltext('<li><a href="%s/%s">%s</a>, %s</li>') % (
|
|
f.formdef.url_name, f.id, f.formdef.name,
|
|
misc.localstrftime(f.receipt_time))
|
|
r += htmltext('</ul>')
|
|
|
|
# with workflows
|
|
workflows = [Workflow.get_default_workflow()] + Workflow.select(order_by = 'name')
|
|
for workflow in workflows:
|
|
# XXX: seperate endpoints from non-endpoints
|
|
for status in workflow.possible_status:
|
|
fms = [x for x in user_forms if \
|
|
not x.formdef.private_status_and_history and \
|
|
x.formdef.workflow.id == workflow.id and \
|
|
(x.get_visible_status() == status)]
|
|
if not fms:
|
|
continue
|
|
r += htmltext('<h2>%s</h2>') % _('Your forms with status "%s"') % status.name
|
|
r += htmltext('<ul>')
|
|
for f in fms:
|
|
r += htmltext('<li><a href="%s/%s/">%s</a>, %s</li>') % (
|
|
f.formdef.url_name, f.id, f.formdef.name,
|
|
misc.localstrftime(f.receipt_time))
|
|
r += htmltext('</ul>')
|
|
return r.getvalue()
|
|
|
|
|
|
def form_list(self, list, category = None, title = None,
|
|
session = None, user_forms = None, advertised_forms = []):
|
|
r = TemplateIO(html=True)
|
|
|
|
keywords = {}
|
|
for formdef in list:
|
|
for keyword in formdef.keywords_list:
|
|
keywords[keyword] = True
|
|
|
|
div_attrs = {'class': 'category'}
|
|
if keywords:
|
|
div_attrs['data-keywords'] = ' '.join(keywords)
|
|
|
|
if title:
|
|
div_attrs['id'] = 'category-%s' % misc.simplify(title)
|
|
elif category:
|
|
div_attrs['id'] = 'category-%s' % category.url_name
|
|
else:
|
|
div_attrs['id'] = 'category-misc'
|
|
|
|
r += htmltext('<div %s>' % ' '.join(
|
|
['%s="%s"' % x for x in div_attrs.items()]))
|
|
if title:
|
|
r += htmltext('<h2>%s</h2>') % title
|
|
elif category:
|
|
r += htmltext('<h2>%s</h2>') % category.name
|
|
|
|
formdefs_data = None
|
|
if self.category:
|
|
url_prefix = ''
|
|
r += self.category.get_description_html_text()
|
|
elif category:
|
|
url_prefix = '%s/' % category.url_name
|
|
else:
|
|
url_prefix = ''
|
|
r += htmltext('<ul class="catforms">')
|
|
for formdef in list:
|
|
if formdef.only_allow_one and user_forms:
|
|
if formdefs_data is None:
|
|
formdefs_data = [x.formdef.id for x in user_forms
|
|
if x.formdef.only_allow_one and not x.is_draft()]
|
|
r += htmltext('<li data-keywords="%s">') % ' '.join(formdef.keywords_list)
|
|
if formdefs_data and formdef.id in formdefs_data:
|
|
# form has already been completed
|
|
r += htmltext('%s (%s, <a href="%s%s/">%s</a>)') % (
|
|
formdef.name, _('already completed'),
|
|
url_prefix, formdef.url_name, _('review'))
|
|
else:
|
|
classes = []
|
|
if formdef.is_disabled() and formdef.disabled_redirection:
|
|
classes.append('redirection')
|
|
r += htmltext('<a class="%s" href="%s%s/">%s</a>') % (
|
|
' '.join(classes), url_prefix, formdef.url_name, formdef.name)
|
|
|
|
if formdef.description:
|
|
r += htmltext('<div class="description">%s</div>' % formdef.description)
|
|
r += htmltext('</li>')
|
|
|
|
for formdef in advertised_forms:
|
|
r += htmltext('<li class="required-authentication" data-keywords="%s">'
|
|
) % ' '.join(formdef.keywords_list)
|
|
r += htmltext('<a href="%s%s/">%s</a>') % (url_prefix, formdef.url_name, formdef.name)
|
|
r += htmltext('<span> (%s)</span>') % _('authentication required')
|
|
if formdef.description:
|
|
r += htmltext('<div class="description">%s</div>' % formdef.description)
|
|
r += htmltext('</li>')
|
|
r += htmltext('</ul>')
|
|
r += htmltext('</div>')
|
|
return r.getvalue()
|
|
|
|
def json(self):
|
|
# backward compatibility
|
|
from wcs.api import ApiFormdefsDirectory
|
|
return ApiFormdefsDirectory(self.category)._q_index()
|
|
|
|
def get_context(self):
|
|
from wcs.api import is_url_signed, get_user_from_api_query_string
|
|
user = get_user_from_api_query_string() or get_request().user
|
|
list_all_forms = (user and user.is_admin) or (is_url_signed() and user is None)
|
|
|
|
list_forms = []
|
|
|
|
if self.category:
|
|
formdefs = FormDef.select(lambda x: (
|
|
str(x.category_id) == str(self.category.id) and (
|
|
not x.is_disabled() or x.disabled_redirection)),
|
|
order_by = 'name')
|
|
else:
|
|
formdefs = FormDef.select(lambda x: not x.is_disabled() or x.disabled_redirection,
|
|
order_by='name',
|
|
ignore_errors=True)
|
|
|
|
charset = get_publisher().site_charset
|
|
|
|
for formdef in formdefs:
|
|
authentication_required = False
|
|
if formdef.roles and not list_all_forms:
|
|
if not user:
|
|
if not formdef.always_advertise:
|
|
continue
|
|
authentication_required = True
|
|
elif logged_users_role().id not in formdef.roles:
|
|
for q in user.roles or []:
|
|
if q in formdef.roles:
|
|
break
|
|
else:
|
|
if not formdef.always_advertise:
|
|
continue
|
|
authentication_required = True
|
|
|
|
formdict = {'title': unicode(formdef.name, charset),
|
|
'slug': formdef.url_name,
|
|
'url': formdef.get_url(),
|
|
'authentication_required': authentication_required}
|
|
|
|
formdict['redirection'] = bool(formdef.is_disabled() and
|
|
formdef.disabled_redirection)
|
|
|
|
# we include the count of submitted forms so it's possible to sort
|
|
# them by popularity
|
|
formdict['count'] = formdef.data_class().count()
|
|
|
|
if formdef.category:
|
|
formdict['category'] = unicode(formdef.category.name, charset)
|
|
formdict['category_position'] = (formdef.category.position or 0)
|
|
else:
|
|
formdict['category_position'] = sys.maxint
|
|
|
|
list_forms.append(formdict)
|
|
|
|
list_forms.sort(lambda x, y: cmp(x['category_position'], y['category_position']))
|
|
for formdict in list_forms:
|
|
del formdict['category_position']
|
|
|
|
return list_forms
|
|
|
|
def get_categories(self, user):
|
|
result = []
|
|
formdefs = FormDef.select(
|
|
lambda x: not x.is_disabled() or x.disabled_redirection,
|
|
order_by='name', ignore_errors=True)
|
|
list_forms, advertised_forms = self.get_list_of_forms(formdefs, user)
|
|
list_forms = list_forms + advertised_forms
|
|
cats = Category.select()
|
|
Category.sort_by_position(cats)
|
|
for c in cats:
|
|
if [x for x in list_forms if str(x.category_id) == str(c.id)]:
|
|
result.append(c)
|
|
return result
|
|
|
|
def categories(self):
|
|
if self.category:
|
|
raise errors.TraversalError()
|
|
if get_request().is_json():
|
|
return self.categories_json()
|
|
template.html_top(_('Categories'))
|
|
r = TemplateIO(html=True)
|
|
user = get_request().user
|
|
for category in self.get_categories(user):
|
|
r += htmltext('<h2>%s</h2>') % category.name
|
|
r += category.get_description_html_text()
|
|
r += htmltext('<p><a href="%s/">%s</a></p>') % (category.url_name, _('All forms'))
|
|
return r.getvalue()
|
|
|
|
def categories_json(self):
|
|
# backward compatibility
|
|
from wcs.api import ApiCategoriesDirectory
|
|
return ApiCategoriesDirectory()._q_index()
|
|
|
|
def _q_lookup(self, component):
|
|
return FormPage(component)
|
|
|
|
|
|
class PublicFormStatusPage(FormStatusPage):
|
|
_q_exports_orig = ['', 'download', 'status']
|
|
form_page_class = FormPage
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
FormStatusPage.__init__(self, *args, **kwargs)
|
|
if self.filled.anonymised:
|
|
if get_session() and get_session().is_anonymous_submitter(self.filled):
|
|
return
|
|
raise errors.TraversalError()
|
|
|
|
def status(self):
|
|
return redirect('%sbackoffice/%s/%s/' % (
|
|
get_publisher().get_root_url(),
|
|
self.formdef.url_name,
|
|
str(self.filled.id)))
|
|
|
|
def form_status_buttons(self):
|
|
if not get_response().iframe_mode:
|
|
r = TemplateIO(html=True)
|
|
r += htmltext('<div class="back-home-button">')
|
|
r += htmltext('<a href="%s">%s</a>') % (get_publisher().get_root_url(), _('Back Home'))
|
|
r += htmltext('</div>')
|
|
return r.getvalue()
|
|
|
|
|
|
TextsDirectory.register('welcome-logged',
|
|
N_('Welcome text on home page for logged users'))
|
|
|
|
TextsDirectory.register('welcome-unlogged',
|
|
N_('Welcome text on home page for unlogged users'))
|
|
|
|
TextsDirectory.register('captcha-page',
|
|
N_('Explanation text before the CAPTCHA'),
|
|
default = N_('''<h3>Verification</h3>
|
|
|
|
<p>
|
|
In order to submit the form you need to complete this simple question.
|
|
</p>'''))
|
|
|
|
|
|
TextsDirectory.register('form-recorded',
|
|
N_('Message when a form has been recorded'),
|
|
hint = N_('Available variables: date, number'),
|
|
category = N_('Forms'),
|
|
default = N_('The form has been recorded on [date] with the number [number].'))
|
|
|
|
TextsDirectory.register('form-recorded-allow-one',
|
|
N_('Message when a form has been recorded, and the form is set to only allow one per user'),
|
|
hint = N_('Available variable: date'),
|
|
category = N_('Forms'),
|
|
default = N_('The form has been recorded on [date].'))
|
|
|
|
TextsDirectory.register('check-before-submit',
|
|
N_('Message when a form is displayed before validation'),
|
|
category = N_('Forms'),
|
|
default = N_('Check values then click submit.'))
|
|
|
|
TextsDirectory.register('tracking-code-email-dialog',
|
|
N_('Message in tracking code popup dialog'),
|
|
category = N_('Forms'),
|
|
default = N_('You can get a reminder of the tracking code by email.'))
|
|
|
|
TextsDirectory.register('tracking-code-short-text',
|
|
N_('Short text in the tracking code box'),
|
|
category=N_('Forms'))
|
|
|
|
EmailsDirectory.register('tracking-code-reminder',
|
|
N_('Tracking Code'),
|
|
N_('Available variables: email, form_name, form_tracking_code'),
|
|
category = N_('Miscellaneous'),
|
|
default_subject = N_('Tracking Code reminder'),
|
|
default_body = N_('''\
|
|
Hello,
|
|
|
|
As a reminder your tracking code for [form_name] is [form_tracking_code].
|
|
|
|
'''))
|