wcs/wcs/backoffice/management.py

4026 lines
169 KiB
Python

# w.c.s. - web application for online forms
# Copyright (C) 2005-2015 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 csv
import datetime
import io
import json
import re
import time
import types
import urllib.parse
import zipfile
import vobject
from django.utils.encoding import force_text
from quixote import get_publisher, get_request, get_response, get_session, redirect
from quixote.directory import Directory
from quixote.html import TemplateIO, htmlescape, htmltext
from quixote.http_request import parse_query
from wcs.admin.forms import UpdateDigestAfterJob
from wcs.admin.settings import UserFieldsFormDef
from wcs.api_utils import get_user_from_api_query_string
from wcs.carddef import CardDef
from wcs.categories import Category
from wcs.conditions import Condition
from wcs.formdata import FormData
from wcs.formdef import FormDef
from wcs.forms.backoffice import FormDefUI
from wcs.forms.common import FormStatusPage
from wcs.roles import logged_users_role
from wcs.variables import LazyFieldVar
from wcs.workflows import ActionsTracingEvolutionPart, WorkflowStatusItem, item_classes, template_on_formdata
from ..qommon import _, emails, errors, ezt, force_str, get_cfg, get_logger, misc, ngettext, ods, sms
from ..qommon.admin.emails import EmailsDirectory
from ..qommon.admin.menu import command_icon
from ..qommon.afterjobs import AfterJob
from ..qommon.backoffice.listing import pagination_links
from ..qommon.backoffice.menu import html_top
from ..qommon.evalutils import make_datetime
from ..qommon.form import (
CheckboxWidget,
DateWidget,
EmailWidget,
Form,
HiddenWidget,
HtmlWidget,
MapWidget,
RadiobuttonsWidget,
SingleSelectWidget,
StringWidget,
TextWidget,
WidgetList,
WysiwygTextWidget,
)
from ..qommon.misc import C_, ellipsize
from ..qommon.storage import (
And,
Contains,
Equal,
FtsMatch,
GreaterOrEqual,
ILike,
Intersects,
LessOrEqual,
NotEqual,
NotNull,
Null,
Or,
)
from ..qommon.substitution import CompatibilityNamesDict
from ..qommon.template import Template
from ..qommon.upload_storage import PicklableUpload
from .submission import FormFillPage
def geojson_formdatas(formdatas, geoloc_key='base', fields=None):
geojson = {'type': 'FeatureCollection', 'features': []}
for formdata in formdatas:
if not formdata.geolocations or geoloc_key not in formdata.geolocations:
continue
coords = misc.normalize_geolocation(formdata.geolocations[geoloc_key])
status = formdata.get_status()
try:
status_colour = status.colour
except AttributeError:
status_colour = 'ffffff'
display_fields = []
formdata_backoffice_url = formdata.get_url(backoffice=True)
if fields:
for field in fields:
if field.type == 'map':
continue
html_value = formdata.get_field_view_value(field, max_length=60)
html_value = html_value.replace('[download]', '%sdownload' % formdata_backoffice_url)
value = formdata.get_field_view_value(field)
if not html_value and not value:
continue
display_fields.append(
{
'varname': field.varname,
'label': field.label,
'value': str(value),
'html_value': str(htmlescape(html_value)),
}
)
feature = {
'type': 'Feature',
'properties': {
'name': str(htmlescape(formdata.get_display_name())),
'url': formdata_backoffice_url,
'status_name': str(htmlescape(status.name)),
'status_colour': '#%s' % status_colour,
'view_label': force_text(_('View')),
},
'geometry': {
'type': 'Point',
'coordinates': [coords['lon'], coords['lat']],
},
}
if display_fields:
feature['properties']['display_fields'] = display_fields
geojson['features'].append(feature)
return geojson
class SendCodeFormdefDirectory(Directory):
formdef = None
def __init__(self, formdef):
self.formdef = formdef
def _q_lookup(self, component):
html_top('management', _('Management'))
formdata = self.formdef.data_class().get(component)
submitter_email = formdata.formdef.get_submitter_email(formdata)
mail_subject = EmailsDirectory.get_subject('tracking-code-reminder')
mail_body = EmailsDirectory.get_body('tracking-code-reminder')
form = Form()
form.add(TextWidget, 'text', title=_('Message'), required=True, cols=60, rows=5, value=mail_body)
form.add(EmailWidget, 'email', title=_('Email'), required=False, value=submitter_email)
sms_class = None
if get_publisher().use_sms_feature:
sms_class = sms.SMS.get_sms_class()
if sms_class:
form.add(StringWidget, 'sms', title=_('SMS Number'), required=False)
form.add(
RadiobuttonsWidget,
'method',
options=[('email', _('Email')), ('sms', _('SMS'))],
value='email',
required=True,
extra_css_class='widget-inline-radio',
)
form.add_submit('submit', _('Send'))
form.add_submit('cancel', _('Cancel'))
if not form.is_submitted() or form.has_errors():
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Send tracking code')
r += form.render()
return r.getvalue()
if not formdata.tracking_code:
tracking_code = get_publisher().tracking_code_class()
tracking_code.formdata = formdata # this stores both objects
message = form.get_widget('text').parse()
data = {
'form_tracking_code': formdata.tracking_code,
'tracking_code': formdata.tracking_code,
'email': form.get_widget('email').parse(),
}
data.update(self.formdef.get_substitution_variables(minimal=True))
if sms_class and form.get_widget('method').parse() == 'sms':
# send sms
sitename = get_cfg('misc', {}).get('sitename') or 'w.c.s.'
sms_cfg = get_cfg('sms', {})
sender = sms_cfg.get('sender', sitename)[:11]
message = Template(message).render(data)
try:
sms_class.send(sender, [form.get_widget('sms').parse()], message)
except errors.SMSError as e:
get_logger().error(e)
get_session().message = ('info', _('SMS with tracking code sent to the user'))
else:
# send mail
emails.template_email(
mail_subject, message, mail_body_data=data, email_rcpt=form.get_widget('email').parse()
)
get_session().message = ('info', _('Email with tracking code sent to the user'))
return redirect('../..')
class SendCodeDirectory(Directory):
def _q_lookup(self, component):
try:
formdef = FormDef.get_by_urlname(component)
if not formdef.enable_tracking_codes:
raise errors.TraversalError()
return SendCodeFormdefDirectory(formdef)
except KeyError:
raise errors.TraversalError()
class UserViewDirectory(Directory):
_q_exports = ['', 'sendcode']
sendcode = SendCodeDirectory()
user = None
def __init__(self, user):
self.user = user
def _q_index(self):
get_response().breadcrumb.append(('%s/' % self.user.id, self.user.display_name))
html_top('management', _('Management'))
# display list of open formdata for the user
formdefs = [x for x in FormDef.select(lightweight=True) if not x.skip_from_360_view]
user_roles = set([logged_users_role().id] + get_request().user.get_roles())
criterias = [
Equal('is_at_endpoint', False),
Equal('user_id', str(self.user.id)),
Contains('formdef_id', [x.id for x in formdefs]),
]
from wcs import sql
formdatas = sql.AnyFormData.select(criterias, order_by='receipt_time')
criterias = [
Equal('is_at_endpoint', False),
Equal('user_id', str(self.user.id)),
Intersects('concerned_roles_array', user_roles),
]
viewable_formdatas = sql.AnyFormData.select(criterias)
viewable_formdatas_ids = {}
for viewable_formdata in viewable_formdatas:
viewable_formdatas_ids[(viewable_formdata.formdef.id, viewable_formdata.id)] = True
r = TemplateIO(html=True)
r += get_session().display_message()
r += htmltext('<div class="bo-block">')
r += htmltext('<h2>%s</h2>') % self.user.display_name
formdef = UserFieldsFormDef()
r += htmltext('<div class="form">')
for field in formdef.get_all_fields():
if not hasattr(field, 'get_view_value'):
continue
value = self.user.form_data.get(field.id)
if not value:
continue
r += htmltext('<div class="title">')
r += field.label
r += htmltext('</div>')
r += htmltext('<div class="StringWidget content">')
r += field.get_view_value(value)
r += htmltext('</div>')
r += htmltext('</div>')
r += htmltext('</div>')
if formdatas:
categories = {}
formdata_by_category = {}
for formdata in formdatas:
if formdata.formdef.category_id not in categories:
categories[formdata.formdef.category_id] = formdata.formdef.category
formdata_by_category[formdata.formdef.category_id] = []
formdata_by_category[formdata.formdef.category_id].append(formdata)
cats = list(categories.values())
Category.sort_by_position(cats)
for cat in cats:
r += htmltext('<div class="bo-block">')
if len(cats) > 1:
if cat is None:
r += htmltext('<h2>%s</h2>') % _('Misc')
cat_formdatas = formdata_by_category[None]
else:
r += htmltext('<h2>%s</h2>') % cat.name
cat_formdatas = formdata_by_category[cat.id]
else:
cat_formdatas = formdatas
r += htmltext('<ul class="biglist c-360-user-view">')
for formdata in cat_formdatas:
status_label = formdata.get_status_label()
submit_date = misc.strftime(misc.date_format(), formdata.receipt_time)
formdata_key_id = (formdata.formdef.id, formdata.id)
if formdata_key_id in viewable_formdatas_ids:
r += htmltext(
'<li><a href="%s">%s, '
'<span class="datetime">%s</span> '
'<span class="status">(%s)</span></a>'
% (
formdata.get_url(backoffice=True),
formdata.formdef.name,
submit_date,
status_label,
)
)
else:
r += htmltext(
'<li><span>%s, '
'<span class="datetime">%s</span> '
'<span class="status">(%s)</span></span>'
% (formdata.formdef.name, submit_date, status_label)
)
if formdata.formdef.enable_tracking_codes:
r += htmltext('<p class="commands">')
r += command_icon(
'sendcode/%s/%s' % (formdata.formdef.url_name, formdata.id),
'export',
label=_('Send tracking code'),
popup=True,
)
r += htmltext('</p>')
r += htmltext('</ul>')
r += htmltext('</div>')
r += htmltext('</div>')
return r.getvalue()
class UsersViewDirectory(Directory):
_q_exports = ['']
def _q_traverse(self, path):
if not get_publisher().is_using_postgresql():
raise errors.TraversalError()
get_response().breadcrumb.append(('users', _('Per User View')))
return super()._q_traverse(path)
def get_search_sidebar(self, offset=None, limit=None, order_by=None):
get_response().add_javascript(['wcs.listing.js'])
r = TemplateIO(html=True)
r += htmltext('<form id="listing-settings" action=".">')
if offset or limit:
if not offset:
offset = 0
r += htmltext('<input type="hidden" name="offset" value="%s"/>') % offset
if limit:
r += htmltext('<input type="hidden" name="limit" value="%s"/>') % limit
if order_by is None:
order_by = ''
r += htmltext('<input type="hidden" name="order_by" value="%s"/>') % order_by
r += htmltext('<h3>%s</h3>') % _('Search')
if get_request().form.get('q'):
q = force_text(get_request().form.get('q'))
r += htmltext('<input name="q" value="%s">') % force_str(q)
else:
r += htmltext('<input name="q">')
r += htmltext('<input type="submit" value="%s"/>') % _('Search')
return r.getvalue()
def _q_index(self):
html_top('management', _('Management'))
r = TemplateIO(html=True)
limit = misc.get_int_or_400(
get_request().form.get('limit', get_publisher().get_site_option('default-page-size')) or 20
)
offset = misc.get_int_or_400(get_request().form.get('offset', 0))
order_by = misc.get_order_by_or_400(get_request().form.get('order_by', None)) or 'name'
query = get_request().form.get('q')
get_response().filter['sidebar'] = self.get_search_sidebar(
limit=limit, offset=offset, order_by=order_by
)
if not query:
r += htmltext('<div id="listing">')
r += htmltext('<div class="big-msg-info">')
r += htmltext('<p>%s</p>') % _('Use the search field on the right ' 'to look for an user.')
r += htmltext('</div>')
r += htmltext('</div>')
if get_request().form.get('ajax') == 'true':
get_response().filter = {'raw': True}
return r.getvalue()
formdef = UserFieldsFormDef()
criteria_fields = [
ILike('name', query),
ILike('ascii_name', misc.simplify(query, ' ')),
ILike('email', query),
]
for field in formdef.get_all_fields():
if field.type in ('string', 'text', 'email'):
criteria_fields.append(ILike('f%s' % field.id, query))
if get_publisher().is_using_postgresql():
criteria_fields.append(FtsMatch(query))
criterias = [Or(criteria_fields)]
users = get_publisher().user_class.select(
order_by=order_by, clause=criterias, limit=limit, offset=offset
)
total_count = get_publisher().user_class.count(criterias)
users_cfg = get_cfg('users', {})
include_name_column = not (users_cfg.get('field_name'))
include_email_column = not (users_cfg.get('field_email'))
r += htmltext('<div id="listing">')
r += htmltext('<table class="main">')
r += htmltext('<thead>')
r += htmltext('<tr>')
if include_name_column:
r += htmltext('<th data-field-sort-key="name"><span>%s</span></th>') % _('Name')
if include_email_column:
r += htmltext('<th data-field-sort-key="email"><span>%s</span></th>') % _('Email')
for field in formdef.get_all_fields():
if field.include_in_listing:
r += htmltext('<th data-field-sort-key="f%s"><span>%s</span></th>') % (field.id, field.label)
r += htmltext('</tr>')
r += htmltext('</thead>')
r += htmltext('<tbody>')
for user in users:
r += htmltext('<tr data-link="%s/">') % user.id
if include_name_column:
r += htmltext('<td>%s</td>') % (user.name or '')
if include_email_column:
r += htmltext('<td>%s</td>') % (user.email or '')
for field in formdef.get_all_fields():
if field.include_in_listing:
r += htmltext('<td>%s</td>') % (user.form_data.get(field.id) or '')
r += htmltext('</tr>')
r += htmltext('</tbody>')
r += htmltext('</table>')
if get_publisher().is_using_postgresql():
r += pagination_links(offset, limit, total_count)
r += htmltext('</div>')
if get_request().form.get('ajax') == 'true':
get_response().filter = {'raw': True}
return r.getvalue()
return r.getvalue()
def _q_lookup(self, component):
try:
user = get_publisher().user_class.get(component)
return UserViewDirectory(user)
except KeyError:
pass
raise errors.TraversalError()
class ManagementDirectory(Directory):
_q_exports = ['', 'forms', 'listing', 'statistics', 'lookup', 'count', 'users', 'geojson', 'map']
users = UsersViewDirectory()
def add_breadcrumb(self):
get_response().breadcrumb.append(('management/', _('Management')))
def is_accessible(self, user):
return user.can_go_in_backoffice()
def _q_traverse(self, path):
self.add_breadcrumb()
return super()._q_traverse(path)
def _q_index(self):
if get_publisher().has_site_option('default-to-global-view'):
return redirect('listing')
else:
return redirect('forms')
def forms(self):
html_top('management', _('Management'))
formdefs = FormDef.select(order_by='name', ignore_errors=True, lightweight=True)
if len(formdefs) == 0:
return self.empty_site_message(_('Forms'))
get_response().filter['sidebar'] = self.get_sidebar(formdefs)
r = TemplateIO(html=True)
r += get_session().display_message()
user = get_request().user
user_roles = [logged_users_role().id] + (user.get_roles() if user else [])
forms_without_pending_stuff = []
forms_with_pending_stuff = []
using_postgresql = get_publisher().is_using_postgresql()
if using_postgresql:
from wcs import sql
actionable_counts = sql.get_actionable_counts(user_roles)
total_counts = sql.get_total_counts(user_roles)
def append_form_entry(formdef):
if using_postgresql:
count_forms = total_counts.get(formdef.id) or 0
waiting_forms_count = actionable_counts.get(formdef.id) or 0
else:
formdef_data_class = formdef.data_class()
count_forms = formdef_data_class.count() - len(
formdef_data_class.get_ids_with_indexed_value('status', 'draft')
)
waiting_forms_count = formdef_data_class.get_actionable_count(user_roles)
if waiting_forms_count == 0:
forms_without_pending_stuff.append((formdef, waiting_forms_count, count_forms))
else:
forms_with_pending_stuff.append((formdef, waiting_forms_count, count_forms))
if user:
for formdef in formdefs:
if user.is_admin or formdef.is_of_concern_for_user(user):
append_form_entry(formdef)
def top_action_links(r):
r += htmltext('<span class="actions">')
if (
get_publisher().is_using_postgresql()
and get_publisher().get_site_option('postgresql_views') != 'false'
):
r += htmltext('<a href="listing">%s</a>') % _('Global View')
if get_publisher().has_site_option('per-user-view'):
r += htmltext(' <a href="users/">%s</a>') % _('Per User View')
for formdef in formdefs:
if formdef.geolocations:
r += htmltext(' <a href="map">%s</a>') % _('Map View')
break
r += htmltext('</span>')
if forms_with_pending_stuff:
r += htmltext('<div id="appbar">')
r += htmltext('<h2>%s</h2>') % _('Forms in your care')
top_action_links(r)
r += htmltext('</div>')
r += htmltext('<div class="bo-block" id="forms-in-your-care">')
r += self.display_forms(forms_with_pending_stuff)
r += htmltext('</div>')
if forms_without_pending_stuff:
r += htmltext('<div id="appbar">')
r += htmltext('<h2>%s</h2>') % _('Other Forms')
if not forms_without_pending_stuff:
top_action_links(r)
r += htmltext('</div>')
r += htmltext('<div class="bo-block" id="other-forms">')
r += self.display_forms(forms_without_pending_stuff)
r += htmltext('</div>')
if not (forms_with_pending_stuff or forms_without_pending_stuff):
r += htmltext('<div id="appbar">')
r += htmltext('<h2>%s</h2>') % _('Forms')
top_action_links(r)
r += htmltext('</div>')
return r.getvalue()
def get_sidebar(self, formdefs):
r = TemplateIO(html=True)
r += self.get_lookup_sidebox()
r += htmltext('<div class="bo-block">')
r += htmltext('<ul id="sidebar-actions">')
r += htmltext('<li class="stats"><a href="statistics">%s</a></li>') % _('Global statistics')
r += htmltext('</ul>')
r += htmltext('</div>')
return r.getvalue()
def lookup(self):
query = get_request().form.get('query', '').strip()
if get_publisher().is_using_postgresql():
from wcs import sql
formdatas = sql.AnyFormData.select([Equal('id_display', query)])
if formdatas:
return redirect(formdatas[0].get_url(backoffice=True))
if any(x for x in FormDef.select(lightweight=True) if x.enable_tracking_codes):
try:
tracking_code = get_publisher().tracking_code_class.get(query)
formdata = tracking_code.formdata
get_session().mark_anonymous_formdata(formdata)
return redirect(formdata.get_url(backoffice=True))
except KeyError:
pass
if re.match(r'^\d+-\d{1,10}$', query):
formdef_id, formdata_id = query.split('-')
try:
formdef = FormDef.get(formdef_id)
formdata = formdef.data_class().get(formdata_id)
except KeyError:
pass
else:
return redirect(formdata.get_url(backoffice=True))
get_session().message = ('error', _('No such tracking code or identifier.'))
return redirect(get_request().form.get('back') or '.')
def get_lookup_sidebox(self, back_place=''):
r = TemplateIO(html=True)
r += htmltext('<div id="lookup-box">')
r += htmltext('<h3>%s</h3>' % _('Look up by tracking code or identifier'))
r += htmltext('<form action="lookup">')
r += htmltext('<input type="hidden" name="back" value="%s"/>') % back_place
r += htmltext('<input class="inline-input" size="12" name="query"/>')
r += htmltext('<button>%s</button>') % _('Look up')
r += htmltext('</form>')
r += htmltext('</div>')
return r.getvalue()
def get_global_listing_sidebar(self, limit=None, offset=None, order_by=None):
get_response().add_javascript(['jquery.js'])
DateWidget.prepare_javascript()
form = Form(use_tokens=False, id='listing-settings')
params = get_request().form
form.add(
SingleSelectWidget,
'status',
title=_('Status'),
options=[
('waiting', _('Waiting for an action'), 'waiting'),
('open', C_('formdata|Open'), 'open'),
('done', _('Done'), 'done'),
('all', _('All'), 'all'),
],
value=params.get('status'),
)
form.add(DateWidget, 'start', title=_('Start Date'), value=params.get('start'))
form.add(DateWidget, 'end', title=_('End Date'), value=params.get('end'))
categories = Category.select()
if categories:
Category.sort_by_position(categories)
category_options = [(None, C_('categories|All'), '')] + [(x.id, x.name, x.id) for x in categories]
category_slugs = (params.get('category_slugs') or '').split(',')
category_slugs = [c.strip() for c in category_slugs if c.strip()]
for i, category in enumerate([c for c in categories if c.url_name in category_slugs]):
params['category_ids$element%s' % i] = category.id
form.add(
WidgetList,
'category_ids',
title=_('Categories'),
element_type=SingleSelectWidget,
add_element_label=_('Add Category'),
element_kwargs={
'render_br': False,
'options': category_options,
},
)
if bool(get_publisher().get_site_option('welco_url', 'variables')):
form.add(
SingleSelectWidget,
'submission_channel',
title=_('Channel'),
options=[(None, C_('channel|All'), '')]
+ [(x, y, x) for x, y in FormData.get_submission_channels().items()],
value=params.get('submission_channel'),
)
form.add(StringWidget, 'q', title=_('Text'), value=params.get('q'))
if not offset:
offset = 0
if not limit:
limit = int(get_publisher().get_site_option('default-page-size') or 20)
if not order_by:
order_by = get_publisher().get_site_option('default-sort-order') or '-receipt_time'
form.add_hidden('offset', offset)
form.add_hidden('limit', limit)
form.add_hidden('order_by', order_by)
form.add_submit('submit', _('Submit'))
r = TemplateIO(html=True)
r += self.get_lookup_sidebox('listing')
r += htmltext('<div>')
r += htmltext('<h3>%s</h3>') % _('Filters')
r += form.render()
r += htmltext('</div>')
return r.getvalue()
def get_stats_sidebar(self):
get_response().add_javascript(['jquery.js'])
DateWidget.prepare_javascript()
form = Form(use_tokens=False)
form.add(DateWidget, 'start', title=_('Start Date'))
form.add(DateWidget, 'end', title=_('End Date'))
form.add_submit('submit', _('Submit'))
r = TemplateIO(html=True)
r += htmltext('<h3>%s</h3>') % _('Period')
r += form.render()
r += htmltext('<h3>%s</h3>') % _('Shortcuts')
r += htmltext('<ul>') # presets
current_month_start = datetime.datetime.now().replace(day=1)
start = current_month_start.strftime(misc.date_format())
r += htmltext(' <li><a href="?start=%s">%s</a>') % (start, _('Current Month'))
previous_month_start = current_month_start - datetime.timedelta(days=2)
previous_month_start = previous_month_start.replace(day=1)
start = previous_month_start.strftime(misc.date_format())
end = current_month_start.strftime(misc.date_format())
r += htmltext(' <li><a href="?start=%s&end=%s">%s</a>') % (start, end, _('Previous Month'))
current_year_start = datetime.datetime.now().replace(month=1, day=1)
start = current_year_start.strftime(misc.date_format())
r += htmltext(' <li><a href="?start=%s">%s</a>') % (start, _('Current Year'))
previous_year_start = current_year_start.replace(year=current_year_start.year - 1)
start = previous_year_start.strftime(misc.date_format())
end = current_year_start.strftime(misc.date_format())
r += htmltext(' <li><a href="?start=%s&end=%s">%s</a>') % (start, end, _('Previous Year'))
return r.getvalue()
def statistics(self):
html_top('management', _('Global statistics'))
get_response().breadcrumb.append(('statistics', _('Global statistics')))
if FormDef.count() == 0:
r = TemplateIO(html=True)
r += htmltext('<div class="top-title">')
r += htmltext('<h2>%s</h2>') % _('Global statistics')
r += htmltext('</div>')
r += htmltext('<div class="big-msg-info">')
r += htmltext('<p>%s</p>') % _(
'This site is currently empty. It is required to first add forms.'
)
r += htmltext('</div>')
return r.getvalue()
get_response().filter['sidebar'] = self.get_stats_sidebar()
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Global statistics')
formdefs = FormDef.select(order_by='name', ignore_errors=True)
counts = {}
parsed_values = {}
criterias = get_global_criteria(get_request(), parsed_values)
period_start = parsed_values.get('period_start')
period_end = parsed_values.get('period_end')
if get_publisher().is_using_postgresql():
from wcs import sql
formdef_totals = sql.get_formdef_totals(period_start, period_end, criterias)
counts = {str(x): y for x, y in formdef_totals}
else:
for formdef in formdefs:
values = formdef.data_class().select(criterias)
counts[formdef.id] = len(values)
do_graphs = False
if (
get_publisher().is_using_postgresql()
and get_publisher().get_site_option('postgresql_views') != 'false'
):
do_graphs = True
r += htmltext('<p>%s %s</p>') % (_('Total count:'), sum(counts.values()))
if do_graphs:
r += htmltext('<div class="splitcontent-left">')
cats = Category.select()
for cat in cats:
category_formdefs = [x for x in formdefs if x.category_id == cat.id]
r += self.category_global_stats(cat.name, category_formdefs, counts)
category_formdefs = [x for x in formdefs if x.category_id is None]
r += self.category_global_stats(_('Misc'), category_formdefs, counts)
if do_graphs:
r += htmltext('</div>')
r += htmltext('<div class="splitcontent-right">')
r += do_graphs_section(period_start, period_end, criterias=[NotEqual('status', 'draft')])
r += htmltext('</div>')
return r.getvalue()
def category_global_stats(self, title, category_formdefs, counts):
r = TemplateIO(html=True)
category_formdefs_ids = [x.id for x in category_formdefs]
if not category_formdefs:
return
cat_counts = {x: y for x, y in counts.items() if x in category_formdefs_ids}
if sum(cat_counts.values()) == 0:
return
r += htmltext('<div class="bo-block">')
r += htmltext('<h3>%s</h3>') % title
r += htmltext('<p>%s %s</p>') % (_('Count:'), sum(cat_counts.values()))
r += htmltext('<ul>')
for category_formdef in category_formdefs:
if not counts.get(category_formdef.id):
continue
r += htmltext('<li>%s %s</li>') % (
_('%s:') % category_formdef.name,
counts.get(category_formdef.id),
)
r += htmltext('</ul>')
r += htmltext('</div>')
return r.getvalue()
def display_forms(self, forms_list):
r = TemplateIO(html=True)
r += htmltext('<ul class="biglist">')
cats = Category.select()
Category.sort_by_position(cats)
for c in cats + [None]:
if c is None:
l2 = [x for x in forms_list if not x[0].category_id]
cat_name = _('Misc')
else:
l2 = [x for x in forms_list if x[0].category_id == c.id]
cat_name = c.name
if not l2:
continue
r += htmltext('<li><h3>%s</h3></li>') % cat_name
for formdef, no_pending, no_total in l2:
r += htmltext('<li><strong><a href="%s/">%s</a></strong>') % (formdef.url_name, formdef.name)
klass = ''
if no_total:
klass = 'badge'
r += htmltext('<p class="details %s">' % klass)
if no_pending:
r += _('%(pending)s open on %(total)s') % {'pending': no_pending, 'total': no_total}
else:
r += ngettext('%(total)s item', '%(total)s items', no_total) % {'total': no_total}
r += htmltext('</p>')
r += htmltext('</li>')
r += htmltext('</ul>')
return r.getvalue()
def get_global_listing_criterias(self, ignore_user_roles=False, include_own=False, include_drafts=False):
parsed_values = {}
user_roles = [logged_users_role().id]
if get_request().user:
user_roles.extend(get_request().user.get_roles())
criterias = get_global_criteria(get_request(), parsed_values, include_drafts=include_drafts)
query_parameters = (get_request().form or {}).copy()
query_parameters.pop('callback', None) # when using jsonp
status = query_parameters.get('status', 'waiting')
if query_parameters.get('waiting') == 'yes':
# compatibility with ?waiting=yes|no parameter, still used in
# the /count endpoint used for indicators
status = 'waiting'
elif query_parameters.get('waiting') == 'no':
status = 'open'
roles_array_column = 'concerned_roles_array'
status_criterias = []
if status == 'waiting':
status_criterias.append(Equal('is_at_endpoint', False))
roles_array_column = 'actions_roles_array'
elif status == 'open':
status_criterias.append(Equal('is_at_endpoint', False))
elif status == 'done':
status_criterias.append(Equal('is_at_endpoint', True))
if not ignore_user_roles:
roles_criteria = Intersects(roles_array_column, user_roles)
if include_own and get_request().user and not get_request().user.is_api_user:
roles_criteria = Or(
[
roles_criteria,
And(
[
Equal('user_id', str(get_request().user.id)),
Intersects(roles_array_column, ['_submitter']),
]
),
]
)
status_criterias.append(roles_criteria)
if include_drafts and get_request().user and not get_request().user.is_api_user:
# include user drafts, without any endpoint / roles array criteria
status_criterias = [
Or(
[
And(status_criterias),
And(
[
Equal('status', 'draft'),
Equal('user_id', str(get_request().user.id)),
]
),
]
)
]
criterias.extend(status_criterias)
name_id = query_parameters.get('filter-user-uuid')
if name_id:
nameid_users = get_publisher().user_class.get_users_with_name_identifier(name_id)
if nameid_users:
criterias.append(Equal('user_id', str(nameid_users[0].id)))
if get_request().form.get('submission_channel'):
if get_request().form.get('submission_channel') == 'web':
criterias.append(Null('submission_channel'))
else:
criterias.append(Equal('submission_channel', get_request().form.get('submission_channel')))
category_slugs = []
category_ids = []
if get_request().form:
prefix = 'category_ids$element'
category_slugs = (get_request().form.get('category_slugs') or '').split(',')
category_slugs = [c.strip() for c in category_slugs if c.strip()]
if category_slugs:
category_ids = [c.id for c in Category.select() if c.url_name in category_slugs]
else:
category_ids = [
get_request().form.get(k) for k in get_request().form.keys() if k.startswith(prefix)
]
category_ids = [v for v in category_ids if v]
if category_slugs or category_ids:
criterias.append(Contains('category_id', category_ids))
if get_request().form.get('q'):
criterias.append(FtsMatch(get_request().form.get('q')))
return criterias
def empty_site_message(self, title):
r = TemplateIO(html=True)
r += htmltext('<div class="top-title">')
r += htmltext('<h2>%s</h2>') % title
r += htmltext('</div>')
r += htmltext('<div class="big-msg-info">')
r += htmltext('<p>%s</p>') % _('This site is currently empty. It is required to first add forms.')
r += htmltext('</div>')
return r.getvalue()
def listing(self):
if not get_publisher().is_using_postgresql():
raise errors.TraversalError()
get_response().add_javascript(['wcs.listing.js'])
from wcs import sql
html_top('management', _('Management'))
if FormDef.count() == 0:
return self.empty_site_message(_('Global View'))
limit = misc.get_int_or_400(
get_request().form.get('limit', get_publisher().get_site_option('default-page-size') or 20)
)
offset = misc.get_int_or_400(get_request().form.get('offset', 0))
order_by = misc.get_order_by_or_400(
get_request().form.get(
'order_by', get_publisher().get_site_option('default-sort-order') or '-receipt_time'
)
)
criterias = self.get_global_listing_criterias()
criterias.append(Null('anonymised')) # exclude anonymised forms
total_count = sql.AnyFormData.count(criterias)
if offset > total_count:
get_request().form['offset'] = '0'
return redirect('listing?' + urllib.parse.urlencode(get_request().form))
formdatas = sql.AnyFormData.select(criterias, order_by=order_by, limit=limit, offset=offset)
include_submission_channel = bool(get_publisher().get_site_option('welco_url', 'variables'))
r = TemplateIO(html=True)
r += htmltext('<table id="listing" class="main">')
r += htmltext('<thead><tr>')
r += htmltext('<th data-field-sort-key="criticality_level"><span></span></th>')
if include_submission_channel:
r += htmltext('<th data-field-sort-key="submission_channel"><span>%s</span></th>') % _('Channel')
r += htmltext('<th data-field-sort-key="formdef_name"><span>%s</span></th>') % _('Form')
r += htmltext('<th><span>%s</span></th>') % _('Reference')
r += htmltext('<th data-field-sort-key="receipt_time"><span>%s</span></th>') % _('Created')
r += htmltext('<th data-field-sort-key="last_update_time"><span>%s</span></th>') % _('Last Modified')
r += htmltext('<th data-field-sort-key="user_name"><span>%s</span></th>') % C_('frontoffice|User')
r += htmltext('<th class="nosort"><span>%s</span></th>') % _('Status')
r += htmltext('</tr></thead>')
r += htmltext('<tbody>')
workflows = {}
session = get_session()
visited_objects = session.get_visited_objects(exclude_user=session.user)
for formdata in formdatas:
if formdata.formdef.workflow_id not in workflows:
workflows[formdata.formdef.workflow_id] = formdata.formdef.workflow
classes = ['status-%s-%s' % (formdata.formdef.workflow.id, formdata.status)]
if formdata.get_object_key() in visited_objects:
classes.append('advisory-lock')
if formdata.backoffice_submission:
classes.append('backoffice-submission')
style = ''
try:
level = formdata.get_criticality_level_object()
except IndexError:
pass
else:
classes.append('criticality-level')
style = 'style="border-left-color: #%s;"' % level.colour
r += htmltext(
'<tr class="%s" data-link="%s">' % (' '.join(classes), formdata.get_url(backoffice=True))
)
r += htmltext('<td %s></td>' % style) # lock
if include_submission_channel:
r += htmltext('<td>%s</td>') % formdata.get_submission_channel_label()
r += htmltext('<td>%s') % formdata.formdef.name
if formdata.default_digest:
r += htmltext(' <small>%s</small>') % formdata.default_digest
r += htmltext('</td>')
r += htmltext('<td><a href="%s">%s</a></td>') % (
formdata.get_url(backoffice=True),
formdata.get_display_id(),
)
r += htmltext('<td class="cell-time">%s</td>') % misc.localstrftime(formdata.receipt_time)
r += htmltext('<td class="cell-time">%s</td>') % misc.localstrftime(formdata.last_update_time)
value = formdata.get_user_label()
if value:
r += htmltext('<td class="cell-user">%s</td>') % value
else:
r += htmltext('<td class="cell-user cell-no-user">-</td>')
r += htmltext('<td class="cell-status">%s</td>') % formdata.get_status_label()
r += htmltext('</tr>\n')
if workflows:
colours = []
for workflow in workflows.values():
for status in workflow.possible_status:
if status.colour and status.colour != 'FFFFFF':
fg_colour = misc.get_foreground_colour(status.colour)
colours.append((workflow.id, status.id, status.colour, fg_colour))
if colours:
r += htmltext('<style>')
for workflow_id, status_id, bg_colour, fg_colour in colours:
r += htmltext(
'tr.status-%s-wf-%s td.cell-status { '
'background-color: #%s !important; color: %s !important; }\n'
% (workflow_id, status_id, bg_colour, fg_colour)
)
r += htmltext('</style>')
r += htmltext('</tbody></table>')
if (offset > 0) or (total_count > limit > 0):
r += pagination_links(offset, limit, total_count)
if get_request().form.get('ajax') == 'true':
get_response().filter = {'raw': True}
return r.getvalue()
get_response().filter['sidebar'] = self.get_global_listing_sidebar(
limit=limit, offset=offset, order_by=order_by
)
rt = TemplateIO(html=True)
rt += htmltext('<div id="appbar">')
rt += htmltext('<h2>%s</h2>') % _('Global View')
rt += htmltext('<span class="actions">')
rt += htmltext('<a href="forms">%s</a>') % _('Forms View')
for formdef in FormDef.select(lightweight=True):
if formdef.geolocations:
rt += htmltext(' <a href="map">%s</a>') % _('Map View')
break
rt += htmltext('</span>')
rt += htmltext('</div>')
rt += get_session().display_message()
rt += r.getvalue()
r = rt
return rt.getvalue()
def count(self):
if not get_publisher().is_using_postgresql():
raise errors.TraversalError()
if FormDef.count() == 0:
return misc.json_response({'count': 0})
from wcs import sql
criterias = self.get_global_listing_criterias()
count = sql.AnyFormData.count(criterias)
return misc.json_response({'count': count})
def geojson(self):
from wcs import sql
criterias = self.get_global_listing_criterias()
formdatas = sql.AnyFormData.select(criterias + [NotNull('geoloc_base_x')])
fields = [
FakeField('name', 'display_name', _('Name')),
FakeField('status', 'status', _('Status')),
]
get_response().set_content_type('application/json')
return json.dumps(geojson_formdatas(formdatas, fields=fields), cls=misc.JSONEncoder)
def map(self):
if not get_publisher().is_using_postgresql():
raise errors.TraversalError()
get_response().add_javascript(['wcs.listing.js', 'qommon.map.js'])
html_top('management', _('Global Map'))
r = TemplateIO(html=True)
get_response().breadcrumb.append(('map', _('Global Map')))
attrs = {
'class': 'qommon-map',
'id': 'backoffice-map',
'data-readonly': True,
'data-geojson-url': '%s/geojson?%s' % (get_request().get_url(1), get_request().get_query()),
}
attrs.update(get_publisher().get_map_attributes())
get_response().filter['sidebar'] = self.get_global_listing_sidebar()
r += htmltext('<h2>%s</h2>') % _('Global Map')
r += htmltext('<div %s></div>' % ' '.join(['%s="%s"' % x for x in attrs.items()]))
return r.getvalue()
def _q_lookup(self, component):
return FormPage(component)
class FormPage(Directory):
_q_exports = [
'',
'live',
'csv',
'stats',
'ods',
'json',
'export',
'map',
'geojson',
('export-spreadsheet', 'export_spreadsheet'),
('filter-options', 'filter_options'),
('save-view', 'save_view'),
('delete-view', 'delete_view'),
]
_view = None
default_view = None
use_default_view = False
admin_permission = 'forms'
formdef_class = FormDef
search_label = _('Search in form content')
WCS_SYNC_EXPORT_LIMIT = 100 # Arbitrary threshold
def __init__(self, component=None, formdef=None, view=None, update_breadcrumbs=True):
self.view_type = None
if component:
try:
self.formdef = self.formdef_class.get_by_urlname(component)
except KeyError:
raise errors.TraversalError()
if update_breadcrumbs:
get_response().breadcrumb.append((component + '/', self.formdef.name))
else:
self.formdef = formdef
self._view = view
if update_breadcrumbs:
get_response().breadcrumb.append((view.get_url_slug() + '/', view.title))
self.set_default_view()
def set_default_view(self):
if not get_request():
return
custom_views = list(self.get_custom_views())
# search for first default user custom view
for view in custom_views:
if view.visibility != 'owner':
continue
if not view.is_default:
continue
self.default_view = view
return
# default user custom view not found, search in 'any' custom views
for view in custom_views:
if view.visibility != 'any':
continue
if not view.is_default:
continue
self.default_view = view
return
@property
def view(self):
view = self._view
if self.use_default_view:
view = view or self.default_view
return view
def check_access(self, api_name=None):
session = get_session()
user = get_request().user
if user is None and get_publisher().user_class.count() == 0:
user = get_publisher().user_class()
user.is_admin = True
if not user:
raise errors.AccessUnauthorizedError()
if not user.is_admin and not self.formdef.is_of_concern_for_user(user):
if session.user:
raise errors.AccessForbiddenError()
raise errors.AccessUnauthorizedError()
def get_custom_views(self, criterias=None):
for view in get_publisher().custom_view_class.select(clause=criterias):
if view.match(get_request().user, self.formdef):
yield view
def get_formdata_sidebar_actions(self, qs=''):
r = TemplateIO(html=True)
if not self.formdef.category or self.formdef.category.has_permission('export', get_request().user):
r += htmltext(
' <li><a rel="popup" data-base-href="export-spreadsheet" data-autoclose-dialog="true" '
'href="export-spreadsheet%s">%s</a></li>'
) % (
qs,
_('Export a Spreadsheet'),
)
if self.formdef.geolocations:
r += htmltext(' <li><a data-base-href="map" href="map%s">%s</a></li>') % (qs, _('Plot on a Map'))
if 'stats' in self._q_exports and (
not self.formdef.category
or self.formdef.category.has_permission('statistics', get_request().user)
):
r += htmltext(' <li class="stats"><a href="stats">%s</a></li>') % _('Statistics')
return r.getvalue()
def get_formdata_sidebar(self, qs=''):
r = TemplateIO(html=True)
r += htmltext('<ul id="sidebar-actions">')
r += self.get_formdata_sidebar_actions(qs=qs)
r += htmltext('</ul>')
views = list(self.get_custom_views())
if views:
r += htmltext('<h3>%s</h3>') % _('Custom Views')
r += htmltext('<ul id="sidebar-custom-views">')
view_type = 'map' if self.view_type == 'map' else ''
for view in sorted(views, key=lambda x: getattr(x, 'title')):
if self._view:
active = bool(self._view.get_url_slug() == view.get_url_slug())
r += htmltext('<li class="active">' if active else '<li>')
r += htmltext('<a href="../%s/%s">%s</a>') % (view.get_url_slug(), view_type, view.title)
else:
r += htmltext('<li><a href="%s/%s">%s</a>') % (view.get_url_slug(), view_type, view.title)
if view.visibility == 'datasource':
r += htmltext(' <span class="as-data-source">(%s)</span>') % _('for data sources')
elif self.default_view and view.id == self.default_view.id:
r += htmltext(' <span class="default-custom-view">(%s)</span>') % _('default')
r += htmltext('</li>')
r += htmltext('</ul>')
return r.getvalue()
def get_default_filters(self, mode):
if self.view:
return self.view.get_default_filters()
if mode == 'listing':
# enable status filter by default
return ('status',)
if mode == 'stats':
# enable period filters by default
return ('start', 'end')
return ()
def get_item_filter_options(self, filter_field, selected_filter, criterias):
criterias = (criterias or [])[:]
# remove potential filter on self (Equal for item, Intersects for items)
criterias = [
x
for x in criterias
if not (isinstance(x, (Equal, Intersects)) and x.attribute == 'f%s' % filter_field.id)
]
# apply other filters
criterias.append(Null('anonymised'))
if selected_filter == 'all':
criterias.append(NotEqual('status', 'draft'))
elif selected_filter in ('waiting', 'pending'):
statuses = ['wf-%s' % x.id for x in self.formdef.workflow.get_not_endpoint_status()]
criterias.append(Contains('status', statuses))
if selected_filter == 'waiting':
user = get_request().user
user_roles = [logged_users_role().id] + user.get_roles()
criterias.append(Intersects('actions_roles_array', user_roles))
elif selected_filter == 'done':
statuses = ['wf-%s' % x.id for x in self.formdef.workflow.get_endpoint_status()]
criterias.append(Contains('status', statuses))
else:
criterias.append(Equal('status', 'wf-%s' % selected_filter))
from wcs import sql
criterias.append(NotNull(sql.get_field_id(filter_field)))
options = self.formdef.data_class().select_distinct(
[sql.get_field_id(filter_field), '%s_display' % sql.get_field_id(filter_field)], clause=criterias
)
if filter_field.type == 'items':
# unnest key/values
exploded_options = {}
for option_keys, option_label in options:
if option_keys and option_label:
for option_key, option_label in zip(option_keys, option_label.split(', ')):
exploded_options[option_key] = option_label
options = list(sorted(exploded_options.items(), key=lambda x: x[1]))
options = [(force_str(x), force_str(y)) for x, y in options if x and y]
return options
def filter_options(self):
get_request().is_json_marker = True
field_id = get_request().form.get('filter_field_id')
for filter_field in self.get_formdef_fields():
if filter_field.id == field_id:
break
else:
raise errors.TraversalError()
selected_filter = self.get_filter_from_query()
criterias = self.get_criterias_from_query()
options = self.get_item_filter_options(filter_field, selected_filter, criterias)
if get_request().form.get('_search'): # select2
term = get_request().form.get('_search')
if term:
options = [x for x in options if term.lower() in x[1].lower()]
options = options[:15]
if self.view and self.view.visibility == 'datasource':
options.append(('{}', _('custom value')))
get_response().set_content_type('application/json')
return json.dumps(
{'err': 0, 'data': [{'id': x[0], 'text': x[1]} for x in options]}, cls=misc.JSONEncoder
)
def get_filterable_field_types(self):
types = ['string', 'email', 'item', 'bool', 'items', 'period-date', 'user-id', 'submission-agent-id']
if get_publisher().is_using_postgresql():
types.append('date')
return types
def get_filter_sidebar(self, selected_filter=None, mode='listing', query=None, criterias=None):
r = TemplateIO(html=True)
waitpoint_status = self.formdef.workflow.get_waitpoint_status()
fake_fields = [
FakeField('start', 'period-date', _('Start')),
FakeField('end', 'period-date', _('End')),
FakeField('user', 'user-id', _('User')),
FakeField('submission-agent', 'submission-agent-id', _('Submission Agent'), addable=False),
]
default_filters = self.get_default_filters(mode)
filter_fields = []
for field in fake_fields + list(self.get_formdef_fields()):
field.enabled = False
if field.type not in self.get_filterable_field_types() + ['status']:
continue
if field.type == 'status' and not waitpoint_status:
continue
filter_fields.append(field)
if get_request().form:
field.enabled = ('filter-%s' % field.id in get_request().form) or (
'filter-%s' % field.varname in get_request().form
)
if 'filter-%s' % field.varname in get_request().form and (
'filter-%s-value' % field.varname not in get_request().form
):
# if ?filter-<varname>= is used, take the value and put it
# into filter-<field id>-value so it is used to fill the
# fields.
get_request().form['filter-%s-value' % field.id] = get_request().form.get(
'filter-%s' % field.varname
)
if (
field.varname in ('start', 'end', 'user', 'submission-agent')
and get_request().form['filter-%s-value' % field.id] == 'on'
):
# reset start/end to an empty value when they're just
# being enabled
get_request().form['filter-%s-value' % field.id] = ''
if not field.enabled and self.view and get_request().form.get('keep-view-filters'):
# keep-view-filters=on is used to initialize page with
# filters from both the custom view and the query string.
field.enabled = field.id in default_filters
else:
field.enabled = field.id in default_filters
if not self.view and field.type in ('item', 'items'):
field.enabled = field.in_filters
r += htmltext('<h3><span>%s</span>') % _('Current view')
r += htmltext('<span class="change">(')
r += htmltext('<a id="filter-settings">%s</a>') % _('filters')
if self.view_type in ('table', 'map'):
if self.view_type == 'table':
columns_settings_labels = (_('Columns Settings'), _('columns'))
elif self.view_type == 'map':
columns_settings_labels = (_('Marker Settings'), _('markers'))
r += htmltext(' - <a id="columns-settings" title="%s">%s</a>') % columns_settings_labels
r += htmltext(')</span></h3>')
filters_dict = {}
if self.view:
filters_dict.update(self.view.get_filters_dict())
filters_dict.update(get_request().form)
for filter_field in filter_fields:
if not filter_field.enabled:
continue
filter_field_key = 'filter-%s-value' % filter_field.id
filter_field_value = filters_dict.get(filter_field_key)
if filter_field.type == 'status':
r += htmltext('<div class="widget">')
r += htmltext('<div class="title">%s</div>') % _('Status to display')
r += htmltext('<div class="content">')
r += htmltext('<select name="filter">')
filters = [
('waiting', _('Waiting for an action'), None),
('all', _('All'), None),
('pending', C_('formdata|Open'), None),
('done', _('Done'), None),
]
for status in waitpoint_status:
filters.append((status.id, status.name, status.colour))
for filter_id, filter_label, filter_colour in filters:
if filter_id == selected_filter:
selected = ' selected="selected"'
else:
selected = ''
style = ''
if filter_colour and filter_colour != 'FFFFFF':
fg_colour = misc.get_foreground_colour(filter_colour)
style = 'style="background: #%s; color: %s;"' % (filter_colour, fg_colour)
r += htmltext('<option value="%s"%s %s>' % (filter_id, selected, style))
r += htmltext('%s</option>') % filter_label
r += htmltext('</select>')
r += htmltext('</div>')
r += htmltext('</div>')
elif filter_field.type == 'period-date':
r += DateWidget(
filter_field_key, title=filter_field.label, value=filter_field_value, render_br=False
).render()
elif filter_field.type == 'user-id':
options = [
('', _('None'), ''),
('__current__', _('Current user'), '__current__'),
]
if filter_field_value and filter_field_value != '__current__':
try:
filtered_user = get_publisher().user_class.get(filter_field_value)
except KeyError:
filtered_user = None
filtered_user_value = filtered_user.display_name if filtered_user else _('Unknown')
options += [(filter_field_value, filtered_user_value, filter_field_value)]
r += SingleSelectWidget(
filter_field_key,
title=filter_field.label,
options=options,
value=filter_field_value,
render_br=False,
).render()
elif filter_field.type == 'submission-agent-id':
r += HiddenWidget(filter_field_key, value=filter_field_value).render()
if filter_field_value:
filtered_user = get_publisher().user_class.get(filter_field_value, ignore_errors=True)
widget = StringWidget(
'_' + filter_field_key,
title=filter_field.label,
value=filtered_user.display_name if filtered_user else _('Unknown'),
readonly=True,
render_br=False,
)
widget._parsed = True # make sure value is not replaced by request query
r += widget.render()
elif filter_field.type in ('item', 'items'):
filter_field.required = False
if get_publisher().is_using_postgresql():
# Get options from existing formdatas.
# This allows for options that don't appear anymore in the
# data source to be listed (for example because the field
# is using a parametrized URL depending on unavailable
# variables, or simply returning different results now).
display_mode = 'select'
if filter_field.type == 'item' and filter_field.get_display_mode() == 'autocomplete':
display_mode = 'select2'
if display_mode == 'select':
options = self.get_item_filter_options(filter_field, selected_filter, criterias)
options = [(x[0], x[1], x[0]) for x in options]
options.insert(0, (None, '', ''))
attrs = {'data-refresh-options': str(filter_field.id)}
if self.view and self.view.visibility == 'datasource':
options.append(('{}', _('custom value'), '{}'))
if filter_field_value and filter_field_value not in [x[0] for x in options]:
options.append((filter_field_value, filter_field_value, filter_field_value))
attrs['data-allow-template'] = 'true'
else:
current_filter = filters_dict.get('filter-%s-value' % filter_field.id)
options = [(current_filter, '', current_filter or '')]
attrs = {'data-remote-options': str(filter_field.id)}
get_response().add_javascript(
['jquery.js', '../../i18n.js', 'qommon.forms.js', 'select2.js']
)
get_response().add_css_include('../js/select2/select2.css')
r += SingleSelectWidget(
filter_field_key,
title=filter_field.label,
options=options,
value=filter_field_value,
render_br=False,
attrs=attrs,
).render()
else:
# In pickle environments, get options from data source
options = filter_field.get_options()
if options:
if len(options[0]) == 2:
options = [(x[0], x[1], x[0]) for x in options]
options.insert(0, (None, '', ''))
r += SingleSelectWidget(
filter_field_key,
title=filter_field.label,
options=options,
value=filter_field_value,
render_br=False,
).render()
else:
# and fall back on a string widget if there are none.
r += StringWidget(
filter_field_key,
title=filter_field.label,
value=filter_field_value,
render_br=False,
).render()
elif filter_field.type == 'bool':
options = [(None, '', ''), (True, _('Yes'), 'true'), (False, _('No'), 'false')]
if filter_field_value == 'true':
filter_field_value = True
elif filter_field_value == 'false':
filter_field_value = False
r += SingleSelectWidget(
filter_field_key,
title=filter_field.label,
options=options,
value=filter_field_value,
render_br=False,
).render()
elif filter_field.type in ('string', 'email'):
r += StringWidget(
filter_field_key, title=filter_field.label, value=filter_field_value, render_br=False
).render()
elif filter_field.type == 'date':
r += DateWidget(
filter_field_key, title=filter_field.label, value=filter_field_value, render_br=False
).render()
# field filter dialog content
r += htmltext('<div style="display: none;">')
r += htmltext('<ul id="field-filter">')
for field in filter_fields:
addable = getattr(field, 'addable', True)
r += htmltext('<li %s>') % ('' if addable else 'hidden')
r += htmltext('<input type="checkbox" name="filter-%s"') % field.id
if field.enabled:
r += htmltext(' checked="checked"')
r += htmltext(' id="fields-filter-%s"') % field.id
r += htmltext('/>')
r += htmltext('<label for="fields-filter-%s">%s</label>') % (
field.id,
misc.ellipsize(field.label, 70),
)
r += htmltext('</li>')
r += htmltext('</ul>')
r += htmltext('</div>')
return r.getvalue()
def get_fields_sidebar(
self, selected_filter, fields, offset=None, limit=None, order_by=None, query=None, criterias=None
):
get_response().add_javascript(['wcs.listing.js'])
r = TemplateIO(html=True)
r += htmltext('<form id="listing-settings" action=".">')
if offset or limit:
if not offset:
offset = 0
r += htmltext('<input type="hidden" name="offset" value="%s"/>') % offset
if limit:
r += htmltext('<input type="hidden" name="limit" value="%s"/>') % limit
if get_publisher().is_using_postgresql():
if order_by is None:
order_by = get_publisher().get_site_option('default-sort-order') or '-receipt_time'
r += htmltext('<input type="hidden" name="order_by" value="%s"/>') % order_by
if get_publisher().is_using_postgresql():
r += htmltext('<h3>%s</h3>') % self.search_label
if get_request().form.get('q'):
q = force_text(get_request().form.get('q'))
r += htmltext('<input class="inline-input" name="q" value="%s">') % force_str(q)
else:
r += htmltext('<input class="inline-input" name="q">')
r += htmltext('<input type="submit" class="side-button" value="%s"/>') % _('Search')
r += self.get_filter_sidebar(selected_filter=selected_filter, query=query, criterias=criterias)
r += htmltext('<button class="refresh" hidden>%s</button>') % _('Refresh')
if self.view_type in ('table', 'map'):
# column settings dialog content
r += htmltext('<div style="display: none;">')
r += htmltext('<ul id="columns-filter" class="objects-list columns-filter">')
column_order = []
field_ids = [x.id for x in fields]
def get_column_position(x):
if x.id in field_ids:
return field_ids.index(x.id)
return 9999
seen_parents = set()
for field in sorted(self.get_formdef_fields(), key=get_column_position):
if not hasattr(field, 'get_view_value'):
continue
classnames = ''
attrs = ''
if isinstance(field, RelatedField):
classnames = 'related-field'
if field.parent_field_id in seen_parents:
classnames += ' collapsed'
attrs = 'data-relation-attr="%s"' % field.parent_field_id
elif getattr(field, 'has_relations', False):
classnames = 'has-relations-field'
attrs = 'data-field-id="%s"' % field.id
seen_parents.add(field.id)
r += htmltext('<li class="%s" %s><span class="handle">⣿</span>' % (classnames, attrs))
r += htmltext('<label><input type="checkbox" name="%s"') % field.id
if field.id in field_ids:
r += htmltext(' checked="checked"')
r += htmltext('/>')
r += htmltext('%s</label>') % misc.ellipsize(field.label, 70)
if getattr(field, 'has_relations', False):
r += htmltext('<button class="expand-relations"></button>')
r += htmltext('</li>')
column_order.append(str(field.id))
r += htmltext('</ul>')
r += htmltext('</div>')
r += htmltext('<input type="hidden" name="columns-order" value="%s">' % ','.join(column_order))
r += htmltext('</form>')
r += self.get_custom_view_form().render()
r += htmltext('<button id="save-view">%s</button>') % _('Save')
if self.can_delete_view():
r += htmltext(' <a data-popup id="delete-view" href="./delete-view" class="button">%s</a>') % _(
'Delete'
)
return r.getvalue()
def get_custom_view_form(self):
form = Form(method='post', id='save-custom-view', hidden='hidden', action='save-view')
form.add(HiddenWidget, 'qs', value=get_request().get_query())
form.add(
StringWidget,
'title',
title=_('Title'),
required=True,
value=self.view.title if self.view else None,
)
if get_publisher().get_backoffice_root().is_accessible(self.admin_permission):
# admins can create views accessible to everyone
options = [
('owner', _('to me only'), 'owner'),
('any', _('to any users'), 'any'),
]
if isinstance(self.formdef, CardDef) and self.formdef.default_digest_template:
options.append(('datasource', _('as data source'), 'datasource'))
form.add(
RadiobuttonsWidget,
'visibility',
title=_('Visibility'),
value=self.view.visibility if self.view else 'owner',
options=options,
attrs={'data-dynamic-display-parent': 'true'},
extra_css_class='widget-inline-radio',
)
form.add(
CheckboxWidget,
'is_default',
title=_('Set as default view'),
value=self.view.is_default if self.view else False,
attrs={
'data-dynamic-display-child-of': 'visibility',
'data-dynamic-display-value-in': 'owner|any',
},
)
if isinstance(self.formdef, CardDef):
digest_template = None
if self.view:
templates = self.formdef.digest_templates or {}
digest_template = templates.get('custom-view:%s' % self.view.get_url_slug())
form.add(
StringWidget,
'digest_template',
title=_('Digest'),
value=digest_template or self.formdef.default_digest_template,
size=50,
attrs={
'data-dynamic-display-child-of': 'visibility',
'data-dynamic-display-value-in': 'datasource|any',
},
)
else:
form.add(
CheckboxWidget,
'is_default',
title=_('Set as default view'),
value=self.view.is_default if self.view else False,
)
if self.view and (
self.view.user_id == get_request().user.id
or get_publisher().get_backoffice_root().is_accessible(self.admin_permission)
):
form.add(CheckboxWidget, 'update', title=_('Update existing view settings'), value=True)
form.add_submit('submit', _('Save View'))
form.add_submit('cancel', _('Cancel'))
return form
def save_view(self):
form = self.get_custom_view_form()
if form.get_widget('update') and form.get_widget('update').parse():
custom_view = self.view
else:
custom_view = get_publisher().custom_view_class()
custom_view.title = form.get_widget('title').parse()
if not custom_view.title:
get_session().message = ('error', _('Missing title.'))
return redirect('.')
custom_view.user = get_request().user
custom_view.formdef = self.formdef
custom_view.set_from_qs(form.get_widget('qs').parse())
if not custom_view.columns['list']:
get_session().message = ('error', _('Views must have at least one column.'))
return redirect('.')
if form.get_widget('is_default'):
custom_view.is_default = form.get_widget('is_default').parse()
if form.get_widget('visibility'):
custom_view.visibility = form.get_widget('visibility').parse()
if custom_view.visibility == 'datasource':
custom_view.is_default = False
custom_view.store()
if form.get_widget('digest_template'):
if not self.formdef.digest_templates:
self.formdef.digest_templates = {}
old_value = self.formdef.digest_templates.get('custom-view:%s' % custom_view.get_url_slug())
new_value = form.get_widget('digest_template').parse()
if old_value != new_value:
self.formdef.digest_templates[
'custom-view:%s' % custom_view.get_url_slug()
] = form.get_widget('digest_template').parse()
self.formdef.store()
if self.formdef.data_class().count():
get_response().add_after_job(UpdateDigestAfterJob(formdef=self.formdef))
if custom_view.is_default and custom_view.visibility != 'datasource':
# need to clean other views to have only one default per owner/any visibility
for view in self.get_custom_views():
if view.id == custom_view.id:
continue
if custom_view.visibility == view.visibility and view.is_default:
view.is_default = False
view.store()
if self.view:
return redirect('../' + custom_view.get_url_slug() + '/')
else:
return redirect(custom_view.get_url_slug() + '/')
def can_delete_view(self):
if not self.view:
return False
if str(self.view.user_id) == str(get_request().user.id):
return True
return get_publisher().get_backoffice_root().is_accessible(self.admin_permission)
def delete_view(self):
if not self.can_delete_view():
raise errors.AccessForbiddenError()
form = Form(enctype='multipart/form-data')
form.widgets.append(
HtmlWidget('<p>%s</p>' % _('You are about to remove the \"%s\" custom view.') % self.view.title)
)
if self.view.visibility == 'any':
form.widgets.append(
HtmlWidget(
'<div class="warningnotice"<p>%s</p></div>'
% _('Beware this view is available to all users, and will thus be removed for everyone.')
)
)
form.add_submit('delete', _('Delete'))
form.add_submit('cancel', _('Cancel'))
if form.get_widget('cancel').parse():
return redirect('.')
if not form.is_submitted() or form.has_errors():
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % (_('Delete Custom View'))
r += form.render()
return r.getvalue()
else:
self.view.remove_self()
return redirect('..')
def get_formdef_fields(self):
yield FakeField('id', 'id', _('Number'))
if get_publisher().get_site_option('welco_url', 'variables'):
yield FakeField('submission_channel', 'submission_channel', _('Channel'))
if self.formdef.backoffice_submission_roles:
yield FakeField('submission_agent', 'submission_agent', _('Submission By'))
yield FakeField('time', 'time', _('Created'))
yield FakeField('last_update_time', 'last_update_time', _('Last Modified'))
# user fields
yield FakeField('user-label', 'user-label', _('User Label'))
if get_publisher().is_using_postgresql():
for field in get_publisher().user_class.get_fields():
if not hasattr(field, 'get_view_value'):
continue
field.has_relations = True
yield UserRelatedField(field)
for field in self.formdef.get_all_fields():
yield field
if not get_publisher().is_using_postgresql():
continue
if not (
field.type == 'item'
and field.data_source
and field.data_source.get('type', '').startswith('carddef:')
):
continue
try:
carddef = CardDef.get_by_urlname(field.data_source['type'][8:])
except KeyError:
continue
for card_field in carddef.get_all_fields():
if not hasattr(card_field, 'get_view_value'):
continue
field.has_relations = True
yield RelatedField(carddef, card_field, field)
yield FakeField('status', 'status', _('Status'))
yield FakeField('anonymised', 'anonymised', _('Anonymised'))
def get_default_columns(self):
if self.view:
field_ids = self.view.get_columns()
else:
field_ids = ['id', 'time', 'last_update_time', 'user-label']
for field in self.formdef.get_all_fields():
if hasattr(field, 'get_view_value') and field.include_in_listing:
field_ids.append(field.id)
field_ids.append('status')
return field_ids
def get_fields_from_query(self, ignore_form=False):
field_ids = [x for x in get_request().form.keys()]
if not field_ids or ignore_form:
field_ids = self.get_default_columns()
fields = []
for field in self.get_formdef_fields():
if field.id in field_ids:
fields.append(field)
if 'columns-order' in get_request().form or self.view:
if ignore_form or 'columns-order' not in get_request().form:
field_order = field_ids
else:
field_order = get_request().form['columns-order'].split(',')
def field_position(x):
if x.id in field_order:
return field_order.index(x.id)
return 9999
fields.sort(key=field_position)
if not fields and not ignore_form:
return self.get_fields_from_query(ignore_form=True)
return fields
def get_filter_from_query(self, default='waiting'):
if 'filter' in get_request().form:
return get_request().form['filter']
if self.view:
view_filter = self.view.get_filter()
if view_filter:
return view_filter
if self.formdef.workflow.possible_status:
return default
return 'all'
def get_criterias_from_query(self):
query_overrides = get_request().form
return self.get_view_criterias(query_overrides, request=get_request())
def get_view_criterias(self, query_overrides=None, request=None):
fake_fields = [
FakeField('start', 'period-date', _('Start')),
FakeField('end', 'period-date', _('End')),
FakeField('start-mtime', 'period-date', _('Start (modification time)')),
FakeField('end-mtime', 'period-date', _('End (modification time)')),
FakeField('user', 'user-id', _('User')),
FakeField('submission-agent', 'submission-agent-id', _('Submission Agent')),
]
criterias = []
filters_dict = {}
if self.view:
filters_dict.update(self.view.get_filters_dict() or {})
filters_dict.update(query_overrides or {})
if request and request.form:
request_form = request.form
else:
request_form = {}
for filter_field in fake_fields + list(self.get_formdef_fields()):
if filter_field.type not in self.get_filterable_field_types():
continue
filter_field_key = None
if filter_field.varname:
# if this is a field with a varname and filter-%(varname)s is
# present in the query string, enable this filter.
if filters_dict.get('filter-%s' % filter_field.varname):
filter_field_key = 'filter-%s' % filter_field.varname
if filter_field.type == 'user-id':
# convert uuid based filter into local id filter
name_id = filters_dict.get('filter-user-uuid')
if name_id:
nameid_users = get_publisher().user_class.get_users_with_name_identifier(name_id)
request_form['filter-user'] = filters_dict['filter-user'] = 'on'
if nameid_users:
filters_dict['filter-user-value'] = str(nameid_users[0].id)
request_form['filter-user-value'] = filters_dict['filter-user-value']
else:
filters_dict['filter-user-value'] = (
'__current__' if name_id == '__current__' else '-1'
)
request_form['filter-user-value'] = (
'__current__' if name_id == '__current__' else '-1'
)
if filter_field.type == 'submission-agent-id':
# convert uuid based filter into local id filter
name_id = filters_dict.get('filter-submission-agent-uuid')
if name_id:
nameid_users = get_publisher().user_class.get_users_with_name_identifier(name_id)
request_form['filter-submission-agent'] = filters_dict['filter-submission-agent'] = 'on'
if nameid_users:
filters_dict['filter-submission-agent-value'] = str(nameid_users[0].id)
request_form['filter-submission-agent-value'] = filters_dict[
'filter-submission-agent-value'
]
else:
filters_dict['filter-submission-agent-value'] = '-1'
request_form['filter-submission-agent-value'] = '-1'
if filters_dict.get('filter-%s' % filter_field.id):
# if there's a filter-%(id)s, it is used to enable the actual
# filter, and the value will be found in filter-%s-value.
filter_field_key = 'filter-%s-value' % filter_field.id
if not filter_field_key:
# if there's not known filter key, skip.
continue
filter_field_value = filters_dict.get(filter_field_key)
if not filter_field_value:
continue
if filter_field.type == 'period-date':
try:
filter_date_value = misc.get_as_datetime(filter_field_value).timetuple()
except ValueError:
continue
if filter_field.id == 'start':
criterias.append(GreaterOrEqual('receipt_time', filter_date_value))
elif filter_field.id == 'end':
criterias.append(LessOrEqual('receipt_time', filter_date_value))
elif filter_field.id == 'start-mtime':
criterias.append(GreaterOrEqual('last_update_time', filter_date_value))
elif filter_field.id == 'end-mtime':
criterias.append(LessOrEqual('last_update_time', filter_date_value))
criterias[-1]._label = '%s: %s' % (filter_field.label, filter_field_value)
elif filter_field.type == 'user-id':
if filter_field_value == '__current__' and get_request().user:
filter_field_value = str(get_request().user.id)
criterias.append(Equal('user_id', filter_field_value))
elif filter_field.type == 'submission-agent-id':
criterias.append(Equal('submission_agent_id', filter_field_value))
elif filter_field.type in ('item', 'items') and filter_field_value is not None:
if filter_field.type == 'item':
criterias.append(Equal('f%s' % filter_field.id, filter_field_value))
elif filter_field.type == 'items':
criterias.append(Intersects('f%s' % filter_field.id, [filter_field_value]))
field_options = filter_field.get_options()
if field_options and type(field_options[0]) in (list, tuple):
for option in field_options:
if filter_field_value in (option[0], option[-1]):
filter_field_value = option[1]
break
criterias[-1]._label = '%s: %s' % (filter_field.label, filter_field_value)
elif filter_field.type == 'bool' and filter_field_value is not None:
if filter_field_value == 'true':
criterias.append(Equal('f%s' % filter_field.id, True))
elif filter_field_value == 'false':
criterias.append(Equal('f%s' % filter_field.id, False))
elif filter_field.type in ('string', 'email') and filter_field_value is not None:
criterias.append(Equal('f%s' % filter_field.id, filter_field_value))
elif filter_field.type == 'date' and filter_field_value is not None:
try:
filter_field_value = misc.get_as_datetime(filter_field_value).date()
except ValueError:
pass
else:
criterias.append(Equal('f%s' % filter_field.id, filter_field_value))
return criterias
def listing_top_actions(self):
return ''
@classmethod
def get_multi_actions(cls, formdef, user, status_filter):
global_actions = formdef.workflow.get_global_manual_actions()
if status_filter not in ('open', 'waiting', 'done', 'all'):
# when the listing is filtered on a specific status, include
# manual jumps with identifiers
try:
status = formdef.workflow.get_status(status_filter)
except KeyError:
status = None
else:
global_actions.extend(status.get_status_manual_actions())
mass_actions = []
for action_dict in global_actions:
# filter actions to get those that can be run by the user,
# either because of actual roles, or because the action is
# accessible to functions.
if logged_users_role().id not in (action_dict.get('roles') or []):
action_dict['roles'] = [x for x in user.get_roles() if x in (action_dict.get('roles') or [])]
if action_dict['roles']:
# action is accessible with user roles, remove mentions of functions
action_dict['functions'] = []
if action_dict['functions'] or action_dict['roles']:
mass_actions.append(action_dict)
return mass_actions
def _q_index(self):
self.view_type = 'table'
self.check_access()
get_logger().info('backoffice - form %s - listing' % self.formdef.name)
if 'job' in get_request().form:
return self.job_multi()
self.use_default_view = True
fields = self.get_fields_from_query()
selected_filter = self.get_filter_from_query()
criterias = self.get_criterias_from_query()
if get_publisher().is_using_postgresql():
# only enable pagination in SQL mode, as we do not have sorting in
# the other case.
limit = misc.get_int_or_400(
get_request().form.get('limit', get_publisher().get_site_option('default-page-size') or 20)
)
else:
limit = misc.get_int_or_400(get_request().form.get('limit', 0))
offset = misc.get_int_or_400(get_request().form.get('offset', 0))
order_by = misc.get_order_by_or_400(get_request().form.get('order_by'))
if self.view and not order_by:
order_by = self.view.order_by
if not order_by:
order_by = get_publisher().get_site_option('default-sort-order') or '-receipt_time'
query = get_request().form.get('q')
qs = ''
if get_request().get_query():
qs = '?' + get_request().get_query()
multi_actions = self.get_multi_actions(
self.formdef, get_request().user, status_filter=selected_filter
)
multi_form = Form(id='multi-actions')
for action in multi_actions:
attrs = {}
if action.get('functions'):
for function in action.get('functions'):
# dashes are replaced by underscores to prevent HTML5
# normalization to CamelCase.
attrs['data-visible_for_%s' % function.replace('-', '_')] = 'true'
else:
attrs['data-visible_for_all'] = 'true'
if getattr(action['action'], 'require_confirmation', False):
attrs['data-ask-for-confirmation'] = 'true'
multi_form.add_submit(
'button-action-%s' % action['action'].id, action['action'].name, attrs=attrs
)
if not get_request().form.get('ajax') == 'true':
if multi_form.is_submitted() and get_request().form.get('select[]'):
for action in multi_actions:
if multi_form.get_submit() == 'button-action-%s' % action['action'].id:
return self.submit_multi(
action, selected_filter=selected_filter, query=query, criterias=criterias
)
table = FormDefUI(self.formdef).listing(
fields=fields,
selected_filter=selected_filter,
limit=limit,
offset=offset,
query=query,
order_by=order_by,
criterias=criterias,
include_checkboxes=bool(multi_actions),
)
if get_response().status_code == 302:
# catch early redirect
return table
multi_form.widgets.append(HtmlWidget(table))
if not multi_actions:
multi_form.widgets.append(HtmlWidget('<div class="buttons"></div>'))
if get_request().form.get('ajax') == 'true':
get_response().filter = {'raw': True}
return multi_form.render()
view_name = self.view.title if self.view else _('Listing')
html_top('management', '%s - %s' % (view_name, self.formdef.name))
r = TemplateIO(html=True)
r += htmltext('<div id="appbar">')
r += htmltext('<h2>%s - %s</h2>') % (self.formdef.name, view_name)
r += get_session().display_message()
r += self.listing_top_actions()
r += htmltext('</div>')
r += multi_form.render()
get_response().filter['sidebar'] = self.get_formdata_sidebar(qs) + self.get_fields_sidebar(
selected_filter,
fields,
limit=limit,
query=query,
criterias=criterias,
offset=offset,
order_by=order_by,
)
return r.getvalue()
def submit_multi(self, action, selected_filter, query, criterias):
item_ids = get_request().form['select[]']
if '_all' in item_ids:
if get_publisher().is_using_postgresql():
criterias.append(Null('anonymised'))
item_ids = FormDefUI(self.formdef).get_listing_item_ids(
selected_filter, user=get_request().user, query=query, criterias=criterias
)
job = get_response().add_after_job(
MassActionAfterJob(
label=_('Executing task "%s" on forms') % action['action'].name,
formdef=self.formdef,
user_id=get_request().user.id,
status_filter=selected_filter,
query_string=get_request().get_query(),
action_id=action['action'].id,
item_ids=item_ids,
return_url=get_request().get_path_query(),
)
)
job.store()
return redirect(job.get_processing_url())
def export_spreadsheet(self):
self.check_access()
if self.formdef.category and not self.formdef.category.has_permission('export', get_request().user):
raise errors.AccessForbiddenError()
form = Form()
form.add_hidden('query_string', get_request().get_query())
form.add(
RadiobuttonsWidget,
'format',
options=[('ods', _('OpenDocument (.ods)'), 'ods'), ('csv', _('Text (.csv)'), 'csv')],
value='ods',
required=True,
extra_css_class='widget-inline-radio',
)
form.add(
CheckboxWidget,
'include_header_line',
title=_('Include header line'),
value=True,
)
form.add_submit('submit', _('Export'))
form.add_submit('cancel', _('Cancel'))
if not form.is_submitted() or form.has_errors():
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Spreadsheet Options')
r += form.render()
return r.getvalue()
get_request().form = parse_query(form.get_widget('query_string').parse() or '', 'utf-8')
get_request().form['skip_header_line'] = not (form.get_widget('include_header_line').parse())
file_format = form.get_widget('format').parse()
if file_format == 'csv':
return self.csv()
else:
return self.ods()
def csv(self):
self.check_access()
if (
not get_request().is_api_url()
and self.formdef.category
and not self.formdef.category.has_permission('export', get_request().user)
):
raise errors.AccessForbiddenError()
fields = self.get_fields_from_query()
selected_filter = self.get_filter_from_query()
user = get_request().user
query = get_request().form.get('q')
criterias = self.get_criterias_from_query()
order_by = misc.get_order_by_or_400(get_request().form.get('order_by', None))
skip_header_line = bool(get_request().form.get('skip_header_line'))
get_logger().info('backoffice - form %s - listing csv' % self.formdef.name)
count = self.formdef.data_class().count()
job = CsvExportAfterJob(
self.formdef,
fields=fields,
selected_filter=selected_filter,
user_id=user.id,
query=query,
criterias=criterias,
order_by=order_by,
skip_header_line=skip_header_line,
)
if count > self.WCS_SYNC_EXPORT_LIMIT:
job = get_response().add_after_job(job)
job.store()
return redirect(job.get_processing_url())
else:
job.execute()
response = get_response()
response.set_content_type('text/plain')
response.set_header('content-disposition', 'attachment; filename=%s.csv' % self.formdef.url_name)
return job.file_content
def export(self):
self.check_access()
try:
job = AfterJob.get(get_request().form.get('download'))
except KeyError:
return redirect('.')
if not job.status == 'completed':
raise errors.TraversalError()
response = get_response()
response.set_content_type(job.content_type)
response.set_header('content-disposition', 'attachment; filename=%s' % job.file_name)
return job.file_content
def ods(self):
self.check_access()
if get_request().has_anonymised_data_api_restriction():
# api/ will let this pass but we don't want that.
raise errors.AccessForbiddenError()
if (
not get_request().is_api_url()
and self.formdef.category
and not self.formdef.category.has_permission('export', get_request().user)
):
raise errors.AccessForbiddenError()
fields = self.get_fields_from_query()
selected_filter = self.get_filter_from_query()
user = get_user_from_api_query_string() or get_request().user
query = get_request().form.get('q')
criterias = self.get_criterias_from_query()
order_by = misc.get_order_by_or_400(get_request().form.get('order_by', None))
skip_header_line = bool(get_request().form.get('skip_header_line'))
count = self.formdef.data_class().count()
job = OdsExportAfterJob(
self.formdef,
fields=fields,
selected_filter=selected_filter,
user_id=user.id,
query=query,
criterias=criterias,
order_by=order_by,
skip_header_line=skip_header_line,
)
if count > self.WCS_SYNC_EXPORT_LIMIT and not get_request().is_api_url():
job = get_response().add_after_job(job)
job.store()
return redirect(job.get_processing_url())
else:
job.execute()
response = get_response()
response.set_content_type('application/vnd.oasis.opendocument.spreadsheet')
response.set_header('content-disposition', 'attachment; filename=%s.ods' % self.formdef.url_name)
return job.file_content
def json(self):
anonymise = get_request().has_anonymised_data_api_restriction()
self.check_access()
get_response().set_content_type('application/json')
user = get_user_from_api_query_string() or get_request().user if not anonymise else None
selected_filter = self.get_filter_from_query(default='all')
criterias = self.get_criterias_from_query()
order_by = misc.get_order_by_or_400(get_request().form.get('order_by', None))
if self.view and not order_by:
order_by = self.view.order_by
query = get_request().form.get('q') if not anonymise else None
offset = None
if 'offset' in get_request().form:
offset = misc.get_int_or_400(get_request().form['offset'])
limit = None
if 'limit' in get_request().form:
limit = misc.get_int_or_400(get_request().form['limit'])
items = FormDefUI(self.formdef).get_listing_items(
None,
selected_filter,
user=user,
query=query,
criterias=criterias,
order_by=order_by,
anonymise=anonymise,
offset=offset,
limit=limit,
)[0]
if get_publisher().is_using_postgresql():
self.formdef.data_class().load_all_evolutions(items)
if get_request().form.get('full') == 'on':
output = []
for filled in items:
data = filled.get_json_export_dict(include_files=False, anonymise=anonymise, user=user)
data.pop('digests')
data['digest'] = filled.default_digest
output.append(data)
else:
output = [
{
'id': filled.id,
'display_id': filled.get_display_id(),
'display_name': filled.get_display_name(),
'digest': filled.default_digest,
'text': filled.get_display_label(),
'url': filled.get_url(),
'receipt_time': datetime.datetime(*filled.receipt_time[:6]),
'last_update_time': datetime.datetime(*filled.last_update_time[:6]),
}
for filled in items
]
if isinstance(self.formdef, CardDef) or self.view:
# for cards and custom views return results in a dictionary, as it
# provides a better path for evolutions
output = {'data': output}
return json.dumps(output, cls=misc.JSONEncoder)
def geojson(self):
if not self.formdef.geolocations:
raise errors.TraversalError()
if get_request().has_anonymised_data_api_restriction():
# api/ will let this pass but we don't want that.
raise errors.AccessForbiddenError()
self.check_access('geojson')
get_response().set_content_type('application/json')
user = get_request().user
if not user:
user = get_user_from_api_query_string('geojson')
selected_filter = self.get_filter_from_query()
if get_request().form.get('full') == 'on':
fields = list(self.get_formdef_fields())
else:
fields = self.get_fields_from_query()
criterias = self.get_criterias_from_query()
query = get_request().form.get('q')
items = FormDefUI(self.formdef).get_listing_items(
fields, selected_filter, user=user, query=query, criterias=criterias
)[0]
return json.dumps(geojson_formdatas(items, fields=fields), cls=misc.JSONEncoder)
def ics(self):
if get_request().has_anonymised_data_api_restriction():
# api/ will let this pass but we don't want that.
raise errors.AccessForbiddenError()
charset = get_publisher().site_charset
self.check_access('ics')
user = get_request().user
if not (user and user.is_admin):
user = get_user_from_api_query_string('ics') or user
formdef = self.formdef
selected_filter = self.get_filter_from_query()
fields = self.get_fields_from_query()
criterias = self.get_criterias_from_query()
query = get_request().form.get('q')
class IcsDirectory(Directory):
# ics/<component> with <component> being the identifier (varname)
# of the field to use as start date (may be a date field or a
# string field).
# ics/<component>/<component2> with <component2> as the identifier
# of the field to use as end date (ditto, date or string field)
def _q_traverse(self, path):
if not path[-1]:
# allow trailing slash
path = path[:-1]
if len(path) not in (1, 2):
raise errors.TraversalError()
start_date_field_varname = path[0]
end_date_field_varname = None
if len(path) == 2:
end_date_field_varname = path[1]
start_date_field_id = None
end_date_field_id = None
for field in formdef.get_all_fields():
if getattr(field, 'varname', None) == start_date_field_varname:
start_date_field_id = field.id
if end_date_field_varname and getattr(field, 'varname', None) == end_date_field_varname:
end_date_field_id = field.id
if not start_date_field_id:
raise errors.TraversalError()
if end_date_field_varname and not end_date_field_id:
raise errors.TraversalError()
formdatas = FormDefUI(formdef).get_listing_items(
fields, selected_filter, user=user, query=query, criterias=criterias
)[0]
cal = vobject.iCalendar()
cal.add('prodid').value = '-//Entr\'ouvert//NON SGML Publik'
for formdata in formdatas:
if not formdata.data.get(start_date_field_id):
continue
try:
dtstart = make_datetime(formdata.data[start_date_field_id])
dtend = None
if end_date_field_id and formdata.data.get(end_date_field_id):
dtend = make_datetime(formdata.data[end_date_field_id])
except ValueError:
continue
vevent = vobject.newFromBehavior('vevent')
vevent.add('uid').value = '%s-%s-%s' % (
get_request().get_server().lower(),
formdef.url_name,
formdata.id,
)
summary = force_text(formdata.get_display_name(), charset)
if formdata.default_digest:
summary += ' - %s' % force_text(formdata.default_digest, charset)
vevent.add('summary').value = summary
vevent.add('dtstart').value = dtstart
if dtend:
vevent.add('dtend').value = dtend
vevent.dtstart.value_param = 'DATE'
backoffice_url = formdata.get_url(backoffice=True)
vevent.add('url').value = backoffice_url
form_name = force_text(formdef.name, charset)
status_name = force_text(formdata.get_status_label(), charset)
description = '%s | %s | %s\n' % (form_name, formdata.get_display_id(), status_name)
if formdata.default_digest:
description += '%s\n' % force_text(formdata.default_digest, charset)
description += backoffice_url
# TODO: improve performance by loading all users in one
# single query before the loop
if formdata.user:
description += '\n%s' % force_text(formdata.user.get_display_name(), charset)
vevent.add('description').value = description
cal.add(vevent)
get_response().set_content_type('text/calendar')
return cal.serialize()
return IcsDirectory()
def map(self):
self.view_type = 'map'
get_response().add_javascript(['qommon.map.js'])
html_top('management', '%s - %s' % (_('Form'), self.formdef.name))
r = TemplateIO(html=True)
get_response().breadcrumb.append(('map', _('Map')))
attrs = {
'class': 'qommon-map',
'id': 'backoffice-map',
'data-readonly': True,
'data-geojson-url': '%s/geojson?%s' % (get_request().get_url(1), get_request().get_query()),
}
attrs.update(get_publisher().get_map_attributes())
fields = self.get_fields_from_query()
selected_filter = self.get_filter_from_query()
qs = ''
if get_request().get_query():
qs = '?' + get_request().get_query()
get_response().filter['sidebar'] = self.get_formdata_sidebar(qs) + self.get_fields_sidebar(
selected_filter, fields
)
r += htmltext('<h2>%s - %s</h2>') % (self.formdef.name, _('Map'))
r += htmltext('<div %s></div>' % ' '.join(['%s="%s"' % x for x in attrs.items()]))
return r.getvalue()
def get_stats_sidebar(self, selected_filter):
get_response().add_javascript(['wcs.listing.js'])
r = TemplateIO(html=True)
r += htmltext('<form id="listing-settings" action="stats">')
r += self.get_filter_sidebar(selected_filter=selected_filter, mode='stats')
r += htmltext('<button class="refresh">%s</button>') % _('Refresh')
r += htmltext('</form>')
return r.getvalue()
def stats(self):
self.check_access()
if self.formdef.category and not self.formdef.category.has_permission(
'statistics', get_request().user
):
raise errors.AccessForbiddenError()
get_logger().info('backoffice - form %s - stats' % self.formdef.name)
html_top('management', '%s - %s' % (_('Form'), self.formdef.name))
r = TemplateIO(html=True)
get_response().breadcrumb.append(('stats', _('Statistics')))
selected_filter = self.get_filter_from_query(default='all')
criterias = self.get_criterias_from_query()
get_response().filter['sidebar'] = self.get_formdata_sidebar() + self.get_stats_sidebar(
selected_filter
)
do_graphs = get_publisher().is_using_postgresql()
displayed_criterias = None
if selected_filter and selected_filter != 'all':
if selected_filter == 'pending':
applied_filters = ['wf-%s' % x.id for x in self.formdef.workflow.get_not_endpoint_status()]
criteria_label = _('Status: %s') % _('Pending')
elif selected_filter == 'done':
applied_filters = ['wf-%s' % x.id for x in self.formdef.workflow.get_endpoint_status()]
criteria_label = _('Status: %s') % _('Done')
else:
try:
criteria_label = _('Status: %s') % self.formdef.workflow.get_status(selected_filter).name
applied_filters = ['wf-%s' % selected_filter]
except KeyError:
applied_filters = None
if applied_filters:
criterias.append(Or([Equal('status', x) for x in applied_filters]))
criterias[-1]._label = criteria_label
displayed_criterias = criterias
if not displayed_criterias:
displayed_criterias = criterias
criterias = [NotEqual('status', 'draft')] + displayed_criterias
values = self.formdef.data_class().select(criterias)
if get_publisher().is_using_postgresql():
# load all evolutions in a single batch, to avoid as many query as
# there are formdata when computing resolution times statistics.
self.formdef.data_class().load_all_evolutions(values)
r += htmltext('<div id="statistics">')
if displayed_criterias:
r += htmltext('<div class="criterias bo-block">')
r += htmltext('<h2>%s</h2>') % _('Filters')
r += htmltext('<ul>')
for criteria in displayed_criterias:
criteria_label = getattr(criteria, '_label', None)
if criteria_label:
r += htmltext('<li>%s</li>') % criteria_label
r += htmltext('</ul></div>')
if do_graphs:
r += htmltext('<div class="splitcontent-left">')
no_forms = len(values)
r += htmltext('<div class="bo-block">')
r += htmltext('<p>%s %d</p>') % (_('Total number of records:'), no_forms)
if self.formdef.workflow:
r += htmltext('<ul>')
for status in self.formdef.workflow.possible_status:
r += htmltext('<li>%s: %d</li>') % (
status.name,
len([x for x in values if x.status == 'wf-%s' % status.id]),
)
r += htmltext('</ul>')
r += htmltext('</div>')
excluded_fields = []
for criteria in displayed_criterias:
if not isinstance(criteria, Equal):
continue
excluded_fields.append(criteria.attribute[1:])
stats_for_fields = self.stats_fields(values, excluded_fields=excluded_fields)
if stats_for_fields:
r += htmltext('<div class="bo-block">')
r += stats_for_fields
r += htmltext('</div>')
stats_times = self.stats_resolution_time(values)
if stats_times:
r += htmltext('<div class="bo-block">')
r += stats_times
r += htmltext('</div>')
if do_graphs:
r += htmltext('</div>')
r += htmltext('<div class="splitcontent-right">')
criterias.append(Equal('formdef_id', int(self.formdef.id)))
r += do_graphs_section(criterias=criterias)
r += htmltext('</div>')
r += htmltext('</div>') # id="statistics"
if get_request().form.get('ajax') == 'true':
get_response().filter = {'raw': True}
return r.getvalue()
page = TemplateIO(html=True)
page += htmltext('<h2>%s - %s</h2>') % (self.formdef.name, _('Statistics'))
page += htmltext(r)
page += htmltext('<a class="back" href=".">%s</a>') % _('Back')
return page.getvalue()
def stats_fields(self, values, excluded_fields=None):
r = TemplateIO(html=True)
had_page = False
last_page = None
last_title = None
for f in self.formdef.get_all_fields():
if excluded_fields and f.id in excluded_fields:
continue
if f.type == 'page':
last_page = f.label
last_title = None
continue
if f.type == 'title':
last_title = f.label
continue
if not f.stats:
continue
t = f.stats(values)
if not t:
continue
if last_page:
if had_page:
r += htmltext('</div>')
r += htmltext('<div class="page">')
r += htmltext('<h3>%s</h3>') % last_page
had_page = True
last_page = None
if last_title:
r += htmltext('<h3>%s</h3>') % last_title
last_title = None
r += t
if had_page:
r += htmltext('</div>')
return r.getvalue()
def stats_resolution_time(self, values):
possible_status = [('wf-%s' % x.id, x.id) for x in self.formdef.workflow.possible_status[1:]]
if len(possible_status) < 2:
return
r = TemplateIO(html=True)
r += htmltext('<h2>%s</h2>') % _('Resolution time')
for dummy, status_id in possible_status:
res_time_forms = []
for filled in values:
for evo in filled.evolution or []:
if evo.status == 'wf-%s' % status_id:
res_time_forms.append(time.mktime(evo.time) - time.mktime(filled.receipt_time))
break
if not res_time_forms:
continue
res_time_forms.sort()
sum_times = sum(res_time_forms)
len_times = len(res_time_forms)
min_times = res_time_forms[0]
max_times = res_time_forms[-1]
r += htmltext('<h3>%s</h3>') % (
_('To Status "%s"') % self.formdef.workflow.get_status(status_id).name
)
r += htmltext('<ul>')
r += htmltext(' <li>%s %s</li>') % (_('Count:'), len_times)
r += htmltext(' <li>%s %s</li>') % (_('Minimum Time:'), format_time(min_times))
r += htmltext(' <li>%s %s</li>') % (_('Maximum Time:'), format_time(max_times))
r += htmltext(' <li>%s %s</li>') % (_('Range:'), format_time(max_times - min_times))
mean = sum_times // len_times
r += htmltext(' <li>%s %s</li>') % (_('Mean:'), format_time(mean))
if len_times % 2:
median = res_time_forms[len_times // 2]
else:
midpt = len_times // 2
median = (res_time_forms[midpt - 1] + res_time_forms[midpt]) // 2
r += htmltext(' <li>%s %s</li>') % (_('Median:'), format_time(median))
# variance...
x = 0
for t in res_time_forms:
x += (t - mean) ** 2.0
try:
variance = x // (len_times + 1)
except Exception:
variance = 0
# not displayed since in square seconds which is not easy to grasp
from math import sqrt
# and standard deviation
std_dev = sqrt(variance)
r += htmltext(' <li>%s %s</li>') % (_('Standard Deviation:'), format_time(std_dev))
r += htmltext('</ul>')
return r.getvalue()
def _q_lookup(self, component):
if component == 'ics':
return self.ics()
if not self.view:
for view in self.get_custom_views():
if view.get_url_slug() == component:
return self.__class__(formdef=self.formdef, view=view)
if component.startswith('user-'):
get_session().message = (
'warning',
_(
'A missing or invalid custom view was referenced; '
'you have been automatically redirected.'
),
)
# remove custom view reference from path
# (ignore the fact that some form/card could itself be named
# user-whatever)
url = get_request().get_path_query().replace('/%s/' % component, '/')
return misc.QLookupRedirect(url)
try:
filled = self.formdef.data_class().get(component)
except KeyError:
raise errors.TraversalError()
return FormBackOfficeStatusPage(self.formdef, filled)
def live(self):
return FormBackofficeEditPage(self.formdef.url_name).live()
class FormBackofficeEditPage(FormFillPage):
def create_form(self, *args, **kwargs):
form = super().create_form(*args, **kwargs)
form.attrs['data-live-url'] = self.formdef.get_url(backoffice=True) + 'live'
return form
class FormBackOfficeStatusPage(FormStatusPage):
_q_exports_orig = [
'',
'download',
'json',
'action',
'live',
'inspect',
('inspect-tool', 'inspect_tool'),
('download-as-zip', 'download_as_zip'),
('lateral-block', 'lateral_block'),
('user-pending-forms', 'user_pending_forms'),
]
form_page_class = FormBackofficeEditPage
sidebar_recorded_message = _('The form has been recorded on %(date)s with the number %(number)s.')
sidebar_recorded_by_agent_message = _(
'The form has been recorded on %(date)s with the number %(number)s by %(agent)s.'
)
def html_top(self, title=None):
return html_top('management', title)
def _q_index(self):
if self.filled.status == 'draft':
if self.filled.backoffice_submission and self.formdef.backoffice_submission_roles:
for role in get_request().user.get_roles():
if role in self.formdef.backoffice_submission_roles:
return redirect(
'../../../submission/%s/%s/' % (self.formdef.url_name, self.filled.id)
)
raise errors.AccessForbiddenError()
get_response().filter['sidebar'] = self.get_sidebar()
return self.status()
def receipt(self, *args, **kwargs):
r = TemplateIO(html=True)
if get_session() and get_session().is_anonymous_submitter(self.filled):
r += htmltext('<div class="infonotice">')
r += _(
'This form has been accessed via its tracking code, it is '
'therefore displayed like you were also its owner.'
)
r += htmltext('</div>')
r += super().receipt(*args, **kwargs)
return r.getvalue()
def get_sidebar(self):
return self.get_extra_context_bar()
def get_workflow_form(self, user):
form = super().get_workflow_form(user)
if form:
form.attrs['data-live-url'] = self.filled.get_url(backoffice=True) + 'live'
return form
def lateral_block(self):
self.check_receiver()
get_response().filter = {'raw': True}
response = self.get_lateral_block()
return response
def user_pending_forms(self):
self.check_receiver()
get_response().filter = {'raw': True}
response = self.get_user_pending_forms()
# preemptive locking of forms
all_visitors = get_session().get_object_visitors(self.filled)
visitors = [x for x in all_visitors if x[0] != get_session().user]
me_in_visitors = bool(get_session().user in [x[0] for x in all_visitors])
if not visitors or me_in_visitors:
related_user_forms = getattr(self.filled, 'related_user_forms', None) or []
user_roles = set(get_request().user.get_roles())
session = get_session()
for user_formdata in related_user_forms:
if user_roles.intersection(user_formdata.actions_roles):
session.mark_visited_object(user_formdata)
return response
def get_extra_context_bar(self, parent=None):
formdata = self.filled
r = TemplateIO(html=True)
if not formdata.is_draft():
r += htmltext('<p><a class="button" href="..">%s</a></p>') % _('Back to Listing')
r += htmltext('<div class="extra-context">')
if (
formdata.backoffice_submission
and formdata.submission_agent_id == str(get_request().user.id)
and formdata.tracking_code
and time.time() - time.mktime(formdata.receipt_time) < 30 * 60
):
# keep displaying tracking code to submission agent for 30
# minutes after submission
r += htmltext('<h3>%s</h3>') % _('Tracking Code')
r += htmltext('<p>%s</p>') % formdata.tracking_code
r += htmltext('<h3>%s</h3>') % _('General Information')
r += htmltext('<p>')
tm = misc.localstrftime(formdata.receipt_time)
agent_user = None
if formdata.submission_agent_id:
agent_user = get_publisher().user_class.get(formdata.submission_agent_id, ignore_errors=True)
if agent_user:
r += self.sidebar_recorded_by_agent_message % {
'date': tm,
'number': formdata.get_display_id(),
'agent': agent_user.get_display_name(),
}
else:
r += self.sidebar_recorded_message % {'date': tm, 'number': formdata.get_display_id()}
r += htmltext('</p>')
try:
status_colour = formdata.get_status().colour
except AttributeError:
status_colour = None
status_colour = status_colour or 'ffffff'
fg_colour = misc.get_foreground_colour(status_colour)
r += htmltext(
'<p class="current-status"><span class="item" style="background: #%s; color: %s;"></span>'
% (status_colour, fg_colour)
)
r += htmltext('<span>%s %s') % (_('Status:'), formdata.get_status_label())
status = formdata.get_status()
if status and status.get_visibility_restricted_roles():
r += htmltext('<span class="visibility-off" title="%s"></span>') % _(
'This status is not visible to users.'
)
r += htmltext('</span></p>')
if formdata.formdef.workflow.criticality_levels:
try:
level = formdata.get_criticality_level_object()
except IndexError:
pass
else:
r += htmltext('<p class="current-level">')
if level.colour:
r += htmltext('<span class="item" style="background: #%s;"></span>' % level.colour)
r += htmltext('<span>%s %s</span></p>') % (_('Criticality Level:'), level.name)
if formdata.anonymised:
r += htmltext('<div class="infonotice">')
r += htmltext(_('This form has been anonymised on %(date)s.')) % {
'date': formdata.anonymised.strftime(misc.date_format())
}
r += htmltext('</div>')
if formdata.formdef.include_download_all_button:
has_attached_files = False
for value in (formdata.data or {}).values():
if isinstance(value, PicklableUpload):
has_attached_files = True
if isinstance(value, dict) and isinstance(value.get('data'), list):
# block fields
for subvalue in value.get('data'):
for subvalue_elem in subvalue.values():
if isinstance(subvalue_elem, PicklableUpload):
has_attached_files = True
break
if has_attached_files:
break
if has_attached_files:
r += htmltext('<p><a class="button" href="download-as-zip">%s</a></p>') % _(
'Download all files as .zip'
)
r += htmltext('</div>')
r += self.get_extra_submission_context_bar()
r += self.get_extra_submission_channel_bar()
r += self.get_extra_submission_user_id_bar(parent=parent)
r += self.get_extra_geolocation_bar()
if formdata.formdef.lateral_template:
r += htmltext('<div data-async-url="%slateral-block"></div>' % formdata.get_url(backoffice=True))
if (
not isinstance(formdata.formdef, CardDef)
and formdata.user_id
and get_publisher().is_using_postgresql()
):
r += htmltext(
'<div data-async-url="%suser-pending-forms"></div>' % formdata.get_url(backoffice=True)
)
if not formdata.is_draft() and (
get_publisher().get_backoffice_root().is_accessible('forms')
or get_publisher().get_backoffice_root().is_accessible('workflows')
):
r += htmltext('<div class="extra-context">')
r += htmltext('<p><a href="%sinspect">' % formdata.get_url(backoffice=True))
r += htmltext('%s</a></p>') % _('Data Inspector')
r += htmltext('</div>')
return r.getvalue()
def get_extra_submission_context_bar(self):
formdata = self.filled
r = TemplateIO(html=True)
if formdata.submission_context or formdata.submission_channel:
extra_context = formdata.submission_context or {}
r += htmltext('<div class="extra-context">')
if extra_context.get('orig_formdef_id'):
object_type = extra_context.get('orig_object_type', 'formdef')
if object_type == 'formdef':
r += htmltext('<h3>%s</h3>') % _('Original form')
object_class = FormDef
else:
r += htmltext('<h3>%s</h3>') % _('Original card')
object_class = CardDef
try:
orig_formdata = (
object_class.get(extra_context.get('orig_formdef_id'))
.data_class()
.get(extra_context.get('orig_formdata_id'))
)
except KeyError:
r += htmltext('<p>%s</p>') % _('(deleted)')
else:
r += htmltext('<p><a href="%s">%s</a></p>') % (
orig_formdata.get_url(backoffice=True),
orig_formdata.get_display_name(),
)
return r.getvalue()
def get_extra_submission_channel_bar(self):
formdata = self.filled
r = TemplateIO(html=True)
if formdata.submission_channel:
extra_context = formdata.submission_context or {}
r += htmltext('<h3>%s</h3>') % '%s: %s' % (_('Channel'), formdata.get_submission_channel_label())
if extra_context.get('caller'):
r += htmltext('<h3>%s: %s</h3>') % (_('Phone'), extra_context['caller'])
if extra_context.get('thumbnail_url'):
r += htmltext('<p class="thumbnail"><img src="%s" alt=""/></p>') % extra_context.get(
'thumbnail_url'
)
if extra_context.get('mail_url'):
r += htmltext('<p><a href="%s">%s</a></p>') % (extra_context.get('mail_url'), _('Open'))
if extra_context.get('comments'):
r += htmltext('<h3>%s</h3>') % _('Comments')
r += htmltext('<p>%s</p>') % extra_context.get('comments')
if extra_context.get('summary_url'):
r += htmltext('<div data-content-url="%s"></div>' % (extra_context.get('summary_url')))
r += htmltext('</div>')
return r.getvalue()
def get_extra_submission_user_id_bar(self, parent):
formdata = self.filled
r = TemplateIO(html=True)
if formdata and formdata.user_id and formdata.get_user():
r += htmltext('<div class="extra-context">')
r += htmltext('<h3>%s</h3>') % _('Associated User')
users_cfg = get_cfg('users', {})
sidebar_user_template = users_cfg.get('sidebar_template')
if sidebar_user_template:
variables = get_publisher().substitutions.get_context_variables(mode='lazy')
sidebar_user = Template(sidebar_user_template).render(variables)
if not sidebar_user.startswith('<'):
sidebar_user = htmltext('<p>%s</p>' % sidebar_user)
r += sidebar_user
else:
r += htmltext('<p>%s</p>') % formdata.get_user().display_name
r += htmltext('</div>')
elif parent and parent.has_user_support and parent.edit_mode:
r += self.get_extra_submission_user_selection_bar(parent=parent)
return r.getvalue()
def get_extra_submission_user_selection_bar(self, parent=None):
r = TemplateIO(html=True)
r += htmltext('<div class="submit-user-selection" style="display: none;">')
get_response().add_javascript(['select2.js'])
r += htmltext('<h3>%s</h3>') % _('Associated User')
r += htmltext('<select>')
if parent and parent.selected_user_id:
r += htmltext('<option value="%s">%s</option>') % (
parent.selected_user_id,
get_publisher().user_class.get(parent.selected_user_id, ignore_errors=True),
)
r += htmltext('</select>')
r += htmltext('</div>')
return r.getvalue()
def get_extra_geolocation_bar(self):
formdata = self.filled
r = TemplateIO(html=True)
if formdata.formdef.geolocations and formdata.geolocations:
r += htmltext('<div class="geolocations">')
for geoloc_key in formdata.formdef.geolocations:
if geoloc_key not in formdata.geolocations:
continue
r += htmltext('<h3>%s</h3>') % formdata.formdef.geolocations[geoloc_key]
geoloc_value = formdata.geolocations[geoloc_key]
map_widget = MapWidget(
'geoloc_%s' % geoloc_key,
readonly=True,
value='%(lat)s;%(lon)s' % geoloc_value,
render_br=False,
)
r += map_widget.render()
r += htmltext('</div>')
return r.getvalue()
def download_as_zip(self):
formdata = self.filled
zip_content = io.BytesIO()
counter = {'value': 0}
def add_zip_file(upload, zip_file):
counter['value'] += 1
filename = '%s_%s' % (counter['value'], upload.base_filename)
zip_file.writestr(filename, upload.get_content())
with zipfile.ZipFile(zip_content, 'w') as zip_file:
for value in formdata.data.values():
if isinstance(value, PicklableUpload):
add_zip_file(value, zip_file)
if isinstance(value, dict) and isinstance(value.get('data'), list):
for subvalue in value.get('data'):
for subvalue_elem in subvalue.values():
if isinstance(subvalue_elem, PicklableUpload):
add_zip_file(subvalue_elem, zip_file)
response = get_response()
response.set_content_type('application/zip')
response.set_header(
'content-disposition', 'attachment; filename=files-%s.zip' % formdata.get_display_id()
)
return zip_content.getvalue()
def get_user_pending_forms(self):
from wcs import sql
formdata = self.filled
r = TemplateIO(html=True)
user_roles = [logged_users_role().id] + get_request().user.get_roles()
criterias = [
Equal('is_at_endpoint', False),
Equal('user_id', str(formdata.user_id)),
Intersects('concerned_roles_array', user_roles),
]
formdatas = sql.AnyFormData.select(criterias, order_by='receipt_time')
self.filled.related_user_forms = formdatas
if formdatas:
r += htmltext('<div class="user-pending-forms">')
r += htmltext('<h3>%s</h3>') % _('User Pending Forms')
categories = {}
formdata_by_category = {}
for formdata in formdatas:
if formdata.formdef.category_id not in categories:
categories[formdata.formdef.category_id] = formdata.formdef.category
formdata_by_category[formdata.formdef.category_id] = []
formdata_by_category[formdata.formdef.category_id].append(formdata)
cats = list(categories.values())
Category.sort_by_position(cats)
if self.formdef.category_id in categories:
# move current category to the top
cats.remove(categories[self.formdef.category_id])
cats.insert(0, categories[self.formdef.category_id])
for cat in cats:
if len(cats) > 1:
if cat is None:
r += htmltext('<h4>%s</h4>') % _('Misc')
cat_formdatas = formdata_by_category[None]
else:
r += htmltext('<h4>%s</h4>') % cat.name
cat_formdatas = formdata_by_category[cat.id]
else:
cat_formdatas = formdatas
r += htmltext('<ul class="user-formdatas">')
for formdata in cat_formdatas:
status = formdata.get_status()
if status:
status_label = status.name
else:
status_label = _('Unknown')
submit_date = misc.strftime(misc.date_format(), formdata.receipt_time)
if str(formdata.formdef_id) == str(self.formdef.id) and (
str(formdata.id) == str(self.filled.id)
):
r += (
htmltext('<li class="self"><span class="formname">%s</span> ')
% formdata.formdef.name
)
else:
r += htmltext('<li><a href="%s">%s</a> ') % (
formdata.get_url(backoffice=True),
formdata.formdef.name,
)
r += htmltext('(<span class="id">%s</span>), ') % formdata.get_display_id()
r += htmltext('<span class="datetime">%s</span> ' '<span class="status">(%s)</span>') % (
submit_date,
status_label,
)
if formdata.default_digest:
r += htmltext('<small>%s</small>') % formdata.default_digest
r += htmltext('</li>')
r += htmltext('</ul>')
r += htmltext('</div>')
return r.getvalue()
def get_lateral_block(self):
r = TemplateIO(html=True)
lateral_block = self.filled.get_lateral_block()
if lateral_block:
r += htmltext('<div class="lateral-block">')
r += htmltext(lateral_block)
r += htmltext('</div>')
return r.getvalue()
def test_tools_form(self):
form = Form(use_tokens=False)
form.add(
RadiobuttonsWidget,
'test_mode',
options=[
('django-condition', _('Condition (Django)'), 'django-condition'),
('python-condition', _('Condition (Python)'), 'python-condition'),
('template', '%s / %s' % (_('Template'), _('Django Expression')), 'template'),
('html_template', _('HTML Template (WYSIWYG)'), 'html_template'),
],
value='django-condition',
attrs={'data-dynamic-display-parent': 'true'},
extra_css_class='widget-inline-radio',
)
form.add(
StringWidget,
'django-condition',
extra_css_class='grid-1-1',
attrs={
'data-dynamic-display-child-of': 'test_mode',
'data-dynamic-display-value': 'django-condition',
},
)
form.add(
StringWidget,
'python-condition',
attrs={
'data-dynamic-display-child-of': 'test_mode',
'data-dynamic-display-value': 'python-condition',
},
)
form.add(
WysiwygTextWidget,
'html_template',
attrs={
'data-dynamic-display-child-of': 'test_mode',
'data-dynamic-display-value': 'html_template',
},
)
form.add(
TextWidget,
'template',
attrs={'data-dynamic-display-child-of': 'test_mode', 'data-dynamic-display-value': 'template'},
)
form.add_submit('submit', _('Evaluate'))
return form
def test_tool_result(self):
form = self.test_tools_form()
r = TemplateIO(html=True)
if form.is_submitted() and not form.has_errors():
# show test result
test_mode = form.get_widget('test_mode').parse()
if test_mode in ('django-condition', 'python-condition'):
condition = Condition(
{'value': form.get_widget(test_mode).parse(), 'type': test_mode.split('-')[0]}
)
condition.log_errors = False
condition.record_errors = False
try:
result = condition.unsafe_evaluate()
except Exception as exception:
r += htmltext('<div class="errornotice">')
r += htmltext('<p>%s</p>') % _('Failed to evaluate condition')
try:
r += htmltext('<p>%s <code>%s: %s</code></p>') % (
_('Error message:'),
exception.__class__.__name__,
str(exception),
)
except UnicodeEncodeError:
r += htmltext('<p>%s <code>%s</code></p>') % (_('Error message:'), repr(exception))
r += htmltext('</div>')
else:
r += htmltext('<div class="test-tool-result infonotice">')
r += htmltext('<h3>%s</h3>') % _('Condition result:')
r += htmltext('<p><span class="result-%s">%s</span>') % (
str(bool(result)).lower(),
_('True') if result else _('False'),
)
if condition.type == 'python':
r += htmltext(
' &mdash; %s <strong><code>%r</code></strong> ' '<span class="type">(%r)</span>'
) % (_('Python actual result is'), result, type(result))
r += htmltext('</p>')
r += htmltext('</div>')
elif test_mode == 'template':
try:
template = form.get_widget('template').parse() or ''
with get_publisher().complex_data():
result = WorkflowStatusItem.compute(
template, raises=True, record_errors=False, allow_complex=True
)
has_complex_result = get_publisher().has_cached_complex_data(result)
complex_result = get_publisher().get_cached_complex_data(result)
result = re.sub(r'[\uE000-\uF8FF]', '', result)
except Exception as exception:
r += htmltext('<div class="errornotice">')
r += htmltext('<p>%s</p>') % _('Failed to evaluate template')
r += htmltext('<p>%s <code>%s: %s</code></p>') % (
_('Error message:'),
exception.__class__.__name__,
str(exception),
)
else:
r += htmltext('<div class="test-tool-result infonotice">')
r += htmltext('<h3>%s</h3>') % _('Template rendering:')
if result and result[0] == '<': # seems to be HTML
r += htmltext('<div class="test-tool-result-html">')
r += htmltext(result)
r += htmltext('</div>')
r += htmltext('<h3>%s</h3>') % _('HTML Source:')
r += htmltext('<pre class="test-tool-result-plain">%s</pre>') % result
else:
r += htmltext('<div class="test-tool-result-plain">%s</div>') % result
if has_complex_result:
r += htmltext('<h3>%s</h3>') % _('Also rendered as an object:')
r += htmltext('<div class="test-tool-result-plain">%s (%s)</div>') % (
complex_result,
complex_result.__class__.__name__,
)
r += htmltext('</div>')
elif test_mode == 'html_template':
try:
html_template = form.get_widget('html_template').parse() or ''
result = template_on_formdata(
self.filled, html_template, raises=True, ezt_format=ezt.FORMAT_HTML
)
except Exception as exception:
r += htmltext('<div class="errornotice">')
r += htmltext('<p>%s</p>') % _('Failed to evaluate HTML template')
r += htmltext('<p>%s <code>%s: %s</code></p>') % (
_('Error message:'),
exception.__class__.__name__,
str(exception),
)
r += htmltext('</div>')
else:
r += htmltext('<div class="test-tool-result infonotice">')
r += htmltext('<h3>%s</h3>') % _('Template rendering:')
r += htmltext('<div class="test-tool-result-html">')
r += htmltext(result)
r += htmltext('</div>')
r += htmltext('<h3>%s</h3>') % _('HTML Source:')
r += htmltext('<pre class="test-tool-result-plain">%s</pre>') % result
r += htmltext('</div>')
return r.getvalue()
def inspect(self):
if not (
get_publisher().get_backoffice_root().is_accessible('forms')
or get_publisher().get_backoffice_root().is_accessible('workflows')
):
raise errors.AccessForbiddenError()
charset = get_publisher().site_charset
get_response().breadcrumb.append(('inspect', _('Data Inspector')))
self.html_top(self.formdef.name)
r = TemplateIO(html=True)
r += htmltext('<div id="appbar">')
r += htmltext('<h2>%s</h2>') % _('Data Inspector')
r += htmltext('<span class="actions">')
if self.formdef._names == 'formdefs':
if get_publisher().get_backoffice_root().is_accessible('forms'):
r += htmltext(' <a href="%s">%s</a>') % (self.formdef.get_admin_url(), _('View Form'))
elif self.formdef._names == 'carddefs':
if get_publisher().get_backoffice_root().is_accessible('cards'):
r += htmltext(' <a href="%s">%s</a>') % (self.formdef.get_admin_url(), _('View Card'))
if get_publisher().get_backoffice_root().is_accessible('workflows'):
r += htmltext(' <a href="%s">%s</a>') % (
self.formdef.workflow.get_admin_url(),
_('View Workflow'),
)
r += htmltext('</span>')
r += htmltext('</div>')
r += get_session().display_message()
r += htmltext('<div id="inspect-test-tools" class="section">')
r += htmltext('<h2>%s</h2>') % _('Test tools')
r += htmltext('<div>')
form = self.test_tools_form()
r += form.render()
r += htmltext('<div id="test-tool-result">')
r += self.test_tool_result()
r += htmltext('</div>')
r += htmltext('</div></div>')
r += htmltext('<div id="inspect-variables" class="section">')
r += htmltext('<h2>%s</h2>') % _('Variables')
r += htmltext('<ul class="form-inspector biglist">')
substvars = CompatibilityNamesDict()
substvars.update(self.filled.get_substitution_variables())
def safe(v):
if isinstance(v, str):
try:
force_text(v, charset)
except UnicodeDecodeError:
v = repr(v)
else:
try:
v = force_text(v).encode(charset)
except Exception:
v = repr(v)
return v
for k in sorted(substvars.get_flat_keys()):
if k in ('attachments',):
# blacklist, legacy
continue
k = safe(k)
v = substvars[k]
if isinstance(v, LazyFieldVar):
r += htmltext('<li><code title="%s">%s</code>') % (k, k)
r += htmltext(' <div class="value"><span>%s</span>') % v
elif isinstance(v, (types.FunctionType, types.MethodType)):
continue
elif k.endswith('form_parent') and isinstance(v, CompatibilityNamesDict) and ('form' in v):
r += htmltext('<li><code title="%s">%s_…</code>') % (k, k)
r += htmltext(
' <div class="value"><span>%(caption)s '
'(<a href="%(inspect_url)s">%(display_name)s</a>)</span>'
) % {
'caption': htmltext('<var>%s</var>') % _('variables from parent\'s request'),
'inspect_url': v['form'].backoffice_url + 'inspect',
'display_name': v['form_display_name'],
}
elif hasattr(v, 'inspect_keys'):
# skip as there are expanded identifiers
continue
else:
if isinstance(v, dict):
# only display dictionaries if they have invalid keys
# (otherwise the expanded identifiers are a better way
# to get to the values).
if all(CompatibilityNamesDict.valid_key_regex.match(k) for k in v):
continue
r += htmltext('<li><code title="%s">%s</code>') % (k, k)
r += htmltext(' <div class="value"><span>%s</span>') % ellipsize(safe(v), 10000)
if not isinstance(v, str):
r += htmltext(' <span class="type">(%r)</span>') % type(v)
r += htmltext('</div></li>')
r += htmltext('</div>')
# assigned functions
if self.formdef.workflow.roles:
workflow = self.formdef.workflow
r += htmltext('<div id="inspect-functions" class="section">')
r += htmltext('<h2>%s</h2></li>\n') % _('Functions')
r += htmltext('<ul class="form-inspector biglist">')
for key, label in (workflow.roles or {}).items():
r += htmltext('<li><span class="label">%s</span>') % label
r += htmltext('<div class="value">')
acting_role_ids = self.filled.get_function_roles(key)
acting_role_names = []
for acting_role_id in acting_role_ids:
try:
if acting_role_id.startswith('_user:'):
acting_role = get_publisher().user_class.get(acting_role_id.split(':')[1])
else:
acting_role = get_publisher().role_class.get(acting_role_id)
acting_role_names.append(acting_role.name)
except KeyError:
acting_role_names.append('%s (%s)' % (acting_role_id, _('deleted')))
if acting_role_names:
acting_role_names.sort()
r += ', '.join(acting_role_names)
else:
r += htmltext('<span class="unset">%s</span>') % _('unset')
r += htmltext('</div>')
r += htmltext('</li>\n')
r += htmltext('</ul>')
r += htmltext('</div>')
has_tracing = False
for evolution in self.filled.evolution or []:
if evolution.parts and any(isinstance(x, ActionsTracingEvolutionPart) for x in evolution.parts):
has_tracing = True
break
if has_tracing:
action_classes = {x.key: x.description for x in item_classes}
r += htmltext('<div id="inspect-timeline" class="section">')
r += htmltext('<h2>%s</h2></li>\n') % _('Actions Tracing')
r += htmltext('<ul class="form-inspector biglist">')
wf_status = None
status_admin_base_url = '#'
for evolution in self.filled.evolution:
if evolution.status and evolution.status != wf_status:
for part in evolution.parts or []:
if isinstance(part, ActionsTracingEvolutionPart):
if part.actions:
r += (
htmltext('<li><span class="event">%s</span></li>')
% part.get_event_label()
)
break
try:
status = self.filled.formdef.workflow.get_status(evolution.status)
status_label = status.name
status_admin_base_url = status.get_admin_url()
except KeyError:
status_label = _('Unavailable status (%s)') % evolution.status
status_admin_base_url = '#missing'
r += htmltext(
'<li><span class="datetime">%s</span> '
'<a class="tracing-link" href="%s"><strong>%s</strong></a></li>'
) % (
time.strftime('%Y-%m-%d %H:%M:%S', evolution.time) if evolution.time else '-',
status_admin_base_url,
status_label,
)
if evolution.status:
wf_status = evolution.status
first_part = True
for part in evolution.parts or []:
if isinstance(part, ActionsTracingEvolutionPart):
if not first_part and part.actions:
r += htmltext('<li><span class="event">%s</span></li>') % part.get_event_label()
first_part = False
for action_ts, action_key, action_id in part.actions:
action_label = action_classes.get(action_key, action_key)
try:
url = '%sitems/%s/' % (
part.get_base_url(self.filled.formdef.workflow, wf_status),
action_id,
)
except KeyError:
url = '#missing-%s' % action_id
r += htmltext(
'<li><span class="datetime">%s</span> '
'<a class="tracing-link" href="%s">%s</a></li>'
) % (
action_ts.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],
url,
action_label,
)
r += htmltext('</ul>')
r += htmltext('</div>')
# markers stack
if '_markers_stack' in (self.filled.workflow_data or {}):
r += htmltext('<div id="inspect-markers" class="section">')
r += htmltext('<h2>%s</h2>') % _('Markers Stack')
r += htmltext('<ul class="form-inspector biglist">')
for marker in reversed(self.filled.workflow_data['_markers_stack']):
status = self.filled.get_status(marker['status_id'])
if status:
r += htmltext('<li><span class="status">%s</span></li>') % status.name
else:
r += htmltext('<li><span class="status">%s</span></li>') % _('Unknown')
r += htmltext('</ul>')
r += htmltext('</div>')
children = list(self.filled.iter_target_datas())
if children:
r += htmltext('<div id="inspect-related" class="section">')
r += htmltext('<h2>%s</h2>') % _('Related Forms/Cards')
r += htmltext('<ul class="form-inspector biglist">')
for child, origin in children:
if isinstance(child, str):
r += htmltext('<li><a href="">%s (%s)</a></li>') % (child, origin)
else:
r += htmltext('<li><a href="%s">%s (%s)</a></li>') % (
child.get_url(backoffice=True),
child.get_display_name(),
origin,
)
r += htmltext('</ul>')
r += htmltext('</div>')
return r.getvalue()
def inspect_tool(self):
if not (
get_publisher().get_backoffice_root().is_accessible('forms')
or get_publisher().get_backoffice_root().is_accessible('workflows')
):
raise errors.AccessForbiddenError()
get_response().filter = {'raw': True}
return self.test_tool_result()
class FakeField:
def __init__(self, id, type_, label, addable=True):
self.id = id
self.type = type_
self.label = force_text(label)
self.fake = True
self.varname = id.replace('-', '_')
self.store_display_value = None
self.addable = addable
def get_view_value(self, value):
# just here to quack like a duck
return None
def get_csv_heading(self):
return [self.label]
def get_csv_value(self, element, **kwargs):
return [element]
@property
def has_relations(self):
return bool(self.id == 'user-label')
class RelatedField:
is_related_field = True
type = 'related-field'
store_display_value = None
varname = None
def __init__(self, carddef, field, parent_field):
self.carddef = carddef
self.related_field = field
self.parent_field = parent_field
self.parent_field_id = parent_field.id
@property
def id(self):
return '%s$%s' % (self.parent_field_id, self.related_field.id)
@property
def label(self):
return '%s - %s' % (self.parent_field.label, self.related_field.label)
def __repr__(self):
return '<%s (card: %r, parent: %r, related: %r)>' % (
self.__class__.__name__,
self.carddef,
self.parent_field.label,
self.related_field.label,
)
def get_view_value(self, value, **kwargs):
if value is None:
return ''
if isinstance(value, bool):
return _('Yes') if value else _('No')
if isinstance(value, datetime.date):
return misc.strftime(misc.date_format(), value)
return value
def get_view_short_value(self, value, max_len=30, **kwargs):
return self.get_view_value(value)
def get_csv_heading(self):
return [self.label]
def get_csv_value(self, value, **kwargs):
return [self.get_view_value(value)]
class UserRelatedField(RelatedField):
# it is named 'user-label' and not 'user' for compatibility with existing
# listings, as the 'classic' user column is named 'user-label'.
parent_field_id = 'user-label'
def __init__(self, field):
self.related_field = field
@property
def label(self):
return _('%s of User') % self.related_field.label
def do_graphs_section(period_start=None, period_end=None, criterias=None):
from wcs import sql
r = TemplateIO(html=True)
monthly_totals = sql.get_monthly_totals(period_start, period_end, criterias)[-12:]
yearly_totals = sql.get_yearly_totals(period_start, period_end, criterias)[-10:]
if not monthly_totals:
monthly_totals = [('%s-%s' % datetime.date.today().timetuple()[:2], 0)]
if not yearly_totals:
yearly_totals = [(datetime.date.today().year, 0)]
weekday_totals = sql.get_weekday_totals(period_start, period_end, criterias)
hour_totals = sql.get_hour_totals(period_start, period_end, criterias)
r += htmltext(
'''<script>
var weekday_line = %(weekday_line)s;
var hour_line = %(hour_line)s;
var month_line = %(month_line)s;
var year_line = %(year_line)s;
</script>'''
% {
'weekday_line': json.dumps(weekday_totals, cls=misc.JSONEncoder),
'hour_line': json.dumps(hour_totals, cls=misc.JSONEncoder),
'month_line': json.dumps(monthly_totals, cls=misc.JSONEncoder),
'year_line': json.dumps(yearly_totals, cls=misc.JSONEncoder),
}
)
if len(yearly_totals) > 1:
r += htmltext('<h3>%s</h3>') % _('Submissions by year')
r += htmltext('<div id="chart_years" style="height:160px; width:100%;"></div>')
r += htmltext('<h3>%s</h3>') % _('Submissions by month')
r += htmltext('<div id="chart_months" style="height:160px; width:100%;"></div>')
r += htmltext('<h3>%s</h3>') % _('Submissions by weekday')
r += htmltext('<div id="chart_weekdays" style="height:160px; width:100%;"></div>')
r += htmltext('<h3>%s</h3>') % _('Submissions by hour')
r += htmltext('<div id="chart_hours" style="height:160px; width:100%;"></div>')
get_response().add_javascript(
[
'jquery.js',
'jqplot/jquery.jqplot.min.js',
'jqplot/plugins/jqplot.canvasTextRenderer.min.js',
'jqplot/plugins/jqplot.canvasAxisLabelRenderer.min.js',
'jqplot/plugins/jqplot.canvasAxisTickRenderer.min.js',
'jqplot/plugins/jqplot.categoryAxisRenderer.min.js',
'jqplot/plugins/jqplot.barRenderer.min.js',
]
)
get_response().add_javascript_code(
'''
function wcs_draw_graphs() {
$.jqplot ('chart_weekdays', [weekday_line], {
series:[{renderer:$.jqplot.BarRenderer}],
axesDefaults: {
tickRenderer: $.jqplot.CanvasAxisTickRenderer,
tickOptions: { angle: -30, }
},
axes: { xaxis: { renderer: $.jqplot.CategoryAxisRenderer } }
});
$.jqplot ('chart_hours', [hour_line], {
axesDefaults: {
tickRenderer: $.jqplot.CanvasAxisTickRenderer,
tickOptions: { angle: -30, }
},
axes: { xaxis: { renderer: $.jqplot.CategoryAxisRenderer }, yaxis: {min: 0} }
});
$.jqplot ('chart_months', [month_line], {
axesDefaults: {
tickRenderer: $.jqplot.CanvasAxisTickRenderer,
tickOptions: { angle: -30, }
},
axes: { xaxis: { renderer: $.jqplot.CategoryAxisRenderer }, yaxis: {min: 0} }
});
if ($('#chart_years').length) {
$.jqplot ('chart_years', [year_line], {
series:[{renderer:$.jqplot.BarRenderer}],
axesDefaults: {
tickRenderer: $.jqplot.CanvasAxisTickRenderer,
tickOptions: { angle: -30, }
},
axes: { xaxis: { renderer: $.jqplot.CategoryAxisRenderer }, yaxis: {min: 0} }
});
}
}
$(document).ready(function(){
wcs_draw_graphs();
});
'''
)
return r.getvalue()
def get_global_criteria(request, parsed_values=None, include_drafts=False):
"""
Parses the request query string and returns a list of criterias suitable
for select() usage. The parsed_values parameter can be given a dictionary,
to be filled with the parsed values.
"""
criterias = []
if not include_drafts:
criterias.append(NotEqual('status', 'draft'))
try:
period_start = misc.get_as_datetime(request.form.get('start')).timetuple()
criterias.append(GreaterOrEqual('receipt_time', period_start))
parsed_values['period_start'] = period_start
except (ValueError, TypeError):
pass
try:
period_end = misc.get_as_datetime(request.form.get('end')).timetuple()
criterias.append(LessOrEqual('receipt_time', period_end))
parsed_values['period_end'] = period_end
except (ValueError, TypeError):
pass
return criterias
def format_time(t, units=2):
days = int(t / 86400)
hours = int((t - days * 86400) / 3600)
minutes = int((t - days * 86400 - hours * 3600) / 60)
seconds = t % 60
if units == 1:
if days:
return _('%d day(s)') % days
if hours:
return _('%d hour(s)') % hours
if minutes:
return _('%d minute(s)') % minutes
elif units == 2:
if days:
return _('%(days)d day(s) and %(hours)d hour(s)') % {'days': days, 'hours': hours}
if hours:
return _('%(hours)d hour(s) and %(minutes)d minute(s)') % {'hours': hours, 'minutes': minutes}
if minutes:
return _('%(minutes)d minute(s) and %(seconds)d seconds') % {
'minutes': minutes,
'seconds': seconds,
}
return _('%d seconds') % seconds
class MassActionAfterJob(AfterJob):
def __init__(self, formdef, **kwargs):
super().__init__(formdef_class=formdef.__class__, formdef_id=formdef.id, **kwargs)
self.query_string = kwargs.get('query_string')
def execute(self):
formdef = self.kwargs['formdef_class'].get(self.kwargs['formdef_id'])
item_ids = self.kwargs['item_ids']
action_id = self.kwargs['action_id']
status_filter = self.kwargs['status_filter']
user = get_publisher().user_class.get(self.kwargs['user_id'])
multi_actions = FormPage.get_multi_actions(formdef, user, status_filter)
for action in multi_actions:
if action['action'].id == action_id:
break
else:
# action not found
return
formdatas = formdef.data_class().get_ids(item_ids, order_by='receipt_time')
self.completion_status = f'0/{len(formdatas)}'
self.store()
publisher = get_publisher()
if formdatas:
oldest_lazy_form = formdatas[0].get_as_lazy()
for i, formdata in enumerate(formdatas):
publisher.substitutions.reset()
publisher.substitutions.feed(publisher)
publisher.substitutions.feed(user)
publisher.substitutions.feed(formdef)
publisher.substitutions.feed(formdata)
publisher.substitutions.feed(
{
'oldest_form': oldest_lazy_form,
'mass_action_index': i,
'mass_action_length': len(formdatas),
}
)
if getattr(action['action'], 'status_action', False):
# manual jump action
from wcs.wf.jump import jump_and_perform
jump_and_perform(formdata, action['action'].action)
else:
# global action
formdata.perform_global_action(action['action'].id, user)
self.completion_status = f'{i + 1}/{len(formdatas)}'
self.store()
def done_action_url(self):
return self.kwargs['return_url']
def done_action_label(self):
return _('Back to Listing')
def done_button_attributes(self):
return {'data-redirect-auto': 'true'}
class CsvExportAfterJob(AfterJob):
label = _('Exporting to CSV file')
def __init__(self, formdef, **kwargs):
super().__init__(formdef_class=formdef.__class__, formdef_id=formdef.id, **kwargs)
self.file_name = '%s.csv' % formdef.url_name
def csv_tuple_heading(self, fields):
heading_fields = [] # '#id', _('time'), _('userlabel'), _('status')]
for field in fields:
heading_fields.extend(field.get_csv_heading())
return heading_fields
def get_spreadsheet_line(self, fields, data):
elements = []
for field in fields:
element = data.get_field_view_value(field) or ''
display_value = None
if field.store_display_value:
display_value = data.data.get('%s_display' % field.id) or ''
for value in field.get_csv_value(element, display_value=display_value):
elements.append({'field': field, 'value': value, 'native_value': element})
return elements
def execute(self):
formdef = self.kwargs['formdef_class'].get(self.kwargs['formdef_id'])
selected_filter = self.kwargs['selected_filter']
fields = self.kwargs['fields']
query = self.kwargs['query']
criterias = self.kwargs['criterias']
order_by = self.kwargs['order_by']
user = get_publisher().user_class.get(self.kwargs['user_id'])
items, total_count = FormDefUI(formdef).get_listing_items(
fields, selected_filter, user=user, query=query, criterias=criterias, order_by=order_by
)
return self.create_export(formdef, fields, items, total_count)
def create_export(self, formdef, fields, items, total_count):
output = io.StringIO()
csv_output = csv.writer(output)
if not self.kwargs.get('skip_header_line'):
csv_output.writerow(self.csv_tuple_heading(fields))
for filled in items:
csv_output.writerow(tuple(x['value'] for x in self.get_spreadsheet_line(fields, filled)))
self.file_content = output.getvalue()
self.content_type = 'text/csv'
self.store()
def done_action_url(self):
formdef = self.kwargs['formdef_class'].get(self.kwargs['formdef_id'])
return formdef.get_url(backoffice=True) + 'export?download=%s' % self.id
def done_action_label(self):
return _('Download Export')
def done_button_attributes(self):
return {'download': self.file_name}
class OdsExportAfterJob(CsvExportAfterJob):
label = _('Exporting to ODS file')
def __init__(self, formdef, **kwargs):
super().__init__(formdef=formdef, **kwargs)
self.file_name = '%s.ods' % formdef.url_name
def create_export(self, formdef, fields, items, total_count):
workbook = ods.Workbook(encoding='utf-8')
ws = workbook.add_sheet(formdef.name)
header_line_counter = 0
if not self.kwargs.get('skip_header_line'):
header_line_counter = 1
for i, field in enumerate(self.csv_tuple_heading(fields)):
ws.write(0, i, field)
for i, formdata in enumerate(items):
for j, item in enumerate(self.get_spreadsheet_line(fields, formdata)):
ws.write(
i + header_line_counter,
j,
item['value'],
formdata=formdata,
data_field=item['field'],
native_value=item['native_value'],
)
output = io.BytesIO()
workbook.save(output)
self.file_content = output.getvalue()
self.content_type = 'application/vnd.oasis.opendocument.spreadsheet'
self.store()