[IMP] edi_oca: add encoding management for exchange type

Author: JordiMForgeFlow

edi_oca/models/edi_backend.py

@@ -214,9 +214,13 @@ def exchange_generate(self, exchange_record, store=True, force=False, **kw):
         self._check_exchange_generate(exchange_record, force=force)
         output = self._exchange_generate(exchange_record, **kw)
         message = None
+        encoding = exchange_record.type_id.encoding or "UTF-8"
+        encoding_error_handler = (
+            exchange_record.type_id.encoding_out_error_handler or "strict"
+        )
         if output and store:
             if not isinstance(output, bytes):
-                output = output.encode()
+                output = output.encode(encoding, errors=encoding_error_handler)
             exchange_record.update(
                 {
                     "exchange_file": base64.b64encode(output),

edi_oca/models/edi_exchange_record.py

@@ -225,11 +225,17 @@ def _get_file_content(
     ):
         """Handy method to not have to convert b64 back and forth."""
         self.ensure_one()
+        encoding = self.type_id.encoding or "UTF-8"
+        decoding_error_handler = self.type_id.encoding_in_error_handler or "strict"
         if not self[field_name]:
             return ""
         if binary:
             res = base64.b64decode(self[field_name])
-            return res.decode() if not as_bytes else res
+            return (
+                res.decode(encoding, errors=decoding_error_handler)
+                if not as_bytes
+                else res
+            )
         return self[field_name]
 
     def name_get(self):

edi_oca/models/edi_exchange_type.py

@@ -152,6 +152,36 @@ class EDIExchangeType(models.Model):
             "Use it directly or within models rules (domain or snippet)."
         ),
     )
+    # https://docs.python.org/3/library/codecs.html#standard-encodings
+    encoding = fields.Char(
+        help="Encoding to be applied to generate/process the exchanged file.\n"
+        "Example: UTF-8, Windows-1252, ASCII...(default is always 'UTF-8')",
+    )
+    # https://docs.python.org/3/library/codecs.html#codec-base-classes
+    encoding_out_error_handler = fields.Selection(
+        string="Encoding Error Handler",
+        selection=[
+            ("strict", "Raise Error"),
+            ("ignore", "Ignore"),
+            ("replace", "Replace with Replacement Marker"),
+            ("backslashreplace", "Replace with Backslashed Escape Sequences"),
+            ("surrogateescape", "Replace Byte with Individual Surrogate Code"),
+            ("xmlcharrefreplace", "Replace with XML/HTML Numeric Character Reference"),
+        ],
+        help="Handling of encoding errors on generate (default is always 'Raise Error').",
+    )
+    # https://docs.python.org/3/library/codecs.html#codec-base-classes
+    encoding_in_error_handler = fields.Selection(
+        string="Decoding Error Handler",
+        selection=[
+            ("strict", "Raise Error"),
+            ("ignore", "Ignore"),
+            ("replace", "Replace with Replacement Marker"),
+            ("backslashreplace", "Replace with Backslashed Escape Sequences"),
+            ("surrogateescape", "Replace Byte with Individual Surrogate Code"),
+        ],
+        help="Handling of decoding errors on process (default is always 'Raise Error').",
+    )
 
     _sql_constraints = [
         (

edi_oca/tests/__init__.py

@@ -13,3 +13,4 @@
 from . import test_security
 from . import test_quick_exec
 from . import test_exchange_type_deprecated_fields
+from . import test_exchange_type_encoding

edi_oca/tests/test_exchange_type_encoding.py

@@ -0,0 +1,84 @@
+# Copyright 2024 ForgeFlow S.L. (https://www.forgeflow.com)
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+import base64
+
+import chardet
+
+from .common import EDIBackendCommonComponentRegistryTestCase
+from .fake_components import FakeOutputGenerator
+
+
+class EDIBackendTestOutputCase(EDIBackendCommonComponentRegistryTestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls._build_components(
+            cls,
+            FakeOutputGenerator,
+        )
+        vals = {
+            "model": cls.partner._name,
+            "res_id": cls.partner.id,
+        }
+        cls.record = cls.backend.create_record("test_csv_output", vals)
+
+    def setUp(self):
+        super().setUp()
+        FakeOutputGenerator.reset_faked()
+
+    def test_encoding_default(self):
+        """
+        Test default output/input encoding (UTF-8). Use string with special
+        character to test the encoding applied.
+        """
+        self.backend.with_context(fake_output="Palmotićeva").exchange_generate(
+            self.record
+        )
+        # Test decoding is applied correctly
+        self.assertEqual(self.record._get_file_content(), "Palmotićeva")
+        # Test encoding used
+        content = base64.b64decode(self.record.exchange_file)
+        encoding = chardet.detect(content)["encoding"].lower()
+        self.assertEqual(encoding, "utf-8")
+
+    def test_encoding(self):
+        """
+        Test specific output/input encoding. Use string with special
+        character to test the encoding applied.
+        """
+        self.exchange_type_out.write({"encoding": "UTF-16"})
+        self.backend.with_context(fake_output="Palmotićeva").exchange_generate(
+            self.record
+        )
+        # Test decoding is applied correctly
+        self.assertEqual(self.record._get_file_content(), "Palmotićeva")
+        # Test encoding used
+        content = base64.b64decode(self.record.exchange_file)
+        encoding = chardet.detect(content)["encoding"].lower()
+        self.assertEqual(encoding, "utf-16")
+
+    def test_encoding_error_handler(self):
+        self.exchange_type_out.write({"encoding": "ascii"})
+        # By default, error handling raises error
+        with self.assertRaises(UnicodeEncodeError):
+            self.backend.with_context(fake_output="Palmotićeva").exchange_generate(
+                self.record
+            )
+        self.exchange_type_out.write({"encoding_out_error_handler": "ignore"})
+        self.backend.with_context(fake_output="Palmotićeva").exchange_generate(
+            self.record
+        )
+        self.assertEqual(self.record._get_file_content(), "Palmotieva")
+
+    def test_decoding_error_handler(self):
+        self.backend.with_context(fake_output="Palmotićeva").exchange_generate(
+            self.record
+        )
+        # Change encoding to ascii to check the decoding
+        self.exchange_type_out.write({"encoding": "ascii"})
+        # By default, error handling raises error
+        with self.assertRaises(UnicodeDecodeError):
+            content = self.record._get_file_content()
+        self.exchange_type_out.write({"encoding_in_error_handler": "ignore"})
+        content = self.record._get_file_content()
+        self.assertEqual(content, "Palmotieva")

edi_oca/views/edi_exchange_type_views.xml

@@ -41,6 +41,15 @@
                             <field name="partner_ids" widget="many2many_tags" />
                             <field name="job_channel_id" />
                             <field name="quick_exec" />
+                            <field name="encoding" />
+                            <field
+                                name="encoding_out_error_handler"
+                                attrs="{'invisible': [('direction', '=', 'input')]}"
+                            />
+                            <field
+                                name="encoding_in_error_handler"
+                                attrs="{'invisible': [('direction', '=', 'output')]}"
+                            />
                         </group>
                     </group>
                     <field name="deprecated_rule_fields_still_used" invisible="1" />