624 lines
23 KiB
Plaintext
624 lines
23 KiB
Plaintext
# w.c.s. - web application for online forms
|
|
# Copyright (C) 2005-2010 Entr'ouvert
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 2 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import os
|
|
import csv
|
|
import cStringIO
|
|
import time
|
|
|
|
from quixote import get_session, get_session_manager, get_publisher, get_request, get_response, redirect
|
|
from quixote.directory import Directory, AccessControlled
|
|
|
|
from qommon.backoffice import BackofficeRootDirectory
|
|
from qommon.backoffice.menu import html_top
|
|
|
|
from qommon import misc, get_logger
|
|
from qommon.afterjobs import AfterJob
|
|
from qommon import errors
|
|
|
|
from wcs.forms.common import FormStatusPage
|
|
|
|
from wcs.categories import Category
|
|
from wcs.formdef import FormDef
|
|
|
|
from wcs.forms.backoffice import FormDefUI
|
|
|
|
try:
|
|
import xlwt
|
|
except ImportError:
|
|
xlwt = None
|
|
|
|
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 RootDirectory(BackofficeRootDirectory):
|
|
_q_exports = ['', 'forms', 'pending']
|
|
|
|
items = [
|
|
('forms', N_('Forms')),
|
|
('/', N_('WCS Form Server'))]
|
|
|
|
def _q_index [html] (self):
|
|
return self.forms()
|
|
|
|
def forms [html] (self):
|
|
get_response().breadcrumb.append(('forms', _('Forms')))
|
|
html_top('forms', _('Forms'))
|
|
get_response().filter['sidebar'] = self.get_sidebar()
|
|
|
|
user = get_request().user
|
|
|
|
forms_without_pending_stuff = []
|
|
forms_with_pending_stuff = []
|
|
|
|
def append_form_entry(formdef):
|
|
formdef_data_class = formdef.data_class()
|
|
count_forms = formdef_data_class.count() - len(formdef_data_class.get_ids_with_indexed_value('status', 'draft'))
|
|
not_endpoint_status = formdef.workflow.get_not_endpoint_status()
|
|
not_endpoint_status_ids = ['wf-%s' % x.id for x in not_endpoint_status]
|
|
pending_forms = []
|
|
for status in not_endpoint_status_ids:
|
|
pending_forms.extend(formdef_data_class.get_ids_with_indexed_value(
|
|
'status', status))
|
|
if len(pending_forms) == 0:
|
|
forms_without_pending_stuff.append((formdef, len(pending_forms), count_forms))
|
|
else:
|
|
forms_with_pending_stuff.append((formdef, len(pending_forms), count_forms))
|
|
|
|
if user:
|
|
for formdef in FormDef.select(order_by='name', ignore_errors=True):
|
|
if formdef.disabled:
|
|
continue
|
|
if user.is_admin or formdef.is_of_concern_for(user):
|
|
append_form_entry(formdef)
|
|
|
|
if forms_with_pending_stuff:
|
|
'<div class="bo-block">'
|
|
'<h2>%s</h2>' % _('Forms in your care')
|
|
self.display_forms(forms_with_pending_stuff)
|
|
'</div>'
|
|
|
|
if forms_without_pending_stuff:
|
|
'<div class="bo-block">'
|
|
'<h2>%s</h2>' % _('Other Forms')
|
|
self.display_forms(forms_without_pending_stuff)
|
|
'</div>'
|
|
|
|
def display_forms [html] (self, forms_list):
|
|
'<ul>'
|
|
cats = Category.select(order_by = 'name')
|
|
one = False
|
|
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
|
|
'<li>%s</li>' % cat_name
|
|
'<ul>'
|
|
for formdef, no_pending, no_total in l2:
|
|
'<li><a href="%s/">%s</a>' % (formdef.url_name, formdef.name)
|
|
if no_pending:
|
|
_(': %(pending)s open on %(total)s') % {'pending': no_pending,
|
|
'total': no_total}
|
|
else:
|
|
_(': %(total)s items') % {'total': no_total}
|
|
'</li>'
|
|
one = True
|
|
'</ul>'
|
|
'</ul>'
|
|
|
|
def pending [html] (self):
|
|
# kept as a redirection for compatibility with possible bookmarks
|
|
return redirect('.')
|
|
|
|
def _q_lookup(self, component):
|
|
return FormPage(component)
|
|
|
|
|
|
class FakeField:
|
|
def __init__(self, id, type_, label):
|
|
self.id = id
|
|
self.type = type_
|
|
self.label = label
|
|
|
|
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):
|
|
return [element]
|
|
|
|
class FormPage(Directory):
|
|
_q_exports = ['', 'csv', 'stats', 'xls', 'pending', 'export']
|
|
|
|
def __init__(self, component):
|
|
try:
|
|
self.formdef = FormDef.get_by_urlname(component)
|
|
except KeyError:
|
|
raise errors.TraversalError()
|
|
|
|
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):
|
|
if session.user:
|
|
raise errors.AccessForbiddenError()
|
|
else:
|
|
raise errors.AccessUnauthorizedError()
|
|
get_response().breadcrumb.append( (component + '/', self.formdef.name) )
|
|
|
|
def get_formdata_sidebar [html] (self, qs=''):
|
|
'<ul>'
|
|
#' <li><a href="list%s">%s</a></li>' % (qs, _('List of results'))
|
|
' <li><a href="csv%s">%s</a></li>' % (qs, _('CSV Export'))
|
|
if xlwt:
|
|
'<li><a href="xls%s">%s</a></li>' % (qs, _('Excel Export'))
|
|
' <li><a href="stats">%s</a></li>' % _('Statistics')
|
|
'</ul>'
|
|
|
|
def get_fields_sidebar [html] (self, selected_filter, fields, offset=None, count=None):
|
|
'<form>'
|
|
if offset:
|
|
'<input type="hidden" name="offset" value="%s"/>' % offset
|
|
if count:
|
|
'<input type="hidden" name="count" value="%s"/>' % count
|
|
waitpoint_status = self.formdef.workflow.get_waitpoint_status()
|
|
if waitpoint_status:
|
|
'<h3>%s</h3>' % _('Status to display')
|
|
'<select name="filter" onchange="this.form.submit()">'
|
|
filters = [('all', _('All')), ('pending', _('Pending')), ('done', _('Done'))]
|
|
for status in waitpoint_status:
|
|
filters.append((status.id, status.name))
|
|
for filter_id, filter_label in filters:
|
|
if filter_id == selected_filter:
|
|
selected = ' selected="selected"'
|
|
else:
|
|
selected = ''
|
|
'<option value="%s"%s/>%s</option>' % (filter_id, selected, filter_label)
|
|
'</select>'
|
|
'<noscript><input type="submit" value="%s"/></noscript>' % _('Refresh')
|
|
|
|
'<h3>%s</h3>' % _('Fields to display')
|
|
'<ul id="fields-filter">'
|
|
for field in self.get_formdef_fields():
|
|
if not hasattr(field, str('get_view_value')):
|
|
continue
|
|
'<li><input type="checkbox" name="%s"' % field.id
|
|
if field.id in [x.id for x in fields]:
|
|
' checked="checked"'
|
|
' id="fields-filter-%s"' % field.id
|
|
'/>'
|
|
'<label for="fields-filter-%s">%s</label>' % (field.id, field.label)
|
|
'</li>'
|
|
'</ul>'
|
|
'<input type="submit" value="%s"/>' % _('Refresh')
|
|
'</form>'
|
|
|
|
def get_formdef_fields(self):
|
|
fields = []
|
|
fields.append(FakeField('id', 'id', _('Identifier')))
|
|
fields.append(FakeField('time', 'time', _('Time')))
|
|
fields.append(FakeField('user-label', 'user-label', _('User Label')))
|
|
fields.extend(self.formdef.fields)
|
|
fields.append(FakeField('status', 'status', _('Status')))
|
|
|
|
return fields
|
|
|
|
def get_fields_from_query(self):
|
|
field_ids = [x for x in get_request().form.keys()]
|
|
if not field_ids:
|
|
field_ids = ['id', 'time', 'user-label']
|
|
for field in self.formdef.fields:
|
|
if hasattr(field, str('get_view_value')) and field.in_listing:
|
|
field_ids.append(field.id)
|
|
field_ids.append('status')
|
|
|
|
fields = []
|
|
for field in self.get_formdef_fields():
|
|
if field.id in field_ids:
|
|
fields.append(field)
|
|
|
|
return fields
|
|
|
|
def get_filter_from_query(self):
|
|
if 'filter' in get_request().form:
|
|
return get_request().form['filter']
|
|
if self.formdef.workflow.possible_status:
|
|
return 'pending'
|
|
return 'all'
|
|
|
|
def _q_index [html] (self):
|
|
get_logger().info('backoffice - form %s - listing' % self.formdef.name)
|
|
|
|
fields = self.get_fields_from_query()
|
|
selected_filter = self.get_filter_from_query()
|
|
|
|
count = get_request().form.get('count', 0)
|
|
offset = get_request().form.get('offset', 0)
|
|
|
|
qs = ''
|
|
if get_request().get_query():
|
|
qs = '?' + get_request().get_query()
|
|
|
|
html_top('forms', '%s - %s' % (_('Listing'), self.formdef.name))
|
|
'<h2>%s - %s</h2>' % (self.formdef.name, _('Listing'))
|
|
FormDefUI(self.formdef).listing(fields=fields,
|
|
selected_filter=selected_filter, include_form=True,
|
|
count=int(count), offset=int(offset))
|
|
|
|
get_response().filter['sidebar'] = self.get_formdata_sidebar(qs) + \
|
|
self.get_fields_sidebar(selected_filter, fields, count=count, offset=offset)
|
|
|
|
def pending [html] (self):
|
|
get_logger().info('backoffice - form %s - pending' % self.formdef.name)
|
|
get_response().breadcrumb.append( ('pending', _('Pending Forms')) )
|
|
html_top('forms', '%s - %s' % (_('Pending Forms'), self.formdef.name))
|
|
'<h2>%s - %s</h2>' % (self.formdef.name, _('Pending Forms'))
|
|
|
|
not_endpoint_status = [('wf-%s' % x.id, x.name) for x in \
|
|
self.formdef.workflow.get_not_endpoint_status()]
|
|
|
|
nb_status = len(not_endpoint_status)
|
|
column2 = nb_status/2
|
|
'<div class="splitcontent-left">'
|
|
for i, (status_id, status_label) in enumerate(not_endpoint_status):
|
|
if i > 0 and i == column2:
|
|
'</div>'
|
|
'<div class="splitcontent-right">'
|
|
status_forms = self.formdef.data_class().get_with_indexed_value(
|
|
str('status'), status_id)
|
|
if not status_forms:
|
|
continue
|
|
status_forms.sort(lambda x,y: cmp(getattr(x, str('receipt_time')),
|
|
getattr(y, str('receipt_time'))))
|
|
status_forms.reverse()
|
|
'<div class="bo-block">'
|
|
'<h3>%s</h3>' % _('Forms with status "%s"') % status_label
|
|
|
|
'<ul>'
|
|
for f in status_forms:
|
|
try:
|
|
u = get_publisher().user_class.get(f.user_id)
|
|
userlabel = u.display_name
|
|
except KeyError:
|
|
userlabel = _('unknown user')
|
|
'<li><a href="%s/">%s, %s</a></li>' % (
|
|
f.id,
|
|
misc.localstrftime(f.receipt_time),
|
|
userlabel)
|
|
'</ul>'
|
|
'</div>'
|
|
'</div>'
|
|
|
|
'<p class="clear"><a href=".">%s</a></p>' % _('Back')
|
|
|
|
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 csv_tuple(self, fields, data):
|
|
elements = []
|
|
for field in fields:
|
|
if field.type == 'id':
|
|
element = str(data.id)
|
|
elif field.type == 'time':
|
|
element = misc.localstrftime(data.receipt_time)
|
|
elif field.type == 'user-label':
|
|
try:
|
|
element = get_publisher().user_class.get(data.user_id).display_name
|
|
except:
|
|
element = '-'
|
|
pass
|
|
elif field.type == 'status':
|
|
element = data.get_status_label()
|
|
else:
|
|
element = data.data.get(field.id, '') or ''
|
|
elements.extend(field.get_csv_value(element))
|
|
return elements
|
|
|
|
def csv(self):
|
|
fields = self.get_fields_from_query()
|
|
selected_filter = self.get_filter_from_query()
|
|
|
|
class Exporter:
|
|
def __init__(self, formpage, formdef, fields, selected_filter):
|
|
self.formpage = formpage
|
|
self.formdef = formdef
|
|
self.fields = fields
|
|
self.selected_filter = selected_filter
|
|
|
|
def export(self, job=None):
|
|
self.output = cStringIO.StringIO()
|
|
csv_output = csv.writer(self.output)
|
|
|
|
csv_output.writerow(self.formpage.csv_tuple_heading(self.fields))
|
|
|
|
items, total_count = FormDefUI(self.formdef).get_listing_items(self.selected_filter)
|
|
|
|
for filled in items:
|
|
csv_output.writerow(self.formpage.csv_tuple(self.fields, filled))
|
|
|
|
if job:
|
|
job.file_content = self.output.getvalue()
|
|
job.content_type = 'text/csv'
|
|
job.file_name = '%s.csv' % self.formdef.url_name
|
|
job.store()
|
|
|
|
get_logger().info('backoffice - form %s - listing csv' % self.formdef.name)
|
|
|
|
count = self.formdef.data_class().count()
|
|
exporter = Exporter(self, self.formdef, fields, selected_filter)
|
|
if count > 100: # Arbitrary threshold
|
|
job = get_response().add_after_job(
|
|
str(N_('Exporting forms in CSV')),
|
|
exporter.export)
|
|
return redirect('export?job=%s' % job.id)
|
|
else:
|
|
exporter.export()
|
|
|
|
response = get_response()
|
|
response.set_content_type('text/csv')
|
|
response.set_header('content-disposition', 'attachment; filename=%s.csv' % self.formdef.url_name)
|
|
return exporter.output.getvalue()
|
|
|
|
def export [html] (self):
|
|
if get_request().form.get('download'):
|
|
return self.export_download()
|
|
|
|
try:
|
|
job = AfterJob.get(get_request().form.get('job'))
|
|
except KeyError:
|
|
return redirect('.')
|
|
|
|
html_top('forms', title=_('Exporting'))
|
|
get_session().display_message()
|
|
get_response().add_javascript(['jquery.js', 'interface.js', 'afterjob.js'])
|
|
'<dl class="job-status">'
|
|
'<dt>'
|
|
_(job.label)
|
|
'</dt>'
|
|
'<dd>'
|
|
'<span class="afterjob" id="%s">' % job.id
|
|
_(job.status)
|
|
'</span>'
|
|
'</dd>'
|
|
'</dl>'
|
|
|
|
'<div class="done">'
|
|
'<a href="export?download=%s">%s</a>' % (job.id, _('Download Export'))
|
|
'</div>'
|
|
|
|
def export_download(self):
|
|
try:
|
|
job = AfterJob.get(get_request().form.get('download'))
|
|
except KeyError:
|
|
return redirect('.')
|
|
|
|
if not job.status == 'completed':
|
|
raise TraversalError()
|
|
response = get_response()
|
|
response.set_content_type(job.content_type)
|
|
response.set_header('content-disposition',
|
|
'attachment; filename=%s' % job.file_name)
|
|
return job.file_content
|
|
|
|
def xls(self):
|
|
if xlwt is None:
|
|
raise errors.TraversalError()
|
|
|
|
fields = self.get_fields_from_query()
|
|
selected_filter = self.get_filter_from_query()
|
|
|
|
class Exporter:
|
|
def __init__(self, formpage, formdef, fields, selected_filter):
|
|
self.formpage = formpage
|
|
self.formdef = formdef
|
|
self.fields = fields
|
|
self.selected_filter = selected_filter
|
|
|
|
def export(self, job=None):
|
|
w = xlwt.Workbook(encoding=get_publisher().site_charset)
|
|
ws = w.add_sheet('1')
|
|
|
|
for i, f in enumerate(self.formpage.csv_tuple_heading(self.fields)):
|
|
ws.write(0, i, f)
|
|
|
|
items, total_count = FormDefUI(self.formdef).get_listing_items(self.selected_filter)
|
|
|
|
for i, filled in enumerate(items):
|
|
for j, elem in enumerate(self.formpage.csv_tuple(fields, filled)):
|
|
ws.write(i+1, j, elem)
|
|
|
|
self.output = cStringIO.StringIO()
|
|
w.save(self.output)
|
|
|
|
if job:
|
|
job.file_content = self.output.getvalue()
|
|
job.content_type = 'application/vnd.ms-excel'
|
|
job.file_name = '%s.xls' % self.formdef.url_name
|
|
job.store()
|
|
|
|
get_logger().info('backoffice - form %s - as excel' % self.formdef.name)
|
|
|
|
count = self.formdef.data_class().count()
|
|
exporter = Exporter(self, self.formdef, fields, selected_filter)
|
|
if count > 100: # Arbitrary threshold
|
|
job = get_response().add_after_job(
|
|
str(N_('Exporting forms in Excel format')),
|
|
exporter.export)
|
|
return redirect('export?job=%s' % job.id)
|
|
else:
|
|
exporter.export()
|
|
|
|
response = get_response()
|
|
response.set_content_type('application/vnd.ms-excel')
|
|
response.set_header('content-disposition', 'attachment; filename=%s.xls' % self.formdef.url_name)
|
|
return exporter.output.getvalue()
|
|
|
|
def stats [html] (self):
|
|
get_logger().info('backoffice - form %s - stats' % self.formdef.name)
|
|
html_top('forms', '%s - %s' % (_('Form'), self.formdef.name))
|
|
get_response().breadcrumb.append( ('stats', _('Statistics')) )
|
|
'<h2>%s - %s</h2>' % (self.formdef.name, _('Statistics'))
|
|
values = self.formdef.data_class().select(lambda x: x.status != 'draft')
|
|
|
|
no_forms = len(values)
|
|
'<p>%s %d</p>' % (_('Total number of records:'), no_forms)
|
|
|
|
if self.formdef.workflow:
|
|
'<ul>'
|
|
for status in self.formdef.workflow.possible_status:
|
|
'<li>%s: %d</li>' % (status.name,
|
|
len([x for x in values if x.status == 'wf-%s' % status.id]))
|
|
'</ul>'
|
|
|
|
self.stats_fields(values)
|
|
self.stats_resolution_time(values)
|
|
|
|
'<a href=".">%s</a>' % _('Back')
|
|
|
|
def stats_fields [html] (self, values):
|
|
had_page = False
|
|
last_page = None
|
|
last_title = None
|
|
for f in self.formdef.fields:
|
|
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:
|
|
'</div>'
|
|
'<div class="page">'
|
|
'<h3>%s</h3>' % last_page
|
|
had_page = True
|
|
last_page = None
|
|
if last_title:
|
|
'<h3>%s</h3>' % last_title
|
|
last_title = None
|
|
t
|
|
|
|
if had_page:
|
|
'</div>'
|
|
|
|
def stats_resolution_time [html] (self, values):
|
|
possible_status = [('wf-%s' % x.id, x.id) for x in self.formdef.workflow.possible_status]
|
|
|
|
if len(possible_status) < 2:
|
|
return
|
|
|
|
'<h2>%s</h2>' % _('Resolution time')
|
|
|
|
for status, status_id in possible_status:
|
|
res_time_forms = [
|
|
(time.mktime(x.evolution[-1].time) - time.mktime(x.receipt_time)) \
|
|
for x in values if x.status == status and x.evolution]
|
|
if not res_time_forms:
|
|
continue
|
|
res_time_forms.sort()
|
|
sum_times = sum(res_time_forms)
|
|
len_times = len(res_time_forms)
|
|
'<h3>%s</h3>' % (_('To Status "%s"') % self.formdef.workflow.get_status(status_id).name)
|
|
'<ul>'
|
|
' <li>%s %s</li>' % (_('Minimum Time:'), format_time(min(res_time_forms)))
|
|
' <li>%s %s</li>' % (_('Maximum Time:'), format_time(max(res_time_forms)))
|
|
' <li>%s %s</li>' % (_('Range:'), format_time(max(res_time_forms)-min(res_time_forms)))
|
|
mean = sum_times/len_times
|
|
' <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
|
|
' <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:
|
|
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)
|
|
' <li>%s %s</li>' % (_('Standard Deviation:'), format_time(std_dev))
|
|
|
|
'</ul>'
|
|
|
|
def _q_lookup(self, component):
|
|
try:
|
|
filled = self.formdef.data_class().get(component)
|
|
except KeyError:
|
|
raise errors.TraversalError()
|
|
|
|
return FormBackOfficeStatusPage(self.formdef, filled)
|
|
|
|
|
|
|
|
class FormBackOfficeStatusPage(FormStatusPage):
|
|
def html_top(self, title = None):
|
|
return html_top('forms', title)
|
|
|
|
def _q_index [html] (self):
|
|
return self.status()
|
|
|