Files
gzth/soong_workflow/models/soong_workflow_node.py
T
2024-12-02 14:34:45 +00:00

728 lines
30 KiB
Python

# -*- 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
_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)"
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")
# Custom colors
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_carbon_copy = fields.Boolean(string="是否抄送")
employee_ids = fields.Many2many('hr.employee', string='抄送人员')
@api.model_create_multi
def create(self, values):
for val in values:
if val['node_name'] != 'end':
if not val["button_ids"]:
if val['node_name'] == 'start':
val["button_ids"] = [
(0, 0, {
"name": _("转交"),
"workflow_id": val['workflow_id'],
"sequence": 1,
},),
(0, 0, {
"name": _("收回"),
"workflow_id": val['workflow_id'],
"sequence": 2,
"action_type": "code",
"code": "obj.action_withdraw()",
},)]
elif val['node_name'] == '分发':
val["button_ids"] = [
(0, 0, {
"name": _("分发"),
"workflow_id": val['workflow_id'],
"sequence": 1,
"action_type": "code",
"has_comment": True,
},)]
else:
val["button_ids"] = [
(0, 0, {
"name": _("转交"),
"workflow_id": val['workflow_id'],
"sequence": 1,
"action_type": "code",
"has_comment": True,
},),
(0, 0, {
"name": _("收回"),
"workflow_id": val['workflow_id'],
"sequence": 2,
"action_type": "code",
"code": "obj.action_withdraw()",
},),
# (0, 0, {
# "name": _("加签"),
# "workflow_id": val['workflow_id'],
# "sequence": 3,
# "action_type": "code",
# "has_comment": True,
# },),
# (0, 0, {
# "name": _("转交"),
# "workflow_id": val['workflow_id'],
# "sequence": 3,
# "action_type": "code",
# "btn_hide": True,
# "review_type": 'expression',
# # "reviewer_expression": "obj.action_withdraw()", # todo domain
# "has_comment": True,
# },),
(0, 0, {
"name": _("退回"),
"workflow_id": val['workflow_id'],
"sequence": 2,
"action_type": "code",
"has_comment": True,
},)]
return super().create(values)
@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):
# Variables
field_obj = self.env["ir.model.fields"]
for rec in self:
# Clear Fields List
rec.field_ids.unlink()
# Load Fields
fields_objs = field_obj.search([("model_id", "=", rec.model_id.id)])
for field in fields_objs:
rec.field_ids.create(
{
"model_id": rec.model_id.id,
"node_id": rec.id,
"name": field.id,
}
)
def all_readonly_fields(self):
self.field_ids.write({"readonly": True})
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..", )
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=False)
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="Validated by",
default="user_role",
selection=[
("individual", "Specific user"),
("group", "Any user in a specific group"),
("expression", "Python Expression"),
('user_role', '角色'),
('invisible', '隐藏条件'),
],
)
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="主角色", domain=[('user_type', '!=', 'lv_role')])
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.Text(
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"),
],
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,
ondelete="cascade",
)
is_automatic = fields.Boolean('是否自动匹配', default=True, help='该节点默认拟办人员的部门相应人员。必须是系统全局角色')
@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 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
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,
)