[ADD] survey_record_generation_extra_field save file in record binary field

survey_record_generation/__manifest__.py

@@ -11,7 +11,7 @@ Allow to create record of any model when sending the form :
 * Associate question with fields
 * For x2m fields : Associate values to questions
 """, 
-    "version": "16.0.1.0.3",
+    "version": "16.0.1.0.2",
     "license": "AGPL-3",
     "author": "Elabore",
     "website": "https://www.elabore.coop",

survey_record_generation/migrations/16.0.1.0.3/end-migration.py

@@ -1,153 +0,0 @@
-# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
-"""Relink survey.user_input.partner_id/email to the contact that actually
-matches each participant's own answers.
-
-Before this version, survey.user_input.partner_id/email were only updated
-when _mark_done() created a *new* res.partner. If the answer already had a
-partner_id/email (e.g. inherited from the Odoo user who was logged in when
-the /survey/start link was opened, or from some other unidentified cause),
-_mark_done() silently kept that value even when the survey_record_creation
-config found or created a different, correct contact from the participant's
-own answers. This backfill re-runs that resolution for every already-done
-answer and fixes partner_id/email accordingly.
-
-It never re-triggers the other survey.user_input._mark_done() overrides
-(crm lead / event registration / notification modules, ...): it only calls
-the res.partner-matching helpers directly, so it can't create duplicate
-leads, registrations, etc. for historical submissions.
-
-A res.partner is expected to already exist for every done submission (it
-was necessarily found or created the first time _mark_done() ran), so this
-backfill never creates one: if none is found, the record is left untouched
-and logged for manual review instead.
-"""
-import logging
-
-from odoo import SUPERUSER_ID, api
-
-_logger = logging.getLogger(__name__)
-
-
-def migrate(cr, version):
-    env = api.Environment(cr, SUPERUSER_ID, {})
-    user_input_model = env["survey.user_input"]
-
-    surveys_with_partner_creation = env["survey.record.creation"].search(
-        [("model_id.model", "=", "res.partner")]
-    ).survey_id
-
-    user_inputs = user_input_model.search(
-        [
-            ("survey_id", "in", surveys_with_partner_creation.ids),
-            ("state", "=", "done"),
-        ]
-    )
-
-    _logger.info(
-        "survey_record_generation: relinking partner_id/email on %d done "
-        "survey.user_input records",
-        len(user_inputs),
-    )
-
-    fixed_count = 0
-    not_found_count = 0
-
-    for user_input in user_inputs:
-        record_creations = user_input.survey_id.survey_record_creation_ids.filtered(
-            lambda rc: rc.model_id.model == "res.partner"
-        ).sorted("sequence")
-
-        record_creation = record_creations[:1]
-        if not record_creation:
-            continue
-
-        # 1) Prefer the res.partner this very submission created, if any
-        # (the first one, by id, in the rare case there is more than one):
-        # it's the exact record _mark_done() produced for this answer, no
-        # guessing involved. This is also the only option for surveys whose
-        # record creation has neither update_existing_records nor any
-        # unicity_check field configured, since find_existing_record()/
-        # find_duplicate...() can then never find anything (nothing to
-        # search on).
-        record = False
-        for generated in user_input.generated_record_ids.sorted("id"):
-            if (
-                generated.survey_record_creation_id == record_creation
-                and generated.created_record_id
-                and generated.created_record_id._name == "res.partner"
-                and generated.created_record_id.exists()
-            ):
-                # The referenced partner may have since been deleted (e.g.
-                # merged into another contact): in that case it's not usable
-                # and we fall through to the search-based lookup below.
-                record = generated.created_record_id
-                break
-
-        # 2) Otherwise, this submission matched an already-existing partner
-        # instead of creating one (find_existing_record()/find_duplicate...()
-        # branch of _mark_done()): re-derive it the same way.
-        if not record:
-            # Only compute the fields find_existing_record()/find_duplicate...()
-            # actually read (the search field, and any unicity_check field),
-            # not every field of the record creation: other fields (e.g. a
-            # "record" reference to a model defined in a module that depends
-            # on this one) may not be loadable yet at this point of the
-            # upgrade, and are useless here anyway since this backfill never
-            # writes to res.partner.
-            needed_field_names = set()
-            if (
-                record_creation.update_existing_records
-                and record_creation.field_to_retrieve_existing_records
-            ):
-                needed_field_names.add(
-                    record_creation.field_to_retrieve_existing_records.name
-                )
-            unicity_field_values = record_creation.field_values_ids.filtered(
-                lambda field_value: field_value.unicity_check
-            )
-            needed_field_names.update(unicity_field_values.mapped("field_id.name"))
-
-            vals = {}
-            for field_value in record_creation.field_values_ids:
-                if field_value.field_id.name not in needed_field_names:
-                    continue
-                value, __ = user_input_model.get_value_based_on_value_origin(
-                    field_value=field_value,
-                    user_input=user_input,
-                    created_records={},
-                    model="res.partner",
-                    other_record_fields_to_update=[],
-                )
-                vals[field_value.field_id.name] = value
-
-            existing_record = user_input_model.find_existing_record(
-                record_creation, vals
-            )
-            duplicate = (
-                user_input_model.find_duplicate_if_there_are_fields_with_unicity_check(
-                    "res.partner", record_creation, vals
-                )
-            )
-            record = duplicate or existing_record
-
-        if not record:
-            _logger.warning(
-                "survey_record_generation: no existing res.partner found for "
-                "survey.user_input %s while backfilling partner_id/email, "
-                "leaving it untouched",
-                user_input.id,
-            )
-            not_found_count += 1
-            continue
-
-        if user_input.partner_id != record or user_input.email != record.email:
-            user_input.partner_id = record.id
-            user_input.email = record.email
-            fixed_count += 1
-
-    _logger.info(
-        "survey_record_generation: fixed %d survey.user_input records "
-        "(%d without a matching res.partner)",
-        fixed_count,
-        not_found_count,
-    )

survey_record_generation/models/survey_user_input.py

@@ -39,7 +39,6 @@ class SurveyUserInput(models.Model):
         for user_input in self:
             created_records = {}
             other_record_fields_to_update: list[SurveyRecordCreationFieldValues] = []
-            partner_linked_in_this_run = False
 
             record_creation: SurveyRecordCreation
             for (
@@ -86,6 +85,8 @@ class SurveyUserInput(models.Model):
                     try:
                         with self.env.cr.savepoint():
                             record = self.env[model].create(vals)
+                            if model == "res.partner" and not self.partner_id:
+                                self.partner_id = record.id
                     except Exception:
                         # This a broad exception because it could be IntegrityError,
                         # EmptyNamesError in case partner_firstname is installed etc...
@@ -102,17 +103,6 @@ class SurveyUserInput(models.Model):
                         }
                     )
 
-                if model == "res.partner" and not partner_linked_in_this_run:
-                    # Always reflect the partner actually matched/created from
-                    # this participant's own answers, even if partner_id/email
-                    # were already set on the answer (e.g. inherited from the
-                    # logged-in user when the survey link was opened). Only the
-                    # first res.partner record creation of this run wins, in
-                    # case several are configured on the same survey.
-                    user_input.partner_id = record.id
-                    user_input.email = record.email
-                    partner_linked_in_this_run = True
-
                 created_records[record_creation.id] = record
 
             # update linked record
@@ -325,44 +315,28 @@ class SurveyUserInput(models.Model):
                 }
             )
 
-
-    def _get_boolean_true_values(self):
-        """Tokens interpreted as true, normalized to lowercase.
-
-        Instance method so that a third-party module or a localization can
-        extend the list without rewriting get_boolean_value.
-        """
-        return {"1", "true", "vrai", "yes", "y", "oui", "o", "on", "x"}
-
-    def _get_boolean_false_values(self):
-        """Tokens interpreted as false, normalized to lowercase."""
-        return {"0", "false", "faux", "no", "n", "non", "off", ""}
-
-    def get_boolean_value(self, answer_value_char: str, question_title: str) -> bool:
-        """Convert the technical value of an answer into a boolean.
-
-        Allows a boolean field to be filled from a "simple_choice" question
-        whose suggested answers carry yes/no values.
-        """
-        if not answer_value_char:
-            # Empty answer: an unset boolean is false, this is not a
-            # configuration error.
-            return False
-
-        token = str(answer_value_char).strip().casefold()
-
-        if token in self._get_boolean_true_values():
-            return True
-        if token in self._get_boolean_false_values():
-            return False
-
-        raise UserError(
-            _(
-                "[Survey record generation] The boolean value %(value)s is not "
-                "supported (for question %(question)s)"
+    @staticmethod
+    def get_boolean_value(answer_value_char: str, question_title: str) -> bool:
+        # Below code is a trick to be able to use "simple_choice" question
+        # with values 'yes' and 'no' and transform it to boolean.
+        if boolean_value := answer_value_char in [
+            "1",
+            "True",
+            "true",
+            "Oui",
+            "oui",
+            "Yes",
+            "yes",
+        ]:
+            return boolean_value
+        else:
+            raise UserError(
+                _(
+                    "[Survey record generation] The boolean value %s(value)s "
+                    "is not supported (for question %(question)s)."
+                )
+                % {
+                    "value": answer_value_char,
+                    "question": question_title,
+                }
             )
-            % {
-                "value": answer_value_char,
-                "question": question_title,
-            }
-        )

survey_record_generation_extra_fields/__init__.py

@@ -0,0 +1 @@
+from . import models

survey_record_generation_extra_fields/__manifest__.py

@@ -0,0 +1,20 @@
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+
+{
+    "name": "Survey record generation extra fields",
+    "summary": "Allow to save files in any model when submitting the survey",
+    "description": "", 
+    "version": "16.0.1.0.0",
+    "license": "AGPL-3",
+    "author": "Elabore",
+    "website": "https://elabore.coop",
+    "maintainer": "Elabore",
+    "category": "Survey",
+    "depends": ["survey_record_generation", "survey_extra_fields"],
+    "data": [        
+    ],    
+    "installable": True,
+    # Install this module automatically if all dependency have been previously
+    # and independently installed.  Used for synergetic or glue modules.
+    "auto_install": True,
+}

survey_record_generation_extra_fields/models/__init__.py

@@ -0,0 +1,2 @@
+from . import survey_record_creation_field_values
+from . import survey_user_input

survey_record_generation_extra_fields/models/survey_record_creation_field_values.py

@@ -0,0 +1,65 @@
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+
+from odoo import _, api, fields, models
+from odoo.exceptions import ValidationError
+
+BINARY_QUESTION_TYPES = ["file"]
+
+
+class SurveyRecordCreationFieldValues(models.Model):
+    _inherit = "survey.record.creation.field.values"
+
+    # add file type to the domain
+    field_id = fields.Many2one(
+        domain="[('model_id','=',model_id),"
+        "('ttype','in',['char','selection','text','html','integer','float',"
+        "'date','datetime','many2one','many2many','boolean','binary'])]",
+    )
+
+    fixed_value_binary = fields.Binary("Value")
+    fixed_value_binary_fname = fields.Char("Filename")
+
+    @api.depends("survey_id")
+    def _compute_allowed_question_ids(self):
+        super()._compute_allowed_question_ids()
+        for record in self.filtered(lambda r: r.field_id.ttype == "binary"):
+            if not record.survey_id:
+                record.allowed_question_ids = False
+                continue
+            record.allowed_question_ids = self.env["survey.question"].search([
+                ("survey_id", "=", record.survey_id.id),
+                ("question_type", "in", BINARY_QUESTION_TYPES),
+            ])
+
+    @api.onchange("fixed_value_binary", "fixed_value_binary_fname")
+    def _compute_displayed_value(self):
+        # display filename instead of file content
+        binary_values = self.filtered(
+            lambda r: r.field_id.ttype == "binary" and r.value_origin == "fixed"
+        )
+        super(
+            SurveyRecordCreationFieldValues, self - binary_values
+        )._compute_displayed_value()
+        for record in binary_values:
+            if record.fixed_value_binary_fname:
+                record.displayed_value = record.fixed_value_binary_fname
+            elif record.fixed_value_binary:
+                record.displayed_value = _("File")
+            else:
+                record.displayed_value = None
+
+    def clean_values(self):
+        res = super().clean_values()
+        self.fixed_value_binary = None
+        self.fixed_value_binary_fname = None
+        return res
+
+    @api.constrains("field_id", "unicity_check")
+    def _check_binary_unicity_check(self):
+        for record in self:
+            if record.field_id.ttype == "binary" and record.unicity_check:
+                raise ValidationError(_(
+                    "The unicity constraint cannot be used on the binary field "
+                    "%s: binary fields are stored as attachments and cannot be "
+                    "searched.", record.field_id.display_name
+                ))

survey_record_generation_extra_fields/models/survey_user_input.py

@@ -0,0 +1,26 @@
+from odoo import models
+
+FILE_QUESTION_TYPES = ["file"]
+
+
+class SurveyUserInput(models.Model):
+    _inherit = "survey.user_input"
+
+    def get_value_from_user_answer(self, field_value, user_input):
+        # we check for file type before parent checks
+        if field_value.question_id.question_type not in FILE_QUESTION_TYPES:
+            return super().get_value_from_user_answer(field_value, user_input)
+
+        line = self._get_file_answer_line(field_value, user_input)
+        if not line:
+            return None
+        return line.with_context(bin_size=False).value_file or None
+
+    def _get_file_answer_line(self, field_value, user_input):
+        # return possible lines for this question, otherwise empty recordset
+        lines = user_input.user_input_line_ids.filtered(
+            lambda line: line.question_id == field_value.question_id
+        )
+        if not lines or lines[0].skipped:
+            return self.env["survey.user_input.line"]
+        return lines[0]

survey_record_generation_extra_fields/tests/__init__.py

@@ -0,0 +1 @@
+from . import test_record_generation_extra_fields

survey_record_generation_extra_fields/tests/test_record_generation_extra_fields.py

@@ -0,0 +1,214 @@
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+
+import base64
+
+from odoo.tests import tagged
+
+from odoo.addons.survey.tests.common import SurveyCase
+
+
+@tagged("post_install", "-at_install")
+class TestSurveyRecordCreationFileCommon(SurveyCase):
+    """Socle commun : un questionnaire avec une question "nom" et une question
+    "fichier", mappées vers res.partner.name et vers un champ binaire
+    personnalisé res.partner.x_document.
+    """
+
+    # Champ binaire cible, créé dynamiquement (voir _setup_target_field).
+    # Le préfixe x_ est imposé par la contrainte sur state="manual".
+    FILE_FIELD = "x_document"
+
+    def setUp(self):
+        super().setUp()
+
+        # `_add_answer_line` s'appuie sur cette table pour déduire le couple
+        # (answer_type, champ de stockage) à partir du type de question.
+        # Le type "file" vient de survey_extra_fields, il n'y figure pas.
+        self._type_match["file"] = ("file", "value_file")
+
+        self.res_partner_model = self.env["ir.model"]._get("res.partner")
+        self._setup_target_field()
+        self._setup_survey()
+        self._setup_record_creation()
+
+        self.file_content = b"%PDF-1.4 contenu de test"
+        self.file_b64 = base64.b64encode(self.file_content)
+        self.file_name = "document.pdf"
+
+    # ------------------------------------------------------------------
+    # Setup helpers
+    # ------------------------------------------------------------------
+
+    def _setup_target_field(self):
+        """Crée un champ binaire personnalisé sur res.partner.
+
+        Le module pont doit fonctionner avec n'importe quel champ binaire de
+        n'importe quel modèle : on teste donc avec un champ personnalisé
+        plutôt qu'avec un champ existant comme image_1920, dont le type Image
+        applique un redimensionnement et refuserait un PDF.
+        """
+        self.env["ir.model.fields"].create({
+            "name": self.FILE_FIELD,
+            "model_id": self.res_partner_model.id,
+            "field_description": "Document justificatif",
+            "ttype": "binary",
+            "state": "manual",
+        })
+        # create() a déjà appelé setup_models() + init_models(), mais ne lève
+        # pas ce drapeau (seul unlink() le fait). Sans lui, le reset_changes()
+        # enregistré en addClassCleanup par TransactionCase ne resynchronise
+        # pas le registry et le champ survit au rollback.
+        self.registry.registry_invalidated = True
+
+        self.file_field = self.env["ir.model.fields"]._get(
+            "res.partner", self.FILE_FIELD
+        )
+        self.name_field = self.env["ir.model.fields"]._get("res.partner", "name")
+
+    def _setup_survey(self):
+        self.survey = self.env["survey.survey"].create({"title": "Test Survey"})
+        # res.partner.name est obligatoire : sans cette question, la création
+        # échoue en IntegrityError avant même d'atteindre le fichier.
+        self.question_name = self._add_question(
+            page=None,
+            name="Name",
+            qtype="char_box",
+            survey_id=self.survey.id,
+            sequence=1,
+        )
+        self.question_document = self._add_question(
+            page=None,
+            name="Document",
+            qtype="file",
+            constr_mandatory=False,
+            survey_id=self.survey.id,
+            sequence=2,
+        )
+
+    def _setup_record_creation(self):
+        self.survey_record_creation = self.env["survey.record.creation"].create({
+            "name": "Contact",
+            "survey_id": self.survey.id,
+            "model_id": self.res_partner_model.id,
+        })
+        self.name_field_values = self._map_question(
+            self.name_field, self.question_name
+        )
+        self.file_field_values = self._map_question(
+            self.file_field, self.question_document
+        )
+
+    def _map_question(self, field, question, **kwargs):
+        values = {
+            "survey_record_creation_id": self.survey_record_creation.id,
+            "survey_id": self.survey.id,
+            "model_id": self.res_partner_model.id,
+            "field_id": field.id,
+            "value_origin": "question",
+            "question_id": question.id,
+        }
+        values.update(kwargs)
+        return self.env["survey.record.creation.field.values"].create(values)
+
+    # ------------------------------------------------------------------
+    # Answer helpers
+    # ------------------------------------------------------------------
+
+    def _answer_survey(self, name="Jean", file_b64=None, file_name=None):
+        """Répond au questionnaire et le valide. Retourne le user_input."""
+        answer = self._add_answer(
+            survey=self.survey, partner=False, email="jean@test.fr"
+        )
+        if name is not None:
+            self._add_answer_line(
+                question=self.question_name, answer=answer, answer_value=name
+            )
+        if file_b64 is not None:
+            self._add_answer_line(
+                question=self.question_document,
+                answer=answer,
+                answer_value=file_b64,
+                value_file_fname=file_name or self.file_name,
+            )
+        answer._mark_done()
+        return answer
+
+    def _read_binary(self, record, fname=None):
+        """Relit un champ binaire depuis la base, hors cache et hors bin_size."""
+        fname = fname or self.FILE_FIELD
+        record.invalidate_recordset([fname])
+        return record.with_context(bin_size=False)[fname]
+
+
+class TestSurveyRecordCreationFile(TestSurveyRecordCreationFileCommon):
+
+    def test_file_is_copied_to_binary_field(self):
+        """Le fichier répondu se retrouve dans le champ binaire cible."""
+        self._answer_survey(file_b64=self.file_b64)
+
+        partner = self.env["res.partner"].search([("name", "=", "Jean")])
+        self.assertEqual(len(partner), 1)
+        stored = self._read_binary(partner)
+        self.assertTrue(stored, "Le champ binaire devrait être renseigné")
+        self.assertEqual(base64.b64decode(stored), self.file_content)
+
+    def test_file_stored_as_attachment(self):
+        """Un champ binaire manuel est stocké en pièce jointe, pas en colonne.
+
+        Le domaine doit mentionner res_field, sinon ir.attachment._search
+        injecte ('res_field', '=', False) et masque la pièce jointe.
+        """
+        self._answer_survey(file_b64=self.file_b64)
+        partner = self.env["res.partner"].search([("name", "=", "Jean")])
+
+        attachment = self.env["ir.attachment"].sudo().search([
+            ("res_model", "=", "res.partner"),
+            ("res_id", "=", partner.id),
+            ("res_field", "=", self.FILE_FIELD),
+        ])
+        self.assertEqual(len(attachment), 1)
+        self.assertEqual(attachment.raw, self.file_content)
+
+    def test_no_file_answered(self):
+        """Sans réponse fichier, l'enregistrement est créé, champ binaire vide."""
+        self._answer_survey(file_b64=None)
+
+        partner = self.env["res.partner"].search([("name", "=", "Jean")])
+        self.assertEqual(len(partner), 1)
+        self.assertFalse(self._read_binary(partner))
+
+    def test_skipped_file_answer(self):
+        """Une ligne de réponse explicitement passée ne remplit pas le champ."""
+        answer = self._add_answer(
+            survey=self.survey, partner=False, email="jean@test.fr"
+        )
+        self._add_answer_line(
+            question=self.question_name, answer=answer, answer_value="Jean"
+        )
+        self.env["survey.user_input.line"].create({
+            "user_input_id": answer.id,
+            "question_id": self.question_document.id,
+            "skipped": True,
+        })
+        answer._mark_done()
+
+        partner = self.env["res.partner"].search([("name", "=", "Jean")])
+        self.assertEqual(len(partner), 1)
+        self.assertFalse(self._read_binary(partner))
+
+    def test_file_updates_existing_record(self):
+        """En mode mise à jour, le fichier est écrit sur le partner existant."""
+        partner = self.env["res.partner"].create({"name": "Jean"})
+        self.survey_record_creation.write({
+            "update_existing_records": True,
+            "field_to_retrieve_existing_records": self.name_field.id,
+        })
+
+        self._answer_survey(file_b64=self.file_b64)
+
+        self.assertEqual(
+            self.env["res.partner"].search_count([("name", "=", "Jean")]), 1
+        )
+        self.assertEqual(
+            base64.b64decode(self._read_binary(partner)), self.file_content
+        )