15898628565

Committed 26 Jun 2025 09:48AM UTC coverage: 91.987% (-1.8%) from 93.775%

Build # 15898628565

Build Type

Pull #68

github

Committed by

chris-adam

Commit Message

Added uninstall profile

Pull Request Pull Request #68: Added uninstall profile

Run Details

2181 of 2371 relevant lines covered (91.99%)

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

collective / collective.documentgenerator / 15898628565

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