Public release from ruodoo-project: 19.0 - 2026-07-26 21:17:35 UTC

directive_layer/models/directive_mixin.py

@@ -0,0 +1,330 @@
+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)