Add multiple files (zip) support for odt2pdf converter.

MarcoSteinacher · MarcoSteinacher · commit dbdf3f486265 · 2026-01-30T18:57:48.000+01:00
GitHub issue #42
diff --git a/django/cohiva/tests/test_views.py b/django/cohiva/tests/test_views.py
@@ -0,0 +1,178 @@
+import io
+import os
+import zipfile
+
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.http import FileResponse, HttpResponseNotFound
+from django.test import TestCase
+
+from cohiva.views.generic import UploadedFileProcessorMixin
+
+
+class UploadedFileProcessorMixinTestCase(TestCase):
+    def test_no_input_files(self):
+        cls = UploadedFileProcessorMixin()
+        input_files = [f for f in cls.get_input_files()]
+        self.assertEqual(len(input_files), 0)
+
+    def test_no_output_files(self):
+        cls = UploadedFileProcessorMixin()
+        response = cls.get_response()
+        self.assertTrue(isinstance(response, HttpResponseNotFound))
+
+    def test_single_file(self):
+        test_upload = SimpleUploadedFile(
+            "test.txt",
+            b"Testing",
+            content_type="text/plain",
+        )
+        cls = UploadedFileProcessorMixin()
+        cls.set_uploaded_file(test_upload)
+        input_files = [f for f in cls.get_input_files()]
+        self.assertEqual(len(input_files), 1)
+        self.assertEqual(input_files[0], test_upload)
+
+        response_file = "/tmp/response.pdf"
+        response_content = b"%PDF-1.4\n%Dummy PDF"
+        with open(response_file, "wb") as f:
+            f.write(response_content)
+        cls.add_output_file(response_file, "test.pdf", "application/pdf")
+        response = cls.get_response()
+        self.assertTrue(isinstance(response, FileResponse))
+        self.assertEqual(response.headers["Content-Type"], "application/pdf")
+        self.assertEqual(
+            response.headers["Content-Disposition"], 'attachment; filename="test.pdf"'
+        )
+        self.assertEqual(response.getvalue(), response_content)
+        response.close()
+        # Make sure the file was not deleted
+        self.assertTrue(os.path.isfile(response_file))
+        os.unlink(response_file)
+
+    def test_multiple_files(self):
+        file_like_object = io.BytesIO()
+        with zipfile.ZipFile(file_like_object, "w") as archive:
+            archive.writestr("testA.txt", b"Testing A")
+            archive.writestr("testB.txt", b"Testing B")
+        file_like_object.seek(0)
+        test_upload = SimpleUploadedFile(
+            "test_files.zip",
+            file_like_object.getvalue(),
+            content_type="application/zip",
+        )
+        cls = UploadedFileProcessorMixin()
+        cls.set_uploaded_file(test_upload)
+        input_file = [f for f in cls.get_input_files()]
+        self.assertEqual(len(input_file), 2)
+
+        response = cls.get_response()
+        self.assertTrue(isinstance(response, HttpResponseNotFound))
+
+        response_file_a = "/tmp/responseA.pdf"
+        response_content_a = b"%PDF-1.4\n%Dummy PDF A"
+        with open(response_file_a, "wb") as f:
+            f.write(response_content_a)
+        response_file_b = "/tmp/responseB.pdf"
+        response_content_b = b"%PDF-1.4\n%Dummy PDF B"
+        with open(response_file_b, "wb") as f:
+            f.write(response_content_b)
+        cls.add_output_file(response_file_a, "testA.pdf", "application/pdf")
+        cls.add_output_file(response_file_b, "testB.pdf", "application/pdf")
+        response = cls.get_response()
+        self.assertTrue(isinstance(response, FileResponse))
+        self.assertEqual(response.headers["Content-Type"], "application/zip")
+        self.assertEqual(
+            response.headers["Content-Disposition"],
+            'attachment; filename="test_files_resultat.zip"',
+        )
+        with zipfile.ZipFile(io.BytesIO(response.getvalue())) as zip_file:
+            self.assertEqual(zip_file.read("testA.pdf"), response_content_a)
+            self.assertEqual(zip_file.read("testB.pdf"), response_content_b)
+        # Make sure the files were not deleted
+        self.assertTrue(os.path.isfile(response_file_a))
+        self.assertTrue(os.path.isfile(response_file_b))
+        os.unlink(response_file_a)
+        os.unlink(response_file_b)
+
+    def test_delete_after_single_file(self):
+        cls = UploadedFileProcessorMixin()
+        response_file = "/tmp/response.pdf"
+        response_content = b"%PDF-1.4\n%Dummy PDF"
+        with open(response_file, "wb") as f:
+            f.write(response_content)
+        cls.add_output_file(response_file, "test.pdf", "application/pdf", delete_after=True)
+        response = cls.get_response()
+        self.assertTrue(isinstance(response, FileResponse))
+        response.close()
+        # Make sure the file was deleted
+        self.assertFalse(os.path.isfile(response_file))
+
+    def test_delete_after_multiple_files(self):
+        cls = UploadedFileProcessorMixin()
+        response_file_a = "/tmp/responseA.pdf"
+        response_content_a = b"%PDF-1.4\n%Dummy PDF A"
+        with open(response_file_a, "wb") as f:
+            f.write(response_content_a)
+        response_file_b = "/tmp/responseB.pdf"
+        response_content_b = b"%PDF-1.4\n%Dummy PDF B"
+        with open(response_file_b, "wb") as f:
+            f.write(response_content_b)
+        cls.add_output_file(response_file_a, "testA.pdf", "application/pdf", delete_after=True)
+        cls.add_output_file(response_file_b, "testB.pdf", "application/pdf", delete_after=True)
+        response = cls.get_response()
+        self.assertTrue(isinstance(response, FileResponse))
+        # Make sure the files were deleted
+        self.assertFalse(os.path.isfile(response_file_a))
+        self.assertFalse(os.path.isfile(response_file_b))
+
+    def test_custom_archive_filename(self):
+        cls = UploadedFileProcessorMixin()
+        response_file_a = "/tmp/responseA.pdf"
+        response_content_a = b"%PDF-1.4\n%Dummy PDF A"
+        with open(response_file_a, "wb") as f:
+            f.write(response_content_a)
+        response_file_b = "/tmp/responseB.pdf"
+        response_content_b = b"%PDF-1.4\n%Dummy PDF B"
+        with open(response_file_b, "wb") as f:
+            f.write(response_content_b)
+        cls.add_output_file(response_file_a, "testA.pdf", "application/pdf")
+        cls.add_output_file(response_file_b, "testB.pdf", "application/pdf")
+        response = cls.get_response(archive_file_name="custom.zip")
+        self.assertEqual(
+            response.headers["Content-Disposition"], 'attachment; filename="custom.zip"'
+        )
+        os.unlink(response_file_a)
+        os.unlink(response_file_b)
+
+    def test_no_custom_archive_filename(self):
+        cls = UploadedFileProcessorMixin()
+        response_file_a = "/tmp/responseA.pdf"
+        response_content_a = b"%PDF-1.4\n%Dummy PDF A"
+        with open(response_file_a, "wb") as f:
+            f.write(response_content_a)
+        response_file_b = "/tmp/responseB.pdf"
+        response_content_b = b"%PDF-1.4\n%Dummy PDF B"
+        with open(response_file_b, "wb") as f:
+            f.write(response_content_b)
+        cls.add_output_file(response_file_a, "testA.pdf", "application/pdf")
+        cls.add_output_file(response_file_b, "testB.pdf", "application/pdf")
+        response = cls.get_response()
+        self.assertEqual(
+            response.headers["Content-Disposition"], 'attachment; filename="resultat.zip"'
+        )
+        os.unlink(response_file_a)
+        os.unlink(response_file_b)
+
+    def test_custom_archive_filename_single_file(self):
+        cls = UploadedFileProcessorMixin()
+        response_file_a = "/tmp/responseA.pdf"
+        response_content_a = b"%PDF-1.4\n%Dummy PDF A"
+        with open(response_file_a, "wb") as f:
+            f.write(response_content_a)
+        cls.add_output_file(response_file_a, "testA.pdf", "application/pdf")
+        response = cls.get_response(archive_file_name="custom.zip")
+        self.assertEqual(
+            response.headers["Content-Disposition"], 'attachment; filename="testA.pdf"'
+        )
+        response.close()
+        os.unlink(response_file_a)
diff --git a/django/cohiva/views/generic.py b/django/cohiva/views/generic.py
@@ -1,11 +1,83 @@
 import io
