[ADD] survey_record_generation : migrate from 16 to 18

This commit is contained in:
2026-03-24 17:29:49 +01:00
parent eced098fbb
commit 6f663674d2
28 changed files with 3156 additions and 0 deletions

View File

@@ -0,0 +1,7 @@
from . import survey_question_answer
from . import survey_question
from . import survey_record_creation_field_values
from . import survey_record_creation
from . import survey_survey
from . import survey_user_input
from . import survey_generated_record

View File

@@ -0,0 +1,15 @@
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from odoo import models, fields, api
class SurveyGeneratedRecord(models.Model):
_name = "survey.generated.record"
survey_record_creation_name = fields.Char('Name', readonly=True)
survey_record_creation_id = fields.Many2one('survey.record.creation', 'Survey record creation', readonly=True)
user_input_id = fields.Many2one('survey.user_input', 'Participation', readonly=True)
created_record_id = fields.Reference(string="Referenced record", selection='_selection_target_model', readonly=True)
@api.model
def _selection_target_model(self):
return [(model.model, model.name) for model in self.env['ir.model'].sudo().search([])]

View File

@@ -0,0 +1,49 @@
import logging
import ast
from typing import Literal
from odoo import api, fields, models, _, Command
from odoo.exceptions import UserError
_logger = logging.getLogger(__name__)
AnswerValuesType: Literal["no", "value", "record"]
class SurveyQuestion(models.Model):
_inherit = 'survey.question'
model_id = fields.Many2one('ir.model', string="Model")
model_name = fields.Char(related="model_id.model")
fill_domain = fields.Char("Domain", default="[]")
answer_values_type = fields.Selection([('no', 'No values'),('value','Value'),('record','Record')], string="Associate value to answer", default="no", required=True)
@api.onchange('model_id')
def onchange_model_id(self):
self.fill_domain = "[]"
if self.model_id:
rec = self.env[self.model_id.model].search([], limit=1)
if not rec:
raise UserError(_('No record found in %s',self.model_id.name))
else:
for answer in self.suggested_answer_ids:
answer.record_id = f"{self.model_id.model},{rec.id}"
def fill(self):
for question in self:
if question.model_id:
new_suggested_answer_ids = [Command.clear()]
record_model = question.model_id.model
if question.fill_domain:
domain = ast.literal_eval(question.fill_domain)
else:
domain = []
records = self.env[record_model].search(domain)
new_suggested_answer_ids += [Command.create({'value':record.display_name, 'record_id':f"{record_model},{record.id}"
}) for record in records]
question.suggested_answer_ids = new_suggested_answer_ids

View File

@@ -0,0 +1,42 @@
import logging
from odoo import api, fields, models, _
_logger = logging.getLogger(__name__)
class SurveyQuestionAnswer(models.Model):
_inherit = 'survey.question.answer'
record_id = fields.Reference(string="Referenced record", selection='_selection_target_model')
model_id = fields.Many2one('ir.model', related="question_id.model_id")
answer_values_type = fields.Selection(related="question_id.answer_values_type")
value_char = fields.Char('Value')
@api.model
def _selection_target_model(self):
return [(model.model, model.name) for model in self.env['ir.model'].sudo().search([])]
@api.onchange('record_id')
def onchange_record_id(self):
if self.record_id:
self.value = self.record_id.display_name
@api.model
def default_get(self, fields):
result = super().default_get(fields)
if (
not result.get("model_id")
or "record_id" not in fields
):
return result
model = self.env['ir.model'].browse(result.get("model_id")).model
res = self.env[model].search([], limit=1)
if res:
result["record_id"] = "%s,%s" % (
model,
res.id,
)
return result

View File

