wcs/wcs/forms/root.py

1410 lines
54 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 sys
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, 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.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 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, EditableWorkflowStatusItem
from qommon.admin.texts import TextsDirectory
from backoffice import FormDefUI
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.extend(formdef.data_class().get_with_indexed_value('user_id', user.id))
try:
user_forms.extend(formdef.data_class().get_with_indexed_value('user_hash', user.hash))
except AttributeError:
pass
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)
except KeyError:
raise errors.TraversalError()
formdata = tracking_code.formdata
if formdata.formdef.enable_tracking_codes is False:
raise errors.TraversalError()
if not get_session().user:
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 = ['', 'listing', 'tempfile', 'tokens', 'schema', 'tryauth',
'auth', 'qrcode', 'autosave', 'code']
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.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 step(self, step_no, page_no = 0, log_detail = None, data = None, editing = 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 = 0
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.formdef.confirmation and not editing:
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 initial_captcha_page(self):
form = Form()
form.add_captcha(hint='')
form.add_submit('submit', _('Next'))
form.add_submit('cancel', _('Cancel'))
if form.get_submit() == 'cancel':
return redirect(get_publisher().get_root_url())
if not form.is_submitted() or form.has_errors():
html_top(self.formdef.name)
r = TemplateIO(html=True)
r += TextsDirectory.get_html_text('captcha-page')
r += form.render()
return r.getvalue()
return self.page(0)
def page(self, page_no, page_change = True, log_detail = None, editing = None):
r = TemplateIO(html=True)
displayed_fields = []
session = get_session()
if page_no == 0 and self.formdef.has_captcha:
if not session.get_user() and not session.won_captcha:
return self.initial_captcha_page()
if page_no > 0:
magictoken = get_request().form['magictoken']
self.feed_current_data(magictoken)
form = self.formdef.create_form(page_no, displayed_fields)
if editing:
form.action = 'wfedit'
else:
form.action = '.'
# 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 form_data.get('draft_formdata_id'):
form.attrs['data-has-draft'] = 'yes'
else:
form_data = {}
if page_no == 0 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 == 0 and get_request().form.has_key('cancelurl'):
cancelurl = get_request().form['cancelurl']
session.add_magictoken(magictoken, {'__cancelurl': cancelurl})
if editing and page_no == self.page_number - 1:
form.add_submit('submit', _('Save Changes'))
elif not self.formdef.confirmation and page_no == self.page_number - 1:
form.add_submit('submit', _('Submit'))
else:
form.add_submit('submit', _('Next'))
if page_no > 0:
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
if data.has_key(k):
v = data[k]
elif field.prefill:
v = field.get_prefill_value()
if get_request().is_in_backoffice() and (
field.prefill and field.prefill.get('type') in ('user', 'geoloc')):
# turn off prefilling from user and geolocation
# attributes if the form is filled from the backoffice
v = None
if v:
prefilled = True
form.get_widget('f%s' % k).set_message(
_('Value has been automatically prefilled.'))
form.get_widget('f%s' % k).prefilled = True
if field.prefill:
# 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 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)
req.form['f%s' % k] = v
one = True
if not one:
req.form = {}
self.html_top(self.formdef.name)
r += self.form_side(0, page_no, log_detail=log_detail, data=data, editing=editing)
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 editing:
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=0, log_detail=None, data=None, editing=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:
r += self.tracking_code_box(data)
r += self.step(step_no, page_no, log_detail, data=data, editing=editing)
r += htmltext('</div> <!-- #side -->')
return r.getvalue()
def tracking_code_box(self, data):
'''Create the tracking code box, it displays the current tracking code
or a 'save' button if it has not yet been created.'''
r = TemplateIO(html=True)
draft_formdata_id = data.get('draft_formdata_id')
r += htmltext('<div id="tracking-code">')
r += htmltext('<h3>%s</h3>') % _('Tracking code')
if draft_formdata_id:
formdata = self.formdef.data_class().get(draft_formdata_id)
if formdata.tracking_code:
get_response().add_javascript(['jquery.js', 'jquery-ui.js', 'popup.js'])
r += htmltext('<a rel="popup" href="%s">%s</a>') % (
'code/%s/' % formdata.tracking_code,
formdata.tracking_code)
else:
r += htmltext('<button>%s</button>') % _('Save')
r += TextsDirectory.get_html_text('tracking-code-short-text')
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, {})
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 _q_index(self, log_detail = None, editing = None):
self.check_role()
if self.check_disabled():
return redirect(self.check_disabled())
session = get_session()
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') != '0' 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())
existing_formdata = None
if editing:
existing_formdata = editing.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
editing.feed_session()
token = randbytes(8)
get_request().form['magictoken'] = token
session.add_magictoken(token, editing.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 = 0
return self.page(page_no, True)
return self.page(0, editing = editing)
if form.get_submit() == 'cancel':
get_logger().info('form %s - cancel' % (self.formdef.name))
if editing:
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
return redirect(get_publisher().get_root_url())
try:
step = int(form.get_widget('step').parse())
except TypeError:
step = 0
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.formdef.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 > 0 and form.get_submit() == 'previous':
return self.previous_page(page_no, magictoken, editing = editing)
if self.formdef.enable_tracking_codes and form.get_submit() == 'removedraft':
self.remove_draft()
return redirect(get_publisher().get_root_url())
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('/'))
# 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 = 0
return self.page(page_no, page_change = False, editing = editing)
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
if self.formdef.enable_tracking_codes:
draft_id = session.get_by_magictoken(magictoken, {}).get('draft_formdata_id')
if draft_id:
# if there's a draft, update it with current data
self.autosave_draft(draft_id, page_no, form_data)
else:
# if there's no draft yet, 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 editing:
form = self.formdef.create_view_form(form_data, use_tokens = False)
return self.submitted_existing(form, editing)
if self.formdef.confirmation:
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, editing = editing)
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, editing = editing)
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, editing = editing)
if self.formdef.enable_tracking_codes and form.get_submit() == 'removedraft':
self.remove_draft()
return redirect(get_publisher().get_root_url())
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.formdef.create_view_form(form_data,
use_tokens = self.formdef.confirmation)
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, editing = None):
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, editing = editing)
def remove_draft(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'))
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':
return
formdata.page_no = page_no
formdata.data = form_data
formdata.receipt_time = time.localtime()
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):
return json.dumps({'result': 'error', 'reason': reason})
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')
draft_id = session.get_by_magictoken(magictoken, {}).get('draft_formdata_id')
if not draft_id:
return result_error('missing draft id')
try:
formdata = self.formdef.data_class().get(draft_id)
except KeyError:
return result_error('missing formdata')
if not formdata.is_draft():
return result_error('formdata has already been saved')
formdata.page_no = page_no
form = self.formdef.create_form(page_no)
formdata.data.update(self.formdef.get_data(form))
formdata.receipt_time = time.localtime()
session = get_session()
if session and session.user and not str(session.user).startswith('anonymous-'):
formdata.user_id = session.user
formdata.store()
return json.dumps({'result': 'success'})
def save_draft(self, data, page_no):
filled = self.formdef.data_class()()
filled.data = data
filled.status = 'draft'
filled.page_no = page_no
filled.receipt_time = time.localtime()
session = get_session()
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)
self.keep_tracking_code(filled)
get_logger().info('form %s - saving draft (id: %s)' % (self.formdef.name, filled.id))
return filled
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.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-'):
try:
filled.user_hash = get_request().user.hash
except AttributeError:
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.keep_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 keep_tracking_code(self, formdata):
'''Remove current draft in favour of formdata, conserving the same
tracking code.'''
code = None
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:
old_draft_formdata = self.formdef.data_class().get(draft_formdata_id)
old_draft_formdata.remove_self()
if old_draft_formdata.tracking_code:
code = get_publisher().tracking_code_class.get(old_draft_formdata.tracking_code)
form_data['draft_formdata_id'] = formdata.id
if code is None:
code = get_publisher().tracking_code_class()
code.formdata = formdata # this will .store() the code
def submitted_existing(self, form, editing):
old_data = editing.data
editing.data = self.formdef.get_data(form)
editing.store()
return redirect('.')
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':
image = Image.open(file_pointer)
image.thumbnail((500, 300))
image_thumb_fp = StringIO()
image.save(image_thumb_fp, "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, data=data)
r += TextsDirectory.get_html_text('check-before-submit')
form = self.formdef.create_view_form(data)
token_widget = form.get_widget(form.TOKEN_NAME)
token_widget._parsed = True
form.add_submit('previous', _('Previous'))
form.add_submit('submit', _('Submit'))
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 listing(self):
if not self.formdef.is_user_allowed_read(get_request().user):
raise errors.AccessForbiddenError()
get_response().breadcrumb.append( ('listing', _('Listing')) )
self.html_top('%s - %s' % (_('Listing'), self.formdef.name))
r = TemplateIO(html=True)
fields = []
for field in self.formdef.fields:
if hasattr(field, str('get_view_value')) and field.in_listing:
fields.append(field)
r += FormDefUI(self.formdef).listing(fields, 'all', include_form=True)
r += htmltext('<p><a href="..">%s</a></p>') % _('Back')
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():
return PublicFormStatusPage(self.formdef, filled)
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
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:
formdefs = FormDef.select(
lambda x: str(x.category_id) == str(self.category.id),
order_by='name', ignore_errors=True)
else:
formdefs = FormDef.select(order_by='name', ignore_errors=True)
formdefs = [x for x in 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 list_forms:
if not formdef.is_disabled():
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()]
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)
if title:
r += htmltext('<div class="category" id="category-%s">') % misc.simplify(title)
r += htmltext('<h2>%s</h2>') % title
elif category:
r += htmltext('<div class="category" id="category-%s">') % category.url_name
r += htmltext('<h2>%s</h2>') % category.name
else:
r += htmltext('<div class="category" id="category-misc">')
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()]
if formdefs_data and formdef.id in formdefs_data:
# form has already been completed
r += htmltext('<li>%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('<li><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">')
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_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 = ['', 'download', 'status', 'wfedit']
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()
def wfedit(self):
wf_status = self.filled.get_status()
for item in wf_status.items:
if not isinstance(item, EditableWorkflowStatusItem):
continue
if item.check_auth(self.filled, get_request().user):
f = FormPage(self.formdef.url_name)
get_response().breadcrumb = get_response().breadcrumb[:-1]
get_response().breadcrumb.append( ('wfedit', _('Edit')) )
return f._q_index(editing = self.filled)
raise errors.AccessForbiddenError()
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 on the CAPTCHA page'),
default = N_('''<h3>Verification</h3>
<p>
In order to proceed 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].
'''))