# Copyright 2023 Akretion (https://www.akretion.com). # @author Sébastien BEAU # Copyright 2025 Tecnativa - Víctor Martínez # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). from collections import defaultdict from odoo import Command, api, models, tools from odoo.exceptions import AccessError from ..tools import format_m2m # To avoid conflict with other module and avoid too long function name # specific tracking_manager method are prefixed with _tm class Base(models.AbstractModel): _inherit = "base" @tools.ormcache() def is_tracked_by_o2m(self): return self._name in self.env["ir.model"]._get_model_tracked_by_o2m() def _tm_get_fields_to_notify(self): return ( self.env["ir.model"] ._get_model_tracked_by_o2m() .get(self._name, {}) .get("notify", []) ) def _tm_get_fields_to_track(self): # We track manually # all fields that belong to a model tracked via a one2many # all the many2many fields return ( self.env["ir.model"] ._get_model_tracked_by_o2m() .get(self._name, {}) .get("fields", []) ) def _tm_notify_owner(self, mode, changes=None): """Notify all model that have a one2many linked to the record changed""" self.ensure_one() data = self.env.cr.precommit.data.setdefault( "tracking.manager.data", defaultdict(lambda: defaultdict(lambda: defaultdict(list))), ) for field_name, owner_field_name in self._tm_get_fields_to_notify(): owner = self[field_name] # Skip processing if the owner is not a valid Odoo recordset or is empty. if not (owner and isinstance(owner, models.BaseModel)): continue data[owner._name][owner.id][owner_field_name].append( { "mode": mode, "record": self.display_name, "changes": changes, } ) def _tm_get_field_description(self, field_name): return self._fields[field_name].get_description(self.env)["string"] def _tm_get_changes(self, values): self.ensure_one() changes = [] for field_name, before in values.items(): field = self._fields[field_name] if before != self[field_name]: if field.type == "many2many": old = format_m2m(before) new = format_m2m(self[field_name]) elif field.type == "many2one": old = before.display_name new = self[field_name]["display_name"] else: old = before new = self[field_name] changes.append( { "name": self._tm_get_field_description(field_name), "old": old, "new": new, } ) return changes def _tm_post_message(self, data): for model_name, model_data in data.items(): # check if record has mail.thread mixin if not getattr(self.env[model_name], "message_post_with_source", False): continue for record_id, messages_by_field in model_data.items(): # Avoid error if no record is linked (example: child_ids of res.partner) if not record_id: continue record = self.env[model_name].browse(record_id) messages = [ { "name": record._tm_get_field_description(field_name), "messages": messages, } for field_name, messages in messages_by_field.items() ] # We do not use message_post_with_view() because emails would be sent rendered_template = self.env["ir.qweb"]._render( "tracking_manager.track_o2m_m2m_template", {"lines": messages, "object": record}, minimal_qcontext=True, ) record._message_log(body=rendered_template) def _tm_prepare_o2m_tracking(self): fnames = self._tm_get_fields_to_track() if not fnames: return self.env.cr.precommit.add(self._tm_finalize_o2m_tracking) initial_values = self.env.cr.precommit.data.setdefault( f"tracking.manager.before.{self._name}", {} ) for record in self: values = initial_values.setdefault(record.id, {}) if values is not None: for fname in fnames: try: values.setdefault(fname, record[fname]) except AccessError: # User does not have access to the field (example with groups) continue def _tm_finalize_o2m_tracking(self): initial_values = self.env.cr.precommit.data.pop( f"tracking.manager.before.{self._name}", {} ) for _id, values in initial_values.items(): # Always use sudo in case that the record have been modified using sudo record = self.sudo().browse(_id) if not record.exists(): # if a record have been modify and then deleted # it's not need to track the change so skip it continue changes = record._tm_get_changes(values) if changes: record._tm_notify_owner("update", changes) data = self.env.cr.precommit.data.pop("tracking.manager.data", {}) self._tm_post_message(data) self.flush_model() def _tm_track_create_unlink(self, mode): self.env.cr.precommit.add(self._tm_finalize_o2m_tracking) for record in self: record._tm_notify_owner(mode) def write(self, vals): if self.is_tracked_by_o2m(): self._tm_prepare_o2m_tracking() return super().write(vals) @api.model_create_multi def create(self, list_vals): records = super().create(list_vals) if self.is_tracked_by_o2m(): records._tm_track_create_unlink("create") return records def unlink(self): if self.is_tracked_by_o2m(): self._tm_track_create_unlink("unlink") return super().unlink() # TODO: Remove if merged https://github.com/odoo/odoo/pull/156236 def _mail_track(self, tracked_fields, initial_values): _tracked_fields = tracked_fields tracked_fields_properties = {} for tf_key in list(_tracked_fields.keys()): tracked_field = tracked_fields[tf_key] if tracked_field["type"] == "properties": tracked_fields_properties[tf_key] = tracked_field updated, tracking_value_ids = super()._mail_track( tracked_fields, initial_values ) # Remove unnecessary tracking_value_ids tracking_value_ids_keys_to_delete = [] for tf_key in list(tracked_fields_properties.keys()): field = self.env["ir.model.fields"]._get(self._name, tf_key) for key, vals in enumerate(tracking_value_ids): if vals[2]["field_id"] == field.id: tracking_value_ids_keys_to_delete.append(key) for key in tracking_value_ids_keys_to_delete: tracking_value_ids.pop(key) # Extra things for properties for col_name, _sequence in self._mail_track_order_fields( tracked_fields_properties ): if col_name not in initial_values: continue initial_value, new_value = initial_values[col_name], self[col_name] if new_value == initial_value or (not new_value and not initial_value): continue p_keys = list(initial_value.keys()) if initial_value else [] properties_data = {} for definition in self.read([col_name])[0][col_name]: properties_data[definition["name"]] = definition tracking_value_ids.extend( Command.create( self.env["mail.tracking.value"]._create_tracking_values_property( initial_value[p_key], new_value[p_key], col_name, properties_data[p_key], self, ), ) for p_key in p_keys if ( p_key in properties_data and properties_data[p_key]["type"] != "separator" and initial_value[p_key] != new_value[p_key] ) ) return updated, tracking_value_ids