# -*- coding: utf-8 -*- from odoo import models, fields, api, _ from odoo.tools.safe_eval import safe_eval from odoo.exceptions import ValidationError, UserError from datetime import datetime, date, time, timedelta import random import string import logging from lxml import etree _logger = logging.getLogger(__name__) CONDITION_CODE_TEMP = """# Available locals: # - time, date, datetime, timedelta: Python libraries. # - env: Odoo Environement. # - model: Model of the record on which the action is triggered. # - obj: Record on which the action is triggered if there is one, otherwise None. # - user, Current user object. # - workflow: Workflow engine. # - syslog : syslog(message), function to log debug information to Odoo logging file or console. # - warning: warning(message), Warning Exception to use with raise. # - To return an user, assign: user = env.user""" PYTHON_CODE_TEMP = """# Available locals: # - time, date, datetime, timedelta: Python libraries. # - env: Odoo Environement. # - model: Model of the record on which the action is triggered. # - obj: Record on which the action is triggered if there is one, otherwise None. # - user, Current user object. # - workflow: Workflow engine. # - syslog : syslog(message), function to log debug information to Odoo logging file or console. # - warning: warning(message), Warning Exception to use with raise. # To return an action, assign: action = {...} """ DEFAULT_BG_COLOR = "rgb(255, 255, 255,1)" DEFAULT_LABEL_COLOR = "rgb(81, 81, 81,1)" START_BG_COLOR = "rgba(120,120,120,1)" START_LABEL_COLOR = "rgba(255,255,255,1)" END_BG_COLOR = "rgba(120,120,120,1)" END_LABEL_COLOR = "rgba(255,255,255,1)" END_BG_COLOR_1 = "rgba(0,255,30,1)" class SoongWorkflowNode(models.Model): _name = "soong.workflow.node" _description = "Soong Workflow Nodes" _order = "sequence" _sql_constraints = [ ( "uniq_node_name", "unique(workflow_id,node_name)", _("Workflow and Technical Name must be unique."), ), ] sequence = fields.Integer(string="Sequence", default=10, help="Arrange node by defining sequence.") name = fields.Char(string="Name", translate=True, help="Enter string name of the node.") node_name = fields.Char(string="Technical Name", help="Generated technical name which used by backend code.", ) workflow_id = fields.Many2one(comodel_name="soong.workflow", string="Workflow Ref.", ondelete="cascade", required=True, ) model_id = fields.Many2one("ir.model", string="Model Ref.", domain="[('state','=','base')]", related="workflow_id.model_id", required=True, ) flow_start = fields.Boolean(string="Flow Start", help="Check it if this node is the starting node.") flow_end = fields.Boolean(string="Flow End", help="Check it if this node is the ending node.") is_visible = fields.Boolean(string="Appear in Statusbar", default=True, help="Control visiability of the node/state in view.", ) out_link_ids = fields.One2many(comodel_name="soong.workflow.link", inverse_name="node_from", string="Outgoing Transitions", ) in_link_ids = fields.One2many(comodel_name="soong.workflow.link", inverse_name="node_to", string="Incoming Transitions", ) field_ids = fields.One2many(comodel_name="soong.workflow.node.field", inverse_name="node_id", string="Fields", ) button_ids = fields.One2many(comodel_name="soong.workflow.node.button", inverse_name="node_id", string="Buttons", ) diagram_position = fields.Char(readonly=True, string="Diagram Position") bg_color = fields.Char(default=DEFAULT_BG_COLOR, string="Background Color") label_color = fields.Char(default=DEFAULT_LABEL_COLOR, string="Label Color") use_custom_colors = fields.Boolean(help="Select colors from the palette manually", string="Use Custom Colors") res_bg_color = fields.Char(compute="_compute_custom_colors", readonly=True, string="Backgroung Color") res_label_color = fields.Char(compute="_compute_custom_colors", readonly=True, string="Label Color") is_archive = fields.Boolean(string="归档") illustrate = fields.Text(string="说明") is_apostille = fields.Boolean(string="加签", default=False) is_end = fields.Boolean(string="结束", default=False) is_confirm = fields.Boolean(string="确认", default=False) department_id = fields.Many2many( 'hr.department', string='部门转入范围', relation="soong_workflow_node_transfer_department_rel", column1="node_id", column2="transfer_department_id" ) transfer_comment_id = fields.Many2one('soong.note', string="转交固定意见") return_comment_id = fields.Many2one('soong.note', string="退回固定意见") confirm_comment_id = fields.Many2one('soong.note', string="确认固定意见") end_comment_id = fields.Many2one('soong.note', string="结束固定意见") approval_concatenation = fields.Selection([('chuan', '串申'), ('bing', '并申')], string="审批形式", default="bing", help='并：任何人均给意见，最后一个完成的人转交到下一节点.' '串：都会收到消息，但任意一个人转交后，即到下一节点，其他人将不再显示，相当于收回.') same_user_skip_type = fields.Selection([ ('none', '不跳过'), ('next', '跳过下一个相同人员节点'), ('prev', '跳过上一个相同人员节点'), ('after', '跳过此节点之后的所有相同人员节点'), ], string='相同人员节点处理', default='none') skip_if_node_id = fields.Many2one( 'soong.workflow.node', string='前置节点', domain="[('workflow_id', '=', workflow_id), ('id', '!=', id)]", help="选择一个位于本节点之前的节点：若它在流转时仍出现在可选节点中（即它可以选择），" "则本节点会被自动跳过；本节点被跳过时，其上配置的「相同人员节点处理」不再触发。", ) is_mandatory_node = fields.Boolean(string="必经节点", default=False) @api.onchange("is_apostille") def onchange_is_apostille(self): for record in self: if record.is_apostille: sequence = max([r.sequence for r in record.button_ids] if record.button_ids else [1]) record.button_ids = [ (0, 0, { "name": _("加签"), # "workflow_id": record.workflow_id.ids[0] if record.workflow_id.ids else None, "node_id": record.id, "sequence": sequence + 1, "action_type": "apostille", "btn_hide": False # 对所有人显示 }), (0, 0, { "name": _("加签确认"), # "workflow_id": record.workflow_id.ids[0] if record.workflow_id.ids else None, "node_id": record.id, "sequence": sequence + 2, "action_type": "apostille", "btn_hide": True, # 仅对加签人员显示 "review_type": "apostille", }) ] else: remove_ids = record.button_ids.filtered(lambda r: r.name in ["加签", "加签确认"]) record.button_ids = [(3, remove_id.id) for remove_id in remove_ids] @api.onchange("is_end") def onchange_is_end(self): for record in self: current_buttons = record.button_ids if record.is_end: record.use_custom_colors = True record.bg_color = END_BG_COLOR_1 record.res_bg_color = END_BG_COLOR_1 end_btns = current_buttons.filtered(lambda r: r.name == _("结束")) archive_btns = current_buttons.filtered(lambda r: r.name == _("归档")) remove_commands = [(3, btn.id) for btn in end_btns + archive_btns] sequence = max([r.sequence for r in current_buttons] if current_buttons else [1]) add_end = (0, 0, { "name": _("结束"), "node_id": record.id, "sequence": sequence + 1, "action_type": "code", "btn_hide": False, "has_comment": True, "review_type": "end", }) add_archive = (0, 0, { "name": _("归档"), "node_id": record.id, "sequence": sequence + 2, "action_type": "code", "btn_hide": False, "has_comment": True, }) record.button_ids = remove_commands + [add_end, add_archive] else: record.use_custom_colors = True record.bg_color = DEFAULT_BG_COLOR record.res_bg_color = DEFAULT_BG_COLOR end_btns = current_buttons.filtered(lambda r: r.name == _("结束")) archive_btns = current_buttons.filtered(lambda r: r.name == _("归档")) record.button_ids = [(3, btn.id) for btn in end_btns + archive_btns] @api.onchange("is_confirm") def onchange_is_confirm(self): for record in self: if record.is_confirm: record.use_custom_colors = True sequence = max([r.sequence for r in record.button_ids] if record.button_ids else [1]) record.button_ids = [ (0, 0, { "name": _("确认"), # "workflow_id": record.workflow_id.ids[0] if record.workflow_id.ids else None, "node_id": record.id, "sequence": sequence + 1, "action_type": "confirm", "btn_hide": False, # 对所有人显示 "has_comment": True, # 对所有人显示 "review_type": "confirm", }), ] else: record.use_custom_colors = True record.bg_color = DEFAULT_BG_COLOR record.res_bg_color = DEFAULT_BG_COLOR remove_ids = record.button_ids.filtered(lambda r: r.name in ["确认"]) record.button_ids = [(3, remove_id.id) for remove_id in remove_ids] def btn_role_sync(self): for record in self: now_btn = record.button_ids.filtered(lambda r: r.name in ["转交"]) for btn in now_btn: other_btn = record.button_ids.filtered(lambda r: r.name not in ["转交", "收回", "加签", "加签确认", '确认', '归档']) other_btn.write({ "btn_hide": btn.btn_hide, "review_type": btn.review_type, "reviewer_id": btn.reviewer_id.id, "reviewer_group_id": btn.reviewer_group_id.id, "users_role_id": btn.users_role_id.id, "users_role_group_ids": btn.users_role_group_ids.ids, }) @api.model_create_multi def create(self, values): records = super().create(values) for record in records: if record.node_name != 'end': if not record.button_ids: if record.node_name == 'start': record.write({"button_ids": [ (0, 0, { "name": _("转交"), "workflow_id": record.workflow_id.id, "sequence": 1}), (0, 0, { "name": _("收回"), "workflow_id": record.workflow_id.id, "sequence": 2, "btn_hide": True, "review_type": "withdraw", "action_type": "code", "code": "obj.action_withdraw()", }) ]}) elif '分发' in record.node_name: record.write({"button_ids": [(0, 0, { "name": _("分发"), "workflow_id": record.workflow_id.id, "sequence": 1, "action_type": "code", "has_comment": True, })]}) else: record.write({"button_ids": [ (0, 0, { "name": _("转交"), "workflow_id": record.workflow_id.id, "sequence": 1, "action_type": "code", "has_comment": True}), (0, 0, { "name": _("收回"), "workflow_id": record.workflow_id.id, "sequence": 2, "btn_hide": True, "review_type": "withdraw", "action_type": "code", "code": "obj.action_withdraw()"}), (0, 0, { "name": _("退回"), "workflow_id": record.workflow_id.id, "sequence": 2, "action_type": "code", "has_comment": True}), (0, 0, { "name": _("确认"), "workflow_id": record.workflow_id.id, "sequence": 2, "action_type": "code", "has_comment": True}) ]}) # 运行时执行加载所有form，并且只读 model_id = self.env['soong.workflow'].search([('id', '=', record.workflow_id.id), ('is_enable', '=', True)]).model_id form_view = self.env["ir.ui.view"].search([("model", "=", model_id.model), ("type", "=", "form")], limit=1) if form_view: xml_tree = etree.fromstring(form_view.arch) field_names = set() for field_node in xml_tree.xpath("//field"): field_name = field_node.get("name") if field_name: field_names.add(field_name) for field in self.env["ir.model.fields"].search( [("model_id", "=", model_id.id), ("name", "in", list(field_names))]): self.field_ids.create({ "model_id": model_id.id, "node_id": record.id, "name": field.id, "readonly": True, }) return records @api.depends("bg_color", "label_color", "use_custom_colors", "flow_start", "flow_end") def _compute_custom_colors(self): for rec in self: if rec.use_custom_colors: rec.res_bg_color = rec.bg_color rec.res_label_color = rec.label_color elif rec.flow_start: rec.res_bg_color = START_BG_COLOR rec.res_label_color = START_LABEL_COLOR elif rec.flow_end: rec.res_bg_color = END_BG_COLOR rec.res_label_color = END_LABEL_COLOR else: rec.res_bg_color = DEFAULT_BG_COLOR rec.res_label_color = DEFAULT_LABEL_COLOR @api.constrains("flow_start", "flow_end") def _check_flow_start_end(self): for rec in self: if rec.flow_start and rec.flow_end: raise UserError(_("Start and end cannot be selected for the node")) @api.onchange("name") def _compute_node_name(self): for rec in self: if rec and rec.name: name = rec.name.lower().strip().replace(" ", "_") rec.node_name = name def btn_load_fields(self): # 加载form视图中的字段 for rec in self: rec.field_ids.unlink() form_view = self.env["ir.ui.view"].search([("model", "=", rec.model_id.model), ("type", "=", "form")], limit=1) if form_view: xml_tree = etree.fromstring(form_view.arch) field_names = set() for field_node in xml_tree.xpath("//field"): field_name = field_node.get("name") if field_name and field_name != "name_title": field_names.add(field_name) for field in self.env["ir.model.fields"].search( [("model_id", "=", rec.model_id.id), ("name", "in", list(field_names))]): rec.field_ids.create({ "model_id": rec.model_id.id, "node_id": rec.id, "name": field.id, }) else: raise UserError('该模型未找到form视图') def all_readonly_fields(self): self.field_ids.write({"readonly": True}) def all_required_fields(self): self.field_ids.write({"required": True}) def all_invisible_fields(self): self.field_ids.write({"invisible": True}) def all_false_fields(self): self.field_ids.write({"invisible": False, "required": False, "readonly": False}) class SoongWorkflowNodeButton(models.Model): _name = "soong.workflow.node.button" _description = "Soong Workflow Node Buttons" _order = "sequence" def _generate_key(self): return "".join( random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(10) ) name = fields.Char( string="Button String", translate=True, help="Enter button string name that will appear in the view.", ) sequence = fields.Integer( string="Sequence", default=10, help="Arrange buttons by defining sequence." ) is_highlight = fields.Boolean(string="Is Highlighted", default=True, help="Control highlighting of the button if needs user attention..", ) # is_high_red = fields.Boolean(string="红色突出") has_icon = fields.Boolean( string="Has Icon", help="Enable it to add icon to the button." ) icon = fields.Char(string="Icon", help="Enter icon name like: fa-print, it's recommended to use FontAwesome Icons.", ) has_comment = fields.Boolean(string="意见/弹窗", default=True) btn_key = fields.Char(string="Button Key", default=_generate_key) btn_hide = fields.Boolean( string="Hide Button if Condition isn't fulfilled", help="If condition is false the button will be hidden.", ) review_type = fields.Selection( string="审批类型", default="user_role", selection=[ ("individual", "Specific user"), ("group", "Any user in a specific group"), ("expression", "Python Expression"), ('user_role', '角色'), # ('invisible', '隐藏条件'), ('apostille', '加签确认'), ('end', "结束"), ('confirm', "确认"), ('withdraw', '收回'), ], ) is_review_invisible = fields.Boolean(string="使用范围", default=False) is_node_automatic = fields.Boolean('是否自动选择所有人员', help='默认否，选择是之后，前节点转交是会自动带出本节点的所有匹配的人员。', default=False) illustrate = fields.Text(string="说明") reviewer_id = fields.Many2one(comodel_name="res.users", string="Reviewer") reviewer_group_id = fields.Many2one(comodel_name="res.groups", string="Reviewer group") users_role_id = fields.Many2one(comodel_name="soong.user.role", string="主角色",) users_role_group_ids = fields.Many2many(comodel_name="soong.user.role.group", string="角色组") reviewer_expression = fields.Text( string="Review Expression", default=CONDITION_CODE_TEMP, help="Write Python code that defines the reviewer. " "The result of executing the expression must be a res.users " "recordset.", ) invisible_expression = fields.Char( string="隐藏条件", default="", ) # todo 找翻译文件改变状态 action_type = fields.Selection( [ ("link", "Trigger Link"), ("link_start_statu", "Trigger Link Start Statu"), ("link_action", "Trigger Link Action"), ("link_action_approve", "Trigger Link Action Approve"), ("link_action_refuse", "Trigger Link Action Refuse"), ("code", "Python Code"), ("action", "Server Action"), ("win_act", "Window Action"), ("apostille", "加签按钮"), ('end', "结束"), ('confirm', "确认"), ], string="Action Type", default="link", help="Choose type of action to be trigger by the button.", ) link_id = fields.Many2one(comodel_name="soong.workflow.link", string="Link") code = fields.Text(string="Python Code", default=PYTHON_CODE_TEMP) server_action_id = fields.Many2one( comodel_name="ir.actions.server", string="Server Action" ) win_act_id = fields.Many2one( comodel_name="ir.actions.act_window", string="Window Action" ) node_id = fields.Many2one( comodel_name="soong.workflow.node", string="Workflow Node Ref.", ondelete="cascade", required=True, ) workflow_id = fields.Many2one( "soong.workflow", string="Workflow Ref.", # required=True, store=True, related="node_id.workflow_id", ondelete="cascade", ) invisible_expression_model = fields.Char(related="workflow_id.model_id.model", store=True) is_automatic = fields.Boolean('发起人部门') is_automatic_company = fields.Boolean('发起人公司', default=True) is_automatic_group = fields.Boolean('发起人集团') is_logged_by = fields.Boolean('上一节点部门') is_logged_company = fields.Boolean('上一节点公司') is_logged_group = fields.Boolean('上一节点集团') is_dep = fields.Boolean('按指定部门') role_id = fields.Many2one(comodel_name="soong.user.role.group", string="角色") specify_field_id = fields.Many2one(comodel_name="ir.model.fields", string="指定字段") is_carbon_copy = fields.Boolean(string="抄送/传阅") department1_ids = fields.Many2many( "hr.department", string="抄送部门", relation="soong_workflow_node_department_rel", column1="node_id", column2="department_id" ) user_role_ids = fields.Many2many("soong.user.role", string="抄送角色", ) cc_users_role_group_ids = fields.Many2many( 'soong.user.role.group', 'soong_workflow_node_button_cc_groups_rel', 'button_id', 'group_id', string='抄送角色组' ) employee_ids = fields.Many2many('hr.employee', string='抄送/传阅人员') @api.onchange('department1_ids', 'user_role_ids', 'cc_users_role_group_ids', 'employee1_ids') def _onchange_sync_employee_ids(self): for record in self: final_employees = self.env['hr.employee'] role_group_roles = record.cc_users_role_group_ids.mapped('user_role_ids') all_target_roles = record.user_role_ids | role_group_roles if all_target_roles: role_employees = self.env['hr.employee'].search([ '|', ('user_role_id', 'in', all_target_roles.ids), ('user_role_ids', 'in', all_target_roles.ids) ]) final_employees |= role_employees if record.department1_ids: dept_employees = self.env['hr.employee'].search([ ('department_id', 'in', record.department1_ids.ids) ]) final_employees |= dept_employees record.employee_ids = final_employees if final_employees else [(5, 0, 0)] def update_user_id(self): button_ids = self.env["soong.workflow.node.button"].browse( self._context.get('active_ids', self._context.get('active_id'))) for button_id in button_ids: button_id.write({'reviewer_id': False}) @api.onchange('is_automatic') def _onchange_is_automatic(self): if self.is_automatic: self.is_automatic_company = False self.is_automatic_group = False self.is_logged_by = False self.is_logged_company = False self.specify_field_id = None self.is_dep = False self.is_logged_group = False @api.onchange('is_automatic_company') def _onchange_is_automatic_company(self): if self.is_automatic_company: self.is_automatic = False self.is_automatic_group = False self.is_logged_by = False self.is_dep = False self.specify_field_id = None self.is_logged_company = False self.is_logged_group = False @api.onchange('is_automatic_group') def _onchange_is_automatic_group(self): if self.is_automatic_group: self.is_automatic = False self.is_automatic_company = False self.is_logged_by = False self.is_dep = False self.specify_field_id = None self.is_logged_company = False self.is_logged_group = False @api.onchange('is_logged_by') def _onchange_is_logged_by(self): if self.is_logged_by: self.is_automatic = False self.is_automatic_company = False self.is_automatic_group = False self.specify_field_id = None self.is_dep = False self.is_logged_company = False self.is_logged_group = False @api.onchange('is_logged_company') def _onchange_is_logged_company(self): if self.is_logged_company: self.is_automatic = False self.is_automatic_company = False self.is_automatic_group = False self.is_logged_by = False self.is_dep = False self.specify_field_id = None self.is_logged_group = False @api.onchange('is_logged_group') def _onchange_is_logged_group(self): if self.is_logged_group: self.is_automatic = False self.is_automatic_company = False self.is_automatic_group = False self.is_logged_by = False self.is_dep = False self.specify_field_id = None self.is_logged_company = False @api.onchange('specify_field_id') def _onchange_specify_field_id(self): if self.specify_field_id: self.is_automatic = False self.is_automatic_company = False self.is_automatic_group = False self.is_logged_by = False self.is_dep = False self.is_logged_company = False self.is_logged_group = False @api.onchange('is_dep') def _onchange_is_dep(self): if self.is_dep: self.is_automatic = False self.is_automatic_company = False self.is_automatic_group = False self.is_logged_by = False self.specify_field_id = None self.is_logged_company = False self.is_logged_group = False @api.onchange('users_role_id', 'users_role_group_ids') def _onchange_role_selection(self): if self.users_role_id and self.users_role_group_ids: self.users_role_id = False self.users_role_group_ids = [(5, 0, 0)] return { 'warning': { 'title': _('输入错误'), 'message': _('只能选择主角色或者角色组中的一个，不能同时选择。') } } @api.onchange("node_id") def change_workflow(self): for rec in self: if isinstance(rec.id, int) and rec.node_id and rec.node_id.workflow_id: rec.workflow_id = rec.node_id.workflow_id.id elif self.env.context.get("default_node_id", 0): model_id = ( self.env["soong.workflow.node"] .sudo() .browse(self.env.context.get("default_node_id")) .model_id.id ) rec.workflow_id = ( self.env["soong.workflow"] .sudo() .search([("model_id", "=", model_id), ('is_enable', '=', True)]) ) @api.constrains("btn_key") def validation(self): for rec in self: # Check if there is no duplicate button key res = self.search_count( [ ("id", "!=", rec.id), ("btn_key", "=", rec.btn_key), ] ) if res: rec.btn_key = self._generate_key() def log(self): cx = self.env.context.copy() or {} query_user_id = self.env.user.id is_admin = self.env.user.has_group('soong_workflow.group_soong_workflow_manager') res_model = cx.get("active_model", False) res_id = cx.get("active_id", 0) record_id = self.env[res_model].search([('id', '=', res_id)]) log = self.env['soong.workflow.log'].search([("model", "=", res_model), ("res_id", "=", res_id)]) employee_user_role_ids = self._get_approvers_for_link(record_id) if is_admin and record_id: if hasattr(record_id, 'user_id') and record_id.user_id: query_user_id = record_id.user_id.id elif hasattr(record_id, 'users_id') and record_id.users_id: query_user_id = record_id.users_id.id if len(log) > 1 and is_admin: current_node_log = self.sudo().env["soong.workflow.log"].search( [("model", "=", res_model), ("res_id", "=", res_id), ("node_id", "=", self.node_id.id)]) else: current_node_log = self.sudo().env["soong.workflow.log"].search( [("model", "=", res_model), ("res_id", "=", res_id), ("done_by", "=", query_user_id), ("node_id", "=", self.node_id.id)]) # 创建或更新当前用户的审批记录 if not current_node_log: self.sudo().env["soong.workflow.log"].create({ "model": res_model, "res_id": res_id, "node_id": self.node_id.id, "workflow_id": self.workflow_id.id, "done_by": query_user_id, "node_btn_name": "0", "node_btn_id": self.id, "lase_click_date": fields.Datetime.now(), "approval_state": "2", # 已完成 "comment": '同意', }) else: current_node_log.write({ "comment": '同意', "approval_state": "2", # 已完成 "lase_click_date": fields.Datetime.now() }) expected_employees = employee_user_role_ids # 查找下一个节点 next_node_id = record_id.workflow_id.node_ids.search([ ('name', '=', self.link_id.node_to.name), ('workflow_id', '=', self.workflow_id.id) ]) completed_logs = self.sudo().env["soong.workflow.log"].search([ ("model", "=", res_model), ("res_id", "=", res_id), ("node_id", "=", self.node_id.id), ("node_btn_name", "=", "4"), ("approval_state", "!=", "2") ]) all_completed = len(completed_logs) == 0 if all_completed: for employee in expected_employees: # 检查该员工在下一节点是否已有记录 next_node_employee_log = self.sudo().env["soong.workflow.log"].search([ ("model", "=", res_model), ("res_id", "=",res_id), ("node_id", "=", next_node_id.id), ('node_btn_name', '!=', '1'), ("done_by", "=", employee.user_id.id) ]) if not next_node_employee_log: # 创建初步记录，状态为"待审批" self.sudo().env["soong.workflow.log"].create({ "model": res_model, "res_id":res_id, "node_id": next_node_id.id, "workflow_id": self.workflow_id.id, "done_by": employee.user_id.id, "node_btn_name": "4", "approval_state": "0", # 待审批 }) workflow_record_id = self.env['soong.workflow.record'].search([('code', '=', record_id.code)]) if employee_user_role_ids: workflow_record_id.users_ids = [(4, employee.user_id.id) for employee in employee_user_role_ids if employee.user_id] workflow_record_id.employee_ids = employee_user_role_ids workflow_record_id.approve_state = "pending" # 待审批 related_record = self.env[res_model].browse(res_id) if hasattr(related_record, 'self_users_ids'): user_ids = [employee.user_id.id for employee in employee_user_role_ids if employee.user_id] related_record.self_users_ids = [(6, 0, user_ids)] if user_ids else [(5,)] record_id.write({'state': self.link_id.node_to.name}) for employee in employee_user_role_ids: action_id = self.env['ir.actions.act_window'].search([('res_model', '=', res_model)])[0] # url = f'https://oa.thtzjt.com/web?login={employee.user_id.login}&oauth_uid={employee.wecom_userid}#id={record_id.id}&cids=2&action={action_id.id}&model={record_id._name}&view_type=form&menu_id=10104' # 生成目标URL（不使用login参数，避免session冲突） target_url = f'https://oa.thtzjt.com/web#id={record_id.id}&cids=2&action={action_id.id}&model={record_id._name}&view_type=form&menu_id=10104' # 使用OAuth登录链接，避免session冲突 url = self.sudo().env["wecom.apps"].generate_oauth_login_url(target_url) self.sudo().env["wecom.apps"].sync_send_message_textcard(category='oa', user_id=employee.wecom_userid, title=record_id._description, description="标题: {}\n".format( record_id.name) + self.env.user.name + ' ' + str( fields.Datetime.now() + timedelta( hours=8)), url=url, btntxt="查看详情") def _get_approvers_for_link(self, record_id=None): node = self.workflow_id.node_ids.search([ ('name', '=', self.link_id.node_to.name), ('workflow_id', '=', self.workflow_id.id) ]) button = False if node.button_ids: # 1. 优先找“分发”按钮 distribute_btn = next((btn for btn in node.button_ids if btn.name == '分发'), None) if distribute_btn: button = distribute_btn else: # 2. 找不到“分发”再找“转交” transfer_btn = next((btn for btn in node.button_ids if btn.name == '转交'), None) if transfer_btn: button = transfer_btn else: # 3. 最后找“结束” button = next((btn for btn in node.button_ids if btn.name == '结束'), None) if not button: return self.env['hr.employee'] link_users_id = button.users_role_id link_users_group_ids = button.users_role_group_ids is_automatic = button.is_automatic is_automatic_company = button.is_automatic_company is_automatic_group = button.is_automatic_group is_logged_by = button.is_logged_by is_logged_company = button.is_logged_company is_logged_group = button.is_logged_group is_dep = button.is_dep is_field = button.specify_field_id if button.reviewer_id: employee = self.env['hr.employee'].search([ ('user_id', '=', button.reviewer_id.id), ], limit=1) return employee # 部门和公司筛选条件 department_domain = [] company_domain = [] user_role_domain = [] if record_id: # 获取当前用户和申请人的相关信息 current_user = self.env.user applicant = None # 获取申请人的用户对象 for field_name in ['user_id', 'users_id']: if hasattr(record_id, field_name): applicant = getattr(record_id, field_name) if applicant: break # 根据勾选的条件设置域条件（只勾选一个） if is_automatic and record_id.department_id: # 按当前申请人自动匹配-本部门（保留部门和公司条件） department_domain = ['|', ('department_id', '=', record_id.department_id.id), ('department_ids', 'in', record_id.department_id.id)] company_domain = [('company_id', '=', applicant.company_id.id)] elif is_automatic_company and applicant and hasattr(applicant, 'company_id'): # 按当前申请人自动匹配-本公司（仅保留公司条件） company_id = applicant.company_id.id company_domain = [('company_id', '=', company_id)] elif is_automatic_group and applicant and hasattr(applicant, 'company_id'): # 按当前申请人自动匹配-本集团（仅保留公司条件） parent_company = applicant.company_id.parent_id if parent_company: group_company_ids = self.env['res.company'].search([('id', 'child_of', parent_company.id)]).ids company_domain = [('company_id', 'in', group_company_ids)] elif is_dep and hasattr(record_id, 'department_id1') and record_id.department_id1: department_id1 = record_id.department_id1.id department_domain = ['|', ('department_id', '=', department_id1), ('department_ids', 'in', department_id1)] user_role_domain = ['|', ('user_role_id', '=', button.role_id.user_role_ids.ids), ('user_role_ids', 'in', button.role_id.user_role_ids.ids)] elif is_logged_by and current_user.department_id: # 按当前登录人自动匹配-本部门（保留部门和公司条件） department_domain = ['|', ('department_id', '=', current_user.department_id.id), ('department_ids', 'in', current_user.department_id.id)] company_domain = [('company_id', '=', current_user.company_id.id)] elif is_logged_company: # 按当前登录人自动匹配-本公司（仅保留公司条件） company_id = current_user.company_id.id company_domain = [('company_id', '=', company_id)] elif is_logged_group: # 按当前登录人自动匹配-本集团（仅保留公司条件） parent_company = current_user.company_id.parent_id if parent_company: group_company_ids = self.env['res.company'].search([('id', 'child_of', parent_company.id)]).ids company_domain = [('company_id', 'in', group_company_ids)] elif is_field: field_name = button.specify_field_id.name # 目前只会有入职 field_value = record_id[field_name] department_domain = ['|', ('department_id', '=', field_value.id), ('department_ids', 'in', field_value.id)] if link_users_id: role_domain = ['|', ('user_role_id', '=', link_users_id.id), ('user_role_ids', 'in', link_users_id.id)] if user_role_domain: role_domain = ['|'] + role_domain + user_role_domain domain = role_domain + company_domain + department_domain return self.env['hr.employee'].sudo().search(domain) elif link_users_group_ids: role_domain = ['|', ('user_role_id', 'in', link_users_group_ids.user_role_ids.ids), ('user_role_ids', 'in', link_users_group_ids.user_role_ids.ids)] if user_role_domain: role_domain = ['|'] + role_domain + user_role_domain domain = role_domain + company_domain + department_domain return self.env['hr.employee'].sudo().search(domain) # 特殊分发按钮 if '分发' in link.name and record_id and hasattr(record_id, 'users_id'): return record_id.users_id.employee_id return self.env['hr.employee'] def run(self): for rec in self: # Check Condition Before Executing Action result = False cx = self.env.context.copy() or {} locals_dict = { "env": self.env, "model": self.env[cx.get("active_model", False)], "obj": self.env[cx.get("active_model", False)].browse(cx.get("active_id", 0)), "user": self.env.user, "datetime": datetime, "time": time, "date": date, "timedelta": timedelta, "workflow": self.env["soong.workflow"], "warning": self.warning, "syslog": self.syslog, } if rec.btn_hide: if rec.review_type == "individual": if rec.reviewer_id == self.env.user: result = True elif rec.review_type == "group": if self.env.user in self.reviewer_group_id.users: result = True else: try: safe_eval( rec.reviewer_expression, locals_dict=locals_dict, mode="exec", nocopy=True, ) user = "user" in locals_dict and locals_dict["user"] or False if user and self.env.user in user: result = True except ValidationError as ex: raise ex except SyntaxError as ex: raise UserError( _( "Wrong python code defined.\n\nError: %s\nLine: %s, Column: %s\n\n%s" % ( ex.args[0], ex.args[1][1], ex.args[1][2], ex.args[1][3], ) ) ) else: result = True if result: # Run Proper Action if rec.action_type == 'confirm': rec.action_type = 'code' func = getattr(self, "_run_%s" % rec.action_type) func_result = func() try: user_ids = [] button_ids = rec.link_id.node_to.button_ids for button_id in button_ids: # 根据每个按钮状态判断当前应该的查看人员 if button_id.review_type == "individual": user_ids.append(button_id.reviewer_id.id) elif button_id.review_type == "group": user_ids += button_id.reviewer_group_id.users.ids elif button_id.review_type == "user_role": user_role_id = self.env['res.users'].search( [('employee_id.user_role_id', '=', button_id.users_role_id.id)]) user_ids += user_role_id.ids # # 发送信息提醒 # partner_ids = self.sudo().env['res.users'].search([('id', 'in', user_ids)]).mapped('partner_id').ids # self.sudo().with_context(lang=self.env.lang)._message_auto_subscribe_notify(partner_ids, 'soong_workflow.message_yuthon_workflow_btn_approval_data') except Exception as e: pass return func_result def _run_win_act(self): # Variables cx = self.env.context.copy() or {} win_act_obj = self.env["ir.actions.act_window"] # Run Window Action for rec in self: action = win_act_obj.with_context(cx).browse(rec.win_act_id.id).read()[0] action["context"] = cx return action return False def _run_action(self): # Variables srv_act_obj = self.env["ir.actions.server"] # Run Server Action for rec in self: srv_act_rec = srv_act_obj.browse(rec.server_action_id.id) return srv_act_rec.run() def _run_code(self): # Variables cx = self.env.context.copy() or {} locals_dict = { "env": self.env, "model": self.env[cx.get("active_model", False)], "obj": self.env[cx.get("active_model", False)].browse( cx.get("active_id", 0) ), "user": self.env.user, "datetime": datetime, "time": time, "date": date, "timedelta": timedelta, "workflow": self.env["soong.workflow"], "warning": self.warning, "syslog": self.syslog, } # Run Code for rec in self: try: safe_eval(rec.code, locals_dict=locals_dict, mode="exec", nocopy=True) action = "action" in locals_dict and locals_dict["action"] or False if action: return action except Warning as ex: raise ex except SyntaxError as ex: raise UserError( _( "Wrong python code defined.\n\nError: %s\nLine: %s, Column: %s\n\n%s" % (ex.args[0], ex.args[1][1], ex.args[1][2], ex.args[1][3]) ) ) return True def _run_link(self): for rec in self: # Check Condition Before Executing Action result = False cx = self.env.context.copy() or {} locals_dict = { "env": self.env, "model": self.env[cx.get("active_model", False)], "obj": self.env[cx.get("active_model", False)].browse( cx.get("active_id", 0) ), "user": self.env.user, "datetime": datetime, "time": time, "date": date, "timedelta": timedelta, "workflow": self.env["soong.workflow"], "warning": self.warning, "syslog": self.syslog, } try: safe_eval( rec.link_id.condition_code, locals_dict=locals_dict, mode="exec", nocopy=True, ) result = "result" in locals_dict and locals_dict["result"] or False except ValidationError as ex: raise ex except SyntaxError as ex: raise UserError( _( "Wrong python code defined.\n\nError: %s\nLine: %s, Column: %s\n\n%s" % (ex.args[0], ex.args[1][1], ex.args[1][2], ex.args[1][3]) ) ) if result: # Trigger link function return rec.link_id.trigger_link() def _run_link_start_statu(self): for rec in self: # Check Condition Before Executing Action result = False cx = self.env.context.copy() or {} locals_dict = { "env": self.env, "model": self.env[cx.get("active_model", False)], "obj": self.env[cx.get("active_model", False)].browse( cx.get("active_id", 0) ), "user": self.env.user, "datetime": datetime, "time": time, "date": date, "timedelta": timedelta, "workflow": self.env["soong.workflow"], "warning": self.warning, "syslog": self.syslog, } try: safe_eval( rec.link_id.condition_code, locals_dict=locals_dict, mode="exec", nocopy=True, ) result = "result" in locals_dict and locals_dict["result"] or False except ValidationError as ex: raise ex except SyntaxError as ex: raise UserError( _( "Wrong python code defined.\n\nError: %s\nLine: %s, Column: %s\n\n%s" % (ex.args[0], ex.args[1][1], ex.args[1][2], ex.args[1][3]) ) ) if result: # todo 更改业务数据的状态，成品不建议这样玩 try: active_id = cx.get("active_id", 0) active_model = cx.get("active_model", False) obj = self.env[active_model].browse(active_id) if obj.exists(): obj.write({'statu': 'approval'}) except Exception as e: pass # Trigger link function return rec.link_id.trigger_link() def _run_link_action(self): for rec in self: # Check Condition Before Executing Action result = False cx = self.env.context.copy() or {} locals_dict = { "env": self.env, "model": self.env[cx.get("active_model", False)], "obj": self.env[cx.get("active_model", False)].browse( cx.get("active_id", 0) ), "user": self.env.user, "datetime": datetime, "time": time, "date": date, "timedelta": timedelta, "workflow": self.env["soong.workflow"], "warning": self.warning, "syslog": self.syslog, } try: safe_eval( rec.link_id.condition_code, locals_dict=locals_dict, mode="exec", nocopy=True, ) result = "result" in locals_dict and locals_dict["result"] or False except ValidationError as ex: raise ex except SyntaxError as ex: raise UserError( _( "Wrong python code defined.\n\nError: %s\nLine: %s, Column: %s\n\n%s" % (ex.args[0], ex.args[1][1], ex.args[1][2], ex.args[1][3]) ) ) if result: win_act_obj = self.env["ir.actions.act_window"] # Run Window Action action = win_act_obj.with_context(cx).browse( self.env.ref('soong_workflow.soong_approval_notice_wizard_action').id).read()[0] action["context"] = cx action["context"].update({"default_approval_notice_state": '0'}) # Trigger link function return action def _run_link_action_approve(self): for rec in self: # Check Condition Before Executing Action result = False cx = self.env.context.copy() or {} locals_dict = { "env": self.env, "model": self.env[cx.get("active_model", False)], "obj": self.env[cx.get("active_model", False)].browse( cx.get("active_id", 0) ), "user": self.env.user, "datetime": datetime, "time": time, "date": date, "timedelta": timedelta, "workflow": self.env["soong.workflow"], "warning": self.warning, "syslog": self.syslog, } try: safe_eval( rec.link_id.condition_code, locals_dict=locals_dict, mode="exec", nocopy=True, ) result = "result" in locals_dict and locals_dict["result"] or False except ValidationError as ex: raise ex except SyntaxError as ex: raise UserError( _( "Wrong python code defined.\n\nError: %s\nLine: %s, Column: %s\n\n%s" % (ex.args[0], ex.args[1][1], ex.args[1][2], ex.args[1][3]) ) ) if result: win_act_obj = self.env["ir.actions.act_window"] # Run Window Action action = win_act_obj.with_context(cx).browse( self.env.ref('soong_workflow.soong_approval_approve_notice_wizard_action').id).read()[0] action["context"] = cx action["context"].update({"default_approval_notice_state": '1'}) # Trigger link function return action def _run_link_action_refuse(self): for rec in self: # Check Condition Before Executing Action result = False cx = self.env.context.copy() or {} locals_dict = { "env": self.env, "model": self.env[cx.get("active_model", False)], "obj": self.env[cx.get("active_model", False)].browse( cx.get("active_id", 0) ), "user": self.env.user, "datetime": datetime, "time": time, "date": date, "timedelta": timedelta, "workflow": self.env["soong.workflow"], "warning": self.warning, "syslog": self.syslog, } try: safe_eval( rec.link_id.condition_code, locals_dict=locals_dict, mode="exec", nocopy=True, ) result = "result" in locals_dict and locals_dict["result"] or False except ValidationError as ex: raise ex except SyntaxError as ex: raise UserError( _( "Wrong python code defined.\n\nError: %s\nLine: %s, Column: %s\n\n%s" % (ex.args[0], ex.args[1][1], ex.args[1][2], ex.args[1][3]) ) ) if result: win_act_obj = self.env["ir.actions.act_window"] # Run Window Action action = win_act_obj.with_context(cx).browse( self.env.ref('soong_workflow.soong_approval_refuse_notice_wizard_action').id).read()[0] action["context"] = cx action["context"].update({"default_approval_notice_state": '2'}) # Trigger link function return action def warning(self, msg): if not isinstance(msg, str): msg = str(msg) raise UserError(msg) def syslog(self, msg): if not isinstance(msg, str): msg = str(msg) _logger.info(msg) class SoongWorkflowNodeField(models.Model): _name = "soong.workflow.node.field" _description = "Soong Workflow Node Fields" name = fields.Many2one(comodel_name="ir.model.fields", string="Field") model_id = fields.Many2one( comodel_name="ir.model", string="Model", domain="[('state','=','base')]" ) readonly = fields.Boolean(string="Readonly") required = fields.Boolean(string="Required") invisible = fields.Boolean(string="Invisible") group_ids = fields.Many2many(comodel_name="res.groups", string="Groups") user_ids = fields.Many2many(comodel_name="res.users", string="Users") node_id = fields.Many2one( comodel_name="soong.workflow.node", string="Node Ref.", ondelete="cascade", required=True, )