[16.0][ADD] fs_image: New field to store your images into external filesystems

fs_attachment/fs_stream.py

@@ -21,18 +21,13 @@ def from_fs_attachment(cls, attachment: IrAttachment) -> FsStream:
         attachment.ensure_one()
         if not attachment.fs_filename:
             raise ValueError("Attachment is not stored into a filesystem storage")
-        size = 0
-        if cls._check_use_x_sendfile(attachment):
-            fs, _storage, fname = attachment._get_fs_parts()
-            fs_info = fs.info(fname)
-            size = fs_info["size"]
         return cls(
             mimetype=attachment.mimetype,
             download_name=attachment.name,
             conditional=True,
             etag=attachment.checksum,
             type="fs",
-            size=size,
+            size=attachment.file_size,
             last_modified=attachment["__last_update"],
             fs_attachment=attachment,
         )

fs_attachment/models/ir_attachment.py

@@ -98,9 +98,14 @@ class IrAttachment(models.Model):
     def _compute_internal_url(self) -> None:
         for rec in self:
             filename, extension = os.path.splitext(rec.name)
+            # determine if the file is an image
+            pfx = "/web/content"
+            if rec.mimetype and rec.mimetype.startswith("image/"):
+                pfx = "/web/image"
+
             if not extension:
                 extension = mimetypes.guess_extension(rec.mimetype)
-            rec.internal_url = f"/web/content/{rec.id}/{filename}{extension}"
+            rec.internal_url = f"{pfx}/{rec.id}/{filename}{extension}"
 
     @api.depends("fs_filename")
     def _compute_fs_url(self) -> None:
@@ -441,8 +446,6 @@ def _enforce_meaningful_storage_filename(self) -> None:
             if self.env["fs.storage"]._must_use_filename_obfuscation(storage):
                 attachment.fs_filename = filename
                 continue
