330 lines
11 KiB
Python
330 lines
11 KiB
Python
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) |