wcs/wcs/formdef.py

698 lines
24 KiB
Python

# w.c.s. - web application for online forms
# Copyright (C) 2005-2010 Entr'ouvert
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, see <http://www.gnu.org/licenses/>.
import new
import urllib
try:
set
except NameError:
from sets import Set as set
import sys
try:
import elementtree.ElementTree as ET
except ImportError:
try:
import xml.etree.ElementTree as ET
except ImportError:
ET = None
import datetime
from qommon.storage import StorableObject
from quixote import get_request, get_session, get_publisher
from qommon import emails
from qommon.form import *
from qommon.misc import simplify, date_format, strftime, localstrftime
from qommon import get_cfg
from qommon.substitution import Substitutions
from formdata import FormData
from roles import Role, logged_users_role
from categories import Category
from wcs.workflows import Workflow, CommentableWorkflowStatusItem, \
SendmailWorkflowStatusItem, ChoiceWorkflowStatusItem, \
template_on_formdata
import fields
class FormField:
### only used to unpickle form fields from older (<200603) versions
def __setstate__(self, dict):
type = dict['type']
self.real_field = fields.get_field_class_by_type(type)(**dict)
class FormDef(StorableObject):
_names = 'formdefs'
_indexes = ['url_name']
name = None
url_name = None
fields = None
receiver_id = None
category_id = None
workflow_id = None
workflow_options = None
roles = None
discussion = False
confirmation = True
signing = None
detailed_emails = True
disabled = False
only_allow_one = False
allow_drafts = False
disabled_redirection = None
always_advertise = False
publication_date = None
expiration_date = None
acl_read = 'owner' # one of ('none', 'owner', 'roles', 'all')
private_status_and_history = False
def migrate(self):
changed = False
if self.__dict__.has_key('receiver'):
self.receiver_id = self.__dict__['receiver']
del self.__dict__['receiver']
changed = True
if self.__dict__.has_key('category'):
self.category_id = self.__dict__['category']
del self.__dict__['category']
changed = True
if not self.url_name:
try:
int(self.id)
except ValueError:
self.url_name = self.id
else:
self.url_name = self.get_new_url_name()
changed = True
if self.fields and type(self.fields[0]) is dict:
for f in self.fields:
if f.has_key('name'):
f['label'] = f['name']
del f['name']
self.fields = [FormField(**x) for x in self.fields]
for i, f in enumerate(self.fields):
f.id = str(i)
for formdata in self.data_class().select():
for f in self.fields:
if not formdata.data.has_key(f.label):
continue
formdata.data[f.id] = formdata.data[f.label]
del formdata.data[f.label]
formdata.store()
changed = True
if self.fields and isinstance(self.fields[0], FormField):
# migration from generic FormField to specific Field classes
# (200603)
self.fields = [x.real_field for x in self.fields]
if self.__dict__.has_key('public'):
if self.__dict__.get('public'):
self.acl_read = 'all'
del self.__dict__['public']
changed = True
if changed:
self.store()
def data_class(self):
if hasattr(sys.modules['formdef'], self.url_name.title()):
return getattr(sys.modules['formdef'], self.url_name.title())
if get_publisher().has_site_option('postgresql') and get_cfg('postgresql', {}):
import sql
table_name = 'formdata_' + self.url_name.replace('-', '_')
cls = new.classobj(self.url_name.title(), (sql.SqlFormData,),
{'_formdef': self,
'_table_name': table_name})
else:
cls = new.classobj(self.url_name.title(), (FormData,),
{'_names': 'form-%s' % self.url_name})
setattr(sys.modules['formdef'], self.url_name.title(), cls)
setattr(sys.modules['wcs.formdef'], self.url_name.title(), cls)
return cls
def get_new_url_name(self):
new_url_name = simplify(self.name)
base_new_url_name = new_url_name
suffix_no = 0
while True:
try:
formdef = self.get_by_urlname(new_url_name, ignore_migration=True)
except KeyError:
break
if formdef.id == self.id:
break
suffix_no += 1
new_url_name = '%s-%s' % (base_new_url_name, suffix_no)
return new_url_name
def store(self):
new_url_name = self.get_new_url_name()
if not self.url_name:
self.url_name = new_url_name
if get_publisher().has_site_option('postgresql') and get_cfg('postgresql', {}):
import sql
sql.do_formdef_tables(self)
elif new_url_name != self.url_name:
# title changed, url will be changed only if there are not yet any
# submitted forms
data_class = self.data_class()
if data_class().count() == 0:
self.url_name = new_url_name
return StorableObject.store(self)
def get_receiver(self):
if self.receiver_id:
try:
return Role.get(self.receiver_id)
except KeyError:
return None
else:
return None
def set_receiver(self, role):
if role:
self.receiver_id = role.id
elif self.receiver_id:
self.receiver_id = None
receiver = property(get_receiver, set_receiver)
def get_category(self):
if self.category_id:
try:
return Category.get(self.category_id)
except KeyError:
return None
else:
return None
def set_category(self, category):
if category:
self.category_id = category.id
elif self.category_id:
self.category_id = None
category = property(get_category, set_category)
def get_workflow(self):
if self.workflow_id:
try:
workflow = Workflow.get(self.workflow_id)
except KeyError:
return Workflow.get_unknown_workflow()
return self.get_workflow_with_options(workflow)
else:
return Workflow.get_default_workflow()
def get_workflow_with_options(self, workflow):
# this needs to be kept in sync with admin/forms.ptl,
# FormDefPage::workflow
if not self.workflow_options:
return workflow
for status in workflow.possible_status:
for item in status.items:
prefix = '%s*%s*' % (status.id, item.id)
for parameter in item.get_parameters():
value = self.workflow_options.get(prefix + parameter)
if value:
setattr(item, parameter, value)
return workflow
def set_workflow(self, workflow):
if workflow:
self.workflow_id = workflow.id
elif self.workflow_id:
self.workflow_id = None
workflow = property(get_workflow, set_workflow)
def get_by_urlname(cls, url_name):
return cls.get_on_index(url_name, 'url_name')
get_by_urlname = classmethod(get_by_urlname)
def get_url(self, backoffice = False):
if backoffice:
base_url = get_publisher().get_backoffice_url()
else:
base_url = get_publisher().get_frontoffice_url()
return '%s/%s/' % (base_url, self.url_name)
def create_form(self, page_no = 0, displayed_fields = None):
form = Form(enctype = "multipart/form-data", use_tokens = False)
# had: , use_tokens = not self.confirmation)
self.add_fields_to_form(form, page_no = page_no, displayed_fields = displayed_fields)
return form
def add_fields_to_form(self, form, page_no = 0, displayed_fields = None, form_data = None):
current_page = 0
for field in self.fields:
if field.type == 'page':
if field is self.fields[0]:
continue
current_page += 1
if current_page > page_no:
break
continue
if current_page != page_no:
continue
if type(displayed_fields) is list:
displayed_fields.append(field)
value = None
if form_data:
value = form_data.get(field.id)
field.add_to_form(form, value)
def get_page(self, page_no):
return [x for x in self.fields if x.type == 'page'][page_no]
def create_view_form(self, dict = {}, use_tokens = True, visible = True):
if visible:
form = Form(enctype = 'multipart/form-data', use_tokens = use_tokens)
else:
form = Form(enctype = 'multipart/form-data',
use_tokens = use_tokens,
style = 'display: none;')
on_disabled_page = False
on_page = False
for field in self.fields:
if field.type == 'page':
on_disabled_page = False
if not field.is_visible(dict, self):
on_disabled_page = True
form_field = False
for f in self.fields[self.fields.index(field)+1:]:
if f.key == 'page':
break
if isinstance(f, fields.WidgetField):
form_field = True
break
if form_field is False:
on_disabled_page = True
if on_disabled_page:
continue
if field.type == 'page':
if on_page:
form.widgets.append(HtmlWidget(htmltext('</div>')))
form.widgets.append(HtmlWidget(
htmltext('<div class="page"><h3>%s</h3>' % field.label)))
on_page = True
value = dict.get(field.id, '')
field.add_to_view_form(form, value)
if on_page:
form.widgets.append(HtmlWidget(htmltext('</div>')))
return form
def get_data(self, form):
d = {}
for field in self.fields:
widget = form.get_widget('f%s' % field.id)
if widget:
d[field.id] = widget.parse()
if d.get(field.id) and field.convert_value_from_str:
d[field.id] = field.convert_value_from_str(d[field.id])
if d.get(field.id) and field.store_display_value:
display_value = field.store_display_value(d[field.id])
if display_value:
d['%s_display' % field.id] = display_value
elif d.has_key('%s_display' % field.id):
del d['%s_display' % field.id]
if widget and widget.cleanup:
widget.cleanup()
return d
def export_to_xml(self, include_id=False):
charset = get_publisher().site_charset
root = ET.Element('formdef')
ET.SubElement(root, 'name').text = unicode(self.name, charset)
if self.url_name:
ET.SubElement(root, 'url_name').text = unicode(self.url_name, charset)
if self.category:
ET.SubElement(root, 'category').text = unicode(self.category.name, charset)
for boolean_attribute in ('only_allow_one', 'allow_drafts', 'discussion',
'confirmation', 'signing'):
value = getattr(self, boolean_attribute)
if value:
value = 'true'
else:
value = 'false'
ET.SubElement(root, boolean_attribute).text = value
only_allow_one = False
allow_drafts = False
fields = ET.SubElement(root, 'fields')
for field in self.fields:
fields.append(field.export_to_xml(charset=charset, include_id=include_id))
return root
def import_from_xml(cls, fd):
try:
tree = ET.parse(fd)
except:
raise ValueError()
return cls.import_from_xml_tree(tree)
import_from_xml = classmethod(import_from_xml)
def import_from_xml_tree(cls, tree, include_id=False):
charset = get_publisher().site_charset
formdef = cls()
if tree.find('name') is None or not tree.find('name').text:
raise ValueError()
formdef.name = tree.find('name').text.encode(charset)
formdef.fields = []
for i, field in enumerate(tree.find('fields')):
try:
field_o = fields.get_field_class_by_type(field.findtext('type'))()
except KeyError:
raise ValueError()
field_o.init_with_xml(field, charset, include_id=include_id)
if not field_o.id:
# this assumes all fields will have id, or none of them
field_o.id = str(i)
formdef.fields.append(field_o)
if tree.find('category') is not None:
category = tree.find('category').text.encode(charset)
cats = Category.select()
for c in cats:
if c.name == category:
formdef.category_id = c.id
break
for boolean_attribute in ('only_allow_one', 'allow_drafts', 'discussion',
'confirmation', 'signing'):
value = tree.find(boolean_attribute)
if value is None:
continue
setattr(formdef, boolean_attribute, value.text == 'true')
return formdef
import_from_xml_tree = classmethod(import_from_xml_tree)
def get_detailed_email_form(self, formdata, url):
details = []
display_username = True
# this is custom code so it is possible to mark forms as anonyms, this
# is done through the VoteAnonymity field, this is very specific but
# isn't generalised yet into an useful extension mechanism, as it's not
# clear at the moment what could be useful.
for f in self.fields:
if f.key == 'vote-anonymity':
display_username = False
break
if display_username and formdata.user_id and formdata.user:
details.append(_('User name:'))
details.append(' %s' % formdata.user.name)
details.append('')
data = formdata.data
for field in self.fields:
if isinstance(field, (fields.SubtitleField, fields.TitleField, fields.CommentField,
fields.PageField)):
continue
if data.get(field.id) is None:
continue
details.append(_('%s:') % field.label)
if field.type in ('text', 'file'):
# XXX: howto support preformatted text in a dl in docutils ?
details.append((' %s' % data[field.id]).replace('\n', '\n '))
else:
details.append('%s' % field.get_rst_view_value(data[field.id], indent=' '))
details.append('')
return '\n'.join(details)
def get_submitter_email(self, formdata):
users_cfg = get_cfg('users', {})
field_email = users_cfg.get('field_email') or 'email'
if formdata.user:
if field_email == 'email' and formdata.user.email:
return formdata.user.email
elif formdata.user.form_data and formdata.user.form_data.get(field_email):
return formdata.user.form_data.get(field_email)
else:
# this shouldn't happen, but then data can get unsynced, so
# even if there's some user custom form with an unfilled email
# form we look at the straight email attribute, and use it if
# it exists.
if formdata.user.email:
return formdata.user.email
# if there is no user, or user has no email address, look
# up in submitted form for one that would hold the user
# email (the one set to be prefilled by user email)
fields = formdata.formdef.fields
for field in fields:
if not hasattr(field, 'prefill'):
continue
if field.prefill and field.prefill.get('type') == 'user':
if field.prefill.get('value') == field_email:
v = formdata.data.get(field.id)
if v:
return v
return None
def get_sign_text(self, dict = {}):
text = ""
on_disabled_page = False
on_page = False
for field in self.fields:
if field.type != 'page':
value = dict.get(field.id, '')
if field.type == "table":
value = field.get_rst_view_value(value)
text += "[%s] %s\n%s\n--\n" % \
(field.id, field.label, value)
return text
def get_substitution_variables(self):
d = {
'form_name': self.name,
}
if self.category:
d.update(self.category.get_substitution_variables())
return d
def get_detailed_evolution(self, formdata):
if not formdata.evolution:
return None
details = []
evo = formdata.evolution[-1]
if evo.who:
evo_who = None
if evo.who == '_submitter':
if formdata.user_id:
evo_who = formdata.user_id
elif formdata.user_hash:
if formdata.is_submitter(get_request().user):
evo_who = get_request().user.id
else:
evo_who = evo.who
if evo_who:
details.append(_('User name'))
details.append(' %s' % get_publisher().user_class.get(evo_who).name)
if evo.status:
details.append(_('Status'))
details.append(' %s' % formdata.get_status_label())
if evo.comment:
details.append('\n%s\n' % evo.comment)
return '\n\n----\n\n' + '\n'.join(details)
def is_user_allowed_read(self, user, formdata=None):
if self.acl_read == 'all':
return True
if not user:
return False
if user.is_admin:
return True
if user.roles: # set(None) raise an exception for python>2.6
user_roles = set(user.roles)
else:
user_roles = set([])
user_roles.add(logged_users_role().id)
if self.acl_read == 'roles':
form_roles = (self.roles or [])
if self.receiver:
form_roles.append(self.receiver.id)
if user_roles.intersection(form_roles):
return True
elif self.acl_read == 'owner':
if formdata and formdata.is_submitter(user):
return True
if self.receiver and self.receiver.id in user_roles:
return True
elif self.acl_read == 'none':
# no special permission for anybody, but the form will be viewable
# to users with a workflow action available.
pass
if formdata:
if self.workflow_id:
# formdef has workflow, get roles allowed some actions relative to
# current status
wf_status = formdata.get_workflow_status()
status_action_roles = []
for item in wf_status.items or []:
if not hasattr(item, 'by'):
continue
for role in item.by:
if role == '_submitter':
# action for submitter
if formdata and formdata.is_submitter(user):
return True
elif role == '_receiver':
# action for receiver
status_action_roles.append(self.receiver_id)
elif not (type(role) is str and role.startswith('_')):
# action for another group
status_action_roles.append(role)
if user_roles.intersection(status_action_roles or []):
return True
else:
# in default workflow, access is allowed for formdef.receiver_id
return self.receiver_id in user_roles
return False
def is_user_allowed_read_status_and_history(self, user, formdata=None):
if user and user.is_admin:
return True
if user and self.private_status_and_history and not self.receiver_id in (user.roles or []):
return False
return self.is_user_allowed_read(user, formdata=formdata)
def is_disabled(self):
if self.disabled:
return True
if self.publication_date:
publication_datetime = datetime.datetime.strptime(self.publication_date, date_format())
if publication_datetime > datetime.datetime.now():
return True
if self.expiration_date:
expiration_datetime = datetime.datetime.strptime(self.expiration_date, date_format())
if expiration_datetime < datetime.datetime.now():
return True
return False
from qommon.admin.emails import EmailsDirectory
EmailsDirectory.register('new_user', N_('Notification of creation to user'),
enabled = False,
category = N_('Workflow'),
default_subject = N_('New form ([name])'),
default_body = N_('''\
Hello,
[if-any user]
This mail is a reminder about the form you just submitted; you can consult it
with this link: [url]
[else]
This mail is a reminder about the form you just submitted.
[end]
[if-any details]
For reference, here are the details:
[details]
[end]
'''))
EmailsDirectory.register('change_user', N_('Notification of change to user'),
N_('Available variables: user, url, before, after, evolution'),
category = N_('Workflow'),
default_subject = N_('Form status change'),
default_body = N_('''\
Hello,
[if-any form_status_changed]
Status of the form you submitted just changed (from "[before]" to "[after]").
[end]
[if-any user]
You can consult it with this link:
[url]
[end]
[if-any form_comment]New comment: [form_comment][end]
[if-any evolution]
[evolution]
[end]
'''))
EmailsDirectory.register('new_receiver', N_('Notification of creation to receiver'),
N_('Available variables: name, url, details'), enabled = False,
category = N_('Workflow'),
default_subject = N_('New form ([name])'),
default_body = N_('''\
Hello,
A new form has been submitted, you can see it with this link:
[form_url_backoffice]
[if-any details]
For reference, here are the details:
[details]
[end]
'''))
EmailsDirectory.register('change_receiver', N_('Notification of change to receiver'),
N_('Available variables: name, url, before, after, evolution'),
category = N_('Workflow'),
default_subject = N_('Form status change ([name])'),
default_body = N_('''\
Hello,
A form just changed, you can consult it with this link:
[form_url_backoffice]
[if-any form_comment]New comment: [form_comment][end]
[if-any evolution]
[evolution]
[end]
'''))
Substitutions.register('form_name', category=N_('Form'), comment=N_('Form Name'))