black

Author: chris-adam

src/collective/documentgenerator/utils.py

@@ -33,16 +33,12 @@
 import tempfile
 
 
-logger = logging.getLogger('collective.documentgenerator')
+logger = logging.getLogger("collective.documentgenerator")
 
 
-def translate(msgid, domain='collective.documentgenerator'):
+def translate(msgid, domain="collective.documentgenerator"):
     portal = api.portal.get()
-    translation = i18n.translate(
-        msgid,
-        domain=domain,
-        context=portal.REQUEST
-    )
+    translation = i18n.translate(msgid, domain=domain, context=portal.REQUEST)
     return translation
 
 
@@ -51,98 +47,109 @@ def compute_md5(data):
     return md5
 
 
-def update_templates(templates, profile='', force=False):
+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
+    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('/ ')
+        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'))
+            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'))
+            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'))
+            ret.append((ppath, ospath, "os path error"))
             continue
-        with open(ospath, 'rb') as f:
+        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'))
+                ret.append((ppath, ospath, "unchanged"))
                 continue
             elif obj.has_been_modified() and not force:
-                ret.append((ppath, ospath, 'kept'))
+                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'))
+            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}))
+            raise Invalid(
+                _("${error_message} for key = '${key}'", mapping={"error_message": error_message, "key": key})
+            )
 
         original_dict[key] = update_dict[key]
 
 
-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):
+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
+    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')
+        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_')
+        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},
+            "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)
+        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'])))
+            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']:
+            if conf[match]["low"]:
                 repl = repl.lower()
-            custom_format = re.sub('%{}'.format(match), repl, custom_format)
+            custom_format = re.sub("%{}".format(match), repl, custom_format)
 
         # then format date
-        custom_format = custom_format.replace('_p_c_', '%%')
-        formatted_date = date.strftime(custom_format.encode('utf8'))
+        custom_format = custom_format.replace("_p_c_", "%%")
+        formatted_date = date.strftime(custom_format.encode("utf8"))
     return safe_unicode(formatted_date)
 
 
@@ -155,15 +162,15 @@ def remove_tmp_file(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'}
+    """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)
+        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'), ''))
+            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())
@@ -178,56 +185,52 @@ def update_oo_config_after_bigbang(event):
 
 
 def get_site_root_relative_path(obj):
-    return "/" + '/'.join(
-        getToolByName(obj, 'portal_url').getRelativeContentPath(obj)
-    )
+    return "/" + "/".join(getToolByName(obj, "portal_url").getRelativeContentPath(obj))
 
 
-def temporary_file_name(suffix=''):
-    tmp_dir = os.getenv('CUSTOM_TMP', None)
+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=''):
+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, 'w+') as tmp_file:
+    with open(tmp_filename, "w+") 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). """
+    """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')
+        tmp_file = create_temporary_file(odt_file, "-to-clean.odt")
         cleaner = Cleaner(path=tmp_file.name, verbose=1)
         cleaned = cleaner.run()
         if cleaned:
             manually_modified = pod_template.has_been_modified()
-            with open(tmp_file.name, 'rb') as res_file:
+            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)
+                    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)
+            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)
 
 
-def convert_odt(afile, fmt='pdf', gen_context=None, delete_temp_files=True, **kwargs):
+def convert_odt(afile, fmt="pdf", gen_context=None, delete_temp_files=True, **kwargs):
     """
     Convert an odt file to another format using appy.pod.
 
@@ -238,7 +241,7 @@ def convert_odt(afile, fmt='pdf', gen_context=None, delete_temp_files=True, **kw
     :param kwargs: other parameters passed to Renderer, i.e pdfOptions='ExportNotes=True;SelectPdfVersion=1'
     :return: converted file content
     """
-    temp_file = create_temporary_file(afile, '.odt')
+    temp_file = create_temporary_file(afile, ".odt")
     converted_filename = temporary_file_name(suffix=".{extension}".format(extension=fmt))
     if not gen_context:
         gen_context = {}
@@ -265,7 +268,7 @@ def convert_odt(afile, fmt='pdf', gen_context=None, delete_temp_files=True, **kw
                 type="error",
             )
             raise Invalid(u"Conversion failed, no converted file '{}'".format(safe_unicode(converted_filename)))
-        with open(converted_filename, 'rb') as f:
+        with open(converted_filename, "rb") as f:
             converted_file = f.read()
     finally:
         if delete_temp_files:
@@ -287,28 +290,34 @@ def convert_file(afile, fmt="pdf", renderer=False, gen_context=None, delete_temp
     :param delete_temp_files:
     """
     if renderer:
-        if not afile.filename.endswith('.odt'):
+        if not afile.filename.endswith(".odt"):
             message = _(u"Conversion with renderer only works from odt files.")
             raise Invalid(message)
         return convert_odt(afile, fmt=fmt, gen_context=gen_context, delete_temp_files=delete_temp_files)
     from appy.pod import converter
+
     converter_path = converter.__file__.endswith(".pyc") and converter.__file__[:-1] or converter.__file__
-    file_ext = afile.filename.split('.')[-1].lower()
+    file_ext = afile.filename.split(".")[-1].lower()
     temp_file = create_temporary_file(afile, base_name=".{}".format(file_ext))
     converted_filename = temp_file.name.replace(".{}".format(file_ext), ".{}".format(fmt))
     converted_file = ""
     try:
         ports = get_oo_port_list()
         port = ports[0] if ports else DEFAULT_OO_PORT
         command = "{python_uno_path} {converter_path} {temp_file} {fmt} -p {port} -e {server}".format(
-            python_uno_path=get_uno_path(), converter_path=converter_path, temp_file=temp_file.name, fmt=fmt,
-            port=port, server=get_oo_server())
+            python_uno_path=get_uno_path(),
+            converter_path=converter_path,
+            temp_file=temp_file.name,
+            fmt=fmt,
+            port=port,
+            server=get_oo_server(),
+        )
         out, err, code = runCommand(command)
         # This command has no output on success
         if code != 0 or err or not os.path.exists(converted_filename):
             message = _(u"Conversion failed, no converted file '{}'".format(safe_unicode(converted_filename)))
             raise Invalid(message)
-        with open(converted_filename, 'rb') as f:
+        with open(converted_filename, "rb") as f:
             converted_file = f.read()
     except Exception as e:
         api.portal.show_message(message=str(e), request=getSite().REQUEST, type="error")
@@ -320,8 +329,17 @@ def convert_file(afile, fmt="pdf", renderer=False, gen_context=None, delete_temp
     return converted_file
 
 
-def convert_and_save_file(afile, container, portal_type, output_name, fmt='pdf', from_uid=None, attributes=None,
-                          renderer=False, gen_context=None):
+def convert_and_save_file(
+    afile,
+    container,
+    portal_type,
+    output_name,
+    fmt="pdf",
+    from_uid=None,
+    attributes=None,
+    renderer=False,
+    gen_context=None,
+):
     """
     Convert a file to another libreoffice readable format using appy.pod and save it in a NamedBlobFile.
 
@@ -341,11 +359,8 @@ def convert_and_save_file(afile, container, portal_type, output_name, fmt='pdf',
         attributes = {}
     attributes["conv_from_uid"] = from_uid
     new_file = createContentInContainer(
-        container,
-        portal_type,
-        title=safe_unicode(output_name),
-        file=file_object,
-        **attributes)
+        container, portal_type, title=safe_unicode(output_name), file=file_object, **attributes
+    )
     if from_uid:
         annot = IAnnotations(new_file)
         annot["documentgenerator"] = {"conv_from_uid": from_uid}