[IMP] document_quick_access_folder_auto_classification: Add Python-Op…

…enCV as backup check

document_quick_access_folder_auto_classification/__manifest__.py

@@ -11,7 +11,7 @@
     "website": "https://github.com/OCA/server-ux",
     "depends": ["document_quick_access", "edi_storage_oca"],
     "external_dependencies": {
-        "deb": ["libzbar0", "poppler-utils"],
+        "deb": ["libzbar0", "poppler-utils", "libgl1"],
         "python": ["pyzbar", "pdf2image"],
     },
     "data": [

document_quick_access_folder_auto_classification/components/document_quick_access_process.py

@@ -27,6 +27,11 @@
     )
 except (ImportError, IOError) as err:
     _logger.warning(err)
+try:
+    import cv2
+    import numpy
+except (ImportError, IOError):
+    cv2 = False
 
 
 class EdiDocumentQuickAccessProcess(Component):
@@ -75,12 +80,20 @@ def _search_document_pdf(self, datas):
         return records
 
     def _search_pil_image(self, image):
-        results = decode(image, symbols=[ZBarSymbol.QRCODE])
+        zbar_results = decode(image, symbols=[ZBarSymbol.QRCODE])
+        results = [zbar_result.data.decode("utf-8") for zbar_result in zbar_results]
+        if not results and cv2:
+            detector = cv2.QRCodeDetector()
+            is_ok, cv2_results, _vertices, _binary = detector.detectAndDecodeMulti(
+                numpy.asarray(image)
+            )
+            if is_ok:
+                results = cv2_results
         records = []
         rule_obj = self.env["document.quick.access.rule"]
         for result in results:
             record = rule_obj.with_context(no_raise_document_access=True).read_code(
-                result.data.decode("utf-8")
+                result
             )
             if record and record not in records:
                 records += record

document_quick_access_folder_auto_classification/tests/test_document_quick_access_auto_classification.py

@@ -5,6 +5,7 @@
 import shutil
 import uuid
 
+import cv2
 from mock import patch
 
 from odoo import tools
@@ -228,6 +229,48 @@ def test_no_ok_reject(self):
         self.assertEqual(missing.edi_exchange_state, "input_processed")
         self.assertFalse(missing.model)
 
+    def test_no_ok_cv2_ok_multi(self):
+        partners = self.env["res.partner"].create({"name": "Partner 1"})
+        partners |= self.env["res.partner"].create({"name": "Partner 2"})
+        partners |= self.env["res.partner"].create({"name": "Partner 3"})
+        partners |= self.env["res.partner"].create({"name": "Partner 4"})
+        self.test_no_ok_cv2_ok(partners)
+
+    @mute_logger("odoo.addons.queue_job.models.base")
+    def test_no_ok_cv2_ok(self, partners=False):
+        if not partners:
+            partners = self.env["res.partner"].create({"name": "Partner"})
+        file = tools.file_open(
+            "test_file.pdf",
+            mode="rb",
+            subdir="addons/document_quick_access_folder_auto_classification/tests",
+        ).read()
+        self.env["document.quick.access.rule"].create(
+            {
+                "model_id": self.model_id.id,
+                "name": "PARTNER",
+                "priority": 1,
+                "barcode_format": "standard",
+            }
+        )
+        with open(os.path.join(self.tmpdir, "test_file.pdf"), "wb") as f:
+            f.write(file)
+        code = [partner.get_quick_access_code() for partner in partners]
+        with patch.object(cv2.QRCodeDetector, "detectAndDecodeMulti") as ptch:
+            ptch.return_value = [True, code, [], []]
+            self.assertFalse(self.exchange_model.search([]))
+            self.backend._storage_cron_check_pending_input()
+            self.assertEqual(self.exchange_model.search_count([]), 1)
+            self.backend._cron_check_input_exchange_sync()
+            self.assertEqual(ptch.call_count, 1)
+        self.assertTrue(partners)
+        for partner in partners:
+            self.assertTrue(
+                self.env["ir.attachment"].search(
+                    [("res_model", "=", partner._name), ("res_id", "=", partner.id)]
+                )
+            )
+
     @mute_logger("odoo.addons.queue_job.models.base")
     def test_corrupted(self):
         """We will check that corrupted files are stored also"""

test-requirements.txt

@@ -1 +1,3 @@
 odoo_test_helper
+opencv-python
+numpy