+import os
+import tempfile
+import zipfile
 from zipfile import ZipFile
 
 from django.core.exceptions import ImproperlyConfigured
 from django.http import FileResponse, HttpResponseNotFound
 from django.views.generic.base import View
 
 
+class UploadedFileProcessorMixin:
+    def __init__(self, *args, **kwargs):
+        self.uploaded_file = None
+        self.output_files = []
+
+    def set_uploaded_file(self, uploaded_file):
+        self.uploaded_file = uploaded_file
+
+    def get_input_files(self):
+        """If uploaded_file is a zip file, return the unpacked files, otherwise return the uploaded file."""
+        if not self.uploaded_file:
+            return
+        if self.uploaded_file.name.endswith(".zip"):
+            with zipfile.ZipFile(io.BytesIO(self.uploaded_file.read())) as zip_file:
+                for zipinfo in zip_file.infolist():
+                    if not zipinfo.is_dir():
+                        file_like_object = io.BytesIO(zip_file.read(zipinfo))
+                        file_like_object.name = zipinfo.filename
+                        yield file_like_object
+        else:
+            yield self.uploaded_file
+
+    def add_output_file(self, filepath, filename, content_type=None, delete_after=False):
+        self.output_files.append(
+            {
+                "filepath": filepath,
+                "filename": filename,
+                "content_type": content_type,
+                "delete_after": delete_after,
+            }
+        )
+
+    def get_response(self, archive_file_name=None):
+        if len(self.output_files) == 1:
+            if self.output_files[0]["delete_after"]:
+                # Make a temporary copy of the file so we can delete it before sending the response.
+                output_file = tempfile.TemporaryFile()
+                with open(self.output_files[0]["filepath"], "rb") as source_file:
+                    for chunk in iter(lambda: source_file.read(8192), b""):
+                        output_file.write(chunk)
+                output_file.seek(0)
+            else:
+                output_file = open(self.output_files[0]["filepath"], "rb")
+            resp = FileResponse(
+                output_file,
+                as_attachment=True,
+                filename=self.output_files[0]["filename"],
+                content_type=self.output_files[0]["content_type"],
+            )
+        elif len(self.output_files) > 1:
+            file_like_object = io.BytesIO()
+            with ZipFile(file_like_object, "w") as archive:
+                for f in self.output_files:
+                    archive.write(f["filepath"], f["filename"])
+            file_like_object.seek(0)
+            if not archive_file_name:
+                if self.uploaded_file:
+                    archive_file_name = f"{self.uploaded_file.name[0:-4]}_resultat.zip"
+                else:
+                    archive_file_name = "resultat.zip"
+            resp = FileResponse(file_like_object, as_attachment=True, filename=archive_file_name)
+        else:
+            resp = HttpResponseNotFound("Keine Dateien gefunden.")
+        for f in self.output_files:
+            if f["delete_after"]:
+                os.unlink(f["filepath"])
+        return resp
+
+
 class ZipDownloadView(View):
     zipfile_name = "download.zip"
 