@@ -0,0 +1,66 @@
import logging
from odoo import api, fields, models, _
_logger = logging.getLogger(__name__)
class SurveyRecordCreation(models.Model):
"""Configure list of models for wich record will be created on survey submission
"""
_name = 'survey.record.creation'
name = fields.Char('Name')
survey_id = fields.Many2one('survey.survey', string="Survey")
model_id = fields.Many2one('ir.model', "Model", help="Model of generated record")
field_values_ids = fields.One2many('survey.record.creation.field.values', 'survey_record_creation_id', string="Field values")
warning_message = fields.Html('Warning message', compute="_compute_warning_message")
sequence = fields.Integer("sequence")
update_existing_records = fields.Boolean(
string="Update existing records",
)
field_to_retrieve_existing_records = fields.Many2one(
"ir.model.fields",
domain="[('id', 'in', allowed_field_ids)]",
ondelete="cascade",
help="Choose the field you want to use to retrieve the existing record. "
"WARNING: We update only the first record found.",
)
allowed_field_ids = fields.Many2many(
"ir.model.fields",
compute="_compute_allowed_field_ids",
store=True,
string="Allowed Fields",
)
ignore_if_mandatory_field_is_missing = fields.Boolean(
string="Ignore creation if a mandatory field is missing",
help="If a mandatory field is missing when trying to create the record, "
"an error is raised when the survey is submitted. "
"If this option is activated, the error is ignored."
)
@api.depends("field_values_ids.field_id")
def _compute_allowed_field_ids(self):
for record in self:
record.allowed_field_ids = record.field_values_ids.mapped("field_id")
@api.onchange("model_id")
def clear_field_values_ids(self):
self.field_values_ids = None
@api.depends("model_id","field_values_ids")
def _compute_warning_message(self):
for record_creation in self:
# check if all mandatory fields set
if record_creation.model_id:
required_field_ids = self.model_id.field_id.filtered(lambda f:f.required and "property_" not in f.name)
set_field_ids = self.field_values_ids.field_id
missing_fields = required_field_ids - set_field_ids
if missing_fields:
record_creation.warning_message = _("Some required fields are not set : %s",', '.join([f"<b>{f.field_description}</b> (<i>{f.name}</i>)" for f in missing_fields]))
else:
record_creation.warning_message = None
else:
record_creation.warning_message = None

View File

@@ -0,0 +1,197 @@
import logging
from odoo import api, fields, models, _
from odoo.exceptions import UserError
from odoo.tools.misc import format_date
_logger = logging.getLogger(__name__)
type_mapping = { #field types on the left, question types on the right. TODO : what about booleans ?
"char": ["char_box", "numerical_box", "date", "datetime", "simple_choice", "multiple_choice"],
"text": ["char_box", "date", "simple_choice"],
"html": ["text_box", "numerical_box", "datetime", "simple_choice"],
"integer": ["numerical_box"],
"float": ["numerical_box"],
"date": ["date"],
"datetime": ["datetime"],
"many2one": ["simple_choice"],
"many2many": ["multiple_choice"],
"selection": ["char_box", "simple_choice"]
}
class SurveyRecordCreationFieldValues(models.Model):
"""Configure default values of records created on survey submission
"""
_name = 'survey.record.creation.field.values'
survey_record_creation_id = fields.Many2one('survey.record.creation')
survey_id = fields.Many2one('survey.survey', related="survey_record_creation_id.survey_id")
model_id = fields.Many2one('ir.model', related="survey_record_creation_id.model_id")
field_id = fields.Many2one(
'ir.model.fields',
domain="[('model_id','=',model_id),('ttype','in',['char','selection','text','html','integer','float','date','datetime','many2one','many2many', 'boolean'])]",
ondelete="cascade")
field_relation = fields.Char(related='field_id.relation')
field_type = fields.Selection(related="field_id.ttype")
field_help = fields.Html('Help', compute="_compute_field_help")
value_origin = fields.Selection(
[('fixed','Fixed'),('question','Question'),('other_record','From other created record')],
string="Value origin",
required=True,
default='fixed',
help="""* Fixed: you can set the value in value field
* Question: Response of the question will set the value. If you do not see your question, maybe the type of question do not match the type of field
* From other created record: You can set other record creation to link several created records. Can only be used with many2one fields.""")
fixed_value_many2one = fields.Reference(string='Record', selection='_selection_target_model')
fixed_value_many2many = fields.One2many('survey.record.creation.field.values.x2m', "survey_record_creation_field_values_id")
fixed_value_char = fields.Char("Value")
fixed_value_selection = fields.Char("Value")
fixed_value_text = fields.Text("Value")
fixed_value_html = fields.Html("Value")
fixed_value_integer = fields.Integer("Value")
fixed_value_float = fields.Float("Value")
fixed_value_date = fields.Date("Value")
fixed_value_datetime = fields.Datetime("Value")
fixed_value_boolean = fields.Boolean("Value")
displayed_value = fields.Char("Value", compute="_compute_displayed_value")
other_created_record_id = fields.Many2one("survey.record.creation", string="Other record", domain="[('survey_id','=',survey_id),('model_id.model','=',field_relation)]")
allowed_question_ids = fields.Many2many('survey.question', compute='_compute_allowed_question_ids')
question_id = fields.Many2one('survey.question', string="Question", domain="[('id','in',allowed_question_ids)]")
unicity_check = fields.Boolean('Unicity constraint', help="On record creation, if another record exists with same value, record will not be created.")
@api.depends("field_id")
def _compute_field_help(self):
for record in self:
field_help = _("Field type is : <b>%s</b>",record.field_type)
if record.field_type == "selection":
field_help += "<br />"+_("possible values are %s",', '.join([f"<b>{s.value}</b> <i>({s.name})</i>" for s in record.field_id.selection_ids]))
record.field_help = field_help
@api.depends('field_id')
def _compute_allowed_question_ids(self):
for record_creation_field_values in self:
if not record_creation_field_values.survey_id or not record_creation_field_values.field_id:
record_creation_field_values.allowed_question_ids = None
continue
question_domain = [('survey_id','=',record_creation_field_values.survey_id.id)]
if record_creation_field_values.field_id.ttype in ['many2one','many2many']:
question_domain.extend(['|','&',('answer_values_type','=','record'),('model_id','=',record_creation_field_values.field_id.relation),('answer_values_type','=','value')])
if record_creation_field_values.field_id.ttype in type_mapping:
question_domain.append(('question_type','in',type_mapping[record_creation_field_values.field_id.ttype]))
record_creation_field_values.allowed_question_ids = self.env['survey.question'].search(question_domain)
@api.model
def _selection_target_model(self):
return [(model.model, model.name) for model in self.env['ir.model'].sudo().search([])]
def clean_values(self):
# clean values
self.fixed_value_many2many = None
self.fixed_value_many2one = None
self.fixed_value_char = None
self.fixed_value_selection = None
self.fixed_value_text = None
self.fixed_value_html = None
self.fixed_value_integer = None
self.fixed_value_float = None
self.fixed_value_date = None
self.fixed_value_datetime = None
self.other_created_record_id = None
self.question_id = None
@api.onchange('field_id')
def _onchange_field_id(self):
# clean values
self.clean_values()
# Set reference field model and select first record
if self.field_id and self.field_id.ttype == 'many2one' and self.field_id.relation:
rec = self.env[self.field_id.relation].search([], limit=1)
if rec:
self.fixed_value_many2one = f"{self.field_id.relation},{rec.id}"
else:
model_name = self.env['ir.model'].search([('model','=',self.field_id.relation)]).name
self.fixed_value_many2one = None
raise UserError(_('You should append at least one record in %s',model_name))
else:
self.fixed_value_many2one = None
def get_fixed_value_for_record_creation(self):
"""return val used in create() method
"""
if self.value_origin == 'fixed':
if self.field_type == 'many2one':
if self.fixed_value_many2one:
return self.fixed_value_many2one.id
elif self.field_type == 'many2many':
return [m2m.value_reference.id for m2m in self.fixed_value_many2many if m2m.value_reference]
else:
return self["fixed_value_"+self.field_type]
@api.onchange("fixed_value_char","fixed_value_selection","fixed_value_text","fixed_value_html","fixed_value_integer","fixed_value_float","fixed_value_date","fixed_value_datetime",'fixed_value_many2one', "fixed_value_many2many","other_created_record_id","question_id")
def _compute_displayed_value(self):
for record in self:
if record.field_id:
if record.value_origin == 'other_record' and record.other_created_record_id:
record.displayed_value = _("Other created record: ")+record.other_created_record_id.name
elif record.value_origin == 'fixed':
if record.field_id.ttype == "many2one":
if record.fixed_value_many2one:
record.displayed_value = record.fixed_value_many2one.display_name
else:
record.displayed_value = None
elif record.field_id.ttype == "many2many":
if record.fixed_value_many2many:
record.displayed_value = ", ".join([r.value_reference.display_name for r in record.fixed_value_many2many if r.value_reference])
else:
record.displayed_value = None
elif record.field_id.ttype == "date":
record.displayed_value = format_date(self.env, record.fixed_value_date)
elif record.field_id.ttype == "datetime":
record.displayed_value = format_date(self.env, record.fixed_value_datetime)
else:
record.displayed_value = str(record['fixed_value_'+record.field_id.ttype])
else: #value_origin = question
record.displayed_value = _('Answer to question: %s',record.question_id.title)
else:
record.displayed_value = ""
class SurveyRecordCreationFieldValuesX2m(models.Model):
"""O2m an M2m default values
"""
_name = 'survey.record.creation.field.values.x2m'
survey_record_creation_field_values_id = fields.Many2one('survey.record.creation.field.values')
value_reference = fields.Reference(string='Record', selection='_selection_target_model')
@api.model
def _selection_target_model(self):
return [(model.model, model.name) for model in self.env['ir.model'].sudo().search([])]
@api.onchange('survey_record_creation_field_values_id')
def _onchange_model_name(self):
# Set reference field model and select first record
field = self.survey_record_creation_field_values_id.field_id
if field and "2many" in field.ttype and field.relation:
rec = self.env[field.relation].search([], limit=1)
if rec:
self.value_reference = f"{field.relation},{rec.id}"
else:
model_name = self.env['ir.model'].search([('model','=',field.relation)]).name
raise ValueError(_('You should append at least one record in %s',(model_name,)))

