from odoo.exceptions import UserError from datetime import datetime, timedelta import pytz from odoo import api, fields, models TRAFFIC_LIGHT = [ ("white", "White"), ("black", "Black"), ("green", "Green"), ("yellow", "Yellow"), ("red", "Red"), ] class Directive(models.Model): _name = "directive.directive" _description = "Directive" _order = "deadline_at asc, priority desc, id desc" is_simulation = fields.Boolean( string="Simulation Task", default=False, help="Флаг для отделения тестовых данных от реальных" ) PRIORITY = [ ("0", "Low"), ("1", "Normal"), ("2", "High"), ] name = fields.Char(string="Название директивы", required=True) active = fields.Boolean(string="Активна", default=True) deadline_at = fields.Datetime(string="Дедлайн исполнения", required=True) planned_hours = fields.Float(string="Трудоёмкость", required=True) load_percent = fields.Integer(string="Процент загрузки", default=10) executor_type = fields.Selection([ ('human', 'Человек'), ('machine', 'Оборудование / Станки'), ('software', 'Программный агент / API') ], string="Тип исполнителя", default='human', required=True) success_probability = fields.Float(string="Вероятность успеха (%)", default=98.0) resource_cost = fields.Float(string="Стоимость ресурса (у.е.)", default=0.0) retry_policy = fields.Selection([ ('none', 'Остановиться и звать человека'), ('retry_3', 'Повторить 3 раза'), ('backoff', 'Экспоненциальная задержка') ], string="Политика отказа", default='none') priority = fields.Selection( selection=PRIORITY, string="Приоритет", default="0", required=True, ) work_type_id = fields.Many2one( "directive.work.type", string="Тип работ", required=True, ondelete="restrict", index=True, ) executor_role_id = fields.Many2one( "res.users.role", string="Роль исполнителя", ondelete="restrict", index=True, ) requester_role_id = fields.Many2one( "res.users.role", string="Роль заявителя", ondelete="restrict", index=True, ) stage_id = fields.Many2one( "directive.stage", string="Статус (для диспетчера)", default=lambda self: self.env["directive.stage"].search([("code", "=", "draft")], limit=1), ondelete="restrict", index=True, group_expand="_group_expand_stage_id" ) state = fields.Char( related="stage_id.code" ) executor_stage_id = fields.Many2one( "directive.stage", string="Статус для исполнителя", ondelete="restrict", index=True, ) requester_stage_id = fields.Many2one( "directive.stage", string="Статус для заявителя", ondelete="restrict", index=True, ) executor_traffic_light = fields.Selection( selection=TRAFFIC_LIGHT, string="Светофор контроля исполнителя", default="white", required=True, ) resource_traffic_light = fields.Selection( selection=TRAFFIC_LIGHT, string="Светофор контроля ресурсов", default="white", required=True, ) planned_start_at = fields.Datetime(string="Планируемая дата начала") planned_end_at = fields.Datetime(string="Планируемая дата окончания") executor_user_id = fields.Many2one( "res.users", string="Назначенный исполнитель", ondelete="set null", index=True, ) origin_id = fields.Many2one( "directive.origin", string="Источник (породивший документ)", required=True, ondelete="cascade", index=True, ) template_id = fields.Many2one( "directive.template", string="Шаблон директивы", ondelete="set null", index=True, ) policy_id = fields.Many2one( "directive.policy", string="Policy", ondelete="set null", index=True, ) stage_code = fields.Char(related="stage_id.code", store=True, readonly=True) generator_key = fields.Char(index=True) deadline_display = fields.Char(compute="_compute_ui_fields") time_left_display = fields.Char(compute="_compute_ui_fields") is_overdue = fields.Boolean(compute="_compute_ui_fields") priority_stars = fields.Char(compute="_compute_ui_fields") sequence = fields.Integer(string="Planner Sequence", default=10) predecessor_ids = fields.Many2many('directive.directive', 'directive_dependencies_rel', 'directive_id', 'predecessor_id', string="Dependencies") @api.depends("deadline_at", "priority") def _compute_ui_fields(self): tz = self.env.user.tz or "UTC" tzinfo = pytz.timezone(tz) now = datetime.now(tzinfo) for rec in self: if rec.deadline_at: dt = fields.Datetime.context_timestamp(rec.with_context(tz=tz), rec.deadline_at) rec.deadline_display = dt.strftime("%d.%m.%y %-H:%M") rec.is_overdue = dt < now delta = dt - now total_min = int(delta.total_seconds() // 60) if total_min < 0: total_min = abs(total_min) sign = "-" else: sign = "" days = total_min // (60 * 24) hours = (total_min % (60 * 24)) // 60 mins = total_min % 60 parts = [] if days: parts.append(f"{days}д") if hours: parts.append(f"{hours}ч") parts.append(f"{mins}м") rec.time_left_display = f"{sign}{' '.join(parts)}" else: rec.deadline_display = "" rec.time_left_display = "" rec.is_overdue = False p = int(rec.priority or 0) rec.priority_stars = "★" * (p + 1) def write(self, vals): if "stage_id" in vals: new_stage = self.env["directive.stage"].browse(vals["stage_id"]) for rec in self: rec._check_stage_transition(rec.stage_id, new_stage) return super().write(vals) def _check_stage_transition(self, from_stage, to_stage): from_code = (from_stage.code or "").strip() to_code = (to_stage.code or "").strip() allowed = { "draft": {"confirmed", "planned", "cancelled"}, "confirmed": {"planned", "in_progress", "cancelled"}, "planned": {"in_progress", "cancelled"}, "in_progress": {"paused", "done"}, "paused": {"in_progress", "cancelled"}, "done": set(), "cancelled": set(), } if from_code and to_code and to_code not in allowed.get(from_code, set()): raise UserError("Нельзя переместить из '%s' в '%s'.") % (from_stage.name, to_stage.name) @api.model def _group_expand_stage_id(self, stages, domain, order=None): return stages.search([], order=order or "id asc") def action_cancel(self): cancelled = self.env["directive.stage"].search([("code", "=", "cancelled")], limit=1) if not cancelled: raise UserError("Stage with code 'cancelled' not found.") self.write({ "stage_id": cancelled.id, "executor_stage_id": cancelled.id, "requester_stage_id": cancelled.id, }) return True def action_open_delete_wizard(self): self.ensure_one() return { "type": "ir.actions.act_window", "name": "Удаление директивы", "res_model": "directive.delete.wizard", "view_mode": "form", "target": "new", "context": { "active_model": self._name, "active_id": self.id, "active_ids": self.ids, }, } def _action_archive(self): self.write({"active": False}) return True def _action_delete_hard(self): self.unlink() return True def _build_payload(self, event: str) -> dict: self.ensure_one() tmpl = self.template_id return { "event": event, "directive_id": self.id, "directive_name": self.name, "template_id": tmpl.id if tmpl else None, "template_code": tmpl.code if tmpl else None, "work_type_id": self.work_type_id.id if self.work_type_id else None, "executor_user_id": self.executor_user_id.id if self.executor_user_id else None, "planned_start_at": self.planned_start_at, "planned_end_at": self.planned_end_at, "stage_code": self.stage_id.code if self.stage_id else None, "priority": self.priority, "load_percent": self.load_percent, } def _stage_by_code(self, code: str): stage = self.env["directive.stage"].search([("code", "=", code)], limit=1) if not stage: raise UserError(f"Directive stage '{code}' not found.") return stage def _set_stage(self, code: str): stage = self._stage_by_code(code) self.write({ "stage_id": stage.id, "executor_stage_id": stage.id, "requester_stage_id": stage.id, }) def _get_origin_record(self): self.ensure_one() origin = self.origin_id if not origin or not origin.res_model or not origin.res_id: return self.env[self._name].browse() # empty return self.env[origin.res_model].browse(origin.res_id).exists() def action_assign_to_me(self): for rec in self: if rec.executor_user_id: raise UserError("Директива уже назначена исполнителю.") if rec.stage_code not in ("confirmed", "planned"): raise UserError("Назначить на себя можно только подтвержденную или запланированную директиву.") rec.write({"executor_user_id": rec.env.user.id}) return True def action_start(self): now = fields.Datetime.now() for rec in self: if not rec.executor_user_id: raise UserError("Нельзя начать директиву без назначенного исполнителя.") if rec.stage_code not in ("confirmed", "planned"): raise UserError("Начать можно только подтвержденную или запланированную директиву.") if rec.planned_start_at and now < rec.planned_start_at: raise UserError("Начать нельзя раньше планируемой даты начала.") rec._set_stage("in_progress") return True def action_pause(self): for rec in self: if rec.stage_code != "in_progress": raise UserError("Поставить на паузу можно только директиву в состоянии 'В работе'.") rec._set_stage("paused") return True def action_done(self): for rec in self: if rec.stage_code != "in_progress": raise UserError("Завершить можно только директиву в состоянии 'В работе'.") rec._set_stage("done") tmpl = rec.template_id handler_key = (tmpl.handler_key or "").strip() if tmpl else "" if handler_key: origin_record = rec._get_origin_record() if origin_record: payload = rec._build_payload("done") method = getattr(origin_record, handler_key, None) if not method: raise UserError(f"Handler '{handler_key}' не найден в модели источника ({origin_record._name}).") method(payload) return True @api.model def action_quick_change_stage(self, directive_id, target_code): """ Публичный метод для вызова из JS """ directive = self.browse(directive_id) if directive.exists(): directive._set_stage(target_code) return True return False