diff --git a/django/geno/forms.py b/django/geno/forms.py
@@ -1052,13 +1052,17 @@ class TransactionUploadFileForm(forms.Form):
 class Odt2PdfForm(forms.Form):
     """Form for uploading ODT files to convert to PDF."""
 
+    form_id = "odt2pdf-form"
+
     file = forms.FileField(
-        label=_("ODT-Datei"),
+        label=_("ODT- oder ZIP-Datei"),
         required=True,
-        help_text=_("Wählen Sie eine LibreOffice-Datei (.odt) zum Hochladen aus."),
+        help_text=_(
+            "Wähle eine LibreOffice-Datei (.odt) oder eine ZIP-Datei (.odt) zum Hochladen aus."
+        ),
         widget=UnfoldAdminFileFieldWidget(
             attrs={
-                "accept": ".odt",
+                "accept": ".odt,.zip",
             }
         ),
     )
@@ -1073,6 +1077,10 @@ def __init__(self, *args, **kwargs):
             Div("file", css_class="mb-4"),
         )
 
+    @property
+    def button_attrs(self):
+        return {"form": self.form_id}
+
 
 class TransactionUploadProcessForm(forms.Form):
     # Transaction types are loaded from centralized module
diff --git a/django/geno/templates/geno/odt2pdf.html b/django/geno/templates/geno/odt2pdf.html
@@ -10,16 +10,19 @@
     <div class="flex items-start gap-3">
       <span class="material-symbols-outlined text-base-400 dark:text-base-600 text-[20px] flex-shrink-0 mt-0.5">info</span>
       <div class="text-sm text-font-light dark:text-font-dark">
-        <p>{% trans "Laden Sie eine LibreOffice-Datei (.odt) hoch, um sie in ein PDF umzuwandeln." %}</p>
+        <p>{% trans "Lade eine LibreOffice-Datei (.odt) hoch, um sie in ein PDF umzuwandeln." %}</p>
+          <p class="mt-2">
+              {% trans "Die Datei wird auf dem Server verarbeitet und als PDF-Datei zum Download bereitgestellt." %}
+          </p>
         <p class="mt-2">
