# -*- coding: utf-8 -*- from odoo import models, fields, api, _ from odoo.tools.safe_eval import safe_eval from odoo.tools.misc import frozendict from odoo.exceptions import UserError from datetime import datetime, date, time, timedelta from odoo.exceptions import UserError, ValidationError from lxml import etree import logging import ast _logger = logging.getLogger(__name__) class SoongWorkflowMixin(models.AbstractModel): _name = "soong.workflow.mixin" _description = "Soong Workflow Mixin" @api.model def read(self, fields=None, load='_classic_read'): for rec in self: if rec.log_ids: current_user = self.env.user target_logs = rec.log_ids.filtered(lambda l: l.done_by.id == current_user.id and l.approval_state == '0') if target_logs: target_logs.write({'approval_state': '1'}) return super(SoongWorkflowMixin, self).read(fields=fields, load=load) def get_edit_button_state(self): """供前端判断编辑按钮是否显示，避免触发 read() 重写里的日志副作用。""" self.ensure_one() # 没有 self_users_ids / work_end 字段的模型不做限制 if not hasattr(self, 'self_users_ids') or not hasattr(self, 'work_end'): return {'show_edit': True} is_current_user = self.env.user.id in self.self_users_ids.ids # is_modify=True 且当前用户是流程创建人时也允许编辑 is_modify_creator = False if hasattr(self, 'is_modify') and self.is_modify: workflow_record = self.env["soong.workflow.record"].search( [("res_model", "=", self._name), ("res_id", "=", self.id), ('reclaim_state', '=', False)], limit=1) if workflow_record and workflow_record.create_uid.id == self.env.user.id: is_modify_creator = True show_edit = (is_current_user and not self.work_end) or is_modify_creator return { 'show_edit': show_edit, 'self_users_ids': self.self_users_ids.ids, 'work_end': self.work_end, 'is_modify': hasattr(self, 'is_modify') and self.is_modify, 'is_modify_creator': is_modify_creator, } # @api.model # def write(self, vals): # res = super(SoongWorkflowMixin, self).write(vals) # workflow_records = self.env['soong.workflow.record'].search([('name', '=', self.name)]) # for record in workflow_records: # for field, value in vals.items(): # if hasattr(record, field): # setattr(record, field, value) # return res def action_is_modify(self): self.is_modify = False @api.model def _compute_default_state(self): # Objects wkf_obj = self.env["soong.workflow"] rec_model = self._name # Get Flow Start wkf_recs = wkf_obj.search([("model_id", "=", rec_model), ('is_enable', '=', True)]) for wkf_rec in wkf_recs: for node in wkf_rec.node_ids: if node.flow_start: return node.node_name @api.model def _load_state(self): # Objects wkf_obj = self.env["soong.workflow"] # Variables rec_model = self._name ret = [] # Get Status wkf_recs = wkf_obj.search([("model_id", "=", rec_model), ('is_enable', '=', True)]) for rec in wkf_recs: for node in rec.node_ids: ret.append((node.node_name, node.name)) return ret state = fields.Selection( selection="_load_state", string="Status", default=_compute_default_state, track_visibility="onchange", store=True ) is_modify = fields.Boolean(string="可修改") workflow_id = fields.Many2one( comodel_name="soong.workflow", string="Workflow Ref.", compute="_compute_workflow_id", ) approval_users_ids = fields.One2many("soong.approval.users", "res_id", domain=lambda self: [('res_model', '=', self._name)], auto_join=True, string="审批明细") approval_sign_ids = fields.Many2many('res.users', store=False, compute="_compute_approval_sign_ids", string="加签人员") is_import = fields.Boolean(string='是否导入') is_back = fields.Boolean(string='是否允许收回', compute='_compute_is_back', store=False,) is_active = fields.Boolean(string='是否允许归档', compute='_compute_is_active', store=False,) is_confirm = fields.Boolean(string="未确认", compute='_compute_is_confirm', store=False,) is_end = fields.Boolean(string="未结束", compute='_compute_is_confirm', store=False,) is_done = fields.Boolean(string="已办", compute='_compute_is_done', store=False,) is_send = fields.Boolean(string="是否分发", compute='_compute_is_send', store=False,) is_show_add_sign = fields.Boolean(string="显示加签按钮", compute='_compute_is_show_add_sign') is_show_transfer = fields.Boolean(string="显示加签确认按钮", compute='_compute_is_show_transfer') approval_result = fields.Selection([('pass', '通过'), ('fail', '不通过')], string="审批结果") def _compute_is_show_add_sign(self): for record in self: record.is_show_add_sign = True domain_hide_01 = [ ('workflow_id', '=', record.workflow_id.id), ('res_id', '=', record.id), ('name', '=', '加签'), ('approval_state', 'in', ['0', '1']), ] has_01 = record.log_ids.search(domain_hide_01) domain_hide_done = [ ('workflow_id', '=', record.workflow_id.id), ('res_id', '=', record.id), ('name', '=', '加签'), ('approval_state', '=', '2'), ('done_by', '=', self.env.user.id), ] has_done = record.log_ids.search(domain_hide_done) has_user_approved = False user_logs = record.log_ids.filtered( lambda l: l.done_by == self.env.user and l.node_btn_name != '1' ) if user_logs: has_user_approved = all(log.approval_state == '2' for log in user_logs) # 如果有日志，取【第一个节点】的经办人 == 当前登录人 → 隐藏 （归档待办的除外） has_first_user = False if record.log_ids: archive_self_log = record.log_ids.filtered( lambda l: l.name == '归档' and l.done_by == self.env.user and l.approval_state in ['0', '1']) if not archive_self_log: first_log = record.log_ids.sorted('sequence')[:1] if not self.state: has_first_user = True else: if first_log and first_log.done_by == self.env.user and self.state[0] == first_log.name: has_first_user = True # ：判断当前登录人是否在 log_ids2 中 has_log_ids2_user = False if record.log_ids2: has_log_ids2_user = self.env.user.id in record.log_ids2.mapped('done_by.id') # 任意一个条件满足 → 隐藏 if has_01 or has_done or has_first_user or has_log_ids2_user or has_user_approved: record.is_show_add_sign = False def _compute_is_show_transfer(self): for record in self: record.is_show_transfer = False domain_has_sign = [ ('workflow_id', '=', record.workflow_id.id), ('res_id', '=', record.id), ('name', '=', '加签'), ('approval_state', 'in', ['0', '1']), ('done_by', '=', self.env.user.id), ] has_sign_ok = record.log_ids.search(domain_has_sign) if has_sign_ok: record.is_show_transfer = True def _compute_is_send(self): for record in self: record.is_send = False log_ids_domain = [ ('done_by', '=', self.env.user.id), ('workflow_id', '=', record.workflow_id.id), ('res_id', '=', record.id), ('approval_state', 'in', ['0', '1']), ('name', '=', '分发') ] lg1 = record.log_ids.search(log_ids_domain) if lg1: record.is_send = True def _compute_is_back(self): for record in self: record.is_back = False current_uid = self.env.uid user_matched_logs = record.log_ids.filtered( lambda log: log.done_by.id == current_uid ).sorted(key=lambda log: log.create_date) if not user_matched_logs: continue # 取当前登录人对应日志的最后一条（最新的） current_user_last_log = user_matched_logs[-1] if current_user_last_log.name == '加签': record.is_back = False continue all_sorted_logs = record.log_ids.sorted(key=lambda log: log.create_date) all_sorted_list = record.log_ids.sorted(key=lambda log: log.create_date).ids # 找到当前日志在所有日志中的索引 try: current_log_index = all_sorted_list.index(current_user_last_log.id) except ValueError: current_log_index = -1 if current_log_index != -1 and (current_log_index + 1) < len(all_sorted_list): next_log = all_sorted_logs[current_log_index + 1] if next_log.approval_state == "0": record.is_back = True same_node_logs = record.log_ids.filtered(lambda log: log.name == next_log.name) if next_log.name == '加签': if same_node_logs and any(log.approval_state in ['1', '2'] for log in same_node_logs): record.is_back = False if record._name in ['yuthon.official.incoming.normal', 'yuthon.sub.enterprises']: current_user_in_logs2 = any(log.done_by.id == self.env.uid for log in record.log_ids2) organizer_user_in_logs = any(log.done_by.id == self.env.user.id for log in record.log_ids.filtered(lambda l: l.name == '承办')) all_handling_logs_approval_0 = all(log.approval_state == '0' for log in record.log_ids.filtered(lambda l: l.name == '承办')) all_logs2_approval_0 = all(log.approval_state == '0' for log in record.log_ids2) if all_logs2_approval_0 and all_handling_logs_approval_0 and not current_user_in_logs2 and not organizer_user_in_logs: record.is_back = True else: record.is_back = False if self.is_show_transfer == True: record.is_back = False def _compute_is_active(self): _logger.info("_compute_is_active----" + str(fields.datetime.now())) current_user_id = self.env.user.id self.is_active = False if self.log_ids: if self.log_ids[-1].name == '归档' and self.log_ids[-1].done_by.id == current_user_id and self.log_ids[-1].approval_state in ['0', '1']: self.is_active = True _logger.info("_compute_is_active----" + str(fields.datetime.now())) def _compute_is_done(self): _logger.info("_compute_is_done循环开始----" + str(fields.datetime.now())) for record in self: record.is_done = False log_ids_domain = [ ('done_by', '=', self.env.user.id), ('workflow_id', '=', record.workflow_id.id), ('res_id', '=', record.id), ] lg1 = record.log_ids.search(log_ids_domain) if lg1: all_approved = all(log.approval_state == '2' for log in lg1) if all_approved: record.is_done = True workflow_record_id = self.env["soong.workflow.record"].search( [("res_model", "=", self._name), ("res_id", "=", self.id), ('reclaim_state', '=', False)]) if workflow_record_id: if workflow_record_id.approve_state == 'end': record.is_done = True _logger.info("_compute_is_done循环结束----" + str(fields.datetime.now())) def _compute_is_confirm(self): for record in self: record.is_confirm = False record.is_end = False target_models = ['yuthon.official.incoming.normal', 'yuthon.sub.enterprises'] if record._name in target_models: chairman_logs = record.log_ids.filtered( lambda log: '董事长' in (log.name or '') and log.approval_state in ['0', '1'] and log.done_by.id == self.env.user.id ) if chairman_logs: record.is_end = True continue log_ids_domain = [ ('done_by', '=', self.env.user.id), ('workflow_id', '=', record.workflow_id.id), ('res_id', '=', record.id), ('approval_state', 'in', ['0', '1']), ('name', '=', '承办') ] log_ids2_domain = [ ('done_by', '=', self.env.user.id), ('workflow_id', '=', record.workflow_id.id), ('res_id', '=', record.id), ('approval_state', 'in', ['0', '1']) ] lg1 = record.log_ids.search(log_ids_domain) lg2 = record.log_ids2.search(log_ids2_domain) lg3 = record.log_ids.search(log_ids2_domain) if lg1 or lg2: record.is_confirm = True if not record.is_confirm: record.is_end = True if not lg2 and not lg3: record.is_end = False if record.log_ids: if record.log_ids[-1].name == '归档': record.is_end = False @api.depends("node_id", "log_ids") def _compute_approval_sign_ids(self): for record in self: record.approval_sign_ids = record.log_ids.filtered(lambda r: r.node_btn_name == "2" and r.node_id == record.node_id).done_by.ids @api.depends("state") def _compute_workflow_id(self): # Objects wkf_obj = self.env["soong.workflow"] for rec in self: # Variables rec_model = rec._name # Get Workflow rec.workflow_id = wkf_obj.search([("model_id", "=", rec_model), ('is_enable', '=', True)]) node_id = fields.Many2one( comodel_name="soong.workflow.node", string="Workflow Node Ref.", compute="_compute_node_id", ) @api.depends("state") def _compute_node_id(self): # Objects wkf_obj = self.env["soong.workflow"] for rec in self: # Variables rec_model = rec._name # Get Workflow Node rec.node_id = wkf_obj.search( [("model_id", "=", rec_model), ('is_enable', '=', True)] ).node_ids.filtered(lambda n: n.node_name == rec.state) log_ids = fields.One2many(comodel_name="soong.workflow.log", inverse_name="res_id", string=" ", domain=lambda self: [("model", "=", self._name)], auto_join=True) log_ids2 = fields.One2many(comodel_name="soong.workflow.log2", inverse_name="res_id", string=" ", domain=lambda self: [("model", "=", self._name)], auto_join=True) # log_ids3 = fields.One2many(comodel_name="soong.workflow.log3", inverse_name="res_id", string=" ", # domain=lambda self: [("model", "=", self._name)], auto_join=True) soong_record_ids = fields.One2many(comodel_name="soong.workflow.line", inverse_name="res_id", string=" ",) project_ids = fields.One2many(comodel_name="project.project.line", inverse_name="res_id", string=" ",) def get_logs_for_print(self, applicant_first=False): """ 返回用于打印的审批日志列表： - 按与界面相同的顺序（sequence 升序） - 保留所有审批记录（包括重复节点，如申请人多次申请） - 可选：申请人置顶 """ self.ensure_one() ordered = self.log_ids.sorted('sequence') if applicant_first: applicants = [l for l in ordered if (l.name or '') == '申请人'] others = [l for l in ordered if (l.name or '') != '申请人'] ordered = applicants + others return ordered # @api.model_create_multi # def create(self, vals_list): # res = super().create(vals_list) # target_role = self.env['soong.user.role'].search( # [('name', '=', '考勤校对'), ('company_id', '=', res.env.user.employee_id.company_id.id)], # limit=1 # ) # target_role_id = target_role.id if target_role else None # # for record in res: # button_ids = record.sudo().workflow_id.node_ids.button_ids.search( # [('workflow_id', '=', record.workflow_id.id)]) # # 获取当前用户的主角色和辅助角色ID列表 # employee = record.env.user.employee_id # current_user_role_ids = [] # if employee.user_role_id: # current_user_role_ids.append(employee.user_role_id.id) # if employee.user_role_ids: # current_user_role_ids.extend(employee.user_role_ids.ids) # if target_role_id in current_user_role_ids: # continue # for button_id in button_ids: # if button_id.node_id.is_end: # continue # allowed_departments = button_id.node_id.department_id # if allowed_departments and employee.department_id.id not in allowed_departments.ids: # continue # allowed_role_ids = [] # if button_id.users_role_id: # allowed_role_ids.append(button_id.users_role_id.id) # elif button_id.users_role_group_ids: # allowed_role_ids.extend(button_id.users_role_group_ids.user_role_ids.ids) # else: # continue # if current_user_role_ids and any(role_id in allowed_role_ids for role_id in current_user_role_ids): # record.state = button_id.node_id.node_name # return res def _create_update_logs(self, btn_rec): log_obj = self.env["soong.workflow.log"] if btn_rec.name not in ['转交', '分发', '加签', '加签确认', '退回', '收回', '结束', '归档', '确认']: record_obj = log_obj.search( [("model", "=", self._name), ("res_id", "=", self.id), ("node_id", "=", btn_rec.node_id.id), ("workflow_id", "=", btn_rec.workflow_id.id), ("node_btn_id", "=", btn_rec.id), ("node_btn_name", "=", "0") ], order="lase_click_date DESC", limit=1) if not record_obj: record_obj = log_obj.create({ "model": self._name, "res_id": self.id, "node_id": btn_rec.node_id.id, "workflow_id": btn_rec.workflow_id.id, "node_btn_id": btn_rec.id, "done_by": self.env.uid, "node_btn_name": "0", "lase_click_date": fields.Datetime.now(), }) record_obj.write({"node_btn_id": btn_rec.id, "done_by": self.env.uid, "approval_state": "2"}) record_obj = log_obj.search([ ("model", "=", self._name), ("res_id", "=", self.id), ("node_id", "=", btn_rec.node_id.id), ("workflow_id", "=", btn_rec.workflow_id.id), ], order="lase_click_date DESC", limit=1 ) return record_obj def btn_exec_action(self): """ Got invoked when a workflow button is called. :return: button action return value. """ self.ensure_one() # Variables cx = self.env.context.copy() or {} cx.update({"active_id": self.id, "active_ids": self.ids}) wkf_btn_obj = self.env["soong.workflow.node.button"] btn_rec = wkf_btn_obj.search([("btn_key", "=", cx.get("btn_key", False))]) # # 优化弹窗 if btn_rec: log_obj = self._create_update_logs(btn_rec) # # 判断加签 # if btn_rec.name in ["加签"]: # # 弹窗 进行填写 人员 # wizard = self.env.ref("soong_workflow.view_soong_workflow_apostille_wizard") # return { # "name": _("添加加签人员"), # "type": "ir.actions.act_window", # "view_mode": "form", # "res_model": "soong.workflow.apostille.wizard", # "views": [(wizard.id, "form")], # "view_id": wizard.id, # "target": "new", # "context": { # "default_res_id": self.id, # "default_res_model": self._name, # "default_log_ids": log_obj.ids, # "default_node_btn_id": btn_rec.id, # }, # } # if btn_rec.name in ["加签确认"]: # # 弹窗 进行填写 人员 # wizard = self.env.ref("soong_workflow.view_soong_workflow_apostille_ok_wizard") # return { # "name": _("添加加签人员"), # "type": "ir.actions.act_window", # "view_mode": "form", # "res_model": "soong.workflow.apostille.ok.wizard", # "views": [(wizard.id, "form")], # "view_id": wizard.id, # "target": "new", # "context": { # "default_res_id": self.id, # "default_res_model": self._name, # "default_log_ids": log_obj.ids, # "default_node_btn_id": btn_rec.id, # }, # } # if btn_rec.name not in ["加签", "加签确认"]: # # 判断当前是否所有加签都已经结束 # apostille_log_obj = self.sudo().env["soong.workflow.log"].search( # [ # ("model", "=", self._name), # ("res_id", "=", self.id), # ("node_id", "=", btn_rec.node_id.id), # ("workflow_id", "=", btn_rec.workflow_id.id), # ("node_btn_name", "=", "2"), # ("approval_state", "!=", "2") # ] # ) # if apostille_log_obj: # raise UserError("当前加签人员还有没有审批完成的！") # 继续正常流程 # if btn_rec.has_comment: return self.with_context(cx)._add_comment(log_obj, btn_rec) # if btn_rec.name == '收回': # return { # 'name': '收回弹窗', # 'type': 'ir.actions.act_window', # 'view_mode': 'form', # 'res_model': 'soong.bounced.wizard', # "target": "new", # 'context': { # 'default_res_id': self.id, # "default_res_model": self._name, # } # } # else: # btn_rec.with_context(cx).log() # return btn_rec.with_context(cx).run() def _add_comment(self, logs, btn_rec): trans_link_ids = [] wizard = self.env.ref("soong_workflow.view_soong_workflow_comment_wizard") workflow_record_id = self.env["soong.workflow.record"].search([("res_model", "=", self._name), ("res_id", "=", self.id), ('reclaim_state', '=', False)]) # 状态列表 node_list = btn_rec.workflow_id.node_ids.sorted('sequence') if self.name_title != '采购申请单': state = self.state state_list = [i.node_name for i in btn_rec.workflow_id.node_ids.sorted('sequence')] else: if self.state == "purchase": state = "总经理" else: state = self.state state_list = [i.node_name for i in btn_rec.workflow_id.node_ids.sorted('sequence')] + ['purchase'] # 当前节点状态 if self.state == 'draft': state = '询价' self_state = state_list.index(state) # 当前节点 if btn_rec.name not in ['结束', '归档']: current_node = node_list[state_list.index(state)] workflow_record_id = self.env["soong.workflow.record"].search( [("res_model", "=", self._name), ("res_id", "=", self.id)]) self.env['trans.link'].search([('workflow_record_id', '=', workflow_record_id.id)]).unlink() # 退回节点根据日志进行判断 if btn_rec.name == '退回': approved_logs = self.log_ids.filtered(lambda log: log.approval_state == '2') for ret in approved_logs: if not self.env['trans.link'].search([('workflow_record_id', '=', workflow_record_id.id), ('name', '=', ret.name), ('link_type', '=', 'return')]): self.env['trans.link'].create({ 'sequence': ret.sequence, 'workflow_record_id': workflow_record_id.id, 'node_button_id': btn_rec.id, 'name': ret.name, 'link_type': 'return', 'record_id': self.id, }) if btn_rec.name in ['转交', '加签']: required_fields = [ field.name for field in current_node.field_ids if field.required and field.name ] missing_fields = [] if required_fields: for field_name in required_fields: field_value = getattr(self, field_name.name, None) if not field_value: missing_fields.append(field_name.field_description) if missing_fields: raise UserError(f"转交失败！必填字段未填写：{', '.join(missing_fields)}") skipped_node_names = [] # 因没有可用按钮而跳过的节点 node_list = list(btn_rec.workflow_id.node_ids.sorted('sequence')) index = self_state actual_path_nodes = [] #跟踪整个工作流路径，只保留实际会执行的节点 while index < len(node_list): current_node = node_list[index] actual_path_nodes.append(current_node) if not current_node.out_link_ids: index = 20 # 防止死循环 break current_department = self.department_id if btn_rec.link_id in current_node.out_link_ids and len(current_node.out_link_ids) == 1: matched_link = btn_rec.link_id next_node = matched_link.node_to matched_key = next_node.name else: if len(current_node.out_link_ids) > 1: # 筛选与当前节点部门匹配的后续节点 valid_nodes = [] for link in current_node.out_link_ids: if self.name_title in ['汇审意见表', '信息发布申请表',]: category_names = self.information_categories_ids.mapped('name') target_name = link.name.split('->')[-1].strip() if '工会活动' in category_names: if target_name == '工会主席': target_node = link.node_to valid_nodes.append((target_node, link)) else: if target_name == '复核': target_node = link.node_to valid_nodes.append((target_node, link)) if target_name == '承办部门意见': target_node = link.node_to valid_nodes.append((target_node, link)) if '经营业务（财务专用）'in category_names: if target_name == '财务部': target_node = link.node_to valid_nodes.append((target_node, link)) else: target_node = link.node_to if target_node.department_id and any( dept in current_department or current_department in self.get_all_sub_departments(dept) for dept in target_node.department_id ): valid_nodes.append((target_node, link)) continue # ===================== 新增：部门为空 → 走【角色/角色组】匹配 ===================== if not target_node.department_id: if hasattr(self, 'user_id') and self.user_id: applicant_user = self.user_id elif hasattr(self, 'users_id') and self.users_id: applicant_user = self.users_id applicant_roles = set() if applicant_user and applicant_user.employee_id: if applicant_user.employee_id.user_role_id: applicant_roles.add(applicant_user.employee_id.user_role_id.id) if applicant_user.employee_id.user_role_ids: applicant_roles.update(applicant_user.employee_id.user_role_ids.ids) button = None if link.node_to.button_ids: transfer_btn = link.node_to.button_ids.filtered(lambda r: r.name == '转交') if transfer_btn: for btn in transfer_btn: if btn.is_review_invisible: try: domain = eval(btn.invisible_expression) expression_result = self.search(domain) has_self_in_result = self.id in expression_result.ids if has_self_in_result: button = btn break except: continue else: button = btn break if not button: end_btn = link.node_to.button_ids.filtered(lambda r: r.name == '结束') if end_btn: button = end_btn[0] if button: valid_nodes.append((target_node, link)) continue # 尝试父部门匹配 if not valid_nodes and current_department and current_department.parent_id: parent_department = current_department.parent_id for link in current_node.out_link_ids: target_node = link.node_to if target_node.department_id and any( dept in parent_department or parent_department in self.get_all_sub_departments(dept) for dept in target_node.department_id ): valid_nodes.append((target_node, link)) # 如果两种匹配都失败，将所有节点加入有效列表 if not valid_nodes: for link in current_node.out_link_ids: valid_nodes.append((link.node_to, link)) final_valid_nodes = [] for node, link in valid_nodes: transfer_buttons = [btn for btn in node.button_ids if btn.name == '转交'] end_buttons = [btn for btn in node.button_ids if btn.name == '结束'] has_visible_transfer_btn = False has_visible_end_btn = False # 判断转交按钮可见性 if transfer_buttons: for btn in transfer_buttons: if btn.is_review_invisible: try: domain = eval(btn.invisible_expression) expression_result = self.search(domain) if self.id in expression_result.ids: has_visible_transfer_btn = True break except: continue else: has_visible_transfer_btn = True break # 判断结束按钮可见性 if end_buttons: for btn in end_buttons: if btn.is_review_invisible: try: domain = eval(btn.invisible_expression) expression_result = self.search(domain) if self.id in expression_result.ids: has_visible_end_btn = True break except: continue else: has_visible_end_btn = True break # 两个按钮都不可见 → 跳过 if not has_visible_transfer_btn and not has_visible_end_btn: skipped_node_names.append(node.name) _logger.info(f"跳过节点: {node.name}，转交按钮和结束按钮均不可见") else: final_valid_nodes.append((node, link)) # 选择第一个匹配的节点 next_node, matched_link = final_valid_nodes[0] if final_valid_nodes else valid_nodes[0] matched_key = next_node.name else: matched_link = current_node.out_link_ids[0] next_node = matched_link.node_to matched_key = next_node.name transfer_buttons = [ btn for btn in next_node.button_ids if btn.name == '转交' ] end_buttons = [ btn for btn in next_node.button_ids if btn.name == '结束' ] has_visible_transfer_btn = False has_visible_end_btn = False # ========== 第一步：判断转交按钮（transfer_buttons）是否可见 ========== if transfer_buttons: for btn in transfer_buttons: if btn.is_review_invisible: # 解析隐藏表达式，判断当前记录是否在表达式结果中 domain = eval(btn.invisible_expression) expression_result = self.search(domain) has_self_in_result = self.id in expression_result.ids # 转交按钮可见条件：表达式结果包含当前记录 if has_self_in_result: has_visible_transfer_btn = True break # 找到可见按钮，终止循环 else: # 无隐藏控制，转交按钮默认可见 has_visible_transfer_btn = True break # 找到可见按钮，终止循环 # ========== 第二步：判断结束按钮（end_buttons）是否可见 ========== if end_buttons: for btn in end_buttons: if btn.is_review_invisible: # 解析隐藏表达式，判断当前记录是否在表达式结果中 domain = eval(btn.invisible_expression) expression_result = self.search(domain) has_self_in_result = self.id in expression_result.ids # 结束按钮可见条件：表达式结果包含当前记录 if has_self_in_result: has_visible_end_btn = True break # 找到可见按钮，终止循环 else: # 无隐藏控制，结束按钮默认可见 has_visible_end_btn = True break # 找到可见按钮，终止循环 if not has_visible_transfer_btn and not has_visible_end_btn: skipped_node_names.append(next_node.name) _logger.info(f"跳过节点: {next_node.name}，转交按钮和结束按钮均不可见") # 更新索引 index = state_list.index(matched_key) if index >= len(node_list): break #根据实际路径创建trans.link记录并生成domain actual_path_links = [] self.env["soong.workflow.record"].search([("res_model", "=", self._name), ("res_id", "=", self.id)]) for node in actual_path_nodes[1:]: workflow_record_id = self.env["soong.workflow.record"].search([("res_model", "=", self._name), ("res_id", "=", self.id)]) if node.name not in skipped_node_names: link = self.env['trans.link'].create({ 'sequence': node.sequence, 'workflow_record_id': workflow_record_id.id, 'node_button_id': btn_rec.id, 'name': node.name, 'link_type': 'trans', 'record_id': self.id, }) # 节点无可选审批人(employee_ids 为空)也跳过，避免生成无人可审的空流转节点 approvers = self._get_approvers_for_link(link, self) # 纯"结束"节点（无转交/加签/分发/下发按钮）不需要审批人，正常保留，避免误跳结束节点卡死流程 node_approval_btns = node.button_ids.filtered( lambda b: b.name in ('转交', '加签', '分发', '下发')) if not approvers and node_approval_btns: _logger.info(f"跳过节点: {node.name}，无可选审批人(employee_ids 为空)") skipped_node_names.append(node.name) link.with_context(confirm_delete=True).unlink() continue actual_path_links.append(link) if actual_path_links: trans_link_ids = [link.id for link in actual_path_links] trans_domain = [('id', 'in', trans_link_ids)] else: trans_domain = [] is_supervising = False if self.name_title in ['行政发文']: is_supervising = self.is_supervising comment_field_map = { '转交': 'transfer_comment_id', '退回': 'return_comment_id', '确认': 'confirm_comment_id', '结束': 'end_comment_id' } comment_note_id = False if btn_rec.name in comment_field_map: target_field = comment_field_map[btn_rec.name] if hasattr(btn_rec.node_id, target_field) and btn_rec.node_id[target_field]: comment_note_id = btn_rec.node_id[target_field].id trans_link_id = self.env['trans.link'].search(trans_domain, limit=1).sorted('sequence').id if btn_rec.name == '转交' else False return_link_id = self.env['trans.link'].search( [('link_type', '=', 'return'), ('record_id', '=', self.id)], limit=1 ).sorted('sequence')[:1].id if btn_rec.name == '退回' else False all_links = self.env['trans.link'].browse(trans_link_ids) invalid_links = self.env['trans.link'] current_link = self.env['trans.link'].browse(trans_link_id) allowed_link_ids = [] default_trans_link_id = trans_link_id if btn_rec.name in ['转交', '加签'] and trans_link_ids: if current_link and all_links: sorted_all_links = all_links.sorted('sequence') for idx, link in enumerate(sorted_all_links): node_id = self.workflow_id.node_ids.search([ ('name', '=', link.name), ('workflow_id', '=', self.workflow_id.id) ], limit=1) # 本节点指定的"前置节点"仍出现在可选列表中（即它可以选择）→ 本节点直接跳过， # 且本节点的"相同人员节点处理"也随之不触发 if node_id and node_id.skip_if_node_id and node_id.skip_if_node_id != node_id: if any(l.name == node_id.skip_if_node_id.name for l in sorted_all_links): invalid_links |= link continue approvers = self._get_approvers_for_link(link, self) if len(approvers) == 1: current_approver = approvers[0] if node_id: # 下一个相同审批人跳过 if node_id.same_user_skip_type == 'next': if idx + 1 < len(sorted_all_links): next_link = sorted_all_links[idx + 1] next_approvers = self._get_approvers_for_link(next_link, self) if len(next_approvers) == 1 and next_approvers[0] == current_approver: invalid_links |= next_link # 上一个相同审批人跳过 if node_id.same_user_skip_type == 'prev': if idx - 1 >= 0: prev_link = sorted_all_links[idx - 1] prev_approvers = self._get_approvers_for_link(prev_link, self) if len(prev_approvers) == 1 and prev_approvers[0] == current_approver: invalid_links |= prev_link # 此节点之后的所有相同审批人节点跳过 if node_id.same_user_skip_type == 'after': for j in range(idx + 1, len(sorted_all_links)): after_link = sorted_all_links[j] after_approvers = self._get_approvers_for_link(after_link, self) if len(after_approvers) == 1 and after_approvers[0] == current_approver: invalid_links |= after_link allowed_link_ids = [lid for lid in trans_link_ids if lid not in invalid_links.ids] default_trans_link_id = allowed_link_ids[0] if allowed_link_ids else trans_link_id return { "name": _("评论"), "type": "ir.actions.act_window", "view_mode": "form", "res_model": "soong.workflow.comment.wizard", "views": [(wizard.id, "form")], "view_id": wizard.id, "target": "new", "context": { "default_return_link_id": return_link_id, "default_res_id": self.id, 'default_comment_note_id': comment_note_id, "default_code": self.code, "default_workflow_record_id": workflow_record_id.id, 'default_illustrate':btn_rec.node_id.illustrate, "default_res_model": self._name, "default_node_btn_id": btn_rec.id, "default_is_supervising": is_supervising, "default_trans_link_id": default_trans_link_id, "all_workflow_links": allowed_link_ids, "default_invalid_link_ids": invalid_links.ids, }, } def get_all_sub_departments(self, department, level=0, print_tree=True): all_departments = department.child_ids # 打印当前部门的子部门 if print_tree and department.child_ids: indent = " " * level # 使用空格缩进表示层级 print(f"{indent}{department.name} 的子部门:") for child in department.child_ids: child_indent = " " * (level + 1) print(f"{child_indent}- {child.name} (ID: {child.id})") # 递归获取每个子部门的子部门 for child in department.child_ids: child_sub_departments = self.get_all_sub_departments(child, level + 1, print_tree) all_departments |= child_sub_departments return all_departments @api.constrains("state") def _update_wkf_records(self): for rec in self: # 仅匹配未回收的有效记录(回收站里的 reclaim_state=True 旧记录不参与)， # 避免"同一张单 + 并发/state 多次写入"时重复 create record_obj = rec.env["soong.workflow.record"].search( [("res_model", "=", rec._name), ("res_id", "=", rec.id), ('reclaim_state', '=', False)], limit=1) record_id = rec.env[rec._name].search([('id', '=', rec.id)]) current_user_id = rec.env.user.id if rec.state in ['申请人', '拟办', '草稿', 'draft']: if not record_obj: values = { "code": record_id.code, "res_model": rec._name, "department_id": rec.department_id.id, "res_id": rec.id, "workflow_id": rec.workflow_id.id, "node_id": rec.node_id.id, "done_by": rec.env.uid, "users_ids": [(4, current_user_id)], "employee_ids": rec.env.user.employee_id, "write_date2": fields.Datetime.now(), } # 查找匹配的字段 matched_fields = rec.find_fields_by_string(record_id) # 提取"事由"或"正文"字段的值 text_fields = [f for f in matched_fields if '事由' in f['field_string'] or '正文' in f['field_string']] if text_fields: # 优先使用"事由"字段，没有则使用第一个"正文"字段 reason_field = next((f for f in text_fields if '事由' in f['field_string']), text_fields[0]) values["text"] = getattr(rec, reason_field['field_name'], '') # 提取"附件"字段的值 attachment_fields = [f for f in matched_fields if '附件' in f['field_string']] if attachment_fields and attachment_fields[0]['field_type'] in ['many2many', 'one2many']: # 处理附件关系字段 attachment_field = attachment_fields[0] attachment_ids = getattr(rec, attachment_field['field_name'], []) if attachment_ids: values["documents_ids"] = [(6, 0, attachment_ids.ids)] # 创建工作流记录 rec.env["soong.workflow.record"].create(values) if record_obj: # 获取上一节点的结束时间（即log_ids最后一条记录的lase_click_date） node_update_time = False try: if hasattr(rec, 'log_ids') and rec.log_ids: # 按ID排序获取最后一条日志 sorted_logs = rec.log_ids.sorted(lambda log: log.id) if sorted_logs and sorted_logs[-1].lase_click_date: node_update_time = sorted_logs[-1].lase_click_date # 补丁：如果审批状态已经是结束，但还没拿到时间，则尝试用当前时间作为办结时间 if not node_update_time and record_obj.approve_state == 'end': node_update_time = fields.Datetime.now() except Exception as e: _logger.debug(f"Failed to get node_update_time: {e}") # 更新记录 update_vals = { "workflow_id": rec.workflow_id.id, "node_id": rec.node_id.id, "done_by": self.env.uid, "write_date2": fields.Datetime.now(), } if node_update_time: update_vals["node_update_time"] = node_update_time record_obj.update(update_vals) def find_fields_by_string(self,record_id): """ 根据模型名称和记录ID查找字段，其string属性包含特定关键词 :param record_id: 记录id :return: 匹配的字段信息列表 """ # 定义要查找的关键词 keywords = ['正文', '事由', '附件'] # 获取模型的所有字段 fields = record_id.fields_get() # 存储匹配的字段信息 matched_fields = [] # 遍历所有字段，检查string属性 for field_name, field_attrs in fields.items(): field_string = field_attrs.get('string', '') # 检查字段的string是否包含任何关键词 if any(keyword in field_string for keyword in keywords): matched_fields.append({ 'field_name': field_name, 'field_string': field_string, 'field_type': field_attrs.get('type'), }) return matched_fields def unlink(self): self.mapped("log_ids").unlink() self.mapped("log_ids2").unlink() if self.env.user.name != 'Administrator': if len(self.log_ids) > 1: raise UserError("该流程不允许删除，如果要调整请联系管理员") record_objs = self.env["soong.workflow.record"].search([("res_model", "=", self._name), ("res_id", "in", self.ids)]) record_objs.unlink() return super(SoongWorkflowMixin, self).unlink() def action_open_diagram(self): self.ensure_one() current_node_id = self.workflow_id.node_ids.filtered( lambda n: n.node_name == self.state ) highlight_color = self.workflow_id.current_stage_highlight_color return { "type": "ir.actions.client", "tag": "open_diagram_plus_dialog", "params": { "act_window": { "name": _("Workflow: %(name)s") % {"name": self.display_name}, "res_model": "soong.workflow", "type": "ir.actions.act_window", "res_id": self.workflow_id.id, "target": "new", "dialog_size": "medium", "context": { "highlight_node_id": current_node_id.id, "highlight_node_color": highlight_color, "diagram_readonly": True, }, "views": [(False, "diagram_plus")], } } } # return { # "name": _("Workflow: %(name)s") # % { # "name": self.display_name, # }, # "res_model": "soong.workflow", # "type": "ir.actions.act_window", # "res_id": self.workflow_id.id, # "target": "new", # "dialog_size": "medium", # "context": { # "highlight_node_id": current_node_id.id, # "highlight_node_color": highlight_color, # "diagram_readonly": True, # }, # "views": [(False, "diagram_plus")], # } @api.model def get_view(self, view_id=None, view_type="form", **options): res = super().get_view(view_id, view_type, **options) active_id = self._context.get('active_id') if '
= 2: rec[fname] = [val[0], hist_name] elif isinstance(val, dict) and 'id' in val: val['display_name'] = hist_name return result # 收回 def action_withdraw(self): pass def _remove_default_header(self, view_type, res): """ Removes default header element from form view. :param view_type: Type of view now rendering. :param res: View resource data. :return: Updated view resource. """ model = self._name wkf_obj = self.env["soong.workflow"] wkf_rec = wkf_obj.search([("model_id", "=", model), ('is_enable', '=', True)]) if not wkf_rec.remove_default_btn_until_end: return res # Variables arch = etree.fromstring(res["arch"]) # Find Default Header Element & EDIT it if view_type == "form": header_el = arch.xpath("//form/header") header_el = header_el[0] if header_el else False if header_el: for button in header_el.iter("button"): btn_invisible = "state != 'end'" if "invisible" in button.attrib: btn_invisible = btn_invisible + " or " + button.attrib["invisible"] button.set("invisible", btn_invisible) res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res def _remove_default_modifiers(self, view_type, res): """ Removed fields default modifiers & attributes in form view. :param view_type: Type of view now rendering. :param res: View resource data. :return: Updated view resource. """ # Variables model = self._name wkf_obj = self.env["soong.workflow"] wkf_rec = wkf_obj.search([("model_id", "=", model), ('is_enable', '=', True)]) arch = etree.fromstring(res["arch"]) # Validation if not wkf_rec.remove_default_attrs_mod: return res fields_name = wkf_rec.node_ids.field_ids.name.mapped("name") # Find Field & Remove Modifiers if view_type == "form": for field in arch.iter("field"): field_name = field.attrib.get("name", False) if not field_name or field_name not in fields_name: continue for attr in field.attrib.keys(): if attr != "name": field.attrib.pop(attr) res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res def _load_buttons_view(self, view_type, res): """ Adds buttons to state bar in form view. :param view_type: Type of view now rendering. :param res: View resource data. :return: Updated view resource. """ # Variables this = self model = self._name wkf_obj = self.env["soong.workflow"] arch = etree.fromstring(res["arch"]) # Helper Functions def _check_condition_code(btn): locals_dict = { "env": this.env, "model": this.env[model], "obj": None, "user": this.env["res.users"].browse(this.env.uid), "datetime": datetime, "time": time, "date": date, "timedelta": timedelta, "workflow": this.env["soong.workflow"], "warning": this.warning, "syslog": this.syslog, } try: safe_eval( btn.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: return True return False except Warning as ex: raise ex except: raise UserError(_("Wrong python condition defined.")) # Find header & ADD button if view_type == "form": # Get Some Elements form_el = arch.xpath("//form") form_el = form_el[0] if form_el else False header_el = arch.xpath("//form/header") header_el = header_el[0] if header_el else False # Create Header Element If not Exists if header_el is False: header_el = etree.Element("header") form_el.insert(0, header_el) # Construct Buttons & Fields Tags wkf_rec = wkf_obj.search([("model_id", "=", model), ('is_enable', '=', True)]) for node in wkf_rec.node_ids: for button in node.button_ids: if self.env.user.has_group("soong_workflow.group_soong_workflow_manager"): btn_exec_action_el = etree.SubElement(header_el, "button") btn_exec_action_el.set("name", "btn_exec_action") btn_exec_action_el.set("string", button.name) btn_exec_action_el.set("type", "object") if button.is_highlight: btn_exec_action_el.set("class", "oe_highlight") if button.has_icon: btn_exec_action_el.set("icon", button.icon) invisible_condition = "state != '%s'" % node.node_name # 采购订单流程结束后state会回到purchase，仅真正最后一个节点（无下游节点）的确认按钮额外允许该状态 if button.name == '确认' and model == 'purchase.order' and node.is_end and not node.out_link_ids: invisible_condition = "state not in ('%s', 'purchase')" % node.node_name # if model == 'purchase.order': # invisible_condition = invisible_condition.replace("询价", "draft") btn_exec_action_el.set("invisible", invisible_condition) btn_exec_action_el.set("context", "{'btn_key':'%s','active_model':'%s'}" % (button.btn_key, self._name), ) # 添加鼠标悬停提示 if button.name == '转交': btn_exec_action_el.set("title", "审核，转至下一节点继续审批") elif button.name == '结束': btn_exec_action_el.set("title", "审批核准，流程到此结束，无下一节点人员可以选择") elif button.name == '确认': btn_exec_action_el.set("title", "传阅确认，不影响流程结束") elif button.name == '加签': btn_exec_action_el.set("title", "转交给下一节点一人或多人，完成后会返回回来") elif button.name == '归档': btn_exec_action_el.set("title", "按要求上传附件存单") else: is_apostille = False if button.btn_hide: if button.review_type == "individual": if self.env.user != button.reviewer_id: continue elif button.review_type == "group": if self.env.user not in button.reviewer_group_id.users: continue # 如果是加签确认 elif button.review_type == "apostille": is_apostille = True elif button.review_type == "expression": if not _check_condition_code(button): continue btn_exec_action_el = etree.SubElement(header_el, "button") btn_exec_action_el.set("name", "btn_exec_action") btn_exec_action_el.set("string", button.name) btn_exec_action_el.set("type", "object") id_el = etree.Element("field") id_el.set("name", "id") id_el.set("invisible", "1") if button.is_highlight: btn_exec_action_el.set("class", "oe_highlight") if button.has_icon: btn_exec_action_el.set("icon", button.icon) # 接下来是隐藏的配置 invisible = "state != '{0}' ".format(str(node.node_name)) if button.review_type == 'user_role' and model != 'yuthon.seals.approval': # 构建角色条件 role_domain = [] if button.users_role_id: role_domain = [ '|', ('employee_id.user_role_id', '=', button.users_role_id.id), ('employee_id.user_role_ids', 'in', button.users_role_id.id) ] if button.users_role_group_ids: role_ids = button.users_role_group_ids.user_role_ids.ids role_domain = [ '|', ('employee_id.user_role_id', 'in', role_ids), ('employee_id.user_role_ids', 'in', role_ids) ] # 按指定字段匹配（specify_field_id）：补充可见用户范围。 # 修复：主角色/角色组为空 + 按字段匹配时，原逻辑 role_domain 为空导致 # 整个 uid 限制被跳过，任何人都能看到按钮。 # 按指定字段匹配（specify_field_id）：按钮仅对字段指向的用户可见。 # 注意：get_view 是 @api.model，self 为空记录集；且 arch 跨记录缓存， # 因此服务端既读不到 self[field_name]，也不能烘焙 per-record 的 uid 列表。 # 正确做法：拼接一条“客户端按当前记录字段求值”的 invisible 表达式。 field_invisible_expr = "" if button.specify_field_id: field_name = button.specify_field_id.name if field_name in self._fields: if button.specify_field_id.relation == 'res.users': # 字段值为 user id，直接与 uid 比较（前端按当前记录求值） field_invisible_expr = f"uid != {field_name}" elif button.specify_field_id.relation == 'hr.employee': # 员工字段需取 user_id：客户端 m2o 值为 id，无法直接 .user_id， # 退回服务端按 active_id 计算（仅 form 打开某条记录时正确， # 跨记录缓存下存在理论偏差，建议本功能优先用 res.users 字段） active_id = self._context.get('active_id') if active_id: fv = self.browse(active_id)[field_name] if fv and fv.user_id: field_invisible_expr = f"uid != {fv.user_id.id}" # 汇总允许的用户（角色条件 ∪ 字段条件） if role_domain: allowed_user_ids = set(self.env['res.users'].search(role_domain).ids) # 委托补充：角色组内的委托人(如部门经理A)若把流程委托给被委托人(b)， # 被委托人本人未必在该角色组内，也应能看到/操作该按钮 entrust_all = self.env['ir.model'].sudo().search( [('model', '=', 'entrust.all')], limit=1) model_rec = self.env['ir.model'].sudo().search( [('model', '=', model)], limit=1) if model else None entrusts = self.env['soong.workflow.entrust'].sudo().search([ ('is_enable', '=', True), ('entrust_user_id', 'in', list(allowed_user_ids)), ]) for ent in entrusts: ent_model_ids = ent.model_ids.ids in_scope = (entrust_all and entrust_all.id in ent_model_ids) or ( model_rec and model_rec.id in ent_model_ids) if in_scope and ent.entrust_user_id2: allowed_user_ids.add(ent.entrust_user_id2.id) allowed_user_ids = list(allowed_user_ids) else: allowed_user_ids = [] # 只要有角色限制条件，就附加 uid 可见性限制 if allowed_user_ids: invisible += f" or uid not in {allowed_user_ids}" # 字段匹配条件（客户端按当前记录字段求值）追加到 invisible if field_invisible_expr: invisible += f" or {field_invisible_expr}" # 如果有隐藏条件的话，启用隐藏 if button.is_review_invisible and button.invisible_expression: domain = eval(button.invisible_expression) btn_invisible = self.sudo().env[button.workflow_id.model_id.model].search(domain).ids invisible += " or id not in {0} ".format(str(btn_invisible)) if button.name == '确认': btn_exec_action_el.set("class", "btn btn-confirm") id_el = etree.Element("field") id_el.set("name", "log_ids") id_el.set("invisible", "1") # 采购订单流程结束后state会回到purchase，仅真正最后一个节点（无下游节点）的确认按钮额外允许该状态 if model == 'purchase.order' and node.is_end and not node.out_link_ids: invisible = "state not in ('%s', 'purchase') or not is_confirm" % node.node_name else: invisible += "or not is_confirm" if button.name == '结束': btn_exec_action_el.set("class", "btn btn-finish") id_el = etree.Element("field") id_el.set("name", "log_ids") id_el.set("invisible", "1") invisible += " or not is_end or is_show_transfer" if button.name == "退回": btn_exec_action_el.set("class", "btn btn-return") id_el = etree.Element("field") id_el.set("name", "log_ids") id_el.set("invisible", "1") invisible += " or not log_ids or is_done or is_show_transfer" if button.name == "分发": btn_exec_action_el.set("class", "btn btn-distribute") invisible += " or not is_send" if button.name == "转交": btn_exec_action_el.set("class", "btn btn-transfer") invisible += " or is_done or is_show_transfer" if button.name == "加签确认": invisible += "or not is_show_transfer" if button.name == "收回": btn_exec_action_el.set("class", "btn btn-recall") id_el = etree.Element("field") id_el.set("name", "log_ids") id_el.set("invisible", "1") invisible += " or not log_ids or not is_back" if button.name == "加签": btn_exec_action_el.set("class", "btn btn-recall") invisible += "or not is_show_add_sign" if button.name == '归档': btn_exec_action_el.set("class", "btn btn-archive") invisible += "or not is_active" custom_class = btn_exec_action_el.get("class", "") btn_exec_action_el.set("class", custom_class) # 添加鼠标悬停提示 if button.name == '转交': btn_exec_action_el.set("title", "审核，转至下一节点继续审批") elif button.name == '结束': btn_exec_action_el.set("title", "审批核准，流程到此结束，无下一节点人员可以选择") elif button.name == '确认': btn_exec_action_el.set("title", "传阅确认，不影响流程结束") elif button.name == '加签': btn_exec_action_el.set("title", "转交给下一节点一人或多人，完成后会返回回来") elif button.name == '归档': btn_exec_action_el.set("title", "按要求上传附件存单") if model == 'purchase.order': invisible = invisible.replace("询价", "draft") btn_exec_action_el.set("invisible", invisible) btn_exec_action_el.set("context", "{'btn_key':'%s','active_model':'%s'}" % (button.btn_key, self._name),) res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res def _load_button_box_view(self, view_type, res): """ Adds button_box in form view. :param view_type: Type of view now rendering. :param res: View resource data. :return: Updated view resource. """ arch = etree.fromstring(res["arch"]) # Find header & ADD button if view_type == "form": # Get Some Elements form_sheet_el = arch.xpath("//form/sheet") form_sheet_el = form_sheet_el[0] if form_sheet_el else False button_box_el = arch.xpath("//form/sheet/div[hasclass('oe_button_box')]") button_box_el = button_box_el[0] if button_box_el else False # Create button_box Element If not Exists if button_box_el is False: button_box_el = etree.Element("div") button_box_el.set("class", "oe_button_box") button_box_el.set("name", "button_box") if form_sheet_el: form_sheet_el.insert(0, button_box_el) # Add Button to View btn_exec_action_el = etree.SubElement(button_box_el, "button") btn_exec_action_el.set("name", "action_open_diagram") btn_exec_action_el.set("type", "object") btn_exec_action_el.set("string", _("View Flow")) btn_exec_action_el.set("class", "oe_stat_button") btn_exec_action_el.set("icon", "fa-code-fork") res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res def _load_state_view(self, view_type, res): """ Adds state bar to form view. :param view_type: Type of view now rendering. :param res: View resource data. :return: Updated view resource. """ # Variables wkf_obj = self.env["soong.workflow"] model = self._name visible_seq = [] arch = etree.fromstring(res["arch"]) wkf_rec = wkf_obj.search([("model_id", "=", model), ('is_enable', '=', True)]) # Find header & ADD state if view_type == "form": # Get Header Element header_el = arch.xpath("//form/header") header_el = header_el[0] if header_el else False # Create State Element If not Exists end_el = etree.Element("field") end_el.set("name", "is_end") end_el.set("invisible", "1") header_el.append(end_el) import_el = etree.Element("field") import_el.set("name", "is_import") import_el.set("invisible", "1") header_el.append(import_el) back_el = etree.Element("field") back_el.set("name", "is_back") back_el.set("invisible", "1") header_el.append(back_el) confirm_el = etree.Element("field") confirm_el.set("name", "is_confirm") confirm_el.set("invisible", "1") header_el.append(confirm_el) active_el = etree.Element("field") active_el.set("name", "is_active") active_el.set("invisible", "1") header_el.append(active_el) send_el = etree.Element("field") send_el.set("name", "is_send") send_el.set("invisible", "1") header_el.append(send_el) sign_el = etree.Element("field") sign_el.set("name", "is_show_add_sign") sign_el.set("invisible", "1") header_el.append(sign_el) approval_el = etree.Element("field") approval_el.set("name", "approval_result") approval_el.set("invisible", "1") header_el.append(approval_el) is_transfer_el = etree.Element("field") is_transfer_el.set("name", "is_show_transfer") is_transfer_el.set("invisible", "1") header_el.append(is_transfer_el) done_el = etree.Element("field") done_el.set("name", "is_done") done_el.set("invisible", "1") header_el.append(done_el) modify_btn_el = etree.SubElement(header_el, "button") modify_btn_el.set("name", "action_is_modify") modify_btn_el.set("string", "已修正") modify_btn_el.set("type", "object") modify_btn_el.set("invisible", "not is_modify") modify_btn_el.set("class", "btn btn-primary") modify_el = etree.Element("field") modify_el.set("name", "is_modify") modify_el.set("invisible", "1") header_el.append(modify_el) if header_el is not False: state_el = etree.Element("field") state_el.set("name", "state") state_el.set("widget", "statusbar") state_el.set("invisible", "1") # Loop all nodes for node in wkf_rec.node_ids: # Read Visible States if node.is_visible: visible_seq.append(node.node_name) # Set Attributes & Setup Modifiers for State Field if visible_seq: state_el.set("statusbar_visible", ",".join(visible_seq)) header_el.append(state_el) # 审批结果徽章（注入到 header 末尾，inline 不影响 sheet 布局，全局生效） # 通过 → 绿色徽章，不通过 → 红色徽章，无结果时不显示 result_badge = etree.Element("span") result_badge.set("class", "o_approval_result ms-2") result_badge.set("invisible", "approval_result not in ('pass', 'fail')") # 通过：绿色徽章 pass_badge = etree.SubElement(result_badge, "span") pass_badge.set("class", "badge text-bg-success") pass_badge.set("invisible", "approval_result != 'pass'") pass_badge.text = "通过" # 不通过：红色徽章 fail_badge = etree.SubElement(result_badge, "span") fail_badge.set("class", "badge text-bg-danger") fail_badge.set("invisible", "approval_result != 'fail'") fail_badge.text = "不通过" header_el.append(result_badge) res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res def _load_one2many_views(self, view_type, res): # Variables # wkf_obj = self.env["soong.workflow"] # model = self._name arch = etree.fromstring(res["arch"]) # wkf_rec = wkf_obj.search([("model_id", "=", model)]) if view_type == "form": # Locate the group element group_el = arch.xpath("//sheet") if group_el: group_el = group_el[0] else: # If no group element exists, create one and add it to the arch group_el = etree.Element("sheet") # Insert the group element at the beginning or end of the arch arch.insert(0, group_el) # Locate or create the notebook element within the group notebook_el = group_el.xpath(".//notebook") if notebook_el: notebook_el = notebook_el[0] else: # Create a notebook element if it doesn't exist notebook_el = etree.Element("notebook") group_el.append(notebook_el) # Create a new page element page_el = etree.Element("page") page_el.set("string", "审批明细") # Set a label for the new page # Create the field element for the one2many field field_el = etree.Element("field") field_el.set("name", "approval_users_ids") field_el_context = {'tree_view_ref': 'soong_workflow.soong_approval_users_search', 'default_res_model_id': self.env['ir.model']._get(self._name).id} field_el.set("context", str(field_el_context)) # Append the field element to the new page page_el.append(field_el) # Append the new page to the notebook notebook_el.append(page_el) # Convert the modified XML tree back to a string res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res def _load_sign_view(self, view_type, res): arch = etree.fromstring(res["arch"]) if view_type == "form": sheet_el = arch.xpath("//sheet") if sheet_el: sheet_el = sheet_el[0] else: sheet_el = etree.Element("sheet") arch.insert(0, sheet_el) notebook_el = sheet_el.xpath(".//notebook") if notebook_el: notebook_el = notebook_el[0] else: notebook_el = etree.Element("notebook") sheet_el.append(notebook_el) # page_el = etree.Element("page") # page_el.set("string", "加签人员") # 设置页面标签 # page_el.set("invisible", "1") # todo domain # field_m2m_el = etree.Element("field") # field_m2m_el.set("name", "approval_sign_ids") # field_m2m_el.set("widget", "many2many_tags") # page_el.append(field_m2m_el) # notebook_el.append(page_el) res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res # def _load_workflow_approver_view(self, view_type, res): # arch = etree.fromstring(res["arch"]) # if view_type == "form": # sheet_el = arch.xpath("//sheet") # if sheet_el: # sheet_el = sheet_el[0] # else: # sheet_el = etree.Element("sheet") # arch.insert(0, sheet_el) # notebook_el = sheet_el.xpath(".//notebook") # if notebook_el: # notebook_el = notebook_el[0] # else: # notebook_el = etree.Element("notebook") # sheet_el.append(notebook_el) # page_el = etree.Element("page") # page_el.set("string", "审批流程") # 设置页面标签 # page_el.set("invisible", "1") # todo domain # field_m2m_el = etree.Element("field") # field_m2m_el.set("name", "workflow_approver_ids") # page_el.append(field_m2m_el) # notebook_el.append(page_el) # res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") # return res def _update_fields_view(self, view_type, res): """ Updates fields attributes. :param view_type: Type of view now rendering. :param res: View resource data. :return: Updated view resource. """ # Objects wkf_obj = self.env["soong.workflow"] user_obj = self.env["res.users"] # Variables model = self._name uid = self._uid wkf_rec = wkf_obj.search([("model_id", "=", model), ('is_enable', '=', True)]) arch = etree.fromstring(res["arch"]) # Helper Functions def _get_external_id(group): arr = [] ext_ids = group._get_external_ids() _logger.info(f"--------{ext_ids}") for ext_id in ext_ids: _logger.info(f"--------{ext_id}") arr.append(ext_ids[ext_id][0]) return arr if view_type == "form": # Read fields of view for field in arch.iter("field"): field_name = field.attrib.get("name", False) if not field_name: continue # Get Fields Instance field_inst = arch.xpath("//field[@name='%s']" % str(field_name)) field_inst = field_inst[0] if field_inst else False # Scope Variables readonly_arr = [] required_arr = [] invisible_arr = [] if field_inst is not False: # Loop all nodes for node in wkf_rec.node_ids: # Loop Other Nodes for field_attrs in node.field_ids: # Record all states for each attribute if ( field_inst is not False and field_attrs.name.name == field_inst.attrib["name"] ): flag_show = True if field_attrs.name.name in ["code", "name_title"]: field_inst.set("readonly", "1") readonly_arr.append(node.node_name) if field_attrs.readonly: readonly_arr.append(node.node_name) if field_attrs.required: required_arr.append(node.node_name) if field_attrs.invisible: invisible_arr.append(node.node_name) # Check Users & Groups if field_attrs.user_ids: user_rec = user_obj.browse(uid) if user_rec not in field_attrs.user_ids: flag_show = False # if field_attrs.group_ids: # has_group = False # user_rec = user_obj.browse(uid) # _logger.info(f"-----{field_attrs.name}") # _logger.info(f"-----{field_attrs.name.name}") # _logger.info(f"-----{field_attrs.group_ids}") # ext_ids = _get_external_id(field_attrs.group_ids) # for ext_id in ext_ids: # has_group = user_rec.has_group(ext_id) # if not has_group: # flag_show = False if not flag_show and (field_attrs.group_ids or field_attrs.user_ids): invisible_arr.append(node.node_name) if readonly_arr: id_el = etree.Element("field") id_el.set("name", "log_ids") id_el.set("invisible", "1") if self._name in ["purchase.order", "property.management"]: # 原有流程节点状态只读 + 额外增加 state=purchase 也只读 field_inst.set("readonly", "(state in %s) or (state == 'purchase')" % readonly_arr) else: field_inst.set("readonly", "(state in %s) and (is_modify != True)" % readonly_arr) # if required_arr: # field_inst.set("required", "state in %s" % required_arr) if invisible_arr: field_inst.set("invisible", "state in %s" % invisible_arr) if self.env.user.has_group("soong_workflow.group_soong_workflow_manager"): field_inst.set("readonly", "0") current_field_name = field_inst.attrib.get("name") if current_field_name in ["code", "name_title"]: field_inst.set("readonly", "1") res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res def _add_state_tree_view(self, view_type, res): """ Adds state field to tree view. :param view_type: Type of view now rendering. :param res: View resource data. :return: Updated view resource. """ # Variables model = self._name model_obj = self.env[model] arch = etree.fromstring(res["arch"]) # Add State Field to Tree View if not Exists if view_type == "tree": # Get Header Element tree_el = arch.xpath("//tree") tree_el = tree_el[0] if tree_el else False # Get State Element node_el = arch.xpath("//field[@name='node_id']") node_el = node_el[0] if node_el else False if node_el is False: node_el = etree.Element("field") node_el.set("name", "node_id") tree_el.append(node_el) approve_state_el = arch.xpath("//field[@name='approve_state']") approve_state_el = approve_state_el[0] if approve_state_el else False if approve_state_el is False: approve_state_el = etree.Element("field") approve_state_el.set("name", "approve_state") tree_el.append(approve_state_el) tree_el.set("decoration-muted", "approve_state == 'end'") tree_el.set("decoration-warning", "approve_state != 'end'") res["arch"] = etree.tostring(arch, encoding="unicode").replace("\t", "") return res def _add_soong_workflow_log_manager(self, view_type, res, params): View = self.env["ir.ui.view"] doc = etree.XML(res["arch"]) all_models = res["models"].copy() for node in doc.xpath("/form/sheet"): # _add_tier_validation_reviews process str_element = self.env["ir.qweb"]._render("soong_workflow.soong_workflow_log_manager", params) new_node = etree.fromstring(str_element) new_arch, new_models = View.postprocess_and_fields(new_node, self._name) for model in new_models: if model in all_models: continue if model not in res["models"]: all_models[model] = new_models[model] else: all_models[model] = res["models"][model] new_node = etree.fromstring(new_arch) node.append(new_node) res["arch"] = etree.tostring(doc) res["models"] = frozendict(all_models) return res def _add_soong_workflow_log(self, view_type, res, params): View = self.env["ir.ui.view"] doc = etree.XML(res["arch"]) all_models = res["models"].copy() for node in doc.xpath("/form/sheet"): # _add_tier_validation_reviews process str_element = self.env["ir.qweb"]._render( "soong_workflow.soong_workflow_log", params ) new_node = etree.fromstring(str_element) new_arch, new_models = View.postprocess_and_fields(new_node, self._name) for model in new_models: if model in all_models: continue if model not in res["models"]: all_models[model] = new_models[model] else: all_models[model] = res["models"][model] new_node = etree.fromstring(new_arch) node.append(new_node) res["arch"] = etree.tostring(doc) res["models"] = frozendict(all_models) return res 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) def _get_approvers_for_link(self, link, record_id=None): """ 参数: link: trans.link 记录对象 record_id: 关联的业务记录对象 返回: hr.employee 记录集（可能为空） """ if not link or isinstance(link, int): return self.env['hr.employee'] workflow_id = link.node_button_id.workflow_id node = workflow_id.node_ids.search([ ('name', '=', link.name), ('workflow_id', '=', workflow_id.id) ]) button_priority_list = ["分发", "下发", "转交", "结束", "加签"] button = False if node.button_ids: found_target_btn = False for btn_name in button_priority_list: for btn in sorted(node.button_ids, key=lambda x: x.sequence): if btn.name == btn_name: is_in_scope = True if not btn.is_review_invisible: button = btn found_target_btn = True break if btn.is_review_invisible: domain = eval(btn.invisible_expression) model = record_id.env[record_id._name] matched = model.search(domain + [('id', '=', record_id.id)]) is_in_scope = bool(matched) if is_in_scope: button = btn found_target_btn = True break if found_target_btn: break 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: dept_id = record_id.department_id.id # 按当前申请人自动匹配-本部门（保留部门和公司条件） applicant_dept_ids = applicant.employee_ids.department_ids.ids department_domain = [ '|', '|', '|', ('department_id', '=', dept_id), ('department_ids', 'in', [dept_id]), ('department_id', 'in', applicant_dept_ids), ('department_ids', 'in', applicant_dept_ids), ] 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] if field_value: if button.specify_field_id.relation == 'res.users': employee = self.env['hr.employee'].sudo().search( [('user_id', '=', field_value.id)], limit=1 ) if employee: return employee elif button.specify_field_id.relation == 'hr.employee': return field_value else: 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 _get_active_workflow_record(self): """本单据「未进回收站」的工作流记录（没有则返回空记录集）。 按 res_model + res_id 匹配；回收站里的记录 reclaim_state=True，不算有效记录。 """ self.ensure_one() return self.env['soong.workflow.record'].sudo().search([ ('res_model', '=', self._name), ('res_id', '=', self.id), ('reclaim_state', '=', False), ], limit=1) def _has_active_workflow_record(self): """本单据是否存在「未进回收站」的工作流记录（回收站里的不算）""" return bool(self._get_active_workflow_record()) # ---- 公司历史名称展示：web_read 直接改写指向 res.company 的公司字段显示名（见 web_read）---- company_id = fields.Many2one('res.company', string='公司') class SoongWorkflowLine(models.Model): _name = "soong.workflow.line" model = fields.Char(string="Related Document Model", index=True) res_id = fields.Integer(string="Related Document ID", index=True) workflow_record_id = fields.Many2one("soong.workflow.record", string="标题", required=True) code = fields.Char(related="workflow_record_id.code", string="流水号") workflow_id = fields.Many2one(related="workflow_record_id.workflow_id", string="审批类型") company_id = fields.Many2one(related="workflow_record_id.company_id", string="公司") department_id = fields.Many2one(related="workflow_record_id.department_id", string="部门") approvals_group_id = fields.Many2one(related="workflow_record_id.approvals_group_id", string="审批组") create_uid = fields.Many2one(related="workflow_record_id.create_uid", string="创建人") write_date2 = fields.Datetime(related="workflow_record_id.write_date2", string="最后更新日期") def unlink(self): if not self.env.context.get('confirm_delete'): # 提示用户确认删除 message = _("确定要删除这些记录吗？此操作不可撤销。") return { 'name': _('确认删除'), 'type': 'ir.actions.act_window', 'view_mode': 'form', 'res_model': 'workflow.line.wizard', 'target': 'new', 'context': { 'active_model': self._name, 'active_ids': self.ids, 'delete_message': message, } } # 继续执行删除操作 return super(SoongWorkflowLine, self).unlink() # return { # "type": "ir.actions.act_window", # "name": "二次确认", # "res_model": "workflow.line.wizard", # "view_mode": "form", # "target": "new", # "context": { # "default_workflow_id": self.workflow_record_id.id, # "default_workflow_line_id": self.id # }, # } def open_record(self): return { "type": "ir.actions.act_window", "res_model": self.workflow_record_id.res_model, "res_id": self.workflow_record_id.res_id, "name": self.workflow_record_id.name, "view_mode": "form", } class ProjectProjectLine(models.Model): _name = "project.project.line" model = fields.Char(string="Related Document Model", index=True) res_id = fields.Integer(string="Related Document ID", index=True) project_id = fields.Many2one("project.project", string="项目") user_id = fields.Many2one(related="project_id.user_id", string="项目管理员") date_start = fields.Date(related="project_id.date_start", string="日期")