entrouvert/wcs
entrouvert
/
wcs
14
1
Fork 0
Code Issues Pull Requests 36 Releases Activity

general: finish removal of support for non-utf8 charset (#85626)

This commit is contained in:
Frédéric Péters 2024-01-13 16:08:33 +01:00
parent 6cb2f0afef
commit 3ff59c706e
39 changed files with 213 additions and 311 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
tests/utilities.py
View File

@ -83,7 +83,6 @@ def create_temporary_pub(pickle_mode=False, lazy_mode=False):
pub = publisher_class.create_publisher()
# allow saving the user
pub.app_dir = os.path.join(APP_DIR, 'example.net')
pub.site_charset = 'utf-8'
if not pickle_mode:
pub.user_class = sql.SqlUser

3
wcs/admin/forms.py
View File

@ -1544,8 +1544,7 @@ class FormDefPage(Directory, TempfileDirectoryMixin):
elif (
current_field
and new_field
and ET.tostring(current_field.export_to_xml('utf-8'))
!= ET.tostring(new_field.export_to_xml('utf-8'))
and ET.tostring(current_field.export_to_xml()) != ET.tostring(new_field.export_to_xml())
):
# same type, but changes within field
table += htmltext('<tr class="modified-field"><td class="indicator">~</td>')

4
wcs/admin/workflows.py
View File

@ -541,12 +541,12 @@ class WorkflowItemPage(Directory):
status_id = form.get_widget('status').parse()
destination_status = self.workflow.get_status(status_id)
item = self.item.export_to_xml('utf-8')
item = self.item.export_to_xml()
item_type = item.attrib['type']
new_item = destination_status.add_action(item_type)
new_item.parent = destination_status
try:
new_item.init_with_xml(item, 'utf-8', check_datasources=False)
new_item.init_with_xml(item, check_datasources=False)
except WorkflowImportError as e:
reason = _(e.msg) % e.msg_args
if hasattr(e, 'render'):

15
wcs/api.py
View File

@ -23,7 +23,7 @@ import time
import urllib.parse
from django.http import HttpResponse, HttpResponseBadRequest, JsonResponse
from django.utils.encoding import force_bytes, force_str
from django.utils.encoding import force_bytes
from quixote import get_publisher, get_request, get_response, get_session, redirect
from quixote.directory import Directory
from quixote.errors import MethodNotAllowedError, RequestError
@ -760,8 +760,6 @@ class ApiFormdefsDirectory(Directory):
elif category_slugs:
formdefs = [x for x in formdefs if x.category and (x.category.url_name in category_slugs)]
charset = get_publisher().site_charset
include_count = get_query_flag('include-count')
for formdef in formdefs:
@ -792,7 +790,7 @@ class ApiFormdefsDirectory(Directory):
authentication_required = True
formdict = {
'title': force_str(formdef.name, charset),
'title': formdef.name,
'slug': formdef.url_name,
'url': formdef.get_url(),
'description': formdef.description or '',
@ -854,8 +852,8 @@ class ApiFormdefsDirectory(Directory):
formdict['functions'][wf_role_id] = workflow_function
if formdef.category:
formdict['category'] = force_str(formdef.category.name, charset)
formdict['category_slug'] = force_str(formdef.category.url_name, charset)
formdict['category'] = formdef.category.name
formdict['category_slug'] = formdef.category.url_name
list_forms.append(formdict)
@ -918,17 +916,16 @@ class ApiCategoriesDirectory(Directory):
list_all_forms = (user and user.is_admin) or (is_url_signed() and user is None)
backoffice_submission = get_request().form.get('backoffice-submission') == 'on'
list_categories = []
charset = get_publisher().site_charset
categories = Category.select()
Category.sort_by_position(categories)
all_formdefs = FormDef.select(order_by='name', ignore_errors=True, lightweight=True)
for category in categories:
d = {}
d['title'] = force_str(category.name, charset)
d['title'] = category.name
d['slug'] = category.url_name
d['url'] = category.get_url()
if category.description:
d['description'] = force_str(str(category.get_description_html_text()), charset)
d['description'] = str(category.get_description_html_text())
formdefs = ApiFormdefsDirectory(category).get_list_forms(
user,
formdefs=all_formdefs,

28
wcs/backoffice/management.py
View File

@ -2806,7 +2806,6 @@ class FormPage(Directory, TempfileDirectoryMixin):
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
if not get_request().user and get_request().form.get('api-user'):
# custom query string authentification as some Outlook versions and
@ -2924,9 +2923,9 @@ class FormPage(Directory, TempfileDirectoryMixin):
formdef.url_name,
formdata.id,
)
summary = force_str(formdata.get_display_name(), charset)
summary = formdata.get_display_name()
if formdata.default_digest:
summary += ' - %s' % force_str(formdata.default_digest, charset)
summary += ' - %s' % formdata.default_digest
vevent.add('summary').value = summary
vevent.add('dtstart').value = dtstart
if isinstance(dtstart, datetime.datetime):
@ -2941,16 +2940,16 @@ class FormPage(Directory, TempfileDirectoryMixin):
vevent.dtend.value_param = 'DATE'
backoffice_url = formdata.get_url(backoffice=True)
vevent.add('url').value = backoffice_url
form_name = force_str(formdef.name, charset)
status_name = force_str(formdata.get_status_label(), charset)
form_name = formdef.name
status_name = formdata.get_status_label()
description = '%s | %s | %s\n' % (form_name, formdata.get_display_id(), status_name)
if formdata.default_digest:
description += '%s\n' % force_str(formdata.default_digest, charset)
description += '%s\n' % formdata.default_digest
description += backoffice_url
# TODO: improve performance by loading all users in one
# single query before the loop
if formdata.user:
description += '\n%s' % force_str(formdata.user.get_display_name(), charset)
description += '\n%s' % formdata.user.get_display_name()
vevent.add('description').value = description
cal.add(vevent)
@ -3933,21 +3932,14 @@ class FormBackOfficeStatusPage(FormStatusPage):
def inspect_variables(self):
r = TemplateIO(html=True)
charset = get_publisher().site_charset
substvars = CompatibilityNamesDict()
substvars.update(self.filled.get_substitution_variables())
def safe(v):
if isinstance(v, str):
try:
force_str(v, charset)
except UnicodeDecodeError:
v = repr(v)
else:
try:
v = force_str(v).encode(charset)
except Exception:
v = repr(v)
try:
v = force_str(v)
except Exception:
v = repr(v)
return v
access_to_admin_forms = get_publisher().get_backoffice_root().is_global_accessible('forms')

5
wcs/blocks.py
View File

@ -174,7 +174,7 @@ class BlockDef(StorableObject):
fields = ET.SubElement(root, 'fields')
for field in self.fields or []:
fields.append(field.export_to_xml(charset='utf-8', include_id=True))
fields.append(field.export_to_xml(include_id=True))
return root
@ -198,7 +198,6 @@ class BlockDef(StorableObject):
@classmethod
def import_from_xml_tree(cls, tree, include_id=False, check_datasources=True, **kwargs):
charset = 'utf-8'
blockdef = cls()
if tree.find('name') is None or not tree.find('name').text:
raise BlockdefImportError(_('Missing name'))
@ -233,7 +232,7 @@ class BlockDef(StorableObject):
field_type = field.findtext('type')
unknown_field_types.add(field_type)
continue
field_o.init_with_xml(field, charset, include_id=True)
field_o.init_with_xml(field, include_id=True)
blockdef.fields.append(field_o)
BlockCategory.object_category_xml_import(blockdef, tree, include_id=include_id)

8
wcs/compat.py
View File

@ -138,14 +138,6 @@ class CompatHTTPRequest(HTTPRequest):
return self._process_multipart(length, params)
def _process_multipart(self, length, params):
# Make sure request.form doesn't contain unicode strings, converting
# them all to strings in the site charset; it would contain unicode
# strings when the user agent specifies a charset in a mime content
# part, such a behaviour appears with some Nokia phones (6020, 6300)
site_charset = get_publisher().site_charset
# parse multipart data with the charset of the website
if 'charset' not in params:
params['charset'] = site_charset
if not self.form:
self.form = {}
for k in self.django_request.POST:

2
wcs/ctl/management/commands/hobo_deploy.py
View File

@ -177,7 +177,7 @@ class Command(TenantCommand):
def update_configuration(self, service, pub):
if not pub.cfg.get('misc'):
pub.cfg['misc'] = {'charset': 'utf-8'}
pub.cfg['misc'] = {}
pub.cfg['misc']['sitename'] = force_str(service.get('title'))
pub.cfg['misc']['frontoffice-url'] = force_str(service.get('base_url'))
if not pub.cfg.get('language'):

2
wcs/ctl/management/commands/hobo_notify.py
View File

@ -21,7 +21,6 @@ from quixote import get_publisher
from wcs.admin.settings import UserFieldsFormDef
from wcs.qommon import force_str
from wcs.qommon.misc import json_encode_helper
from wcs.qommon.publisher import get_cfg, get_publisher_class
from . import TenantCommand
@ -143,7 +142,6 @@ class Command(TenantCommand):
for o in data:
try:
o = json_encode_helper(o, publisher.site_charset)
if action == 'provision':
if not cls.check_valid_user(o):
raise ValueError('invalid user')

13
wcs/custom_views.py
View File

@ -17,7 +17,6 @@
import urllib.parse
import xml.etree.ElementTree as ET
from django.utils.encoding import force_str
from quixote import get_publisher
from wcs.backoffice.data_management import CardPage
@ -227,7 +226,7 @@ class CustomView(StorableObject):
return criterias
def export_to_xml(self, charset=None, include_id=False):
def export_to_xml(self, include_id=False):
root = ET.Element(self.xml_root_node)
fields = [
'title',
@ -248,14 +247,10 @@ class CustomView(StorableObject):
if not isinstance(field_dict, dict):
continue
for k, v in sorted(field_dict.items()):
text_value = force_str(v, charset, errors='replace')
ET.SubElement(el, k).text = text_value
ET.SubElement(el, k).text = str(v)
elif isinstance(val, dict):
for k, v in sorted(val.items()):
text_value = force_str(v, charset, errors='replace')
ET.SubElement(el, k).text = text_value
elif isinstance(val, str):
el.text = force_str(val, charset, errors='replace')
ET.SubElement(el, k).text = str(v)
else:
el.text = str(val)
@ -276,7 +271,7 @@ class CustomView(StorableObject):
return root
def init_with_xml(self, elem, charset, include_id=False):
def init_with_xml(self, elem, include_id=False):
fields = [
'title',
'slug',

5
wcs/data_sources.py
View File

@ -539,7 +539,6 @@ def _get_structured_items(data_source, mode=None, raise_on_error=False, with_fil
return []
if len(value) == 0:
return []
value = misc.json_encode_helper(value, get_publisher().site_charset)
if isinstance(value[0], (list, tuple)):
if len(value[0]) >= 3:
return [{'id': x[0], 'text': x[1], 'key': x[2]} for x in value]
@ -869,10 +868,10 @@ class NamedDataSource(XmlStorableObject):
section = 'settings'
return '%s/%s/data-sources/%s/' % (base_url, section, self.id)
def export_data_source_to_xml(self, element, attribute_name, charset, **kwargs):
def export_data_source_to_xml(self, element, attribute_name, **kwargs):
data_source = getattr(self, attribute_name)
ET.SubElement(element, 'type').text = data_source.get('type')
ET.SubElement(element, 'value').text = force_str(data_source.get('value') or '', charset)
ET.SubElement(element, 'value').text = data_source.get('value') or ''
def import_data_source_from_xml(self, element, **kwargs):
return {

33
wcs/fields/base.py
View File

@ -305,7 +305,7 @@ class Field:
if attribute in elem:
setattr(self, attribute, elem.get(attribute))
def export_to_xml(self, charset, include_id=False):
def export_to_xml(self, include_id=False):
field = ET.Element('field')
extra_fields = ['default_value'] # specific to workflow variables
if include_id:
@ -313,7 +313,7 @@ class Field:
ET.SubElement(field, 'type').text = self.key
for attribute in self.get_admin_attributes() + extra_fields:
if hasattr(self, '%s_export_to_xml' % attribute):
getattr(self, '%s_export_to_xml' % attribute)(field, charset, include_id=include_id)
getattr(self, '%s_export_to_xml' % attribute)(field, include_id=include_id)
continue
if hasattr(self, attribute) and getattr(self, attribute) is not None:
val = getattr(self, attribute)
@ -322,13 +322,9 @@ class Field:
el = ET.SubElement(field, attribute)
if isinstance(val, dict):
for k, v in sorted(val.items()):
if isinstance(v, str):
text_value = force_str(v, charset, errors='replace')
else:
# field having non str value in dictionnary field must overload
# import_to_xml to handle import
text_value = force_str(v)
ET.SubElement(el, k).text = text_value
# field having non str value in dictionnary field must overload
# import_to_xml to handle import
ET.SubElement(el, k).text = force_str(v)
elif isinstance(val, list):
if attribute[-1] == 's':
atname = attribute[:-1]
@ -336,26 +332,23 @@ class Field:
atname = 'item'
# noqa pylint: disable=not-an-iterable
for v in val:
ET.SubElement(el, atname).text = force_str(v, charset, errors='replace')
elif isinstance(val, str):
el.attrib['type'] = 'str'
el.text = force_str(val, charset, errors='replace')
ET.SubElement(el, atname).text = force_str(v)
else:
el.text = str(val)
if isinstance(val, bool):
el.attrib['type'] = 'bool'
elif isinstance(val, int):
el.attrib['type'] = 'int'
elif isinstance(val, str):
el.attrib['type'] = 'str'
return field
def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def init_with_xml(self, elem, include_id=False, snapshot=False):
extra_fields = ['default_value'] # specific to workflow variables
for attribute in self.get_admin_attributes() + extra_fields:
el = elem.find(attribute)
if hasattr(self, '%s_init_with_xml' % attribute):
getattr(self, '%s_init_with_xml' % attribute)(
el, charset, include_id=include_id, snapshot=False
)
getattr(self, '%s_init_with_xml' % attribute)(el, include_id=include_id, snapshot=False)
continue
if el is None:
continue
@ -395,7 +388,7 @@ class Field:
except Exception:
pass
def condition_init_with_xml(self, node, charset, include_id=False, snapshot=False):
def condition_init_with_xml(self, node, include_id=False, snapshot=False):
self.condition = None
if node is None:
return
@ -407,7 +400,7 @@ class Field:
elif node.text:
self.condition = {'type': 'python', 'value': force_str(node.text).strip()}
def data_source_init_with_xml(self, node, charset, include_id=False, snapshot=False):
def data_source_init_with_xml(self, node, include_id=False, snapshot=False):
self.data_source = {}
if node is None:
return
@ -421,7 +414,7 @@ class Field:
elif self.data_source.get('value') is None:
del self.data_source['value']
def prefill_init_with_xml(self, node, charset, include_id=False, snapshot=False):
def prefill_init_with_xml(self, node, include_id=False, snapshot=False):
self.prefill = {}
if node is not None and node.findall('type'):
self.prefill = {

8
wcs/fields/file.py
View File

@ -249,19 +249,19 @@ class FileField(WidgetField):
if value and hasattr(value, 'token'):
get_request().form[self.field_key + '$token'] = value.token
def export_to_xml(self, charset, include_id=False):
def export_to_xml(self, include_id=False):
# convert some sub-fields to strings as export_to_xml() only supports
# dictionnaries with strings values
if self.document_type and self.document_type.get('mimetypes'):
old_value = self.document_type['mimetypes']
self.document_type['mimetypes'] = '|'.join(self.document_type['mimetypes'])
result = super().export_to_xml(charset, include_id=include_id)
result = super().export_to_xml(include_id=include_id)
if self.document_type and self.document_type.get('mimetypes'):
self.document_type['mimetypes'] = old_value
return result
def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
super().init_with_xml(elem, charset, include_id=include_id)
def init_with_xml(self, elem, include_id=False, snapshot=False):
super().init_with_xml(elem, include_id=include_id)
# translate fields flattened to strings
if self.document_type and self.document_type.get('mimetypes'):
self.document_type['mimetypes'] = self.document_type['mimetypes'].split('|')

4
wcs/fields/item.py
View File

@ -284,8 +284,8 @@ class ItemField(WidgetField, MapOptionsMixin, ItemFieldMixin, ItemWithImageField
changed = True
return changed
def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
super().init_with_xml(elem, charset, include_id=include_id)
def init_with_xml(self, elem, include_id=False, snapshot=False):
super().init_with_xml(elem, include_id=include_id)
if getattr(elem.find('show_as_radio'), 'text', None) == 'True':
self.display_mode = 'radio'

10
wcs/fields/page.py
View File

@ -145,7 +145,7 @@ class PageField(Field):
post_conditions = None
def post_conditions_init_with_xml(self, node, charset, include_id=False, snapshot=False):
def post_conditions_init_with_xml(self, node, include_id=False, snapshot=False):
if node is None:
return
self.post_conditions = []
@ -169,7 +169,7 @@ class PageField(Field):
}
)
def post_conditions_export_to_xml(self, node, charset, include_id=False):
def post_conditions_export_to_xml(self, node, include_id=False):
if not self.post_conditions:
return
@ -178,13 +178,13 @@ class PageField(Field):
post_condition_node = ET.SubElement(conditions_node, 'post_condition')
condition_node = ET.SubElement(post_condition_node, 'condition')
ET.SubElement(condition_node, 'type').text = force_str(
(post_condition['condition'] or {}).get('type') or '', charset, errors='replace'
(post_condition['condition'] or {}).get('type') or ''
)
ET.SubElement(condition_node, 'value').text = force_str(
(post_condition['condition'] or {}).get('value') or '', charset, errors='replace'
(post_condition['condition'] or {}).get('value') or ''
)
ET.SubElement(post_condition_node, 'error_message').text = force_str(
post_condition['error_message'] or '', charset, errors='replace'
post_condition['error_message'] or ''
)
def fill_admin_form(self, form):

7
wcs/fields/string.py
View File

@ -20,7 +20,6 @@ import xml.etree.ElementTree as ET
from django.utils.encoding import force_str
from django.utils.html import urlize
from quixote import get_publisher
from quixote.html import htmltext
from wcs import data_sources
@ -121,8 +120,6 @@ class StringField(WidgetField):
def get_view_value(self, value, **kwargs):
value = value or ''
if isinstance(value, str) and (value.startswith('http://') or value.startswith('https://')):
charset = get_publisher().site_charset
value = force_str(value, charset)
return htmltext(force_str(urlize(value, nofollow=True, autoescape=True)))
return str(value)
@ -172,8 +169,8 @@ class StringField(WidgetField):
changed = True
return changed
def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
super().init_with_xml(elem, charset, include_id=include_id)
def init_with_xml(self, elem, include_id=False, snapshot=False):
super().init_with_xml(elem, include_id=include_id)
self.migrate()
def get_validation_parameter_view_value(self, widget):

32
wcs/formdef.py
View File

@ -1039,13 +1039,12 @@ class FormDef(StorableObject):
def export_to_json(self, include_id=False, indent=None, with_user_fields=False):
from wcs.carddef import CardDef
charset = get_publisher().site_charset
root = {}
root['name'] = force_str(self.name, charset)
root['name'] = self.name
if include_id and self.id:
root['id'] = str(self.id)
if self.category:
root['category'] = force_str(self.category.name, charset)
root['category'] = self.category.name
root['category_id'] = str(self.category.id)
if self.workflow:
root['workflow'] = self.workflow.get_json_export_dict(include_id=include_id)
@ -1160,9 +1159,7 @@ class FormDef(StorableObject):
return json.dumps(root, indent=indent, sort_keys=True, cls=JSONEncoder)
@classmethod
def import_from_json(cls, fd, charset=None, include_id=False):
if charset is None:
charset = get_publisher().site_charset
def import_from_json(cls, fd, include_id=False):
formdef = cls()
def unicode2str(v):
@ -1250,14 +1247,13 @@ class FormDef(StorableObject):
return formdef
def export_to_xml(self, include_id=False):
charset = get_publisher().site_charset
root = ET.Element(self.xml_root_node)
if include_id and self.id:
root.attrib['id'] = str(self.id)
for text_attribute in list(self.TEXT_ATTRIBUTES):
if not hasattr(self, text_attribute) or not getattr(self, text_attribute):
continue
ET.SubElement(root, text_attribute).text = force_str(getattr(self, text_attribute), charset)
ET.SubElement(root, text_attribute).text = str(getattr(self, text_attribute))
for boolean_attribute in self.BOOLEAN_ATTRIBUTES:
if not hasattr(self, boolean_attribute):
continue
@ -1278,7 +1274,7 @@ class FormDef(StorableObject):
if not workflow:
workflow = self.get_default_workflow()
elem = ET.SubElement(root, 'workflow')
elem.text = force_str(workflow.name, charset)
elem.text = workflow.name
if workflow.slug:
elem.attrib['slug'] = str(workflow.slug)
if include_id:
@ -1291,7 +1287,7 @@ class FormDef(StorableObject):
fields = ET.SubElement(root, 'fields')
for field in self.fields or []:
fields.append(field.export_to_xml(charset=charset, include_id=include_id))
fields.append(field.export_to_xml(include_id=include_id))
from wcs.workflows import get_role_name_and_slug
@ -1376,7 +1372,7 @@ class FormDef(StorableObject):
for view in get_publisher().custom_view_class.select_shared_for_formdef(formdef=self):
custom_views.append(view)
for view in custom_views:
custom_view_node = view.export_to_xml(charset=charset, include_id=include_id)
custom_view_node = view.export_to_xml(include_id=include_id)
if custom_view_node is not None:
custom_views_element.append(custom_view_node)
@ -1384,7 +1380,7 @@ class FormDef(StorableObject):
for geoloc_key, geoloc_label in (self.geolocations or {}).items():
element = ET.SubElement(geolocations, 'geolocation')
element.attrib['key'] = geoloc_key
element.text = force_str(geoloc_label, charset)
element.text = geoloc_label
if self.required_authentication_contexts:
element = ET.SubElement(root, 'required_authentication_contexts')
@ -1402,14 +1398,13 @@ class FormDef(StorableObject):
return root
@classmethod
def import_from_xml(cls, fd, charset=None, include_id=False, fix_on_error=False, check_datasources=True):
def import_from_xml(cls, fd, include_id=False, fix_on_error=False, check_datasources=True):
try:
tree = ET.parse(fd)
except Exception:
raise ValueError()
formdef = cls.import_from_xml_tree(
tree,
charset=charset,
include_id=include_id,
fix_on_error=fix_on_error,
check_datasources=check_datasources,
@ -1435,13 +1430,10 @@ class FormDef(StorableObject):
@classmethod
def import_from_xml_tree(
cls, tree, include_id=False, charset=None, fix_on_error=False, snapshot=False, check_datasources=True
cls, tree, include_id=False, fix_on_error=False, snapshot=False, check_datasources=True
):
from wcs.carddef import CardDef
if charset is None:
charset = get_publisher().site_charset
assert charset == 'utf-8'
formdef = cls()
if tree.find('name') is None or not tree.find('name').text:
raise FormdefImportError(_('Missing name'))
@ -1486,7 +1478,7 @@ class FormDef(StorableObject):
else:
unknown_field_types.add(field_type)
continue
field_o.init_with_xml(field, charset, include_id=True)
field_o.init_with_xml(field, include_id=True)
if field_type.startswith('block:'):
field_o.block_slug = field_type.removeprefix('block:')
if fix_on_error or not field_o.id:
@ -1529,7 +1521,7 @@ class FormDef(StorableObject):
formdef._custom_views = []
for view in tree.findall('custom_views/%s' % get_publisher().custom_view_class.xml_root_node):
view_o = get_publisher().custom_view_class()
view_o.init_with_xml(view, charset, include_id=include_id)
view_o.init_with_xml(view, include_id=include_id)
formdef._custom_views.append(view_o)
cls.category_class.object_category_xml_import(formdef, tree, include_id=include_id)

3
wcs/portfolio.py
View File

@ -63,14 +63,13 @@ def push_document(user, filename, stream):
if not user:
return
publisher = get_publisher()
charset = publisher.site_charset
payload = {}
if user.name_identifiers:
payload['user_nameid'] = force_str(user.name_identifiers[0], 'ascii')
elif user.email:
payload['user_email'] = force_str(user.email, 'ascii')
payload['origin'] = urllib.parse.urlparse(publisher.get_frontoffice_url()).netloc
payload['file_name'] = force_str(filename, charset)
payload['file_name'] = filename
stream.seek(0)
payload['file_b64_content'] = force_str(base64.b64encode(stream.read()))
async_post = fargo_post_json_async('/api/documents/push/', payload)

4
wcs/qommon/form.py
View File

@ -2659,10 +2659,8 @@ class WysiwygTextWidget(TextWidget):
self.value = ''
# unescape Django template tags
charset = get_publisher().site_charset
def unquote_django(matchobj):
return force_str(html.unescape(force_str(matchobj.group(0), charset)))
return force_str(html.unescape(matchobj.group(0)))
self.value = re.sub('{[{%](.*?)[%}]}', unquote_django, self.value)
if self.validation_function:

2
wcs/qommon/http_request.py
View File

@ -37,7 +37,7 @@ class HTTPRequest(quixote.http_request.HTTPRequest):
def __init__(self, *args, **kwargs):
quixote.http_request.HTTPRequest.__init__(self, *args, **kwargs)
self.response = HTTPResponse()
self.charset = get_publisher().site_charset
self.charset = 'utf-8'
self.is_json_marker = None
self.ignore_session = False
self.wscalls_cache = {}

27
wcs/qommon/misc.py
View File

@ -146,7 +146,7 @@ def get_provider_label(provider):
name = re.findall('<OrganizationName.*>(.*?)</OrganizationName>', organization)
if not name:
return provider.providerId
return htmltext(name[0].decode('utf8').encode(get_publisher().site_charset))
return htmltext(name[0])
def get_provider(provider_key):
@ -177,10 +177,7 @@ def simplify(s, space='-'):
if s is None:
return ''
if not isinstance(s, str):
if get_publisher() and get_publisher().site_charset:
s = force_str('%s' % s, get_publisher().site_charset, errors='ignore')
else:
s = force_str('%s' % s, 'iso-8859-1', errors='ignore')
s = force_str('%s' % s, 'utf-8', errors='ignore')
s = unidecode.unidecode(s)
s = re.sub(r'[^\w\s\'\-%s]' % space, '', s).strip().lower()
s = re.sub(r'[\s\'\-_%s]+' % space, space, s).strip(space)
@ -275,15 +272,11 @@ def get_as_datetime(s):
def site_encode(s):
if s is None:
return None
if isinstance(s, str):
return s
if not isinstance(s, str):
s = force_str(s)
return s.encode(get_publisher().site_charset)
return force_str(s)
def ellipsize(s, length=30):
s = force_str(s, get_publisher().site_charset, errors='replace')
s = force_str(s)
if s and len(s) > length:
if length > 3:
s = Truncator(s).chars(length, truncate='(…)')
@ -673,13 +666,7 @@ class JSONEncoder(json.JSONEncoder):
return json.JSONEncoder.default(self, o)
def json_encode_helper(d, charset):
'''Encode a JSON structure into local charset'''
# since Python 3 strings are unicode, no need to convert anything
return d
def json_loads(value, charset=None):
def json_loads(value):
return json.loads(force_str(value))
@ -837,9 +824,7 @@ def normalize_geolocation(lat_lon):
def html2text(text):
if isinstance(text, (htmltext, str)):
text = force_str(str(text), get_publisher().site_charset)
return site_encode(html.unescape(strip_tags(text)))
return site_encode(html.unescape(strip_tags(force_str(text))))
def validate_luhn(string_value, length=None):

7
wcs/qommon/publisher.py
View File

@ -789,13 +789,6 @@ class QommonPublisher(Publisher):
return self._app_logger
def sitecharset2utf8(self, str):
'''Convert a string in site encoding to UTF-8'''
return str.decode(self.site_charset).encode('utf-8')
def utf82sitecharset(self, str):
return str.decode('utf-8').encode(self.site_charset)
def get_default_position(self):
default_position = self.cfg.get('misc', {}).get('default-position', None)
if not default_position:

18
wcs/qommon/xml_storage.py
View File

@ -17,7 +17,6 @@
import datetime
import xml.etree.ElementTree as ET
from django.utils.encoding import force_str
from quixote import get_publisher
from .misc import indent_xml, xml_node_text
@ -43,7 +42,6 @@ class XmlStorableObject(StorableObject):
return object.export_to_xml_string(include_id=True)
def export_to_xml(self, include_id=False):
charset = get_publisher().site_charset
root = ET.Element(self.xml_root_node)
if include_id and self.id:
root.attrib['id'] = str(self.id)
@ -53,11 +51,11 @@ class XmlStorableObject(StorableObject):
continue
element = ET.SubElement(root, attribute_name)
export_method = getattr(self, 'export_%s_to_xml' % attribute_type)
export_method(element, attribute_name, charset=charset, include_id=include_id)
export_method(element, attribute_name, include_id=include_id)
return root
def export_str_to_xml(self, element, attribute_name, charset, **kwargs):
element.text = force_str(getattr(self, attribute_name), charset)
def export_str_to_xml(self, element, attribute_name, **kwargs):
element.text = getattr(self, attribute_name)
def export_int_to_xml(self, element, attribute_name, **kwargs):
element.text = str(getattr(self, attribute_name))
@ -86,17 +84,15 @@ class XmlStorableObject(StorableObject):
sub.text = role.name
@classmethod
def import_from_xml(cls, fd, charset=None, include_id=False):
def import_from_xml(cls, fd, include_id=False):
try:
tree = ET.parse(fd)
except Exception:
raise ValueError()
return cls.import_from_xml_tree(tree, charset=charset, include_id=include_id)
return cls.import_from_xml_tree(tree, include_id=include_id)
@classmethod
def import_from_xml_tree(cls, tree, include_id=False, charset=None, **kwargs):
if charset is None:
charset = get_publisher().site_charset
def import_from_xml_tree(cls, tree, include_id=False, **kwargs):
obj = cls()
# if the tree we get is actually a ElementTree for real, we get its
@ -120,7 +116,7 @@ class XmlStorableObject(StorableObject):
setattr(
obj,
attribute_name,
import_method(element, charset=charset, include_id=include_id),
import_method(element, include_id=include_id),
)
return obj

20
wcs/roles.py
View File

@ -18,7 +18,6 @@ import json
import urllib.parse
import xml.etree.ElementTree as ET
from django.utils.encoding import force_str
from quixote import get_publisher
from quixote.html import htmltext
@ -93,27 +92,25 @@ class Role(StorableObject):
return data
def get_json_export_dict(self):
charset = get_publisher().site_charset
return {
'name': force_str(self.name, charset),
'text': force_str(self.name, charset), # generic key
'name': self.name,
'text': self.name, # generic key
'allows_backoffice_access': self.allows_backoffice_access,
'emails': [force_str(email, charset) for email in self.emails or []],
'details': force_str(self.details or '', charset),
'emails': [email for email in self.emails or []],
'details': self.details or '',
'emails_to_members': self.emails_to_members,
'slug': force_str(self.slug, charset),
'slug': self.slug,
'id': self.id,
}
def export_to_xml(self, include_id=False):
charset = get_publisher().site_charset
root = ET.Element(self.xml_root_node)
if include_id and self.id:
root.attrib['id'] = str(self.id)
for text_attribute in list(self.TEXT_ATTRIBUTES):
if not hasattr(self, text_attribute) or not getattr(self, text_attribute):
continue
ET.SubElement(root, text_attribute).text = force_str(getattr(self, text_attribute), charset)
ET.SubElement(root, text_attribute).text = getattr(self, text_attribute)
for boolean_attribute in self.BOOLEAN_ATTRIBUTES:
if not hasattr(self, boolean_attribute):
continue
@ -129,15 +126,12 @@ class Role(StorableObject):
return (json.dumps({'name': self.name, 'slug': self.slug, 'uuid': self.uuid}), 'application/json')
@classmethod
def import_from_xml(cls, fd, charset=None, include_id=False):
def import_from_xml(cls, fd, include_id=False):
try:
tree = ET.parse(fd)
except Exception:
raise ValueError()
if charset is None:
charset = get_publisher().site_charset
assert charset == 'utf-8'
role = cls()
# if the tree we get is actually a ElementTree for real, we get its

5
wcs/variables.py
View File

@ -1302,8 +1302,6 @@ class LazyFieldVar:
value = self._data.get(self._field.id)
if self._field.convert_value_to_str:
return self._field.convert_value_to_str(value)
if isinstance(value, str):
return force_str(value, get_publisher().site_charset)
return value
def __str__(self):
@ -1319,9 +1317,6 @@ class LazyFieldVar:
def __contains__(self, value):
return value in self.get_value()
def __unicode__(self):
return force_str(str(self), get_publisher().site_charset)
def __eq__(self, other):
return force_str(self) == force_str(other)

4
wcs/wf/attachment.py
View File

@ -304,7 +304,7 @@ class AddAttachmentWorkflowStatusItem(WorkflowStatusItem):
advanced=True,
)
def document_type_export_to_xml(self, xml_item, charset, include_id=False):
def document_type_export_to_xml(self, xml_item, include_id=False):
if not self.document_type:
return
node = ET.SubElement(xml_item, 'document_type')
@ -314,7 +314,7 @@ class AddAttachmentWorkflowStatusItem(WorkflowStatusItem):
ET.SubElement(node, 'mimetype').text = mimetype
return node
def document_type_init_with_xml(self, node, charset, include_id=False, snapshot=False):
def document_type_init_with_xml(self, node, include_id=False, snapshot=False):
self.document_type = {}
if node is None:
return

8
wcs/wf/backoffice_fields.py
View File

@ -209,23 +209,23 @@ class SetBackofficeFieldsWorkflowStatusItem(WorkflowStatusItem):
ContentSnapshotPart.take(formdata=formdata, old_data=old_data)
formdata.store()
def fields_export_to_xml(self, item, charset, include_id=False):
def fields_export_to_xml(self, item, include_id=False):
if not self.fields:
return
fields_node = ET.SubElement(item, 'fields')
for field in self.fields:
fields_node.append(FieldNode(field).export_to_xml(charset=charset, include_id=include_id))
fields_node.append(FieldNode(field).export_to_xml(include_id=include_id))
return fields_node
def fields_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def fields_init_with_xml(self, elem, include_id=False, snapshot=False):
fields = []
if elem is None:
return
for field_xml_node in elem.findall('field'):
field_node = FieldNode()
field_node.init_with_xml(field_xml_node, charset, include_id=include_id, snapshot=snapshot)
field_node.init_with_xml(field_xml_node, include_id=include_id, snapshot=snapshot)
fields.append(field_node.as_dict())
if fields:
self.fields = fields

4
wcs/wf/comment.py
View File

@ -186,7 +186,7 @@ class CommentableWorkflowStatusItem(WorkflowStatusItem):
value=self.backoffice_info_text,
)
def button_label_export_to_xml(self, xml_item, charset, include_id=False):
def button_label_export_to_xml(self, xml_item, include_id=False):
if self.button_label == 0:
pass
elif self.button_label is None:
@ -197,7 +197,7 @@ class CommentableWorkflowStatusItem(WorkflowStatusItem):
el = ET.SubElement(xml_item, 'button_label')
el.text = self.button_label
def button_label_init_with_xml(self, element, charset, include_id=False, snapshot=False):
def button_label_init_with_xml(self, element, include_id=False, snapshot=False):
if element is None:
return
# this can be None if element is self-closing, <button_label />, which

4
wcs/wf/create_formdata.py
View File

@ -868,14 +868,14 @@ class CreateFormdataWorkflowStatusItem(WorkflowStatusItem):
else:
field.set_value(formdata.data, value)
def mappings_export_to_xml(self, parent, charset, include_id=False):
def mappings_export_to_xml(self, parent, include_id=False):
container = ET.SubElement(parent, 'mappings')
for mapping in self.mappings or []:
item = ET.SubElement(container, 'mapping')
item.attrib['field_id'] = str(mapping.field_id)
item.text = mapping.expression
def mappings_init_with_xml(self, container, charset, include_id=False, snapshot=False):
def mappings_init_with_xml(self, container, include_id=False, snapshot=False):
self.mappings = []
for child in container:
field_id = child.attrib.get('field_id', '')

28
wcs/wf/dispatch.py
View File

@ -85,11 +85,11 @@ class RuleNode(XmlSerialisable):
def get_parameters(self):
return ('role_id', 'value')
def role_id_export_to_xml(self, item, charset, include_id=False):
self._role_export_to_xml('role_id', item, charset, include_id=include_id, include_missing=True)
def role_id_export_to_xml(self, item, include_id=False):
self._role_export_to_xml('role_id', item, include_id=include_id, include_missing=True)
def role_id_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
self._role_init_with_xml('role_id', elem, charset, include_id=include_id, snapshot=snapshot)
def role_id_init_with_xml(self, elem, include_id=False, snapshot=False):
self._role_init_with_xml('role_id', elem, include_id=include_id, snapshot=snapshot)
class DispatchWorkflowStatusItem(WorkflowStatusItem):
@ -116,37 +116,37 @@ class DispatchWorkflowStatusItem(WorkflowStatusItem):
parameters.remove('role_id')
return parameters
def role_id_export_to_xml(self, item, charset, include_id=False):
self._role_export_to_xml('role_id', item, charset, include_id=include_id)
def role_id_export_to_xml(self, item, include_id=False):
self._role_export_to_xml('role_id', item, include_id=include_id)
def _get_role_id_from_xml(self, elem, charset, include_id=False, snapshot=False):
def _get_role_id_from_xml(self, elem, include_id=False, snapshot=False):
# override to allow for "other" role being a plain string referencing an user
# by its email address.
value = xml_node_text(elem)
if '@' in (xml_node_text(elem) or ''):
return value
return super()._get_role_id_from_xml(elem, charset, include_id=include_id, snapshot=snapshot)
return super()._get_role_id_from_xml(elem, include_id=include_id, snapshot=snapshot)
def role_id_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
self._role_init_with_xml('role_id', elem, charset, include_id=include_id, snapshot=snapshot)
def role_id_init_with_xml(self, elem, include_id=False, snapshot=False):
self._role_init_with_xml('role_id', elem, include_id=include_id, snapshot=snapshot)
def rules_export_to_xml(self, item, charset, include_id=False):
def rules_export_to_xml(self, item, include_id=False):
if self.dispatch_type != 'automatic' or not self.rules:
return
rules_node = ET.SubElement(item, 'rules')
for rule in self.rules:
rules_node.append(RuleNode(rule).export_to_xml(charset=charset, include_id=include_id))
rules_node.append(RuleNode(rule).export_to_xml(include_id=include_id))
return rules_node
def rules_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def rules_init_with_xml(self, elem, include_id=False, snapshot=False):
rules = []
if elem is None:
return
for rule_xml_node in elem.findall('rule'):
rule_node = RuleNode()
rule_node.init_with_xml(rule_xml_node, charset, include_id=include_id, snapshot=snapshot)
rule_node.init_with_xml(rule_xml_node, include_id=include_id, snapshot=snapshot)
rules.append(rule_node.as_dict())
if rules:
self.rules = rules

7
wcs/wf/export_to_model.py
View File

@ -678,8 +678,7 @@ class ExportToModel(WorkflowStatusItem):
user_field_values = {}
def process_text(t):
t = template_on_context(context, force_str(t), autoescape=False)
return force_str(t, get_publisher().site_charset)
return template_on_context(context, force_str(t), autoescape=False)
nodes = []
for node in root.iter():
@ -811,7 +810,7 @@ class ExportToModel(WorkflowStatusItem):
node_value.attrib['{%s}style-name' % OO_TEXT_NS] = 'Field_20_Value'
section_node.append(node_value)
def model_file_export_to_xml(self, xml_item, charset, include_id=False):
def model_file_export_to_xml(self, xml_item, include_id=False):
if not self.model_file:
return
el = ET.SubElement(xml_item, 'model_file')
@ -821,7 +820,7 @@ class ExportToModel(WorkflowStatusItem):
base64.encodebytes(self.model_file.get_file().read())
)
def model_file_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def model_file_init_with_xml(self, elem, include_id=False, snapshot=False):
if elem is None:
return
base_filename = elem.find('base_filename').text

10
wcs/wf/form.py
View File

@ -233,8 +233,8 @@ class FormWorkflowStatusItem(WorkflowStatusItem):
for field in self.formdef.fields:
yield from field.get_dependencies()
def export_to_xml(self, charset, include_id=False):
item = WorkflowStatusItem.export_to_xml(self, charset, include_id=include_id)
def export_to_xml(self, include_id=False):
item = WorkflowStatusItem.export_to_xml(self, include_id=include_id)
if not hasattr(self, 'formdef') or not self.formdef or not self.formdef.fields:
return item
formdef = ET.SubElement(item, 'formdef')
@ -245,12 +245,12 @@ class FormWorkflowStatusItem(WorkflowStatusItem):
fields = ET.SubElement(formdef, 'fields')
for field in self.formdef.fields:
fields.append(field.export_to_xml(charset=charset, include_id=include_id))
fields.append(field.export_to_xml(include_id=include_id))
return item
def init_with_xml(self, elem, charset, include_id=False, snapshot=False, check_datasources=True):
def init_with_xml(self, elem, include_id=False, snapshot=False, check_datasources=True):
super().init_with_xml(
elem, charset, include_id=include_id, snapshot=snapshot, check_datasources=check_datasources
elem, include_id=include_id, snapshot=snapshot, check_datasources=check_datasources
)
el = elem.find('formdef')
if el is None:

2
wcs/wf/jump.py
View File

@ -161,7 +161,7 @@ class JumpWorkflowStatusItem(WorkflowStatusJumpItem):
directory_name = 'jump'
directory_class = JumpDirectory
def timeout_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def timeout_init_with_xml(self, elem, include_id=False, snapshot=False):
if elem is None or elem.text is None:
self.timeout = None
else:

8
wcs/wf/profile.py
View File

@ -117,23 +117,23 @@ class UpdateUserProfileStatusItem(WorkflowStatusItem):
ProfileUpdateTableWidget, '%sfields' % prefix, title=_('Profile Update'), value=self.fields
)
def fields_export_to_xml(self, item, charset, include_id=False):
def fields_export_to_xml(self, item, include_id=False):
if not self.fields:
return
fields_node = ET.SubElement(item, 'fields')
for field in self.fields:
fields_node.append(FieldNode(field).export_to_xml(charset=charset, include_id=include_id))
fields_node.append(FieldNode(field).export_to_xml(include_id=include_id))
return fields_node
def fields_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def fields_init_with_xml(self, elem, include_id=False, snapshot=False):
fields = []
if elem is None:
return
for field_xml_node in elem.findall('field'):
field_node = FieldNode()
field_node.init_with_xml(field_xml_node, charset, include_id=include_id, snapshot=snapshot)
field_node.init_with_xml(field_xml_node, include_id=include_id, snapshot=snapshot)
fields.append(field_node.as_dict())
if fields:
self.fields = fields

8
wcs/wf/roles.py
View File

@ -46,11 +46,11 @@ def sign_ws_url(url):
class RoleMixin:
def role_id_export_to_xml(self, item, charset, include_id=False):
self._role_export_to_xml('role_id', item, charset, include_id=include_id)
def role_id_export_to_xml(self, item, include_id=False):
self._role_export_to_xml('role_id', item, include_id=include_id)
def role_id_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
self._role_init_with_xml('role_id', elem, charset, include_id=include_id, snapshot=snapshot)
def role_id_init_with_xml(self, elem, include_id=False, snapshot=False):
self._role_init_with_xml('role_id', elem, include_id=include_id, snapshot=snapshot)
def get_role_id_parameter_view_value(self):
return self.get_line_details()

10
wcs/wf/sendmail.py
View File

@ -74,7 +74,7 @@ class SendmailWorkflowStatusItem(WorkflowStatusItem):
yield from super().get_dependencies()
yield MailTemplate.get_by_slug(self.mail_template)
def _get_role_id_from_xml(self, elem, charset, include_id=False, snapshot=False):
def _get_role_id_from_xml(self, elem, include_id=False, snapshot=False):
# override to allow for destination set with computed values.
if elem is None:
return None
@ -83,12 +83,12 @@ class SendmailWorkflowStatusItem(WorkflowStatusItem):
if self.get_expression(value)['type'] != 'text' or '@' in value:
return value
return super()._get_role_id_from_xml(elem, charset, include_id=include_id, snapshot=snapshot)
return super()._get_role_id_from_xml(elem, include_id=include_id, snapshot=snapshot)
def to_export_to_xml(self, item, charset, include_id=False):
self._roles_export_to_xml('to', item, charset, include_id=include_id, include_missing=True)
def to_export_to_xml(self, item, include_id=False):
self._roles_export_to_xml('to', item, include_id=include_id, include_missing=True)
def mail_template_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def mail_template_init_with_xml(self, elem, include_id=False, snapshot=False):
self.mail_template = None
if elem is None:
return

20
wcs/wf/wscall.py
View File

@ -669,7 +669,7 @@ class WebserviceCallStatusItem(WorkflowStatusItem):
else:
return _('Error calling webservice')
def _kv_data_export_to_xml(self, xml_item, charset, include_id, attribute):
def _kv_data_export_to_xml(self, xml_item, include_id, attribute):
assert attribute
if not getattr(self, attribute):
return
@ -685,7 +685,7 @@ class WebserviceCallStatusItem(WorkflowStatusItem):
else:
raise AssertionError('unknown type for value (%r)' % key)
def _kv_data_init_with_xml(self, elem, charset, include_id, attribute):
def _kv_data_init_with_xml(self, elem, include_id, attribute):
if elem is None:
return
setattr(self, attribute, {})
@ -694,17 +694,17 @@ class WebserviceCallStatusItem(WorkflowStatusItem):
value = force_str(item.find('value').text or '')
getattr(self, attribute)[key] = value
def post_data_export_to_xml(self, xml_item, charset, include_id=False):
self._kv_data_export_to_xml(xml_item, charset, include_id=include_id, attribute='post_data')
def post_data_export_to_xml(self, xml_item, include_id=False):
self._kv_data_export_to_xml(xml_item, include_id=include_id, attribute='post_data')
def post_data_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
self._kv_data_init_with_xml(elem, charset, include_id=include_id, attribute='post_data')
def post_data_init_with_xml(self, elem, include_id=False, snapshot=False):
self._kv_data_init_with_xml(elem, include_id=include_id, attribute='post_data')
def qs_data_export_to_xml(self, xml_item, charset, include_id=False):
self._kv_data_export_to_xml(xml_item, charset, include_id=include_id, attribute='qs_data')
def qs_data_export_to_xml(self, xml_item, include_id=False):
self._kv_data_export_to_xml(xml_item, include_id=include_id, attribute='qs_data')
def qs_data_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
self._kv_data_init_with_xml(elem, charset, include_id=include_id, attribute='qs_data')
def qs_data_init_with_xml(self, elem, include_id=False, snapshot=False):
self._kv_data_init_with_xml(elem, include_id=include_id, attribute='qs_data')
register_item_class(WebserviceCallStatusItem)

137
wcs/workflows.py
View File

@ -530,8 +530,6 @@ class ContentSnapshotPart(EvolutionPart):
return value, data.get('%s_display' % field.id)
if field.convert_value_to_str:
return value, field.convert_value_to_str(value)
if isinstance(value, str):
return value, force_str(value, get_publisher().site_charset)
return value, value
def diff_field(field, old_data, new_data, block_field=None, block_item_num=0):
@ -1111,13 +1109,12 @@ class Workflow(StorableObject):
StorableObject.remove_self(self)
def export_to_xml(self, include_id=False):
charset = get_publisher().site_charset
root = ET.Element('workflow')
if include_id and self.id and not str(self.id).startswith('_'):
root.attrib['id'] = str(self.id)
ET.SubElement(root, 'name').text = force_str(self.name, charset)
ET.SubElement(root, 'name').text = self.name
if self.slug:
ET.SubElement(root, 'slug').text = force_str(self.slug, charset)
ET.SubElement(root, 'slug').text = self.slug
WorkflowCategory.object_category_xml_export(self, root, include_id=include_id)
@ -1126,21 +1123,21 @@ class Workflow(StorableObject):
for role_id, role_label in sorted(self.roles.items()):
role_node = ET.SubElement(roles_node, 'role')
role_node.attrib['id'] = role_id
role_node.text = force_str(role_label, charset)
role_node.text = role_label
possible_status = ET.SubElement(root, 'possible_status')
for status in self.possible_status:
possible_status.append(status.export_to_xml(charset=charset, include_id=include_id))
possible_status.append(status.export_to_xml(include_id=include_id))
if self.global_actions:
global_actions = ET.SubElement(root, 'global_actions')
for action in self.global_actions:
global_actions.append(action.export_to_xml(charset=charset, include_id=include_id))
global_actions.append(action.export_to_xml(include_id=include_id))
if self.criticality_levels:
criticality_levels = ET.SubElement(root, 'criticality_levels')
for level in self.criticality_levels:
criticality_levels.append(level.export_to_xml(charset=charset))
criticality_levels.append(level.export_to_xml())
if self.variables_formdef:
variables = ET.SubElement(root, 'variables')
@ -1148,7 +1145,7 @@ class Workflow(StorableObject):
ET.SubElement(formdef, 'name').text = '-' # required by formdef xml import
fields = ET.SubElement(formdef, 'fields')
for field in self.variables_formdef.fields:
fields.append(field.export_to_xml(charset=charset, include_id=include_id))
fields.append(field.export_to_xml(include_id=include_id))
if self.backoffice_fields_formdef:
variables = ET.SubElement(root, 'backoffice-fields')
@ -1156,7 +1153,7 @@ class Workflow(StorableObject):
ET.SubElement(formdef, 'name').text = '-' # required by formdef xml import
fields = ET.SubElement(formdef, 'fields')
for field in self.backoffice_fields_formdef.fields:
fields.append(field.export_to_xml(charset=charset, include_id=include_id))
fields.append(field.export_to_xml(include_id=include_id))
return root
@ -1176,7 +1173,6 @@ class Workflow(StorableObject):
@classmethod
def import_from_xml_tree(cls, tree, include_id=False, snapshot=False, check_datasources=True):
charset = get_publisher().site_charset
workflow = cls()
if tree.find('name') is None or not tree.find('name').text:
raise WorkflowImportError(_('Missing name'))
@ -1215,7 +1211,6 @@ class Workflow(StorableObject):
try:
status_o.init_with_xml(
status,
charset,
include_id=include_id,
snapshot=snapshot,
check_datasources=check_datasources,
@ -1235,7 +1230,7 @@ class Workflow(StorableObject):
action_o = WorkflowGlobalAction()
action_o.parent = workflow
try:
action_o.init_with_xml(action, charset, include_id=include_id, snapshot=snapshot)
action_o.init_with_xml(action, include_id=include_id, snapshot=snapshot)
except FormdefImportUnknownReferencedError as e:
for k, v in e._details.items():
unknown_referenced_objects_details[k].update(v)
@ -1249,7 +1244,7 @@ class Workflow(StorableObject):
if criticality_levels is not None:
for level in criticality_levels:
level_o = WorkflowCriticalityLevel()
level_o.init_with_xml(level, charset)
level_o.init_with_xml(level)
workflow.criticality_levels.append(level_o)
variables = tree.find('variables')
@ -1320,14 +1315,13 @@ class Workflow(StorableObject):
return render_list_of_roles(self, roles)
def get_json_export_dict(self, include_id=False):
charset = get_publisher().site_charset
root = {}
root['name'] = force_str(self.name, charset)
root['name'] = self.name
if include_id and self.id:
root['id'] = str(self.id)
roles = root['functions'] = {}
for role, label in self.roles.items():
roles[role] = force_str(label, charset)
roles[role] = label
statuses = root['statuses'] = []
endpoint_status_ids = [s.id for s in self.get_endpoint_status()]
waitpoint_status_ids = [s.id for s in self.get_waitpoint_status()]
@ -1335,7 +1329,7 @@ class Workflow(StorableObject):
statuses.append(
{
'id': status.id,
'name': force_str(status.name, charset),
'name': status.name,
'forced_endpoint': status.forced_endpoint,
'endpoint': status.id in endpoint_status_ids,
'waitpoint': status.id in waitpoint_status_ids,
@ -1467,7 +1461,7 @@ class XmlSerialisable:
node_name = None
key = None
def export_to_xml(self, charset, include_id=False):
def export_to_xml(self, include_id=False):
node = ET.Element(self.node_name)
if self.key:
node.attrib['type'] = self.key
@ -1475,36 +1469,34 @@ class XmlSerialisable:
node.attrib['id'] = self.id
for attribute in self.get_parameters():
if getattr(self, '%s_export_to_xml' % attribute, None):
getattr(self, '%s_export_to_xml' % attribute)(node, charset, include_id=include_id)
getattr(self, '%s_export_to_xml' % attribute)(node, include_id=include_id)
continue
if hasattr(self, attribute) and getattr(self, attribute) is not None:
el = ET.SubElement(node, attribute)
val = getattr(self, attribute)
if isinstance(val, dict):
for k, v in val.items():
ET.SubElement(el, k).text = force_str(v, charset, errors='replace')
ET.SubElement(el, k).text = force_str(v, errors='replace')
elif isinstance(val, list):
if attribute[-1] == 's':
atname = attribute[:-1]
else:
atname = 'item'
for v in val:
ET.SubElement(el, atname).text = force_str(str(v), charset, errors='replace')
ET.SubElement(el, atname).text = force_str(str(v), errors='replace')
elif isinstance(val, str):
el.text = force_str(val, charset, errors='replace')
el.text = force_str(val, errors='replace')
else:
el.text = str(val)
return node
def init_with_xml(self, elem, charset, include_id=False, snapshot=False, check_datasources=True):
def init_with_xml(self, elem, include_id=False, snapshot=False, check_datasources=True):
if include_id and elem.attrib.get('id'):
self.id = elem.attrib.get('id')
for attribute in self.get_parameters():
el = elem.find(attribute)
if getattr(self, '%s_init_with_xml' % attribute, None):
getattr(self, '%s_init_with_xml' % attribute)(
el, charset, include_id=include_id, snapshot=snapshot
)
getattr(self, '%s_init_with_xml' % attribute)(el, include_id=include_id, snapshot=snapshot)
continue
if el is None:
continue
@ -1530,7 +1522,7 @@ class XmlSerialisable:
else:
setattr(self, attribute, xml_node_text(el))
def _roles_export_to_xml(self, attribute, item, charset, include_id=False, include_missing=False):
def _roles_export_to_xml(self, attribute, item, include_id=False, include_missing=False):
if not hasattr(self, attribute) or not getattr(self, attribute):
return
el = ET.SubElement(item, attribute)
@ -1552,18 +1544,18 @@ class XmlSerialisable:
sub.attrib['role_id'] = role_id
sub.text = role_name
def _roles_init_with_xml(self, attribute, elem, charset, include_id=False, snapshot=False):
def _roles_init_with_xml(self, attribute, elem, include_id=False, snapshot=False):
if elem is None:
setattr(self, attribute, [])
else:
imported_roles = []
for child in elem:
imported_roles.append(
self._get_role_id_from_xml(child, charset, include_id=include_id, snapshot=snapshot)
self._get_role_id_from_xml(child, include_id=include_id, snapshot=snapshot)
)
setattr(self, attribute, imported_roles)
def _role_export_to_xml(self, attribute, item, charset, include_id=False, include_missing=False):
def _role_export_to_xml(self, attribute, item, include_id=False, include_missing=False):
if not hasattr(self, attribute) or not getattr(self, attribute):
return
role_id = str(getattr(self, attribute))
@ -1581,7 +1573,7 @@ class XmlSerialisable:
sub.attrib['role_id'] = role_id
sub.text = role_name
def _get_role_id_from_xml(self, elem, charset, include_id=False, snapshot=False):
def _get_role_id_from_xml(self, elem, include_id=False, snapshot=False):
if elem is None:
return None
@ -1624,11 +1616,11 @@ class XmlSerialisable:
role.store()
return role.id
def _role_init_with_xml(self, attribute, elem, charset, include_id=False, snapshot=False):
def _role_init_with_xml(self, attribute, elem, include_id=False, snapshot=False):
setattr(
self,
attribute,
self._get_role_id_from_xml(elem, charset, include_id=include_id, snapshot=snapshot),
self._get_role_id_from_xml(elem, include_id=include_id, snapshot=snapshot),
)
@ -1763,13 +1755,13 @@ class WorkflowGlobalActionManualTrigger(WorkflowGlobalActionTrigger):
return form
def roles_export_to_xml(self, item, charset, include_id=False):
self._roles_export_to_xml('roles', item, charset, include_id=include_id)
def roles_export_to_xml(self, item, include_id=False):
self._roles_export_to_xml('roles', item, include_id=include_id)
def roles_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
self._roles_init_with_xml('roles', elem, charset, include_id=include_id, snapshot=snapshot)
def roles_init_with_xml(self, elem, include_id=False, snapshot=False):
self._roles_init_with_xml('roles', elem, include_id=include_id, snapshot=snapshot)
def statuses_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def statuses_init_with_xml(self, elem, include_id=False, snapshot=False):
if not elem:
return
value = [xml_node_text(x) or '' for x in elem]
@ -2410,25 +2402,25 @@ class WorkflowGlobalAction(SerieOfActionsMixin):
self.triggers.append(o)
return o
def export_to_xml(self, charset, include_id=False):
def export_to_xml(self, include_id=False):
status = ET.Element('action')
ET.SubElement(status, 'id').text = force_str(self.id, charset)
ET.SubElement(status, 'name').text = force_str(self.name, charset)
ET.SubElement(status, 'id').text = self.id
ET.SubElement(status, 'name').text = self.name
if self.backoffice_info_text:
ET.SubElement(status, 'backoffice_info_text').text = force_str(self.backoffice_info_text, charset)
ET.SubElement(status, 'backoffice_info_text').text = self.backoffice_info_text
items = ET.SubElement(status, 'items')
for item in self.items:
items.append(item.export_to_xml(charset=charset, include_id=include_id))
items.append(item.export_to_xml(include_id=include_id))
triggers = ET.SubElement(status, 'triggers')
for trigger in self.triggers or []:
triggers.append(trigger.export_to_xml(charset=charset, include_id=include_id))
triggers.append(trigger.export_to_xml(include_id=include_id))
return status
def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def init_with_xml(self, elem, include_id=False, snapshot=False):
self.id = xml_node_text(elem.find('id'))
self.name = xml_node_text(elem.find('name'))
if elem.find('backoffice_info_text') is not None:
@ -2440,7 +2432,7 @@ class WorkflowGlobalAction(SerieOfActionsMixin):
self.add_action(item_type)
item_o = self.items[-1]
item_o.parent = self
item_o.init_with_xml(item, charset, include_id=include_id, snapshot=snapshot)
item_o.init_with_xml(item, include_id=include_id, snapshot=snapshot)
self.triggers = []
for trigger in elem.find('triggers'):
@ -2450,7 +2442,7 @@ class WorkflowGlobalAction(SerieOfActionsMixin):
if trigger.attrib.get('id'):
trigger_o.id = trigger.attrib['id']
trigger_o.parent = self
trigger_o.init_with_xml(trigger, charset, include_id=include_id, snapshot=snapshot)
trigger_o.init_with_xml(trigger, include_id=include_id, snapshot=snapshot)
def get_dependencies(self):
yield from super().get_dependencies()
@ -2498,15 +2490,15 @@ class WorkflowCriticalityLevel:
self.colour = colour
self.id = str(random.randint(0, 100000))
def export_to_xml(self, charset, include_id=False):
def export_to_xml(self, include_id=False):
level = ET.Element('criticality-level')
ET.SubElement(level, 'id').text = force_str(self.id, charset) if self.id else ''
ET.SubElement(level, 'name').text = force_str(self.name, charset)
ET.SubElement(level, 'id').text = self.id or ''
ET.SubElement(level, 'name').text = self.name
if self.colour:
ET.SubElement(level, 'colour').text = force_str(self.colour, charset)
ET.SubElement(level, 'colour').text = self.colour
return level
def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def init_with_xml(self, elem, include_id=False, snapshot=False):
self.id = xml_node_text(elem.find('id'))
self.name = xml_node_text(elem.find('name'))
if elem.find('colour') is not None:
@ -2708,19 +2700,19 @@ class WorkflowStatus(SerieOfActionsMixin):
del odict['parent']
return odict
def export_to_xml(self, charset, include_id=False):
def export_to_xml(self, include_id=False):
status = ET.Element('status')
ET.SubElement(status, 'id').text = force_str(self.id, charset)
ET.SubElement(status, 'name').text = force_str(self.name, charset)
ET.SubElement(status, 'colour').text = force_str(self.colour, charset)
ET.SubElement(status, 'id').text = str(self.id)
ET.SubElement(status, 'name').text = self.name
ET.SubElement(status, 'colour').text = self.colour
if self.extra_css_class:
ET.SubElement(status, 'extra_css_class').text = force_str(self.extra_css_class, charset)
ET.SubElement(status, 'extra_css_class').text = self.extra_css_class
if self.forced_endpoint:
ET.SubElement(status, 'forced_endpoint').text = 'true'
if self.backoffice_info_text:
ET.SubElement(status, 'backoffice_info_text').text = force_str(self.backoffice_info_text, charset)
ET.SubElement(status, 'backoffice_info_text').text = self.backoffice_info_text
visibility_node = ET.SubElement(status, 'visibility')
for role in self.visibility or []:
@ -2728,10 +2720,10 @@ class WorkflowStatus(SerieOfActionsMixin):
items = ET.SubElement(status, 'items')
for item in self.items:
items.append(item.export_to_xml(charset=charset, include_id=include_id))
items.append(item.export_to_xml(include_id=include_id))
return status
def init_with_xml(self, elem, charset, include_id=False, snapshot=False, check_datasources=True):
def init_with_xml(self, elem, include_id=False, snapshot=False, check_datasources=True):
self.id = xml_node_text(elem.find('id'))
self.name = xml_node_text(elem.find('name'))
if elem.find('colour') is not None:
@ -2757,7 +2749,6 @@ class WorkflowStatus(SerieOfActionsMixin):
try:
item_o.init_with_xml(
item,
charset,
include_id=include_id,
snapshot=snapshot,
check_datasources=check_datasources,
@ -3233,19 +3224,19 @@ class WorkflowStatusItem(XmlSerialisable):
formdata.data[backoffice_filefield_id] = upload
formdata.store()
def by_export_to_xml(self, item, charset, include_id=False):
self._roles_export_to_xml('by', item, charset, include_id=include_id)
def by_export_to_xml(self, item, include_id=False):
self._roles_export_to_xml('by', item, include_id=include_id)
def by_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
self._roles_init_with_xml('by', elem, charset, include_id=include_id, snapshot=snapshot)
def by_init_with_xml(self, elem, include_id=False, snapshot=False):
self._roles_init_with_xml('by', elem, include_id=include_id, snapshot=snapshot)
def to_export_to_xml(self, item, charset, include_id=False):
self._roles_export_to_xml('to', item, charset, include_id=include_id)
def to_export_to_xml(self, item, include_id=False):
self._roles_export_to_xml('to', item, include_id=include_id)
def to_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
self._roles_init_with_xml('to', elem, charset, include_id, snapshot=snapshot)
def to_init_with_xml(self, elem, include_id=False, snapshot=False):
self._roles_init_with_xml('to', elem, include_id, snapshot=snapshot)
def condition_init_with_xml(self, node, charset, include_id=False, snapshot=False):
def condition_init_with_xml(self, node, include_id=False, snapshot=False):
self.condition = None
if node is None:
return
@ -3267,7 +3258,7 @@ class WorkflowStatusItem(XmlSerialisable):
del odict['parent']
return odict
def attachments_init_with_xml(self, elem, charset, include_id=False, snapshot=False):
def attachments_init_with_xml(self, elem, include_id=False, snapshot=False):
if elem is None:
self.attachments = None
else:

8
wcs/wscalls.py
View File

@ -276,16 +276,16 @@ class NamedWsCall(XmlStorableObject):
if self.request.get('post_data'):
yield from self.request.get('post_data').values()
def export_request_to_xml(self, element, attribute_name, charset, **kwargs):
def export_request_to_xml(self, element, attribute_name, **kwargs):
request = getattr(self, attribute_name)
for attr in ('url', 'request_signature_key', 'method', 'timeout', 'cache_duration'):
ET.SubElement(element, attr).text = force_str(request.get(attr) or '', charset)
ET.SubElement(element, attr).text = force_str(request.get(attr) or '')
for attr in ('qs_data', 'post_data'):
data_element = ET.SubElement(element, attr)
for k, v in (request.get(attr) or {}).items():
sub = ET.SubElement(data_element, 'param')
sub.attrib['key'] = force_str(k, charset)
sub.text = force_str(v, charset)
sub.attrib['key'] = str(k)
sub.text = str(v)
if request.get('post_formdata'):
ET.SubElement(element, 'post_formdata')