✨ Media exporter: filter with a query

3 files changed, 98 insertions, 40 deletions

diff --git a/archaeological_operations/tests.py b/archaeological_operations/tests.py
index b8db7e2c7..385337213 100644
--- a/archaeological_operations/tests.py
+++ b/archaeological_operations/tests.py
@@ -91,7 +91,8 @@ from ishtar_common.models import (
     Document,
     ValueFormater,
     Regexp,
-    MediaExporter
+    MediaExporter,
+    SourceType
 )
 from ishtar_common.models_rest import ApiUser, ApiSearchModel, ApiSheetFilter
 from archaeological_files.models import File, FileType
@@ -4101,7 +4102,8 @@ class OperationExportMediaTest(TestCase, TestPermissionQuery, OperationInitTest)
         response = c.get(self.export_url)
         self.assertEqual(response.status_code, 200)
 
-    def _test_files(self, response, filenames):
+    def _test_files(self, response, filenames, exclude=None):
+        exclude = exclude or []
         with tempfile.TemporaryDirectory() as tmpdir:
             f, z = None, None
             try:
@@ -4117,6 +4119,8 @@ class OperationExportMediaTest(TestCase, TestPermissionQuery, OperationInitTest)
                     image = f"{tmpdir}/{filename}"
                     with Image.open(image) as im:
                         self.assertIsNone(im.verify())
+                for excluded in exclude:
+                    self.assertNotIn(exclude, name_list)
             finally:
                 if z:
                     z.close()
@@ -4124,27 +4128,37 @@ class OperationExportMediaTest(TestCase, TestPermissionQuery, OperationInitTest)
                     f.close()
 
     def test_content(self):
+        document = self.documents[0]
+        document.pk = None
+        document.save()
+        self.operation.documents.add(document)
+
         c = Client()
         c.login(username=self.username, password=self.password)
 
         response = c.get(self.export_url)
-        self._test_files(response, [('operation_image_00001.png', True)])
+        self._test_files(response, [
+            ('operation_image_00001.png', True),
+            ('operation_image_00002.png', True),
+        ])
 
         self.exporter.files_to_export = 'A'
         self.exporter.save()
         response = c.get(self.export_url)
-        self._test_files(
-            response,
-            [('operation_image_00001.png', True), ('operation_file_00001.txt', False), ]
-        )
+        self._test_files(response, [
+            ('operation_image_00001.png', True),
+            ('operation_image_00002.png', True),
+            ('operation_file_00001.txt', False),
+        ])
 
         self.exporter.thumbnail_for_images = True
         self.exporter.save()
         response = c.get(self.export_url)
-        self._test_files(
-            response,
-            [('operation_image_00001.jpg', True), ('operation_file_00001.txt', False), ]
-        )
+        self._test_files(response, [
+            ('operation_image_00001.jpg', True),
+            ('operation_image_00002.jpg', True),
+            ('operation_file_00001.txt', False),
+        ])
 
     def _add_find(self):
         cr_data = {
@@ -4189,6 +4203,35 @@ class OperationExportMediaTest(TestCase, TestPermissionQuery, OperationInitTest)
              ('find_image_00001.png', True),]
         )
 
+    def test_query(self):
+        document_type = SourceType.objects.all()[0]
+        document = self.documents[0]
+        document.pk = None
+        document.source_type = document_type
+        document.save()
+        self.operation.documents.add(document)
+
+        c = Client()
+        c.login(username=self.username, password=self.password)
+
+        # classic export
+        response = c.get(self.export_url)
+        self._test_files(
+            response,
+            [('operation_image_00001.png', True),
+             ('operation_image_00002.png', True)]
+        )
+
+        # filter: only one document is relevant
+        self.exporter.query = f"type={document_type.label}"
+        self.exporter.save()
+        response = c.get(self.export_url)
+        self._test_files(
+            response,
+            [('operation_image_00001.png', True)],
+            exclude=[('operation_image_00002.png', True)]
+        )
+
 
 class LabelTest(TestCase, OperationInitTest):
     fixtures = FILE_FIXTURES
diff --git a/ishtar_common/models_imports.py b/ishtar_common/models_imports.py
index 0157b33ee..23fb4f38e 100644
--- a/ishtar_common/models_imports.py
+++ b/ishtar_common/models_imports.py
@@ -100,6 +100,7 @@ from ishtar_common.data_importer import (
     LowerCaseFormater,
 )
 from ishtar_common.utils import task
+from ishtar_common.views_item import get_item
 from ishtar_common.ignf_utils import IGNF
 
 
