1063 lines
44 KiB
Python
1063 lines
44 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 datetime
|
|
import json
|
|
import os
|
|
import time
|
|
import urllib.parse
|
|
|
|
from django.utils.timezone import is_naive, make_aware
|
|
from quixote import get_publisher, get_request, get_response, get_session, redirect
|
|
from quixote.directory import Directory
|
|
from quixote.errors import RequestError
|
|
from quixote.html import TemplateIO, htmltext
|
|
from quixote.util import randbytes
|
|
|
|
from wcs import data_sources
|
|
from wcs.api_utils import get_query_flag, get_user_from_api_query_string, is_url_signed, sign_url_auto_orig
|
|
from wcs.blocks import BlockSubWidget, BlockWidget
|
|
from wcs.qommon.admin.texts import TextsDirectory
|
|
from wcs.wf.editable import EditableWorkflowStatusItem
|
|
from wcs.workflows import RedisplayFormException
|
|
|
|
from ..qommon import _, audit, errors, misc, template
|
|
|
|
|
|
class FileDirectory(Directory):
|
|
_q_exports = []
|
|
_lookup_methods = ['lookup_file_field']
|
|
|
|
def __init__(self, formdata, reference, thumbnails=False):
|
|
self.formdata = formdata
|
|
self.reference = reference
|
|
self.thumbnails = thumbnails
|
|
|
|
def lookup_file_field(self, filename):
|
|
try:
|
|
if '$' in self.reference:
|
|
# path to block field contents
|
|
fn2, idx, sub = self.reference.split('$', 2)
|
|
return self.formdata.data[fn2]['data'][int(idx)][sub]
|
|
else:
|
|
return self.formdata.data[self.reference]
|
|
except (KeyError, ValueError):
|
|
return None
|
|
|
|
def _q_lookup(self, component):
|
|
if component == 'thumbnail':
|
|
self.thumbnails = True
|
|
return self
|
|
for lookup_method_name in self._lookup_methods:
|
|
lookup_method = getattr(self, lookup_method_name)
|
|
file = lookup_method(filename=component)
|
|
if file:
|
|
break
|
|
else:
|
|
# no such file
|
|
raise errors.TraversalError()
|
|
|
|
if component and component not in (file.base_filename, urllib.parse.quote(file.base_filename)):
|
|
raise errors.TraversalError()
|
|
|
|
if file.has_redirect_url():
|
|
redirect_url = file.get_redirect_url(backoffice=get_request().is_in_backoffice())
|
|
if not redirect_url:
|
|
raise errors.TraversalError()
|
|
redirect_url = sign_url_auto_orig(redirect_url)
|
|
return redirect(redirect_url)
|
|
|
|
if not self.thumbnails:
|
|
# do not log access to thumbnails as they will already be accounted for as
|
|
# a view of the formdata/carddata containing them.
|
|
audit('download file', obj=self.formdata, extra_label=component)
|
|
return self.serve_file(file, thumbnail=self.thumbnails)
|
|
|
|
@classmethod
|
|
def serve_file(cls, file, thumbnail=False):
|
|
response = get_response()
|
|
|
|
if thumbnail:
|
|
if file.can_thumbnail():
|
|
try:
|
|
content = misc.get_thumbnail(file.get_fs_filename(), content_type=file.content_type)
|
|
response.set_content_type('image/png')
|
|
return content
|
|
except misc.ThumbnailError:
|
|
raise errors.TraversalError()
|
|
else:
|
|
raise errors.TraversalError()
|
|
|
|
# force potential HTML upload to be used as-is (not decorated with theme)
|
|
# and with minimal permissions
|
|
response.filter = {}
|
|
response.set_header(
|
|
'Content-Security-Policy',
|
|
'default-src \'none\'; img-src %s;' % get_request().build_absolute_uri(),
|
|
)
|
|
|
|
if file.content_type:
|
|
response.set_content_type(file.content_type)
|
|
else:
|
|
response.set_content_type('application/octet-stream')
|
|
if file.charset:
|
|
response.set_charset(file.charset)
|
|
if file.base_filename:
|
|
# remove invalid characters from filename
|
|
filename = file.base_filename.translate(str.maketrans({x: '_' for x in '"\n\r'}))
|
|
content_disposition = 'attachment'
|
|
if file.content_type.startswith('image/') and not file.content_type.startswith('image/svg'):
|
|
content_disposition = 'inline'
|
|
elif file.content_type == 'application/pdf':
|
|
content_disposition = 'inline'
|
|
response.set_header('content-disposition', '%s; filename="%s"' % (content_disposition, filename))
|
|
|
|
return file.get_file_pointer().read()
|
|
|
|
|
|
class FilesDirectory(Directory):
|
|
def __init__(self, formdata):
|
|
self.formdata = formdata
|
|
|
|
def _q_lookup(self, component):
|
|
return FileDirectory(self.formdata, reference=component)
|
|
|
|
|
|
class FormTemplateMixin:
|
|
def get_formdef_template_variants(self, template_names):
|
|
template_part_names = [(os.path.dirname(x), os.path.basename(x)) for x in template_names]
|
|
for dirname, basename in template_part_names:
|
|
for keyword in self.formdef.appearance_keywords_list:
|
|
yield os.path.join(dirname, 'appearance-' + keyword, basename)
|
|
if self.formdef.category_id:
|
|
yield os.path.join(dirname, 'category-' + self.formdef.category.url_name, basename)
|
|
yield os.path.join(dirname, basename)
|
|
|
|
|
|
class FormStatusPage(Directory, FormTemplateMixin):
|
|
_q_exports_orig = ['', 'download', 'json', 'action', 'live', 'tempfile']
|
|
_q_extra_exports = []
|
|
form_page_class = None
|
|
|
|
do_not_call_in_templates = True
|
|
summary_templates = ['wcs/formdata_summary.html']
|
|
history_templates = ['wcs/formdata_history.html']
|
|
status_templates = ['wcs/formdata_status.html']
|
|
|
|
def html_top(self, title=None):
|
|
template.html_top(title=title, default_org=_('Forms'))
|
|
|
|
def __init__(self, formdef, filled, register_workflow_subdirs=True, custom_view=None, parent_view=None):
|
|
get_publisher().substitutions.feed(filled)
|
|
self.formdef = formdef
|
|
self.formdata = filled
|
|
self.filled = filled
|
|
self.custom_view = custom_view
|
|
self.parent_view = parent_view
|
|
self._q_exports = self._q_exports_orig[:]
|
|
for q in self._q_extra_exports:
|
|
if q not in self._q_exports:
|
|
self._q_exports.append(q)
|
|
|
|
if self.formdata and self.formdef.workflow and register_workflow_subdirs:
|
|
for name, directory in self.formdef.workflow.get_subdirectories(self.filled):
|
|
self._q_exports.append(name)
|
|
setattr(self, name, directory)
|
|
|
|
def check_auth(self, api_call=False):
|
|
session = get_session()
|
|
mine = False
|
|
if api_call:
|
|
user = get_user_from_api_query_string() or get_request().user
|
|
if get_request().has_anonymised_data_api_restriction() and (not user or not user.is_api_user):
|
|
if is_url_signed() or (get_request().user and get_request().user.is_admin):
|
|
return None
|
|
else:
|
|
raise errors.AccessUnauthorizedError()
|
|
else:
|
|
user = get_request().user
|
|
if user and not user.anonymous:
|
|
if self.filled.is_submitter(user):
|
|
mine = True
|
|
else:
|
|
if session and session.is_anonymous_submitter(self.filled):
|
|
mine = True
|
|
|
|
self.check_receiver()
|
|
return mine
|
|
|
|
def json(self):
|
|
self.check_auth(api_call=True)
|
|
anonymise = get_request().has_anonymised_data_api_restriction()
|
|
|
|
if self.custom_view:
|
|
# call to management view to get list of possible ids,
|
|
# and check this one is allowed
|
|
from wcs.backoffice.management import FormDefUI, FormPage
|
|
|
|
listing_page = FormPage(formdef=self.formdef, view=self.custom_view)
|
|
selected_filter = listing_page.get_filter_from_query(default='all')
|
|
criterias = listing_page.get_criterias_from_query()
|
|
user = get_user_from_api_query_string() or get_request().user if not anonymise else None
|
|
item_ids = FormDefUI(self.formdef).get_listing_item_ids(
|
|
selected_filter, criterias=criterias, user=user
|
|
)
|
|
if str(self.filled.id) not in [str(x) for x in item_ids]:
|
|
raise errors.TraversalError(_('ID not available in filtered view'))
|
|
|
|
values_at = get_request().form.get('at')
|
|
if values_at:
|
|
try:
|
|
values_at = datetime.datetime.fromisoformat(values_at)
|
|
if is_naive(values_at):
|
|
values_at = make_aware(values_at)
|
|
except ValueError:
|
|
raise RequestError('Invalid value "%s" for "at"' % values_at)
|
|
return self.export_to_json(
|
|
anonymise=anonymise,
|
|
include_evolution=get_query_flag('include-evolution', default=True),
|
|
include_files=get_query_flag('include-files-content', default=True),
|
|
include_roles=get_query_flag('include-roles', default=True),
|
|
include_submission=get_query_flag('include-submission', default=True),
|
|
include_fields=get_query_flag('include-fields', default=True),
|
|
include_unnamed_fields=False,
|
|
include_workflow=get_query_flag('include-workflow', default=True),
|
|
include_workflow_data=get_query_flag('include-workflow-data', default=True),
|
|
values_at=values_at,
|
|
)
|
|
|
|
def tempfile(self):
|
|
# allow for file uploaded via a file widget in a workflow form
|
|
# to be downloaded back from widget
|
|
return self.parent_view.tempfile()
|
|
|
|
def workflow_messages(self, position='top'):
|
|
if self.formdef.workflow:
|
|
workflow_messages = self.filled.get_workflow_messages(position=position, user=get_request().user)
|
|
if workflow_messages:
|
|
r = TemplateIO(html=True)
|
|
if position == 'top':
|
|
r += htmltext('<div id="receipt-intro" class="workflow-messages %s">' % position)
|
|
else:
|
|
r += htmltext('<div class="workflow-messages %s">' % position)
|
|
for workflow_message in workflow_messages:
|
|
r += htmltext(workflow_message)
|
|
r += htmltext('</div>')
|
|
return r.getvalue()
|
|
return ''
|
|
|
|
def actions_workflow_messages(self):
|
|
return self.workflow_messages(position='actions')
|
|
|
|
def bottom_workflow_messages(self):
|
|
return self.workflow_messages(position='bottom')
|
|
|
|
def recorded_message(self):
|
|
r = TemplateIO(html=True)
|
|
# behaviour if workflow doesn't display any message
|
|
if self.filled.receipt_time is not None:
|
|
tm = misc.localstrftime(self.filled.receipt_time)
|
|
else:
|
|
tm = '???'
|
|
|
|
if self.formdef.only_allow_one:
|
|
r += TextsDirectory.get_html_text('form-recorded-allow-one', vars={'date': tm})
|
|
else:
|
|
r += TextsDirectory.get_html_text(
|
|
'form-recorded', vars={'date': tm, 'number': self.filled.get_display_id()}
|
|
)
|
|
|
|
return r.getvalue()
|
|
|
|
def get_handling_role_info_text(self):
|
|
handling_role = self.filled.get_handling_role()
|
|
if not (handling_role and handling_role.details):
|
|
return ''
|
|
r = TemplateIO(html=True)
|
|
endpoint_status = self.formdef.workflow.get_endpoint_status()
|
|
r += htmltext('<p>')
|
|
if self.filled.status in ['wf-%s' % x.id for x in endpoint_status]:
|
|
r += str(_('Your case has been handled by:'))
|
|
else:
|
|
r += str(_('Your case is handled by:'))
|
|
r += htmltext('</p>')
|
|
r += htmltext('<p id="receiver">')
|
|
r += htmltext(handling_role.details.replace('\n', '<br />'))
|
|
r += htmltext('</p>')
|
|
return r.getvalue()
|
|
|
|
def _q_index(self):
|
|
if (
|
|
self.formdef.category_id
|
|
and self.parent_view
|
|
and self.parent_view.ensure_parent_category_in_url
|
|
and get_request().get_method() == 'GET'
|
|
and not self.parent_view.parent_category
|
|
):
|
|
url = self.filled.get_url(include_category=True)
|
|
if get_request().get_query():
|
|
url += '?' + get_request().get_query()
|
|
return redirect(url)
|
|
|
|
if self.filled.is_draft():
|
|
return self.restore_draft()
|
|
|
|
mine = self.check_auth()
|
|
if not mine and not get_request().is_in_backoffice():
|
|
# Access authorized but the form doesn't belong to the user; if the
|
|
# user has access to the backoffice, redirect.
|
|
# Unless ?debug=whatever is set.
|
|
if get_request().user.can_go_in_backoffice() and not get_request().form.get('debug'):
|
|
return redirect(self.filled.get_url(backoffice=True))
|
|
|
|
get_request().view_name = 'status'
|
|
|
|
user = get_request().user
|
|
form = self.get_workflow_form(user)
|
|
try:
|
|
response = self.check_submitted_form(form)
|
|
except RedisplayFormException:
|
|
# don't display errors after "add block" button has been clicked.
|
|
form.clear_errors()
|
|
else:
|
|
if response:
|
|
return response
|
|
|
|
if form:
|
|
form.add_media()
|
|
|
|
get_response().add_javascript(['jquery.js', 'qommon.forms.js', 'qommon.map.js'])
|
|
self.html_top(get_publisher().translate(self.formdef.name))
|
|
get_response().filter['page_title'] = self.filled.get_display_label()
|
|
context = {
|
|
'view': self,
|
|
'mine': mine,
|
|
'formdata': self.filled,
|
|
'workflow_form': form,
|
|
}
|
|
|
|
return template.QommonTemplateResponse(
|
|
templates=list(self.get_formdef_template_variants(self.status_templates)), context=context
|
|
)
|
|
|
|
def get_restore_draft_magictoken(self):
|
|
# restore draft into session
|
|
session = get_session()
|
|
filled = self.filled
|
|
if not (get_request().is_in_backoffice() and filled.backoffice_submission):
|
|
if session.is_anonymous_submitter(filled):
|
|
pass
|
|
elif session.user:
|
|
if str(session.user) != str(filled.user_id):
|
|
raise errors.AccessUnauthorizedError()
|
|
else:
|
|
raise errors.AccessUnauthorizedError()
|
|
|
|
magictoken = randbytes(8)
|
|
filled.feed_session()
|
|
form_data = filled.data
|
|
for field in filled.formdef.fields:
|
|
if field.id not in form_data:
|
|
continue
|
|
if form_data[field.id] is None:
|
|
# remove keys that were not set, this is required when we restore a
|
|
# draft from SQL (where all columns are always defined).
|
|
del form_data[field.id]
|
|
continue
|
|
if field.type == 'file':
|
|
# add back file to session
|
|
tempfile = session.add_tempfile(form_data[field.id], storage=field.storage)
|
|
form_data[field.id].token = tempfile['token']
|
|
form_data['prefilling_data'] = filled.prefilling_data or {}
|
|
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)
|
|
|
|
# restore computed fields data
|
|
computed_data = {}
|
|
for field in self.formdef.fields:
|
|
if field.key != 'computed':
|
|
continue
|
|
if field.id in form_data:
|
|
computed_data[field.id] = form_data[field.id]
|
|
if computed_data:
|
|
session.add_magictoken('%s-computed' % magictoken, computed_data)
|
|
return magictoken
|
|
|
|
def restore_draft(self):
|
|
# redirect to draft
|
|
magictoken = self.get_restore_draft_magictoken()
|
|
return redirect('../?mt=%s' % magictoken)
|
|
|
|
def get_workflow_form(self, user):
|
|
submitted_fields = []
|
|
form = self.filled.get_workflow_form(user, displayed_fields=submitted_fields)
|
|
if form and form.is_submitted():
|
|
with get_publisher().substitutions.temporary_feed(self.filled, force_mode='lazy'):
|
|
# remove fields that could be required but are not visible
|
|
self.filled.evaluate_live_workflow_form(user, form)
|
|
get_publisher().substitutions.invalidate_cache()
|
|
get_publisher().substitutions.feed(self.filled)
|
|
# recreate form to get live data source items
|
|
form = self.filled.get_workflow_form(user, displayed_fields=submitted_fields)
|
|
if form:
|
|
for field in submitted_fields:
|
|
if (
|
|
not field.is_visible(self.filled.data, self.formdef)
|
|
and 'f%s' % field.id in form._names
|
|
):
|
|
del form._names['f%s' % field.id]
|
|
|
|
if form:
|
|
form.attrs['data-live-url'] = self.filled.get_url() + 'live'
|
|
return form
|
|
|
|
def check_submitted_form(self, form):
|
|
if form and form.is_submitted():
|
|
submit_button_name = form.get_submit()
|
|
if submit_button_name:
|
|
submit_button = form.get_widget(submit_button_name)
|
|
if getattr(submit_button, 'ignore_form_errors', False):
|
|
form.clear_errors()
|
|
if form.has_errors():
|
|
return
|
|
url = self.submit(form)
|
|
if url is None:
|
|
url = get_request().get_frontoffice_url()
|
|
status = self.filled.get_status()
|
|
top_alert = False
|
|
for item in status.items or []:
|
|
if (
|
|
item.key == 'displaymsg'
|
|
and item.position == 'top'
|
|
and self.filled.is_for_current_user(item.to)
|
|
):
|
|
top_alert = True
|
|
break
|
|
if top_alert:
|
|
# prevent an existing anchor client side to take effect
|
|
url += '#'
|
|
else:
|
|
url += '#action-zone'
|
|
response = get_response()
|
|
response.set_status(303)
|
|
response.headers['location'] = url
|
|
response.content_type = 'text/plain'
|
|
return "Your browser should redirect you"
|
|
|
|
def export_to_json(
|
|
self,
|
|
anonymise=False,
|
|
include_evolution=True,
|
|
include_files=True,
|
|
include_roles=True,
|
|
include_submission=True,
|
|
include_fields=True,
|
|
include_unnamed_fields=False,
|
|
include_workflow=True,
|
|
include_workflow_data=True,
|
|
values_at=None,
|
|
):
|
|
# noqa pylint: disable=too-many-arguments
|
|
get_response().set_content_type('application/json')
|
|
return self.filled.export_to_json(
|
|
anonymise=anonymise,
|
|
include_evolution=include_evolution,
|
|
include_files=include_files,
|
|
include_roles=include_roles,
|
|
include_submission=include_submission,
|
|
include_fields=include_fields,
|
|
include_unnamed_fields=include_unnamed_fields,
|
|
include_workflow=include_workflow,
|
|
include_workflow_data=include_workflow_data,
|
|
values_at=values_at,
|
|
)
|
|
|
|
def history(self):
|
|
if not self.filled.evolution:
|
|
return
|
|
if not self.formdef.is_user_allowed_read_status_and_history(get_request().user, self.filled):
|
|
return
|
|
|
|
include_authors_in_form_history = (
|
|
get_publisher().get_site_option('include_authors_in_form_history', 'variables') != 'False'
|
|
)
|
|
include_authors = get_request().is_in_backoffice() or include_authors_in_form_history
|
|
return template.render(
|
|
list(self.get_formdef_template_variants(self.history_templates)),
|
|
{
|
|
'formdata': self.filled,
|
|
'include_authors': include_authors,
|
|
'view': self,
|
|
},
|
|
)
|
|
|
|
def check_receiver(self):
|
|
session = get_session()
|
|
if not session or not session.user:
|
|
if not self.filled.formdef.is_user_allowed_read(None, self.filled):
|
|
raise errors.AccessUnauthorizedError()
|
|
user = get_request().user
|
|
if self.filled.formdef is None:
|
|
raise errors.AccessForbiddenError()
|
|
if not self.filled.formdef.is_user_allowed_read(user, self.filled):
|
|
raise errors.AccessForbiddenError()
|
|
return user
|
|
|
|
def should_fold_summary(self, mine, request_user):
|
|
# fold the summary if the form has already been seen by the user, i.e.
|
|
# if it's user own form or if the user is present in the formdata log
|
|
# (evolution).
|
|
if mine or (request_user and self.filled.is_submitter(request_user)):
|
|
return True
|
|
elif request_user and self.filled.evolution:
|
|
for evo in self.filled.evolution:
|
|
if str(evo.who) == str(request_user.id) or (
|
|
evo.who == '_submitter' and self.filled.is_submitter(request_user)
|
|
):
|
|
return True
|
|
return False
|
|
|
|
def should_fold_history(self):
|
|
return False
|
|
|
|
def receipt(self, always_include_user=False, form_url='', mine=True):
|
|
request_user = user = get_request().user
|
|
if not always_include_user and get_request().user and get_request().user.id == self.filled.user_id:
|
|
user = None
|
|
else:
|
|
try:
|
|
user = get_publisher().user_class.get(self.filled.user_id)
|
|
except KeyError:
|
|
user = None
|
|
|
|
return template.render(
|
|
list(self.get_formdef_template_variants(self.summary_templates)),
|
|
{
|
|
'formdata': self.filled,
|
|
'should_fold_summary': self.should_fold_summary(mine, request_user),
|
|
'fields': self.display_fields(form_url=form_url),
|
|
'view': self,
|
|
'user': user,
|
|
},
|
|
)
|
|
|
|
def display_fields(self, fields=None, form_url='', include_unset_required_fields=False):
|
|
import wcs.workflows
|
|
|
|
field_details = self.filled.get_summary_field_details(
|
|
fields, include_unset_required_fields=include_unset_required_fields
|
|
)
|
|
|
|
r = TemplateIO(html=True)
|
|
on_page = None
|
|
for field_value_info in field_details:
|
|
f = field_value_info['field']
|
|
if f.type == 'page':
|
|
if on_page:
|
|
r += htmltext('</div>')
|
|
r += htmltext('</div>')
|
|
r += htmltext('<div class="page">')
|
|
r += htmltext('<h3>%s</h3>') % get_publisher().translate(f.label)
|
|
r += htmltext('<div>')
|
|
on_page = f
|
|
continue
|
|
|
|
if f.type == 'title':
|
|
label = wcs.workflows.template_on_formdata(
|
|
None, get_publisher().translate(f.label), autoescape=False
|
|
)
|
|
r += htmltext('<div class="title %s"><h3>%s</h3></div>') % (f.extra_css_class or '', label)
|
|
continue
|
|
|
|
if f.type == 'subtitle':
|
|
label = wcs.workflows.template_on_formdata(
|
|
None, get_publisher().translate(f.label), autoescape=False
|
|
)
|
|
r += htmltext('<div class="subtitle %s"><h4>%s</h4></div>') % (f.extra_css_class or '', label)
|
|
continue
|
|
|
|
if f.type == 'comment':
|
|
r += htmltext(
|
|
'<div class="comment-field %s">%s</div>' % (f.extra_css_class or '', f.get_text())
|
|
)
|
|
continue
|
|
|
|
css_classes = ['field', 'field-type-%s' % f.key]
|
|
if f.extra_css_class:
|
|
css_classes.append(f.extra_css_class)
|
|
r += htmltext('<div class="%s">' % ' '.join(css_classes))
|
|
if f.key == 'block' and f.label_display == 'subtitle':
|
|
r += htmltext('<div class="subtitle"><h4>%s</h4></div>') % get_publisher().translate(f.label)
|
|
elif not (f.key == 'block' and f.label_display == 'hidden'):
|
|
r += htmltext('<p class="label">%s</p> ') % get_publisher().translate(f.label)
|
|
value, value_details = field_value_info['value'], field_value_info['value_details']
|
|
if value is None:
|
|
if not (f.key == 'block' and f.label_display == 'hidden'):
|
|
r += htmltext('<div class="value"><i>%s</i></div>') % _('Not set')
|
|
else:
|
|
r += htmltext('<div class="value">')
|
|
s = f.get_view_value(
|
|
value,
|
|
summary=True,
|
|
include_unset_required_fields=include_unset_required_fields,
|
|
**value_details,
|
|
)
|
|
s = s.replace('[download]', str('%sdownload' % form_url))
|
|
r += s
|
|
r += htmltext('</div>')
|
|
r += htmltext('</div>')
|
|
|
|
if on_page:
|
|
r += htmltext('</div></div>')
|
|
|
|
return r.getvalue()
|
|
|
|
def backoffice_fields_section(self):
|
|
backoffice_fields = self.formdef.workflow.get_backoffice_fields()
|
|
if not backoffice_fields:
|
|
return
|
|
content = self.display_fields(backoffice_fields, include_unset_required_fields=True)
|
|
if not content:
|
|
return
|
|
r = TemplateIO(html=True)
|
|
r += htmltext('<div class="section foldable">')
|
|
r += htmltext(
|
|
'<h2><span role="button" aria-expanded="true" '
|
|
'aria-controls="sect-backoffice-data" '
|
|
'id="sect-backoffice-data-label">%s</span></h2>'
|
|
) % _('Backoffice Data')
|
|
r += htmltext('<div class="dataview" id="sect-backoffice-data">')
|
|
r += content
|
|
r += htmltext('</div>')
|
|
r += htmltext('</div>')
|
|
return r.getvalue()
|
|
|
|
def status(self):
|
|
if get_request().get_query() == 'unlock':
|
|
# mark user as active visitor of the object, then redirect to self,
|
|
# the unlocked form will appear.
|
|
get_session().mark_visited_object(self.filled)
|
|
return redirect('./#lock-notice')
|
|
|
|
user = self.check_receiver()
|
|
form = self.get_workflow_form(user)
|
|
try:
|
|
response = self.check_submitted_form(form)
|
|
except RedisplayFormException:
|
|
# don't display errors after "add block" button has been clicked.
|
|
form.clear_errors()
|
|
else:
|
|
if response:
|
|
get_session().unmark_visited_object(self.filled)
|
|
return response
|
|
|
|
get_response().add_javascript(['jquery.js', 'qommon.forms.js'])
|
|
audit('view', obj=self.filled)
|
|
self.html_top('%s - %s' % (self.formdef.name, self.filled.id))
|
|
r = TemplateIO(html=True)
|
|
r += get_session().display_message()
|
|
r += htmltext(self.workflow_messages())
|
|
r += self.receipt(always_include_user=True, mine=False)
|
|
r += self.backoffice_fields_section()
|
|
|
|
r += self.history()
|
|
|
|
bottom_workflow_messages = self.bottom_workflow_messages()
|
|
if bottom_workflow_messages or form:
|
|
r += htmltext('<span id="action-zone"></span>')
|
|
|
|
r += htmltext(bottom_workflow_messages)
|
|
|
|
locked = False
|
|
if form:
|
|
all_visitors = get_session().get_object_visitors(self.filled)
|
|
visitors = [x for x in all_visitors if x[0] != get_session().user]
|
|
if visitors:
|
|
current_timestamp = time.time()
|
|
visitor_users = []
|
|
for visitor_id, visitor_timestamp in visitors:
|
|
try:
|
|
visitor_name = get_publisher().user_class.get(visitor_id).display_name
|
|
except KeyError:
|
|
continue
|
|
minutes_ago = int((current_timestamp - visitor_timestamp) / 60)
|
|
if minutes_ago < 1:
|
|
time_ago = _('less than a minute ago')
|
|
else:
|
|
time_ago = _('less than %s minutes ago') % (minutes_ago + 1)
|
|
visitor_users.append('%s (%s)' % (visitor_name, time_ago))
|
|
if visitor_users:
|
|
r += htmltext('<div id="lock-notice" class="infonotice"><p>')
|
|
r += str(
|
|
_('Be warned forms of this user are also being looked at by: %s.')
|
|
% ', '.join(visitor_users)
|
|
)
|
|
r += ' '
|
|
r += htmltext('</p>')
|
|
me_in_visitors = bool(get_session().user in [x[0] for x in all_visitors])
|
|
if not me_in_visitors:
|
|
locked = True
|
|
r += htmltext('<p class="action"><a href="?unlock">%s</a></p>') % _(
|
|
'(unlock actions)'
|
|
)
|
|
r += htmltext('</div>')
|
|
if not locked:
|
|
r += htmltext(self.actions_workflow_messages())
|
|
if form.widgets:
|
|
r += htmltext('<div class="section"><div>')
|
|
r += form.render()
|
|
if form.widgets:
|
|
r += htmltext('</div></div>')
|
|
get_session().mark_visited_object(self.filled)
|
|
|
|
if not locked:
|
|
if (self.filled.get_status() and self.filled.get_status().backoffice_info_text) or (
|
|
form
|
|
and any(getattr(button, 'backoffice_info_text', None) for button in form.get_submit_widgets())
|
|
):
|
|
r += htmltext('<div class="backoffice-description bo-block">')
|
|
if self.filled.get_status().backoffice_info_text:
|
|
r += htmltext(self.filled.get_status().backoffice_info_text)
|
|
if form:
|
|
for button in form.get_submit_widgets():
|
|
if not getattr(button, 'backoffice_info_text', None):
|
|
continue
|
|
r += htmltext('<div class="action-info-text" data-button-name="%s">' % button.name)
|
|
r += htmltext(button.backoffice_info_text)
|
|
r += htmltext('</div>')
|
|
r += htmltext('</div>')
|
|
|
|
r += htmltext('<a href="..">%s</a>') % _('Back to Listing')
|
|
return r.getvalue()
|
|
|
|
def submit(self, form):
|
|
user = get_request().user
|
|
next_url = self.filled.handle_workflow_form(user, form)
|
|
if next_url:
|
|
return next_url
|
|
if form.has_errors():
|
|
return
|
|
try:
|
|
self.check_auth()
|
|
except errors.AccessError:
|
|
# the user no longer has access to the form; redirect to a
|
|
# different page
|
|
if 'backoffice/' in [x[0] for x in get_response().breadcrumb]:
|
|
user = get_request().user
|
|
if user and (user.is_admin or self.formdef.is_of_concern_for_user(user)):
|
|
# user has access to the formdef, redirect to the
|
|
# listing.
|
|
return '..'
|
|
else:
|
|
return get_publisher().get_backoffice_url()
|
|
else:
|
|
return get_publisher().get_root_url()
|
|
|
|
def download(self):
|
|
if not is_url_signed():
|
|
self.check_receiver()
|
|
file = None
|
|
if get_request().form and get_request().form.get('hash'):
|
|
# look in all known formdata files for file with given hash
|
|
file_digest = get_request().form.get('hash')
|
|
for field_data in self.filled.get_all_file_data():
|
|
if not hasattr(field_data, 'file_digest'):
|
|
continue
|
|
if field_data.file_digest() == file_digest:
|
|
thumbnail = bool(get_request().form.get('thumbnail') and field_data.can_thumbnail())
|
|
if not thumbnail:
|
|
# do not log access to thumbnails as they will already be accounted for as
|
|
# a view of the formdata/carddata containing them.
|
|
audit(
|
|
'download file',
|
|
obj=self.filled,
|
|
extra_label=str(field_data),
|
|
file_digest=file_digest,
|
|
)
|
|
return FileDirectory.serve_file(field_data, thumbnail=thumbnail)
|
|
elif get_request().form and get_request().form.get('f'):
|
|
try:
|
|
fn = get_request().form['f']
|
|
if '$' in fn:
|
|
# path to block field contents
|
|
fn2, idx, sub = fn.split('$', 2)
|
|
file = self.filled.data[fn2]['data'][int(idx)][sub]
|
|
else:
|
|
file = self.filled.data[fn]
|
|
except (KeyError, ValueError):
|
|
pass
|
|
|
|
if not hasattr(file, 'content_type'):
|
|
raise errors.TraversalError()
|
|
|
|
if file.has_redirect_url():
|
|
redirect_url = file.get_redirect_url(backoffice=get_request().is_in_backoffice())
|
|
if not redirect_url:
|
|
raise errors.TraversalError()
|
|
redirect_url = sign_url_auto_orig(redirect_url)
|
|
return redirect(redirect_url)
|
|
|
|
file_url = 'files/%s/' % fn
|
|
|
|
if get_request().form.get('thumbnail') == '1':
|
|
if file.can_thumbnail():
|
|
file_url += 'thumbnail/'
|
|
else:
|
|
raise errors.TraversalError()
|
|
|
|
if is_url_signed():
|
|
# serve file directly, no redirect to URL with path ending with filename
|
|
file_directory = FileDirectory(
|
|
self.filled,
|
|
reference=fn,
|
|
thumbnails=bool(get_request().form.get('thumbnail') and file.can_thumbnail()),
|
|
)
|
|
return file_directory._q_lookup(component=None)
|
|
|
|
if getattr(file, 'base_filename'):
|
|
file_url += urllib.parse.quote(file.base_filename)
|
|
|
|
return redirect(file_url)
|
|
|
|
@classmethod
|
|
def live_process_fields(cls, form, formdata, displayed_fields):
|
|
if form is None:
|
|
return json.dumps({'result': {}})
|
|
|
|
result = {}
|
|
for field in displayed_fields:
|
|
result[field.id] = {'visible': field.is_visible(formdata.data, formdata.formdef)}
|
|
|
|
modified_field_varnames = set()
|
|
if get_request().form.get('modified_field_id') == 'init':
|
|
# when page is initialized, <select> will get their first option
|
|
# automatically selected, so mark them all as modified.
|
|
for field in displayed_fields:
|
|
if field.key == 'item' and field.display_mode == 'list' and field.varname:
|
|
modified_field_varnames.add(field.varname)
|
|
elif get_request().form.get('modified_field_id') == 'user' and get_request().is_in_frontoffice():
|
|
# not allowed in frontoffice.
|
|
raise errors.AccessForbiddenError()
|
|
elif get_request().form.get('modified_field_id') == 'user':
|
|
# user selection in sidebar
|
|
formdata.user_id = get_request().form.get('user_id')
|
|
elif get_request().form.get('modified_field_id'):
|
|
for field in displayed_fields:
|
|
if field.id == get_request().form.get('modified_field_id'):
|
|
modified_field_varnames.add(field.varname)
|
|
break
|
|
|
|
for field in displayed_fields:
|
|
if field.key in ('item', 'items') and field.data_source:
|
|
data_source = data_sources.get_object(field.data_source)
|
|
if data_source.type not in ('json', 'geojson') and not data_source.type.startswith(
|
|
'carddef:'
|
|
):
|
|
continue
|
|
varnames = data_source.get_referenced_varnames(field.formdef)
|
|
if (not modified_field_varnames or modified_field_varnames.intersection(varnames)) and (
|
|
field.display_mode == 'autocomplete' and data_source.can_jsonp() and field.type != 'items'
|
|
):
|
|
# computed earlier, in perform_more_widget_changes, when the field
|
|
# was added to the form
|
|
result[field.id]['source_url'] = field.url
|
|
elif modified_field_varnames.intersection(varnames):
|
|
result[field.id]['items'] = field.get_extended_options()
|
|
if field.display_mode == 'timetable':
|
|
# timetables require additional attributes
|
|
# but reduce payload weight by removing the API URLs
|
|
for options in result[field.id]['items']:
|
|
options.pop('api', None)
|
|
|
|
def get_all_field_widgets(form):
|
|
for widget in form.widgets:
|
|
if not getattr(widget, 'field', None):
|
|
continue
|
|
yield (None, None, widget.field, widget)
|
|
if isinstance(widget, BlockWidget):
|
|
block_row = 0
|
|
for subwidget in widget.widgets:
|
|
if isinstance(subwidget, BlockSubWidget):
|
|
for field_widget in subwidget.widgets:
|
|
yield (widget.field, block_row, field_widget.field, field_widget)
|
|
block_row += 1
|
|
|
|
for block, block_row, field, widget in get_all_field_widgets(form):
|
|
if block:
|
|
try:
|
|
block_data = formdata.data.get(block.id)['data'][block_row]
|
|
except (IndexError, TypeError):
|
|
block_data = {}
|
|
|
|
with block.block.visibility_context(block_data, block_row):
|
|
is_visible = field.is_visible({}, formdef=None)
|
|
entry = {'visible': is_visible, 'row': block_row, 'field_id': field.id, 'block_id': block.id}
|
|
result['%s-%s-%s' % (block.id, field.id, block_row)] = entry
|
|
entry['block_id'] = block.id
|
|
entry['field_id'] = field.id
|
|
entry['block_row'] = 'element%s' % block_row
|
|
|
|
else:
|
|
entry = result[field.id]
|
|
if field.key == 'comment':
|
|
entry['content'] = widget.content
|
|
elif field.key == 'block':
|
|
# do not apply live updates to prefilled blocks as this would imply too
|
|
# much javascript (for example new block rows may have to be created).
|
|
pass
|
|
elif field.get_prefill_configuration().get('type') == 'string':
|
|
if 'request.GET' in (field.get_prefill_configuration().get('value') or ''):
|
|
# Prefilling with a value from request.GET cannot be compatible with
|
|
# live updates of prefill values. Skip those. (a "computed data" field
|
|
# should be used as replacement).
|
|
if field.id in result:
|
|
del result[field.id]
|
|
continue
|
|
update_prefill = bool('prefilled_%s' % field.id in get_request().form)
|
|
if update_prefill:
|
|
value = field.get_prefill_value()[0]
|
|
if field.key == 'bool':
|
|
value = field.convert_value_from_str(value)
|
|
elif field.key == 'date' and value:
|
|
try:
|
|
value = field.convert_value_from_anything(value)
|
|
text_content = field.convert_value_to_str(value)
|
|
# convert date to Y-m-d as expected by the <input type=date> field
|
|
value = field.get_json_value(value)
|
|
except ValueError:
|
|
text_content = None
|
|
entry['text_content'] = text_content
|
|
elif field.key == 'item' and value:
|
|
id_value = field.get_id_by_option_text(value)
|
|
if id_value:
|
|
value = id_value
|
|
entry['content'] = value
|
|
elif field.get_prefill_configuration().get('type') == 'user':
|
|
update_prefill = bool(get_request().form.get('modified_field_id') == 'user')
|
|
if update_prefill:
|
|
value, locked = field.get_prefill_value(user=formdata.user)
|
|
entry['content'] = value
|
|
entry['locked'] = locked
|
|
|
|
return json.dumps({'result': result})
|
|
|
|
def live(self):
|
|
get_request().ignore_session = True
|
|
# live evaluation of fields
|
|
get_response().set_content_type('application/json')
|
|
|
|
def result_error(reason):
|
|
return json.dumps({'result': 'error', 'reason': reason})
|
|
|
|
session = get_session()
|
|
if not session:
|
|
return result_error('missing session')
|
|
|
|
displayed_fields = []
|
|
user = get_request().user
|
|
form = self.filled.get_workflow_form(user, displayed_fields=displayed_fields)
|
|
if form is None:
|
|
return result_error('no more form')
|
|
|
|
self.filled.evaluate_live_workflow_form(user, form)
|
|
get_publisher().substitutions.unfeed(lambda x: x is self.filled)
|
|
get_publisher().substitutions.feed(self.filled)
|
|
# reevaluate workflow form according to possible new content
|
|
displayed_fields = []
|
|
form = self.filled.get_workflow_form(user, displayed_fields=displayed_fields)
|
|
return self.live_process_fields(form, self.filled, displayed_fields)
|
|
|
|
def _q_lookup(self, component):
|
|
if component == 'files':
|
|
self.check_receiver()
|
|
return FilesDirectory(self.filled)
|
|
if component.startswith('wfedit-'):
|
|
return self.wfedit(component[len('wfedit-') :])
|
|
return Directory._q_lookup(self, component)
|
|
|
|
def _q_traverse(self, path):
|
|
get_response().breadcrumb.append((str(self.filled.id) + '/', self.filled.get_display_id()))
|
|
return super()._q_traverse(path)
|
|
|
|
def wfedit(self, action_id):
|
|
wf_status = self.filled.get_status()
|
|
for item in wf_status.items:
|
|
if item.id != action_id:
|
|
continue
|
|
if not isinstance(item, EditableWorkflowStatusItem):
|
|
break
|
|
if not item.check_auth(self.filled, get_request().user):
|
|
break
|
|
f = self.form_page_class(self.formdef.url_name)
|
|
f.edit_mode = True
|
|
f.edited_data = self.filled
|
|
f.edit_action = item
|
|
f.action_url = 'wfedit-%s' % item.id
|
|
if get_request().is_in_backoffice():
|
|
get_session().mark_visited_object(self.filled)
|
|
get_response().breadcrumb = get_response().breadcrumb[:-1]
|
|
get_response().breadcrumb.append((f.action_url, _('Edit')))
|
|
return f._q_index()
|
|
|
|
raise errors.AccessForbiddenError()
|
|
|
|
|
|
class FormdefDirectoryBase(Directory):
|
|
user = None
|
|
|
|
def tempfile(self):
|
|
get_request().ignore_session = True
|
|
self.check_access()
|
|
if self.user and not self.user.id == get_session().user:
|
|
self.check_receiver()
|
|
try:
|
|
t = get_request().form['t']
|
|
tempfile = get_session().get_tempfile(t)
|
|
except KeyError:
|
|
raise errors.TraversalError()
|
|
if tempfile is None:
|
|
raise errors.TraversalError()
|
|
response = get_response()
|
|
|
|
# force potential HTML upload to be used as-is (not decorated with theme)
|
|
# and with minimal permissions
|
|
response.filter = {}
|
|
response.set_header(
|
|
'Content-Security-Policy',
|
|
'default-src \'none\'; img-src %s;' % get_request().build_absolute_uri(),
|
|
)
|
|
|
|
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'])
|
|
|
|
if get_request().form.get('thumbnail') == '1':
|
|
try:
|
|
thumbnail = misc.get_thumbnail(
|
|
get_session().get_tempfile_path(t), content_type=tempfile['content_type']
|
|
)
|
|
except misc.ThumbnailError:
|
|
pass
|
|
else:
|
|
response.set_content_type('image/png')
|
|
return thumbnail
|
|
return get_session().get_tempfile_content(t).get_file_pointer().read()
|