import functools from odoo import api, fields, models from odoo.exceptions import UserError import logging _logger = logging.getLogger(__name__) class DirectiveMixin(models.AbstractModel): _name = "directive.mixin" _description = "Directive Mixin" directive_origin_id = fields.Many2one( "directive.origin", string="Directive Origin", readonly=True, index=True, ondelete="set null", ) directive_ids = fields.One2many( "directive.directive", string="Директивы", related="directive_origin_id.directive_ids", readonly=True, ) directive_execution_status = fields.Char( string="Текущий статус исполнения", compute="_compute_directive_execution_status", store=True, ) directive_template_group_id = fields.Many2one( "directive.template.group", string="Группа шаблонов директив", ) @api.model_create_multi def create(self, vals_list): records = super().create(vals_list) records._ensure_directive_origin() return records def write(self, vals): _logger.debug( "[directive] write() called. Model: %s, IDs: %s, Keys in vals: %s", self._name, self.ids, list(vals.keys()) ) policies = self.env['directive.policy'].sudo().search([ ('origin_model', '=', self._name), ('event', '=', 'on_write'), ('active', '=', True) ]) relevant_policies = policies.filtered( lambda p: p.trigger_field and p.trigger_field in vals ) before_map = {} if relevant_policies: watched_fields = set(relevant_policies.mapped('trigger_field')) for rec in self: before_map[rec.id] = {f: rec[f] for f in watched_fields} res = super().write(vals) self._ensure_directive_origin() if relevant_policies: for policy in relevant_policies: field = policy.trigger_field records_to_process = self.env[self._name] for rec in self: before_val = before_map[rec.id].get(field) after_val = rec[field] def _norm(v): return v.id if hasattr(v, 'id') else v b_val = _norm(before_val) a_val = _norm(after_val) if b_val != a_val and policy._policy_should_fire(rec): records_to_process |= rec if records_to_process: _logger.info( "[directive] applying policy '%s' for model=%s, ids=%s", policy.name, self._name, records_to_process.ids ) policy.sudo().apply_to_records(records_to_process) return res def _ensure_directive_origin(self): for rec in self: if not rec.id: continue if rec.directive_origin_id: continue origin = self.env["directive.origin"].search( [("res_model", "=", rec._name), ("res_id", "=", rec.id)], limit=1, ) if not origin: origin = self.env["directive.origin"].create( {"res_model": rec._name, "res_id": rec.id} ) rec.sudo().write({"directive_origin_id": origin.id}) @api.depends( "directive_ids.stage_id", "directive_ids.stage_id.code", "directive_ids.deadline_at", "directive_ids.planned_end_at", ) def _compute_directive_execution_status(self): for rec in self: directives = rec.directive_ids.filtered( lambda d: d.stage_id.code not in ("done", "cancelled") ) if not directives: rec.directive_execution_status = "" continue stage_priority = { "paused": 1, "in_progress": 2, "planned": 3, "confirmed": 3, "draft": 4, } def stage_rank(d): return stage_priority.get(d.stage_id.code, 99) current_directive = sorted(directives, key=stage_rank)[0] stage_name = current_directive.stage_id.name deadlines = directives.mapped("deadline_at") nearest_deadline = min(deadlines) if deadlines else None planned_ends = directives.mapped("planned_end_at") planned_ends = [d for d in planned_ends if d] expected_end = max(planned_ends) if planned_ends else max(deadlines) parts = [stage_name] if nearest_deadline: parts.append( f"до {fields.Datetime.to_string(nearest_deadline)[:10]}" ) if expected_end: parts.append( f"ожидается завершение к {fields.Datetime.to_string(expected_end)[:10]}" ) rec.directive_execution_status = ", ".join(parts) def _directive_on_assign(self, payload: dict): return None def _directive_on_start(self, payload: dict): return None def _directive_on_pause(self, payload: dict): return None def _directive_on_done(self, payload: dict): return None def _directive_create_from_group(self, template_group, generator_key=None, policy=None): self.ensure_one() self._ensure_directive_origin() if not template_group: raise UserError("Template group is required.") draft_stage = self.env["directive.stage"].search([("code", "=", "draft")], limit=1) if not draft_stage: raise UserError("Stage with code 'draft' not found.") lines = template_group.template_line_ids.sorted(key=lambda l: (l.sequence, l.id)) if not lines: raise UserError("Template group has no lines.") directives_vals = [] for line in lines: tmpl = line.template_id directives_vals.append({ "generator_key": generator_key or False, "name": tmpl.name, "deadline_at": fields.Datetime.now(), "planned_hours": tmpl.planned_hours, "load_percent": tmpl.load_percent, "priority": tmpl.priority, "work_type_id": tmpl.work_type_id.id, "executor_role_id": tmpl.executor_role_id.id or False, "requester_role_id": tmpl.requester_role_id.id or False, "stage_id": draft_stage.id, "executor_stage_id": False, "requester_stage_id": False, "origin_id": self.directive_origin_id.id, "template_id": tmpl.id, "policy_id": policy.id if policy else False, }) return self.env["directive.directive"].sudo().create(directives_vals) def _directive_update(self, *args, **kwargs): return None def _directive_cancel(self, *args, **kwargs): return None def _directive_delete(self, *args, **kwargs): return None def action_choose_directive_group(self): self.ensure_one() self._ensure_directive_origin() return { "type": "ir.actions.act_window", "name": "Выбор группы директив", "res_model": "directive.choose.group.wizard", "view_mode": "form", "target": "new", "context": { "active_model": self._name, "active_id": self.id, }, } def action_open_directives(self): self.ensure_one() self._ensure_directive_origin() return { "type": "ir.actions.act_window", "name": "Директивы", "res_model": "directive.directive", "view_mode": "kanban,list,form", "domain": [("origin_id", "=", self.directive_origin_id.id)], "context": { "default_origin_id": self.directive_origin_id.id, }, } @api.model def _directive_generation_spec(self): """Может быть переопределен в модели-источнике.""" return [] @api.model def _register_hook(self): res = super()._register_hook() _logger.info( "[directive] register_hook model=%s", self._name ) self._directive_register_callback_wrappers() return res @api.model def _directive_register_callback_wrappers(self): cls = self.__class__ if cls.__dict__.get("_directive_callbacks_wrapped"): return cls._directive_callbacks_wrapped = True self.env.cr.execute("SELECT count(*) FROM information_schema.tables WHERE table_name = 'directive_policy'") if not self.env.cr.fetchone()[0]: return policies = self.env['directive.policy'].sudo().search([ ('origin_model', '=', self._name), ('event', '=', 'on_callback'), ('active', '=', True) ]) if not policies: return for policy in policies: callback_name = policy.trigger_field if not callback_name: continue orig = getattr(cls, callback_name, None) if not orig or not callable(orig): _logger.warning( "[directive] Не найден метод '%s' в модели %s (из политики %s)", callback_name, self._name, policy.name ) continue if getattr(orig, "_directive_wrapped", False): continue _logger.info( "[directive] Оборачиваем метод %s.%s для политики '%s'", self._name, callback_name, policy.name ) policy_id = policy.id @functools.wraps(orig) def _wrapped(self_recordset, *args, __orig=orig, __policy_id=policy_id, **kwargs): if self_recordset.env.context.get("directive_skip_generation"): return __orig(self_recordset, *args, **kwargs) result = __orig(self_recordset, *args, **kwargs) if result is not False: active_policy = self_recordset.env['directive.policy'].sudo().browse(__policy_id) if active_policy.exists() and active_policy.active: active_policy.sudo().apply_to_records(self_recordset) return result _wrapped._directive_wrapped = True setattr(cls, callback_name, _wrapped)