9287988437

Committed 29 May 2024 02:29PM UTC coverage: 91.882% (+0.5%) from 91.42%

Build # 9287988437

Build Type

Pull #60

github

Committed by

duchenean

Commit Message

isort and add HAS_PLONE_6

Pull Request Pull Request #60: Plone 6 compatibility

Run Details

65 of 86 new or added lines in 15 files covered. (75.58%)

1 existing line in 1 file now uncovered.

2173 of 2365 relevant lines covered (91.88%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.31

/src/collective/documentgenerator/utils.py

# -*- coding: utf-8 -*-
from collective.documentgenerator import _
from imio.helpers.security import fplog
from plone import api
from plone.namedfile.file import NamedBlobFile
from Products.CMFCore.utils import getToolByName
from Products.CMFPlone.utils import safe_unicode
from zope import i18n
from zope.component import getMultiAdapter
from zope.component.hooks import getSite
from zope.component.hooks import setSite
from zope.interface import Interface
from zope.interface import Invalid
from zope.lifecycleevent import Attributes
from zope.lifecycleevent import modified

import hashlib
import logging
import os
import re
import six
import tempfile


if six.PY2:
    from appy.bin.odfclean import Cleaner
else:
    from appy.bin.oclean import Cleaner

logger = logging.getLogger('collective.documentgenerator')


def translate(msgid, domain='collective.documentgenerator'):
    portal = api.portal.get()
    translation = i18n.translate(
        msgid,
        domain=domain,
        context=portal.REQUEST
    )
    return translation


def compute_md5(data):
    md5 = hashlib.md5(data).hexdigest()
    return md5


def update_templates(templates, profile='', force=False):
    """
        function to manage templates update.
        # see http://trac.imio.be/trac/ticket/9383 for full implementation
        :param list templates: list of tuples containing ('plone-template-path', 'os-template-path')
        :param str profile: profile path stored on template (or various identification)
        :param bool force: force overrides of templates
    """
    # Don't use profile now !
    portal = api.portal.getSite()
    ret = []
    for (ppath, ospath) in templates:
        ppath = ppath.strip('/ ')
        obj = portal.unrestrictedTraverse(ppath, default=None)
        if not obj:
            logger.warn("The plone template '%s' was not found" % ppath)
            ret.append((ppath, ospath, 'plone path error'))
            continue
        if not obj.odt_file:
            logger.warn("The plone template '%s' doesn't have odt file" % ppath)
            ret.append((ppath, ospath, 'no odt file'))
            continue
        if not os.path.exists(ospath):
            logger.warn("The template file '%s' doesn't exist" % ospath)
            ret.append((ppath, ospath, 'os path error'))
            continue
        with open(ospath, 'rb') as f:
            data = f.read()
            new_md5 = compute_md5(data)
            if obj.initial_md5 == new_md5:
                ret.append((ppath, ospath, 'unchanged'))
                continue
            elif obj.has_been_modified() and not force:
                ret.append((ppath, ospath, 'kept'))
                continue
            obj.initial_md5 = new_md5
            obj.style_modification_md5 = new_md5
            obj.odt_file.data = data
            modified(obj, Attributes(Interface, 'odt_file'))
            ret.append((ppath, ospath, 'replaced'))
    return ret


def update_dict_with_validation(original_dict, update_dict, error_message=_("Dict update collision on key")):
    for key in update_dict:
        if key in original_dict:
            raise Invalid(_("${error_message} for key = '${key}'",
                            mapping={'error_message': error_message, 'key': key}))

        original_dict[key] = update_dict[key]


def safe_encode(value, encoding='utf-8'):
    """
        Converts a value to encoding, only when it is not already encoded.
    """
    if isinstance(value, six.PY2 and unicode or bytes):  # noqa: F821
        return value.encode(encoding)
    return value


def ulocalized_time(date, long_format=None, time_only=None, custom_format=None,
                    domain='plonelocales', target_language=None, context=None,
                    request=None, month_lc=True, day_lc=True):
    """
        Return for a datetime the string value with week and mont translated.
        Take into account %a, %A, %b, %B
    """
    if not custom_format:
        # use toLocalizedTime
        plone = getMultiAdapter((context, request), name=u'plone')
        formatted_date = plone.toLocalizedTime(date, long_format, time_only)
    else:
        from Products.CMFPlone.i18nl10n import monthname_msgid
        from Products.CMFPlone.i18nl10n import monthname_msgid_abbr
        from Products.CMFPlone.i18nl10n import weekdayname_msgid
        from Products.CMFPlone.i18nl10n import weekdayname_msgid_abbr
        if request is None:
            portal = api.portal.get()
            request = portal.REQUEST
        # first replace parts to translate
        custom_format = custom_format.replace('%%', '_p_c_')

        conf = {
            'a': {'fct': weekdayname_msgid_abbr, 'fmt': '%w', 'low': day_lc},
            'A': {'fct': weekdayname_msgid, 'fmt': '%w', 'low': day_lc},
            'b': {'fct': monthname_msgid_abbr, 'fmt': '%m', 'low': month_lc},
            'B': {'fct': monthname_msgid, 'fmt': '%m', 'low': month_lc},
        }
        matches = re.findall(r'%([aAbB])', custom_format)
        for match in sorted(set(matches)):
            # function( int(date.strftime(format) )
            msgid = conf[match]['fct'](int(date.strftime(conf[match]['fmt'])))
            repl = i18n.translate(msgid, domain, context=request, target_language=target_language)
            if conf[match]['low']:
                repl = repl.lower()
            custom_format = re.sub('%{}'.format(match), repl, custom_format)

        # then format date
        custom_format = custom_format.replace('_p_c_', '%%')
        if six.PY3:
            formatted_date = date.strftime(custom_format)
        else:
            formatted_date = date.strftime(custom_format.encode('utf8'))
    return safe_unicode(formatted_date)


def remove_tmp_file(filename):
    """Do not break if unable to remove temporary file, but log error if any."""
    try:
        os.remove(filename)
    except OSError:
        logger.warn("Could not remove temporary file at {0}".format(filename))


def update_oo_config():
    """ Update config following buildout var """
    key_template = 'collective.documentgenerator.browser.controlpanel.IDocumentGeneratorControlPanelSchema.{}'
    var = {'oo_server': 'OO_SERVER', 'oo_port_list': 'OO_PORT', 'uno_path': 'PYTHON_UNO'}
    for key in var.keys():
        full_key = key_template.format(key)
        configured_oo_option = api.portal.get_registry_record(full_key)
        env_value = os.getenv(var.get(key, 'NO_ONE'), None)
        if env_value:
            new_oo_option = type(configured_oo_option)(os.getenv(var.get(key, 'NO_ONE'), ''))
            if new_oo_option and new_oo_option != configured_oo_option:
                api.portal.set_registry_record(full_key, new_oo_option)
    logger.info("LibreOffice configuration updated for " + getSite().getId())


def update_oo_config_after_bigbang(event):
    setSite(event.object)
    try:
        update_oo_config()
    except Exception:
        logger.error("Update LibreOffice configuration failed", exc_info=1)


def get_site_root_relative_path(obj):
    return "/" + '/'.join(
        getToolByName(obj, 'portal_url').getRelativeContentPath(obj)
    )


def temporary_file_name(suffix=''):
    tmp_dir = os.getenv('CUSTOM_TMP', None)
    if tmp_dir and not os.path.exists(tmp_dir):
        os.mkdir(tmp_dir)
    return tempfile.mktemp(suffix=suffix, dir=tmp_dir)


def create_temporary_file(initial_file=None, base_name=''):
    tmp_filename = temporary_file_name(suffix=base_name)
    # create the file in any case
    with open(tmp_filename, 'wb') as tmp_file:
        if initial_file:
            tmp_file.write(initial_file.data)
    return tmp_file


def clean_notes(pod_template):
    """ Use appy.pod Cleaner to clean notes (comments). """
    cleaned = 0
    odt_file = pod_template.odt_file
    if odt_file:
        # write file to /tmp to be able to use appy.pod Cleaner
        tmp_file = create_temporary_file(odt_file, '-to-clean.odt')
        if six.PY2:
            cleaner = Cleaner(path=tmp_file.name, verbose=1)
        else:
            cleaner = Cleaner(path=tmp_file.name, silent=False)
        cleaned = cleaner.run()
        if cleaned:
            manually_modified = pod_template.has_been_modified()
            with open(tmp_file.name, 'rb') as res_file:
                # update template
                result = NamedBlobFile(
                    data=res_file.read(),
                    contentType=odt_file.contentType,
                    filename=pod_template.odt_file.filename)
            pod_template.odt_file = result
            if not manually_modified:
                pod_template.style_modification_md5 = pod_template.current_md5
            extras = 'pod_template={0} cleaned_parts={1}'.format(
                repr(pod_template), cleaned)
            fplog('clean_notes', extras=extras)
        remove_tmp_file(tmp_file.name)

    return bool(cleaned)

1	# -- coding: utf-8 --
2	from collective.documentgenerator import _	1✔
3	from imio.helpers.security import fplog	1✔
4	from plone import api	1✔
5	from plone.namedfile.file import NamedBlobFile	1✔
6	from Products.CMFCore.utils import getToolByName	1✔
7	from Products.CMFPlone.utils import safe_unicode	1✔
8	from zope import i18n	1✔
9	from zope.component import getMultiAdapter	1✔
10	from zope.component.hooks import getSite	1✔
11	from zope.component.hooks import setSite	1✔
12	from zope.interface import Interface	1✔
13	from zope.interface import Invalid	1✔
14	from zope.lifecycleevent import Attributes	1✔
15	from zope.lifecycleevent import modified	1✔
16
17	import hashlib	1✔
18	import logging	1✔
19	import os	1✔
20	import re	1✔
21	import six	1✔
22	import tempfile	1✔
23
24
25	if six.PY2:	1✔
NEW 26	from appy.bin.odfclean import Cleaner	×
27	else:
28	from appy.bin.oclean import Cleaner	1✔
29
30	logger = logging.getLogger('collective.documentgenerator')	1✔
31
32
33	def translate(msgid, domain='collective.documentgenerator'):	1✔
34	portal = api.portal.get()	1✔
35	translation = i18n.translate(	1✔
36	msgid,
37	domain=domain,
38	context=portal.REQUEST
39	)
40	return translation	1✔
41
42
43	def compute_md5(data):	1✔
44	md5 = hashlib.md5(data).hexdigest()	1✔
45	return md5	1✔
46
47
48	def update_templates(templates, profile='', force=False):	1✔
49	"""
50	function to manage templates update.
51	# see http://trac.imio.be/trac/ticket/9383 for full implementation
52	:param list templates: list of tuples containing ('plone-template-path', 'os-template-path')
53	:param str profile: profile path stored on template (or various identification)
54	:param bool force: force overrides of templates
55	"""
56	# Don't use profile now !
57	portal = api.portal.getSite()	1✔
58	ret = []	1✔
59	for (ppath, ospath) in templates:	1✔
60	ppath = ppath.strip('/ ')	1✔
61	obj = portal.unrestrictedTraverse(ppath, default=None)	1✔
62	if not obj:	1✔
63	logger.warn("The plone template '%s' was not found" % ppath)	1✔
64	ret.append((ppath, ospath, 'plone path error'))	1✔
65	continue	1✔
66	if not obj.odt_file:	1✔
67	logger.warn("The plone template '%s' doesn't have odt file" % ppath)	×
68	ret.append((ppath, ospath, 'no odt file'))	×
69	continue	×
70	if not os.path.exists(ospath):	1✔
71	logger.warn("The template file '%s' doesn't exist" % ospath)	1✔
72	ret.append((ppath, ospath, 'os path error'))	1✔
73	continue	1✔
74	with open(ospath, 'rb') as f:	1✔
75	data = f.read()	1✔
76	new_md5 = compute_md5(data)	1✔
77	if obj.initial_md5 == new_md5:	1✔
78	ret.append((ppath, ospath, 'unchanged'))	1✔
79	continue	1✔
80	elif obj.has_been_modified() and not force:	1✔
81	ret.append((ppath, ospath, 'kept'))	1✔
82	continue	1✔
83	obj.initial_md5 = new_md5	1✔
84	obj.style_modification_md5 = new_md5	1✔
85	obj.odt_file.data = data	1✔
86	modified(obj, Attributes(Interface, 'odt_file'))	1✔
87	ret.append((ppath, ospath, 'replaced'))	1✔
88	return ret	1✔
89
90
91	def update_dict_with_validation(original_dict, update_dict, error_message=_("Dict update collision on key")):	1✔
92	for key in update_dict:	1✔
93	if key in original_dict:	1✔
94	raise Invalid(_("${error_message} for key = '${key}'",	1✔
95	mapping={'error_message': error_message, 'key': key}))
96
97	original_dict[key] = update_dict[key]	1✔
98
99
100	def safe_encode(value, encoding='utf-8'):	1✔
101	"""
102	Converts a value to encoding, only when it is not already encoded.
103	"""
NEW 104	if isinstance(value, six.PY2 and unicode or bytes): # noqa: F821	×
NEW 105	return value.encode(encoding)	×
NEW 106	return value	×
107
108
109	def ulocalized_time(date, long_format=None, time_only=None, custom_format=None,	1✔
110	domain='plonelocales', target_language=None, context=None,
111	request=None, month_lc=True, day_lc=True):
112	"""
113	Return for a datetime the string value with week and mont translated.
114	Take into account %a, %A, %b, %B
115	"""
116	if not custom_format:	1✔
117	# use toLocalizedTime
118	plone = getMultiAdapter((context, request), name=u'plone')	1✔
119	formatted_date = plone.toLocalizedTime(date, long_format, time_only)	1✔
120	else:
121	from Products.CMFPlone.i18nl10n import monthname_msgid	1✔
122	from Products.CMFPlone.i18nl10n import monthname_msgid_abbr	1✔
123	from Products.CMFPlone.i18nl10n import weekdayname_msgid	1✔
124	from Products.CMFPlone.i18nl10n import weekdayname_msgid_abbr	1✔
125	if request is None:	1✔
126	portal = api.portal.get()	×
127	request = portal.REQUEST	×
128	# first replace parts to translate
129	custom_format = custom_format.replace('%%', '_p_c_')	1✔
130
131	conf = {	1✔
132	'a': {'fct': weekdayname_msgid_abbr, 'fmt': '%w', 'low': day_lc},
133	'A': {'fct': weekdayname_msgid, 'fmt': '%w', 'low': day_lc},
134	'b': {'fct': monthname_msgid_abbr, 'fmt': '%m', 'low': month_lc},
135	'B': {'fct': monthname_msgid, 'fmt': '%m', 'low': month_lc},
136	}
137	matches = re.findall(r'%([aAbB])', custom_format)	1✔
138	for match in sorted(set(matches)):	1✔
139	# function( int(date.strftime(format) )
140	msgid = conf[match]['fct'](int(date.strftime(conf[match]['fmt'])))	1✔
141	repl = i18n.translate(msgid, domain, context=request, target_language=target_language)	1✔
142	if conf[match]['low']:	1✔
143	repl = repl.lower()	1✔
144	custom_format = re.sub('%{}'.format(match), repl, custom_format)	1✔
145
146	# then format date
147	custom_format = custom_format.replace('_p_c_', '%%')	1✔
148	if six.PY3:	1✔
149	formatted_date = date.strftime(custom_format)	1✔
150	else:
NEW 151	formatted_date = date.strftime(custom_format.encode('utf8'))	×
152	return safe_unicode(formatted_date)	1✔
153
154
155	def remove_tmp_file(filename):	1✔
156	"""Do not break if unable to remove temporary file, but log error if any."""
157	try:	1✔
158	os.remove(filename)	1✔
159	except OSError:	×
160	logger.warn("Could not remove temporary file at {0}".format(filename))	×
161
162
163	def update_oo_config():	1✔
164	""" Update config following buildout var """
165	key_template = 'collective.documentgenerator.browser.controlpanel.IDocumentGeneratorControlPanelSchema.{}'	1✔
166	var = {'oo_server': 'OO_SERVER', 'oo_port_list': 'OO_PORT', 'uno_path': 'PYTHON_UNO'}	1✔
167	for key in var.keys():	1✔
168	full_key = key_template.format(key)	1✔
169	configured_oo_option = api.portal.get_registry_record(full_key)	1✔
170	env_value = os.getenv(var.get(key, 'NO_ONE'), None)	1✔
171	if env_value:	1✔
172	new_oo_option = type(configured_oo_option)(os.getenv(var.get(key, 'NO_ONE'), ''))	1✔
173	if new_oo_option and new_oo_option != configured_oo_option:	1✔
174	api.portal.set_registry_record(full_key, new_oo_option)	1✔
175	logger.info("LibreOffice configuration updated for " + getSite().getId())	1✔
176
177
178	def update_oo_config_after_bigbang(event):	1✔
179	setSite(event.object)	×
180	try:	×
181	update_oo_config()	×
182	except Exception:	×
183	logger.error("Update LibreOffice configuration failed", exc_info=1)	×
184
185
186	def get_site_root_relative_path(obj):	1✔
187	return "/" + '/'.join(	1✔
188	getToolByName(obj, 'portal_url').getRelativeContentPath(obj)
189	)
190
191
192	def temporary_file_name(suffix=''):	1✔
193	tmp_dir = os.getenv('CUSTOM_TMP', None)	1✔
194	if tmp_dir and not os.path.exists(tmp_dir):	1✔
195	os.mkdir(tmp_dir)	1✔
196	return tempfile.mktemp(suffix=suffix, dir=tmp_dir)	1✔
197
198
199	def create_temporary_file(initial_file=None, base_name=''):	1✔
200	tmp_filename = temporary_file_name(suffix=base_name)	1✔
201	# create the file in any case
202	with open(tmp_filename, 'wb') as tmp_file:	1✔
203	if initial_file:	1✔
204	tmp_file.write(initial_file.data)	1✔
205	return tmp_file	1✔
206
207
208	def clean_notes(pod_template):	1✔
209	""" Use appy.pod Cleaner to clean notes (comments). """
210	cleaned = 0	1✔
211	odt_file = pod_template.odt_file	1✔
212	if odt_file:	1✔
213	# write file to /tmp to be able to use appy.pod Cleaner
214	tmp_file = create_temporary_file(odt_file, '-to-clean.odt')	1✔
215	if six.PY2:	1✔
NEW 216	cleaner = Cleaner(path=tmp_file.name, verbose=1)	×
217	else:
218	cleaner = Cleaner(path=tmp_file.name, silent=False)	1✔
219	cleaned = cleaner.run()	1✔
220	if cleaned:	1✔
221	manually_modified = pod_template.has_been_modified()	1✔
222	with open(tmp_file.name, 'rb') as res_file:	1✔
223	# update template
224	result = NamedBlobFile(	1✔
225	data=res_file.read(),
226	contentType=odt_file.contentType,
227	filename=pod_template.odt_file.filename)
228	pod_template.odt_file = result	1✔
229	if not manually_modified:	1✔
230	pod_template.style_modification_md5 = pod_template.current_md5	1✔
231	extras = 'pod_template={0} cleaned_parts={1}'.format(	1✔
232	repr(pod_template), cleaned)
233	fplog('clean_notes', extras=extras)	1✔
234	remove_tmp_file(tmp_file.name)	1✔
235
236	return bool(cleaned)	1✔

collective / collective.documentgenerator / 9287988437

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous