# -*- coding: utf-8 -*- import logging import datetime from odoo import models, fields, api, _, modules from odoo.exceptions import ValidationError, UserError _logger = logging.getLogger(__name__) MODEL_DOMAIN = """[ ('is_workflow', '=', True), ('state', '=', 'base'), ('transient', '=', False), '!', '|', '|', '|', '|', '|', '|', '|', ('model', '=ilike', 'res.%'), ('model', '=ilike', 'ir.%'), ('model', '=ilike', 'soong.workflow%'), ('model', '=ilike', 'bus.%'), ('model', '=ilike', 'base.%'), ('model', '=ilike', 'base_%'), ('model', '=', 'base'), ('model', '=', '_unknown'), ]""" class SoongWorkflow(models.Model): _name = "soong.workflow" _description = "Soong Workflow" _order = 'sequence asc' _sql_constraints = [ ("uniq_name", "unique(name)", _("Workflow name must be unique.")), # ("uniq_model", "unique(model_id)", _("Model must be unique.")), ] name = fields.Char(string="Name", help="Give workflow a name.") sequence = fields.Integer(string='序号') model_id = fields.Many2one("ir.model", string="Model", domain=MODEL_DOMAIN, help="Enter business model you would like to modify its workflow.", ) node_ids = fields.One2many(comodel_name="soong.workflow.node", inverse_name="workflow_id", string="Nodes") link_ids = fields.One2many(comodel_name="soong.workflow.link", inverse_name="workflow_id", string="Links") remove_default_btn_until_end = fields.Boolean(string="覆盖表头按钮", default=False) current_stage_highlight_color = fields.Char(string="当前阶段颜色", default="rgb(0, 255, 255)", help="The color to highlight the current stage on flow", ) start_date = fields.Date(string="生效日期", copy=False, default=fields.Date.today()) text = fields.Text(string="描述") end_date = fields.Date(string="失效日期", copy=False, default=fields.Date.from_string('2099-01-01')) is_enable = fields.Boolean(string="是否启用", compute="compute_is_enable", store=True) is_enable_return_button = fields.Boolean(string="是否启用收回", default=True) is_enable_transmitted_button = fields.Boolean(string="是否启用转交", default=True) approvals_group_id = fields.Many2one('soong.approval.group', string="审批分组") department_role_ids = fields.Many2many('hr.department', string="授权部门范围", store=True) user_role_ids = fields.Many2many('soong.user.role', 'soong_workflow_user_role_rel', 'workflow_id', 'user_role_id', string="授权范围", store=True) is_user_department_role = fields.Boolean(string="是否有权限查看", store=False, search="_search_is_user_department_role") archive_type = fields.Selection([('applicant', '当前申请人'), ('employee', '人员'), ('role', '角色')], string="归档类型", default="applicant") employee_ids = fields.Many2many('hr.employee', string="人员") role_ids = fields.Many2many('soong.user.role', 'soong_workflow_role_rel', 'workflow_id', 'user_role_id', string="角色") def _search_is_user_department_role(self, operator, value): id_list = [] if self.env.user.has_group("soong_workflow.group_soong_workflow_user"): # 查询主权限的 for user_role_id in self.env.user.employee_id.user_role_id.ids: child_list = [user_role_id] self.search_user_role_child_ids(user_role_id, child_list) print(child_list) for child_id in child_list: id_list += self.sudo().with_context(active_test=False).search([("user_role_ids", "in", [child_id])]).ids # 查询辅助权限 for user_role_id in self.env.user.employee_id.user_role_ids.ids: child_list = [user_role_id] self.search_user_role_child_ids(user_role_id, child_list) print(child_list) for child_id in child_list: id_list += self.sudo().with_context(active_test=False).search([("user_role_ids", "in", [child_id])]).ids # 查询部门的权限 for department_id in self.env.user.employee_id.department_id.ids: child_list = [department_id] self.search_user_department_child_ids(department_id, child_list) print(child_list) for child_id in child_list: id_list += self.sudo().with_context(active_test=False).search([("department_role_ids", "in", [child_id])]).ids # 查询辅助部门的权限 for department_id in self.env.user.employee_id.department_ids.ids: child_list = [department_id] self.search_user_department_child_ids(department_id, child_list) print(child_list) for child_id in child_list: id_list += self.sudo().with_context(active_test=False).search([("department_role_ids", "in", [child_id])]).ids if self.env.user.has_group("soong_workflow.group_soong_workflow_manager"): id_list += self.sudo().with_context(active_test=False).search([]).ids return [('id', "in", id_list)] def search_user_role_child_ids(self, parent_id, id_list): child_ids = self.env["soong.user.role"].sudo().search([("id", "=", parent_id)]).parent_id.ids for child_id in child_ids: id_list.append(child_id) self.search_user_role_child_ids(child_id, id_list) def search_user_department_child_ids(self, parent_id, id_list): child_ids = self.env["hr.department"].sudo().search([("id", "=", parent_id)]).parent_id.ids for child_id in child_ids: id_list.append(child_id) self.search_user_department_child_ids(child_id, id_list) def copy(self, default=None): workflow_id = self.search([('model_id', '=', self.model_id.id)]) workflow_version = len(workflow_id) + 1 default = dict(default or {}) default.update({ 'name': self.name + '-v' + str(workflow_version), 'model_id': False, # 'remove_default_attrs_mod': self.remove_default_attrs_mod, 'current_stage_highlight_color': self.current_stage_highlight_color, }) res = super(SoongWorkflow, self).copy(default) node_dict = dict() button_dict = dict() # 复制节点 for node_id in self.node_ids: new_noode_id = node_id.copy({ 'workflow_id': res.id }) node_dict[node_id.id] = new_noode_id.id # 复制节点-按钮 for button_id in node_id.button_ids: new_button_id = button_id.copy({ 'node_id': new_noode_id.id, 'link_id': False }) if button_id.link_id: if button_id.link_id.id in button_dict: button_dict[button_id.link_id.id] |= new_button_id else: button_dict[button_id.link_id.id] = new_button_id # 复制节点-字段 for field_id in node_id.field_ids: field_id.copy({ 'node_id': new_noode_id.id, }) # 复制链接 for link_id in self.link_ids: new_link_id = link_id.copy({ 'workflow_id': res.id, 'node_from': node_dict[link_id.node_from.id], 'node_to': node_dict[link_id.node_to.id] }) if link_id.id in button_dict: for new_button_id in button_dict[link_id.id]: new_button_id.link_id = new_link_id.id return res @api.depends('start_date', 'end_date') def compute_is_enable(self): for rec in self: if rec.start_date and rec.end_date and self.name: if rec.start_date <= fields.Date.today() < rec.end_date: is_enable_ids = self.search([('model_id', '=', self.model_id.id), ('is_enable', '=', True)]) if len(is_enable_ids) <= 1: rec.is_enable = True else: raise UserError('每个模型只能生效一个流程') else: rec.is_enable = False else: rec.is_enable = False @api.constrains("node_ids") def validate_nodes(self): # Objects wkf_node_obj = self.env["soong.workflow.node"] for rec in self: # Must have one flow start node res = rec.node_ids.search_count( [ ("workflow_id", "=", rec.id), ("flow_start", "=", True), ] ) if res > 1: raise ValidationError(_("Workflow must have only one start node.")) # Nodes' sequence must be unique per workflow for node in rec.node_ids: res = wkf_node_obj.search_count( [ ("id", "!=", node.id), ("workflow_id", "=", rec.id), ("sequence", "=", node.sequence), ] ) if res: raise ValidationError( _("Nodes' sequence must be unique per workflow.") ) def action_workflow_diagram(self): self.ensure_one() action = self.env.ref("soong_workflow.action_soong_workflow").read()[0] action["name"] = _("Workflow: %(type_name)s") % {"type_name": self.display_name} action["context"] = {"default_workflow_id": self.id} action["res_id"] = self.id action["views"] = [(False, "diagram_plus"), (False, "form")] return action def create_request(self): self.ensure_one() return { "type": "ir.actions.act_window", "view_mode": "form", "res_model": self.model_id.model, "views": [[False, "form"]], } def get_soong_workflow_time(self): for record in self.search([]): if record.end_date: if record.end_date < fields.Date.today(): record.is_enable = False