Download zip separated folders

conftest.py

@@ -80,3 +80,14 @@ def _handle_page_goto(
     if not (url.startswith("http://") or url.startswith("https://")):
         url = base_url + url
     return page._goto(url, *args, **kwargs)  # type: ignore
+
+
+@pytest.fixture
+def login(page):
+    page.goto("")
+    page.click("text=Log In")
+    page.click(":nth-match(:text(\"Log In\"), 2)")
+    page.fill("input[name=\"email\"]", "example@example.com")
+    page.press("input[name=\"email\"]", "Tab")
+    page.fill("input[name=\"password\"]", "openlex1234")
+    page.click("input:has-text(\"Log In\")")

controllers/expedientes.py

@@ -4,6 +4,9 @@
 
 import zipfile
 import io
+import tempfile
+from collections import OrderedDict
+from xhtml2pdf import pisa
 LINKED_TABLES = ['movimiento', 'agenda', 'parte']
 ZIP_FILENAME = 'Movimiento.zip'
 CHUNK_SIZE = 4096
@@ -69,26 +72,57 @@ def index():
 
 @auth.requires_login()
 def download():
-    tempfile = io.BytesIO()
-    temparchive = zipfile.ZipFile(tempfile, 'w', zipfile.ZIP_DEFLATED)
-    rows = db(db.movimiento.archivo != None).select()
-    try:
-        for file_id in rows:
-            file_single = file_id.archivo
-            if file_single:
-                file_loc = db.movimiento.archivo.retrieve_file_properties(file_single)['path']+ '/' + file_single
-                file_name = db.movimiento.archivo.retrieve_file_properties(file_single)['filename']
-                temparchive.write(file_loc, file_name)
-    finally:
-        temparchive.close()
-        tempfile.seek(0)
+    tempfiles = io.BytesIO()
+    temparchive = zipfile.ZipFile(tempfiles, 'w', zipfile.ZIP_DEFLATED)
+    temp_dir = tempfile.TemporaryDirectory()
+    # Obtener ID de expediente y guardarlo en una lista.
+    movimiento_full_row = db(db.movimiento).select()
+    expediente_list = [numero.expediente_id for numero in movimiento_full_row]
+    expediente_lists = OrderedDict.fromkeys(expediente_list).keys()
 
+    # Separar archivos por el numero de expediente.
+    for expedient_single_id in expediente_lists:
+        movimiento_filter_row = db(db.movimiento.expediente_id == expedient_single_id).select(
+                                    db.movimiento.archivo,
+                                    db.movimiento.texto,
+                                    db.movimiento.titulo)
+        expedient_number_row = db(db.expediente.id == expedient_single_id).select(db.expediente.numero)
+        expedient_name = [numero.numero for numero in expedient_number_row]
+        try:
+            for file_id in movimiento_filter_row:
+                file_single = file_id.archivo
+                if file_single:
+                    file_loc = db.movimiento.archivo.retrieve_file_properties(file_single)['path'] + '/' + file_single
+                    file_name = db.movimiento.archivo.retrieve_file_properties(file_single)['filename']
+                    temparchive.write(file_loc, "upload/" + str(expedient_name[0]) + "/" + file_name)
+        except Exception as err:
+            response.flash = DIV(T('Fallo en la descarga de archivos'), PRE(str(err)))
+        finally:
+            for text_id in movimiento_filter_row:
+                text_single_text = text_id.texto
+                text_single_title = text_id.titulo
+                status, result_file = convert_html_to_pdf(text_single_text, temp_dir.name + "/" + text_single_title + ".pdf")
+                temparchive.write(result_file.name, "upload/" + str(expedient_name[0]) + "/" + text_single_title + ".pdf")
+    del temp_dir
+    temparchive.close()
+    tempfiles.seek(0)
     response.headers['Content-Type'] = 'application/zip'
     response.headers['Content-Disposition'] = 'attachment; filename = %s' % ZIP_FILENAME
-    res = response.stream(tempfile, CHUNK_SIZE)
+    res = response.stream(tempfiles, CHUNK_SIZE)
     return res
 
 
+def convert_html_to_pdf(source_html, output_filename):
+    # open output file for writing (truncated binary)
+    with open(output_filename, "w+b") as result_file:
+        # convert HTML to PDF
+        pisa_status = pisa.CreatePDF(
+                source_html,  # the HTML to convert
+                dest=result_file)  # file handle to recieve result
+        # return False on success and True on errors
+    return pisa_status, result_file
+
+
 def vista_expediente():
     'muestra un panel a la izquierda que tiene los datos del expediente y permite navegar en él'
     expte = SQLFORM(db.expediente,
@@ -112,9 +146,9 @@ def vista_expediente():
         text = SPAN(k.capitalize() + 's', _class='buttontext button')
         links.append(A(text, _href=url, _type='button',
                        _class='btn btn-default'))
-    url = URL('download', args='movimiento.archivo') #Boton de descarga
-    text1="Descarga"
-    links.append(A(text1, _href=url, _type='button',
+    url = URL('download', args='movimiento.archivo')  # Boton de descarga
+    text_download = "Descarga"
+    links.append(A(text_download, _href=url, _type='button',
                  _class='btn btn-default'))
 
     return dict(links=links, expte=expte)

requirements.txt

@@ -1,4 +1,5 @@
 pytest==6.2.4
 playwright==1.11.2
 pytest-playwright==0.1.1
-pytest-html==3.1.1
+pytest-html==3.1.1
+xhtml2pdf==0.2.5

tests/test_b_expediente_updated.py

@@ -1,19 +1,9 @@
-def test_upload_expedientes(page):
+def test_upload_expedientes(page, login):
     pattern = "movimiento.expediente_id"
-    page.goto("http://127.0.0.1:8020/OpenLex/")
-    login(page)
     page.click("css=[alt=Expedientes]")
     assert page.url.endswith("/expedientes/index")
     page.click("text=Agregar")
     page.fill("input[name=\"numero\"]", "1111")
     page.press("input[name=\"numero\"]", "Tab")
     page.fill("input[name=\"caratula\"]", "ssdd")
     page.click("input:has-text(\"Enviar\")")
-
-def login(page):
-    page.click("text=Log In")
-    page.click(":nth-match(:text(\"Log In\"), 2)")
-    page.fill("input[name=\"email\"]", "example@example.com")
-    page.press("input[name=\"email\"]", "Tab")
-    page.fill("input[name=\"password\"]", "openlex1234")
-    page.click("input:has-text(\"Log In\")")

tests/test_download.py

@@ -0,0 +1,15 @@
+import pytest
+
+@pytest.mark.skip(reason="Este test funciona en local pero falla en github, no toma el evento download")   
+def test_download(page, login):
+    page.goto("expedientes/index")
+    page.click("text=Movimientos")
+    with page.expect_download() as download_info:
+        # Perform the action that initiates download
+        page.click("text=Descarga")
+    download = download_info.value
+    # Wait for the download process to complete
+    name = download.suggested_filename
+    assert "Movimiento.zip" == name
+
+