-          {% trans "Die Datei wird auf dem Server verarbeitet und als PDF-Datei zum Download bereitgestellt." %}
+          {% trans "Du kannst auch mehrere LibreOffice-Dateien in einer ZIP-Datei (.zip) zusammengefasst hochladen. Dann erhältst du wiederum eine ZIP-Datei mit den PDF-Dateien zurück." %}
         </p>
       </div>
     </div>
   </div>
 
   {# Form with custom layout #}
-  <form method="post" enctype="multipart/form-data" id="odt2pdf-form">
+  <form method="post" enctype="multipart/form-data" id="{{ form.form_id }}">
     {% csrf_token %}
 
     {% component "unfold/components/card.html" %}
@@ -35,7 +38,7 @@
     <div class="backdrop-blur-xs bg-white/80 rounded-b-default pb-4 px-4 dark:bg-base-900/80 lg:border-t lg:border-base-200 relative lg:scrollable-top lg:py-0 dark:border-base-800">
       <div class="flex flex-col-reverse gap-3 items-center mx-auto lg:flex-row-reverse container lg:h-[64px]">
         {# Primary action - Submit form using form attribute #}
-        {% component "unfold/components/button.html" with submit=1 attrs="form='odt2pdf-form'" %}
+        {% component "unfold/components/button.html" with submit=1 attrs=form.button_attrs %}
         {% trans "In PDF umwandeln" %}
       {% endcomponent %}
     </div>
diff --git a/django/geno/tests/test_views.py b/django/geno/tests/test_views.py
@@ -1,8 +1,12 @@
 import datetime
+import io
+import zipfile
 from unittest.mock import DEFAULT, patch
 
 from django.apps import apps as django_apps
 from django.conf import settings
+from django.http import FileResponse
+from django.test import tag
 from django.urls import reverse
 
 import geno.tests.data as geno_testdata
@@ -223,3 +227,31 @@ def test_invoice_manual_process_creation_error(self, mock_create_qrbill):
             raw=True,
         )
         self.assertEqual(Invoice.objects.count(), 0)
+
+
+class Odt2PdfViewTest(GenoAdminTestCase):
+    @tag("slow-test")
+    def test_odt2pdf_view_singlefile(self):
+        self.client.login(username="superuser", password="secret")
+        with open("geno/tests/template_test.odt", "rb") as dummy_file:
+            response = self.client.post(reverse("geno:odt2pdf"), {"file": dummy_file})
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(isinstance(response, FileResponse))
+        self.assertEqual(response.headers["Content-Type"], "application/pdf")
+        self.assertInPDF(response.getvalue(), "Template-Test")
+
+    @tag("slow-test")
+    def test_odt2pdf_view_multifile(self):
+        self.client.login(username="superuser", password="secret")
+        with open("/tmp/test.zip", "wb") as zipfile_object:
+            with zipfile.ZipFile(zipfile_object, "w") as archive:
+                archive.write("geno/tests/template_test.odt", "testA.odt")
+                archive.write("geno/tests/template_test.odt", "testB.odt")
+        with open("/tmp/test.zip", "rb") as inputfile:
+            response = self.client.post(reverse("geno:odt2pdf"), {"file": inputfile})
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(isinstance(response, FileResponse))
+        self.assertEqual(response.headers["Content-Type"], "application/zip")
+        with zipfile.ZipFile(io.BytesIO(response.getvalue())) as zip_file:
+            self.assertInPDF(zip_file.read("testA.pdf"), "Template-Test")
+            self.assertInPDF(zip_file.read("testB.pdf"), "Template-Test")
diff --git a/django/geno/utils.py b/django/geno/utils.py
@@ -51,9 +51,11 @@ def odt2pdf(odtfile, instance_tag="default"):
             p = Popen(cmd, stdout=PIPE, stderr=PIPE)
             output, err = p.communicate()
             if p.returncode or len(err):
-                raise Exception("odt2pdf failed: %s - %s (2. attempt)" % (output, err))
+                raise Exception(
+                    "odt2pdf for %s failed: %s - %s (2. attempt)" % (odtfile, output, err)
+                )
         else:
-            raise Exception("odt2pdf failed: %s - %s (1. attempt)" % (output, err))
+            raise Exception("odt2pdf for %s failed: %s - %s (1. attempt)" % (odtfile, output, err))
 
     pdf_file = "%s/%s.pdf" % (path, outfile[0])
     if not os.path.isfile(pdf_file):
diff --git a/django/geno/views.py b/django/geno/views.py
diff --git a/scripts/before-pr.sh b/scripts/before-pr.sh
diff --git a/scripts/cleanup-on-push-to-main.sh b/scripts/cleanup-on-push-to-main.sh