View File

@@ -0,0 +1,13 @@
import logging
from odoo import api, fields, models, _
_logger = logging.getLogger(__name__)
class SurveySurvey(models.Model):
_inherit = 'survey.survey'
survey_record_creation_ids = fields.One2many('survey.record.creation', 'survey_id', 'Records creation', help="List of records created when survey submitted")

View File

@@ -0,0 +1,339 @@
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from typing import TYPE_CHECKING, Any, Literal
from odoo import _, fields, models
from odoo.exceptions import UserError
from odoo.fields import Command
if TYPE_CHECKING:
from .survey_question import AnswerValuesType
from .survey_record_creation import SurveyRecordCreation
from .survey_record_creation_field_values import SurveyRecordCreationFieldValues
class SurveyUserInput(models.Model):
_inherit = "survey.user_input"
generated_record_ids = fields.One2many(
"survey.generated.record", "user_input_id", "Generated records"
)
generated_records_count = fields.Integer(
"Attempts Count", compute="_compute_generated_records_count"
)
def _compute_generated_records_count(self):
for user_input in self:
user_input.generated_records_count = len(user_input.generated_record_ids)
def action_redirect_to_generated_records(self):
self.ensure_one()
action = self.env["ir.actions.act_window"]._for_xml_id(
"survey_record_generation.survey_generated_record_action"
)
action['domain'] = [('user_input_id', '=', self.id)]
return action
def _mark_done(self):
# generate records
for user_input in self:
created_records = {}
other_record_fields_to_update: list[SurveyRecordCreationFieldValues] = []
record_creation: SurveyRecordCreation
for (
record_creation
) in user_input.survey_id.survey_record_creation_ids.sorted("sequence"):
model: str = record_creation.model_id.model
vals: dict = {}
field_value: SurveyRecordCreationFieldValues
for field_value in record_creation.field_values_ids:
value, other_record_fields_to_update = (
self.get_value_based_on_value_origin(
field_value=field_value,
user_input=user_input,
created_records=created_records,
model=model,
other_record_fields_to_update=other_record_fields_to_update,
)
)
field_name: str = field_value.field_id.name
vals[field_name] = value
existing_record = self.find_existing_record(record_creation, vals)
duplicate = self.find_duplicate_if_there_are_fields_with_unicity_check(
model, record_creation, vals
)
if duplicate:
record = duplicate
elif existing_record:
vals_with_keys_not_in_record = {
k: v
for k, v in vals.items()
if not getattr(existing_record, k, False)
}
existing_record.write(vals_with_keys_not_in_record)
record = existing_record
else:
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...
if record_creation.ignore_if_mandatory_field_is_missing:
continue
raise
# Link generated records to user input
self.env["survey.generated.record"].create(
{
"survey_record_creation_name": record_creation.name,
"survey_record_creation_id": record_creation.id,
"user_input_id": user_input.id,
"created_record_id": f"{model},{record.id}",
}
)
created_records[record_creation.id] = record
# update linked record
for field_to_update in other_record_fields_to_update:
record_to_update = created_records.get(
field_to_update.survey_record_creation_id.id
)
if record_to_update:
linked_record = created_records[
field_to_update.other_created_record_id.id
]
value = self.get_value_for_relational_field(
field_to_update, linked_record
)
record_to_update.write({field_to_update.field_id.name: value})
return super()._mark_done()
def find_existing_record(
self, record_creation: "SurveyRecordCreation", vals: dict
) -> Any:
if record_creation.update_existing_records:
model = record_creation.model_id.model
search_field = record_creation.field_to_retrieve_existing_records
user_answer_value = vals.get(search_field.name)
if user_answer_value:
return self.env[model].search(
[(search_field.name, "=", user_answer_value)], limit=1
)
return None
def get_value_based_on_value_origin(
self,
field_value: "SurveyRecordCreationFieldValues",
user_input: "SurveyUserInput",
created_records: dict[Any, Any],
model: str,
other_record_fields_to_update: list["SurveyRecordCreationFieldValues"],
) -> tuple[Any, list["SurveyRecordCreationFieldValues"]]:
value: Any = None
if field_value.value_origin == "fixed":
value = field_value.get_fixed_value_for_record_creation()
elif field_value.value_origin == "question":
value = self.get_value_from_user_answer(field_value, user_input)
elif field_value.value_origin == "other_record":
# if the other_record value is a required field, get it or raise
value = self.get_required_value_from_other_record(
model, created_records, field_value
)
# otherwise, we update the record later (out of this for loop)
if not value:
other_record_fields_to_update.append(field_value)
return value, other_record_fields_to_update
@staticmethod
def get_value_for_relational_field(
field_to_update: "SurveyRecordCreationFieldValues", linked_record
) -> Any:
field_type = field_to_update.field_id.ttype
if field_type == "many2one":
return linked_record.id
else:
# many2many or one2many
return [Command.set(linked_record.ids)]
def find_duplicate_if_there_are_fields_with_unicity_check(
self, model: str, record_creation: "SurveyRecordCreation", vals: dict[Any, Any]
) -> Any:
# check duplicates
unique_fields = [
field_value.field_id.name
for field_value in record_creation.field_values_ids.filtered(
lambda r: r.unicity_check
)
]
duplicate = None
if unique_fields:
uniq_domain = []
for uniq_field in unique_fields:
uniq_domain.append((uniq_field, "=", vals[uniq_field]))
duplicate = self.env[model].search(uniq_domain, limit=1)
return duplicate
def get_required_value_from_other_record(
self,
model: str,
created_records: dict[Any, Any],
field_value: "SurveyRecordCreationFieldValues",
) -> Any:
model_class = self.env[model]
if model_class._fields[field_value.field_id.name].required:
# check if the other record is already created,
# if yes add it to vals, else raise
if (
len(created_records) > 0
and created_records[field_value.other_created_record_id.id]
):
linked_record = created_records[field_value.other_created_record_id.id]
return self.get_value_for_relational_field(field_value, linked_record)
else:
raise UserError(
_(
"The field %(field)s is mandatory for model %(model)s. "
"In Record Creation tab, drag %(record)s "
"on top of the model %(model)s."
)
% {
"field": field_value.field_id.display_name,
"model": model,
"record": field_value.other_created_record_id.name,
}
)
def get_value_from_user_answer(
self,
field_value: "SurveyRecordCreationFieldValues",
user_input: "SurveyUserInput",
) -> Any:
# find user_input_lines (which are user's answers) for the question
user_input_lines = [
user_input_line
for user_input_line in user_input.user_input_line_ids
if user_input_line.question_id == field_value.question_id
]
if not user_input_lines:
# If the question has not been displayed to the user,
# there are no user_input_lines
return None
if user_input_lines[0].skipped:
# The question has been ignored by the user
return None
question_type = field_value.question_id.question_type
if question_type in [
"char_box",
"text_box",
"numerical_box",
"date",
"datetime",
]:
return user_input_lines[0][f"value_{user_input_lines[0].answer_type}"]
elif question_type in ["simple_choice", "multiple_choice", "matrix"]:
answer_values_type = field_value.question_id.answer_values_type
return self.get_value_based_on_answer_values_type(
answer_values_type, field_value, question_type, user_input_lines
)
else:
raise UserError(
_(
"[Survey record generation] The question type %(type)s is not "
"recognized (for question %(question)s)."
)
% {"type": question_type, "question": field_value.question_id.title}
)
def get_value_based_on_answer_values_type(
self,
answer_values_type: "AnswerValuesType",
field_value: "SurveyRecordCreationFieldValues",
question_type: Literal["simple_choice", "multiple_choice", "matrix"],
user_input_lines: list[Any],
) -> Any:
if answer_values_type == "record":
answered_record_ids = []
for user_input_line in user_input_lines:
if (
user_input_line.suggested_answer_id
and user_input_line.suggested_answer_id.record_id
):
answered_record_ids.append(
user_input_line.suggested_answer_id.record_id.id
)
if not answered_record_ids:
return None
if question_type == "simple_choice":
return answered_record_ids[0]
elif question_type == "multiple_choice":
return answered_record_ids
else:
raise UserError(
_(
"[Survey record generation] The question type"
" %(type)s is not supported yet."
)
% {"type": question_type}
)
elif answer_values_type == "value":
answer_value_char = user_input_lines[0].suggested_answer_id.value_char
if field_value.field_id.ttype != "boolean":
return answer_value_char
else:
return self.get_boolean_value(
answer_value_char=answer_value_char,
question_title=field_value.question_id.title,
)
else:
raise UserError(
_(
"[Survey record generation] The answer values type '%(type)s' "
"is not supported (for question %(question)s). Use 'record' or "
"'value' instead."
)
% {
"type": answer_values_type,
"question": field_value.question_id.title,
}
)
@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,
}
)