Merge pull request #72 from collective/SE-267/odfsplit

Added odfsplit utils

Author: sgeulette

CHANGES.rst

@@ -12,6 +12,8 @@ Changelog
   [gbastien]
 - Made subtemplate icon grey.
   [chris-adam]
+- Added odfsplit utils.
+  [chris-adam]
 
 3.45 (2025-06-25)
 -----------------

setup.py

@@ -66,5 +66,7 @@
     entry_points="""
     [z3c.autoinclude.plugin]
     target = plone
+    [console_scripts]
+    odfsplit = appy.bin.odfsplit:main
     """,
 )

src/collective/documentgenerator/__init__.py

@@ -1,11 +1,17 @@
 # -*- coding: utf-8 -*-
 """Init and utils."""
-
 from zope.i18nmessageid import MessageFactory
 
+import os
+
 
 _ = MessageFactory('collective.documentgenerator')
 
+if os.environ.get("ZOPE_HOME", ""):
+    BLDT_DIR = "/".join(os.getenv("INSTANCE_HOME", "").split("/")[:-2])
+else:  # test env
+    BLDT_DIR = os.getenv("PWD", "")
+
 
 def initialize(context):
     """Initializer called when used as a Zope 2 product."""

src/collective/documentgenerator/browser/generation_view.py

@@ -492,13 +492,18 @@ def __call__(self, document_uid='', document_url_path='', generated_doc_title=''
         persisted_doc = self.generate_persistent_doc(self.pod_template, self.output_format)
         self.redirects(persisted_doc)
 
+    def add_mailing_infos(self, doc, gen_context):
+        """ store mailing informations on generated doc """
+        annot = IAnnotations(doc)
+        annot['documentgenerator'] = {"need_mailing": False, "template_uid": self.pod_template.UID(), "mailed": True}
+
     def _get_base_args(self, template_uid, output_format):
         annot = IAnnotations(self.document).get('documentgenerator', '')
         if not annot or 'template_uid' not in annot:
             raise Exception("Cannot find 'template_uid' on document '{0}'".format(self.document.absolute_url()))
         self.orig_template = self.get_pod_template(annot['template_uid'])
-        if (not base_hasattr(self.orig_template, 'mailing_loop_template') or
-                not self.orig_template.mailing_loop_template):
+        if (not base_hasattr(self.orig_template, 'mailing_loop_template')
+                or not self.orig_template.mailing_loop_template):
             raise Exception("Cannot find 'mailing_loop_template' on template '{0}'".format(
                 self.orig_template.absolute_url()))
         loop_template = self.get_pod_template(self.orig_template.mailing_loop_template)

src/collective/documentgenerator/tests/test_mailing.odt

@@ -2,6 +2,7 @@
 from collective.documentgenerator.testing import PODTemplateIntegrationTest
 from collective.documentgenerator.utils import compute_md5
 from collective.documentgenerator.utils import convert_odt
+from collective.documentgenerator.utils import odfsplit
 from collective.documentgenerator.utils import temporary_file_name
 from collective.documentgenerator.utils import update_dict_with_validation
 from collective.documentgenerator.utils import update_oo_config
@@ -210,3 +211,33 @@ def test_convert_odt(self):
         self.assertIn("<body", content)
         self.assertIn("Page 1", content)
         self.assertIn("Page 2", content)
+
+    def test_odfsplit(self):
+        filename = u"test_mailing.odt"
+        current_path = os.path.dirname(__file__)
+        content = open(os.path.join(current_path, filename), "rb").read()
+        code, result, nb = odfsplit(content)
+        self.assertEqual(code, 0)
+        self.assertEqual(nb, 2)
+        results = list(result)
+        self.assertEqual(len(results), 2)
+        self.assertEqual(len(results[0]), 32531)
+        self.assertEqual(len(results[1]), 32352)
+
+        filename = u"test_file.odt"
+        current_path = os.path.dirname(__file__)
+        content = open(os.path.join(current_path, filename), "rb").read()
+        code, result, nb = odfsplit(content)
+        self.assertEqual(code, 0)
+        self.assertEqual(nb, 1)
+        results = list(result)
+        self.assertEqual(len(results), 1)
+        self.assertEqual(len(results[0]), 9729)
+
+        filename = u"test_utils.py"
+        current_path = os.path.dirname(__file__)
+        content = open(os.path.join(current_path, filename), "rb").read()
+        code, result, nb = odfsplit(content)
+        self.assertEqual(code, 1)
+        self.assertEqual(nb, 0)
+        self.assertTrue(result.strip().endswith("zipfile.BadZipfile: File is not a zip file"))

src/collective/documentgenerator/tests/test_utils.py

@@ -3,9 +3,11 @@
 from appy.pod.lo_pool import LoPool
 from appy.pod.renderer import Renderer
 from collective.documentgenerator import _
+from collective.documentgenerator import BLDT_DIR
 from collective.documentgenerator import config
 from imio.helpers.content import uuidToObject
 from imio.helpers.security import fplog
+from imio.pyutils.system import runCommand
 from plone import api
 from plone.dexterity.utils import createContentInContainer
 from plone.namedfile.file import NamedBlobFile
@@ -307,3 +309,43 @@ def need_mailing_value(document=None, document_uid=None):
     if "documentgenerator" in annot and annot["documentgenerator"].get("need_mailing", False):
         return True
     return False
+
+
+def odfsplit(content):
+    """Splits an ODT document into a series of sub-documents. The split is based on page breaks.
+
+    :param content: The binary content of the ODT file to be split.
+    :return: A tuple containing the exit code, a generator yielding the binary content of each subfile and
+             the number of files
+    """
+
+    def get_subfiles(temp_file, nb_files):
+        if nb_files == 1:
+            with open(temp_file, "rb") as f:
+                yield f.read()
+            remove_tmp_file(temp_file)
+            return
+        for i in range(1, nb_files + 1):
+            subfile = temp_file.replace(".odt", ".{}.odt".format(i))
+            with open(subfile, "rb") as sf:
+                yield sf.read()
+            remove_tmp_file(subfile)
+        remove_tmp_file(temp_file)
+
+    temp_file = temporary_file_name(suffix=".odt")
+    with open(temp_file, "wb") as f:
+        f.write(content)
+    command = "{pwd}/bin/odfsplit {temp_file}".format(temp_file=temp_file, pwd=BLDT_DIR)
+    out, err, code = runCommand(command)
+    nb_files = 0
+    if out and code == 0:
+        part0 = out[0].split(" ")[0]  # Ex: "2 files were generated."
+        if part0.isdigit():
+            nb_files = int(part0)
+            value = get_subfiles(temp_file, nb_files)
+        else:
+            nb_files = 1
+            value = get_subfiles(temp_file, 1)
+    else:
+        value = "".join(err)
+    return code, value, nb_files