-            if self._is_fs_filename_meaningful(filename):
-                continue
             new_filename = attachment._build_fs_filename()
             # we must keep the same full path as the original filename
             new_filename_with_path = os.path.join(
@@ -491,17 +494,6 @@ def _fs_parse_store_fname(
         fname = partition[2]
         return fs, storage_code, fname
 
-    @api.model
-    def _is_fs_filename_meaningful(self, filename: str) -> bool:
-        """Return True if the filename is meaningful
-        A filename is meaningful if it's formatted as
-        """
-        parsed = self._parse_fs_filename(filename)
-        if not parsed:
-            return False
-        name, res_id, version, extension = parsed
-        return bool(name and res_id and version is not None and extension)
-
     @api.model
     def _parse_fs_filename(self, filename: str) -> tuple[str, int, int, str] | None:
         """Parse the filename and return the name, id, version and extension

fs_attachment/models/ir_binary.py

@@ -2,7 +2,11 @@
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
 import logging
 
+import werkzeug.http
+
 from odoo import models
+from odoo.http import request
+from odoo.tools.image import image_process
 
 from ..fs_stream import FsStream
 
@@ -39,3 +43,69 @@
         if fs_attachment:
             return FsStream.from_fs_attachment(fs_attachment)
         return super()._record_to_stream(record, field_name)
+
+    def _get_image_stream_from(
+        self,
+        record,
+        field_name="raw",
+        filename=None,
+        filename_field="name",
+        mimetype=None,
+        default_mimetype="image/png",
+        placeholder=None,
+        width=0,
+        height=0,
+        crop=False,
+        quality=0,
+    ):
+        # we need to override this method since if you pass a width or height or
+        # set crop=True, the stream data must be a bytes object, not a
+        # file-like object. In the base implementation, the stream data is
+        # passed to `image_process` method to transform it and this method
+        # expects a bytes object.
+        initial_width = width
+        initial_height = height
+        initial_crop = crop
+        if record._name != "ir.attachment" and field_name:
+            field_def = record._fields[field_name]
+            if field_def.type in ("fs_image", "fs_file"):
+                value = record[field_name]
+                if value:
+                    record = value.attachment
+                    field_name = "raw"
+        stream = super()._get_image_stream_from(
+            record,
+            field_name=field_name,
+            filename=filename,
+            filename_field=filename_field,
+            mimetype=mimetype,
+            default_mimetype=default_mimetype,
+            placeholder=placeholder,
+            width=0,
+            height=0,
+            crop=False,
+            quality=quality,
+        )
+        modified = werkzeug.http.is_resource_modified(
+            request.httprequest.environ,
+            etag=stream.etag,
+            last_modified=stream.last_modified,
+        )
+        if modified and (initial_width or initial_height or initial_crop):
+            if stream.type == "path":
+                with open(stream.path, "rb") as file:
+                    stream.type = "data"
+                    stream.path = None
+                    stream.data = file.read()
+            elif stream.type == "fs":
+                stream.data = stream.read()
+                stream.type = "data"
+            stream.data = image_process(
+                stream.data,
+                size=(initial_width, initial_height),
+                crop=initial_crop,
+                quality=quality,
+            )
+            stream.size = len(stream.data)
+
+        return stream

fs_attachment/tests/__init__.py

@@ -2,3 +2,4 @@
 from . import test_fs_attachment_file_like_adapter
 from . import test_fs_attachment_internal_url
 from . import test_fs_storage
+from . import test_stream

fs_attachment/tests/test_stream.py

@@ -0,0 +1,138 @@
+# Copyright 2023 ACSONE SA/NV (http://acsone.eu).
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
+import io
+import os
+import shutil
+import tempfile
+
+from PIL import Image
+
+from odoo.tests.common import HttpCase
+
+
+class TestStream(HttpCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.env = cls.env(context=dict(cls.env.context, tracking_disable=True))
+        temp_dir = tempfile.mkdtemp()
+        cls.temp_backend = cls.env["fs.storage"].create(
+            {
+                "name": "Temp FS Storage",
+                "protocol": "file",
+                "code": "tmp_dir",
+                "directory_path": temp_dir,
+                "base_url": "http://my.public.files/",
+            }
+        )
+        cls.temp_dir = temp_dir
+        cls.content = b"This is a test attachment"
+        cls.attachment_binary = (
+            cls.env["ir.attachment"]
+            .with_context(
+                storage_location=cls.temp_backend.code,
+                storage_file_path="test.txt",
+            )
+            .create({"name": "test.txt", "raw": cls.content})
+        )
+
+        cls.image = cls._create_image(128, 128)
+        cls.attachment_image = (
+            cls.env["ir.attachment"]
+            .with_context(
+                storage_location=cls.temp_backend.code,
+                storage_file_path="test.png",
+            )
+            .create({"name": "test.png", "raw": cls.image})
+        )
+
+        @cls.addClassCleanup
+        def cleanup_tempdir():
+            shutil.rmtree(temp_dir)
+
+        assert cls.attachment_binary.fs_filename
+        assert cls.attachment_image.fs_filename
+
+    def setUp(self):
+        super().setUp()
+        # enforce temp_backend field since it seems that they are reset on
+        # savepoint rollback when managed by server_environment -> TO Be investigated
+        self.temp_backend.write(
+            {
+                "protocol": "file",
+                "code": "tmp_dir",
+                "directory_path": self.temp_dir,
+                "base_url": "http://my.public.files/",
+            }
+        )
+
+    @classmethod
+    def tearDownClass(cls):
+        super().tearDownClass()
+        for f in os.listdir(cls.temp_dir):
+            os.remove(os.path.join(cls.temp_dir, f))
+
+    @classmethod
+    def _create_image(cls, width, height, color="#4169E1", img_format="PNG"):
+        f = io.BytesIO()
+        Image.new("RGB", (width, height), color).save(f, img_format)
+        f.seek(0)
+        return f.read()
+
+    def assertDownload(
+        self, url, headers, assert_status_code, assert_headers, assert_content=None
+    ):
+        res = self.url_open(url, headers=headers)
+        res.raise_for_status()
+        self.assertEqual(res.status_code, assert_status_code)
+        for header_name, header_value in assert_headers.items():
+            self.assertEqual(
+                res.headers.get(header_name),
+                header_value,
+                f"Wrong value for header {header_name}",
+            )
+        if assert_content:
+            self.assertEqual(res.content, assert_content, "Wong content")
+        return res
+
+    def test_content_url(self):
+        self.authenticate("admin", "admin")
+        url = f"/web/content/{self.attachment_binary.id}"
+        self.assertDownload(
+            url,
+            headers={},
+            assert_status_code=200,
+            assert_headers={
+                "Content-Type": "text/plain; charset=utf-8",
+                "Content-Disposition": "inline; filename=test.txt",
+            },
+            assert_content=self.content,
+        )
+
+    def test_image_url(self):
+        self.authenticate("admin", "admin")
+        url = f"/web/image/{self.attachment_image.id}"
+        self.assertDownload(
+            url,
+            headers={},
+            assert_status_code=200,
+            assert_headers={
+                "Content-Type": "image/png",
+                "Content-Disposition": "inline; filename=test.png",
+            },
+            assert_content=self.image,
+        )
+
+    def test_image_url_with_size(self):
+        self.authenticate("admin", "admin")
+        url = f"/web/image/{self.attachment_image.id}?width=64&height=64"
+        res = self.assertDownload(
+            url,
+            headers={},
+            assert_status_code=200,
+            assert_headers={
+                "Content-Type": "image/png",
+                "Content-Disposition": "inline; filename=test.png",
+            },
+        )
+        self.assertEqual(Image.open(io.BytesIO(res.content)).size, (64, 64))