@@ -3052,8 +3053,8 @@ class MediaExporter(models.Model):
     query = models.TextField(
         _("Filter query"), blank=True, null=False, default="",
         help_text=_(
-            "Use 'text' query used in Ishtar search input. Can be left empty "
-            "to export all."
+            "Use 'text' query used in Ishtar search input. The query is from the "
+            "document point of view. Can be left empty to export all attached media."
         )
     )
     naming = models.TextField(
@@ -3128,13 +3129,7 @@ class MediaExporter(models.Model):
                 down += list(getattr(o, attr).all())
         return down
 
-    def export(self, obj, tmpdir=None):
-        if not tmpdir:
-            tmpdir = tempfile.mkdtemp()
-        if not hasattr(obj, "documents"):
-            # database inconstency - should not occur
-            return
-        q = None
+    def _get_query(self, ishtaruser):
         media_attrs = []
         if self.files_to_export in ("A", "I"):
             q = Q(image__isnull=False)
@@ -3149,24 +3144,18 @@ class MediaExporter(models.Model):
             else:
                 q = q2
             media_attrs.append("associated_file")
-        archive_path = os.path.join(tmpdir, "archive")
-        os.mkdir(archive_path)
-        items = [obj]
-        if self.cascade:
-            down = self._get_down_objects([obj])
-            items += down
-            while down:
-                down = self._get_down_objects(down)
-                items += down
-        counters = {}
-        for item in items:
-            if not hasattr(item, "documents"):
-                continue
-            self._copy_media(item, q, media_attrs, counters, archive_path)
-        now = datetime.datetime.now()
-        archive_name = os.path.join(tmpdir, f"media-{now.strftime('%Y-%m-%d-%H%M%S')}")
-        shutil.make_archive(archive_name, "zip", archive_path)
-        return archive_name + ".zip"
+        # if query is empty do also the query in order to filter with permissions
+        Document = apps.get_model("ishtar_common", "Document")
+        _get_item = get_item(
+            Document,
+            "", "", no_permission_check=True,
+        )
+        query = {"search_vector": self.query or ""}
+        ids = _get_item(
+            None, return_query=True, ishtaruser=ishtaruser, query=query
+        ).values_list("id", flat=True)
+        q = q & Q(pk__in=ids)
+        return q, media_attrs
 
     def _copy_media(self, item, q, media_attrs, counters, archive_path):
         item_type = item.SLUG
@@ -3189,5 +3178,31 @@ class MediaExporter(models.Model):
                     if key not in counters:
                         counters[key] = 0
                     counters[key] += 1
-                    name += f"_{idx + counters[key]:05d}.{ext}"
+                    name += f"_{counters[key]:05d}.{ext}"
                 shutil.copy(media.path, os.path.join(archive_path, name))
+
+    def export(self, obj, ishtaruser, tmpdir=None):
+        if not tmpdir:
+            tmpdir = tempfile.mkdtemp()
+        if not hasattr(obj, "documents"):
+            # database inconstency - should not occur
+            return
+        q, media_attrs = self._get_query(ishtaruser)
+        archive_path = os.path.join(tmpdir, "archive")
+        os.mkdir(archive_path)
+        items = [obj]
+        if self.cascade:
+            down = self._get_down_objects([obj])
+            items += down
+            while down:
+                down = self._get_down_objects(down)
+                items += down
+        counters = {}
+        for item in items:
+            if not hasattr(item, "documents"):
+                continue
+            self._copy_media(item, q, media_attrs, counters, archive_path)
+        now = datetime.datetime.now()
+        archive_name = os.path.join(tmpdir, f"media-{now.strftime('%Y-%m-%d-%H%M%S')}")
+        shutil.make_archive(archive_name, "zip", archive_path)
+        return archive_name + ".zip"
diff --git a/ishtar_common/views.py b/ishtar_common/views.py
index 95cc37990..aa47040aa 100644
--- a/ishtar_common/views.py
+++ b/ishtar_common/views.py
@@ -1674,7 +1674,7 @@ class ExportMediaView(IshtarMixin, LoginRequiredMixin, View):
         if not obj:
             return HttpResponse(content_type="text/plain")
         with tempfile.TemporaryDirectory() as tmpdir:
-            export = exporter.export(obj, tmpdir=tmpdir)
+            export = exporter.export(obj, request.user.ishtaruser, tmpdir=tmpdir)
             if not export:
                 return HttpResponse(content_type="text/plain")
             with open(export, "rb") as f: