Files
zkxw/backend/app/services/oa_engine.py
T
2026-07-15 16:20:48 +08:00

1570 lines
50 KiB
Python

"""OA 审批运行时引擎(钉钉式竖向树)"""
from __future__ import annotations
import json
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.crud import oa_approval as oa_crud
from app.models.models import (
OaApp,
OaAppVersion,
OaInstance,
OaInstanceEvent,
OaRelation,
OaRole,
OaTask,
SysUser,
)
from app.services.oa_assignee import AssigneeResolveError, resolve_assignees
from app.services.oa_formula import FormulaError, apply_formulas_and_linkages, eval_formula
class OaEngineError(ValueError):
"""引擎业务错误"""
def _loads(text: Optional[str], default: Any = None) -> Any:
if text is None or text == "":
return default
if not isinstance(text, str):
return text
try:
return json.loads(text)
except (TypeError, ValueError):
return default
def _dumps(data: Any) -> str:
return json.dumps(data if data is not None else {}, ensure_ascii=False)
def _node_type(node: Dict[str, Any]) -> str:
return (node.get("type") or "").strip()
def _flow_root(flow_tree: Any) -> Dict[str, Any]:
if not isinstance(flow_tree, dict):
raise OaEngineError("流程树无效")
if isinstance(flow_tree.get("root"), dict):
return flow_tree["root"]
return flow_tree
def _find_node(root: Dict[str, Any], node_key: str) -> Optional[Dict[str, Any]]:
if not root or not node_key:
return None
if (root.get("id") or "") == node_key:
return root
child = root.get("child")
if isinstance(child, dict):
found = _find_node(child, node_key)
if found:
return found
for g in root.get("conditionGroups") or []:
if isinstance(g, dict) and isinstance(g.get("child"), dict):
found = _find_node(g["child"], node_key)
if found:
return found
return None
def _flow_node_key_order(root: Dict[str, Any]) -> Dict[str, int]:
"""按流程树先序遍历为 node.id 分配稳定序号(用于详情时间轴排序)。"""
order: Dict[str, int] = {}
counter = 0
def walk(node: Any) -> None:
nonlocal counter
if not isinstance(node, dict):
return
nid = (node.get("id") or "").strip()
if nid and nid not in order:
order[nid] = counter
counter += 1
if _node_type(node) == "condition":
for g in node.get("conditionGroups") or []:
if isinstance(g, dict) and isinstance(g.get("child"), dict):
walk(g["child"])
child = node.get("child")
if isinstance(child, dict):
walk(child)
walk(root)
return order
def _next_task_order_num(db: Session, instance_id: str) -> int:
"""同一实例内任务展示序号取 max+1(无任务从 1 起)。"""
rows = (
db.query(OaTask.order_num)
.filter(OaTask.instance_id == instance_id, OaTask.del_flag == 0)
.all()
)
vals = [int(r[0]) for r in rows if r[0] is not None]
return (max(vals) + 1) if vals else 1
def _sort_tasks_for_detail(
tasks: List[OaTask], *, flow_root: Optional[Dict[str, Any]]
) -> List[OaTask]:
"""详情时间轴:按流程树节点顺序,再 order_num / create_time(兼容旧数据同秒创建)。"""
node_order = _flow_node_key_order(flow_root) if isinstance(flow_root, dict) else {}
far = 10**9
def key(t: OaTask) -> Tuple[int, int, int, datetime, str]:
nk = (t.node_key or "").strip()
return (
int(getattr(t, "round_no", None) or 1),
node_order.get(nk, far),
int(t.order_num) if t.order_num is not None else 0,
t.create_time or datetime.min,
t.id or "",
)
return sorted(tasks, key=key)
def _fmt_dt(dt: Any) -> str:
if not dt:
return ""
if hasattr(dt, "strftime"):
return dt.strftime("%Y-%m-%d %H:%M:%S")
return str(dt)
def _terminal_item(status: str, title: str, inst: Any) -> Dict[str, Any]:
return {
"kind": "terminal",
"roundNo": getattr(inst, "current_round", None) or 1,
"title": title,
"status": status,
"assigneeUserId": "",
"assigneeUserName": "",
"timestamp": _fmt_dt(getattr(inst, "update_time", None)),
"commentText": "",
"nodeKey": "",
"taskType": "",
}
def _task_round(t: Any) -> int:
v = getattr(t, "round_no", None)
try:
if v is None:
return 1
return int(v)
except (TypeError, ValueError):
return 1
def _append_pending_nodes(
items: List[Dict[str, Any]],
*,
flow_root: Dict[str, Any],
reached_keys: set,
round_no: int,
) -> None:
order = _flow_node_key_order(flow_root)
for nid, _idx in sorted(order.items(), key=lambda x: x[1]):
node = _find_node(flow_root, nid)
if not node:
continue
nt = _node_type(node)
if nt not in ("approval", "cc") or nid in reached_keys:
continue
items.append(
{
"kind": "pending_node",
"roundNo": round_no,
"title": node.get("name") or ("抄送人" if nt == "cc" else "审批人"),
"status": "not_reached",
"assigneeUserId": "",
"assigneeUserName": "",
"timestamp": "",
"commentText": "",
"nodeKey": nid,
"taskType": nt,
}
)
def build_progress_items(
*,
inst: Any,
flow_root: Optional[Dict[str, Any]],
tasks: List[Any],
name_map: Dict[str, str],
events: Optional[List[Any]] = None,
) -> List[Dict[str, Any]]:
"""投影审批进度:发起人 → 树序 tasks → 未到达;终态收尾。
钉钉对齐:打回结果落在审批任务节点上,不另插 kind=return;不做同单多轮分段。
"""
events = list(events or [])
sorted_tasks = _sort_tasks_for_detail(list(tasks), flow_root=flow_root)
# 展示压成单轴(roundNo 统一为 1);历史多轮任务仍按树序合并展示
rn = 1
submit_ts = _fmt_dt(getattr(inst, "create_time", None))
for e in events:
if (getattr(e, "event_type", None) or "") == "submit":
submit_ts = _fmt_dt(getattr(e, "create_time", None)) or submit_ts
break
items: List[Dict[str, Any]] = [
{
"kind": "submit",
"roundNo": rn,
"title": "发起人",
"status": "submitted",
"assigneeUserId": inst.initiator_user_id,
"assigneeUserName": name_map.get(inst.initiator_user_id or "", ""),
"timestamp": submit_ts,
"commentText": "",
"nodeKey": "",
"taskType": "",
}
]
reached_keys = {
(t.node_key or "").strip() for t in sorted_tasks if (t.node_key or "").strip()
}
for t in sorted_tasks:
items.append(
{
"kind": "task",
"roundNo": rn,
"title": t.node_name or t.node_key or "-",
"status": t.status,
"assigneeUserId": t.assignee_user_id,
"assigneeUserName": name_map.get(t.assignee_user_id or "", ""),
"timestamp": _fmt_dt(t.update_time or t.create_time),
"commentText": t.comment_text or "",
"nodeKey": t.node_key or "",
"taskType": t.task_type or "",
"resolveSource": getattr(t, "resolve_source", None) or "",
"taskId": t.id,
}
)
if inst.status == "running" and isinstance(flow_root, dict):
_append_pending_nodes(
items, flow_root=flow_root, reached_keys=reached_keys, round_no=rn
)
if inst.status == "completed":
items.append(_terminal_item("done", "流程结束", inst))
elif inst.status == "rejected":
items.append(_terminal_item("rejected", "流程结束(已拒绝)", inst))
elif inst.status == "cancelled":
items.append(_terminal_item("cancelled", "已撤销", inst))
elif inst.status == "returned":
items.append(_terminal_item("returned", "已打回", inst))
return items
def _add_instance_event(
db: Session,
*,
inst: OaInstance,
event_type: str,
actor_user_id: Optional[str],
round_no: int,
comment: Optional[str] = None,
task_id: Optional[str] = None,
node_key: Optional[str] = None,
) -> OaInstanceEvent:
now = datetime.now()
ev = OaInstanceEvent(
id=generate_uuid(),
instance_id=inst.id,
round_no=round_no,
event_type=event_type,
actor_user_id=actor_user_id,
task_id=task_id,
node_key=node_key,
comment_text=(comment or "").strip() or None,
tenant_no=inst.tenant_no,
order_num=0,
remark=None,
create_by=actor_user_id,
create_dept_id=inst.initiator_dept_id or "",
create_time=now,
update_by=actor_user_id,
update_time=now,
del_flag=0,
)
db.add(ev)
return ev
def _advanced(ver: OaAppVersion) -> Dict[str, Any]:
data = _loads(ver.advanced_settings_json, {}) or {}
if not isinstance(data, dict):
return dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
merged = dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
merged.update(data)
return merged
def _check_initiator_scope(app: OaApp, user_id: str) -> None:
scope = (app.initiator_scope or "all").strip()
if scope == "all":
return
ids = _loads(app.initiator_user_ids, []) or []
if user_id not in ids:
raise OaEngineError("您不在该审批的发起范围内")
def _instance_to_dict(inst: OaInstance, *, app_name: str = "") -> Dict[str, Any]:
return {
"id": inst.id,
"appId": inst.app_id,
"appVersionId": inst.app_version_id,
"appName": app_name,
"title": inst.title or "",
"initiatorUserId": inst.initiator_user_id,
"initiatorDeptId": inst.initiator_dept_id or "",
"formDataJson": _loads(inst.form_data_json, {}),
"status": inst.status,
"sourceInstanceId": getattr(inst, "source_instance_id", None) or "",
"currentRound": int(getattr(inst, "current_round", None) or 1),
"createTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "",
"updateTime": inst.update_time.strftime("%Y-%m-%d %H:%M:%S") if inst.update_time else "",
}
def _task_to_dict(t: OaTask, *, assignee_user_name: str = "") -> Dict[str, Any]:
return {
"id": t.id,
"instanceId": t.instance_id,
"nodeKey": t.node_key or "",
"nodeName": t.node_name or "",
"taskType": t.task_type,
"assigneeUserId": t.assignee_user_id,
"assigneeUserName": assignee_user_name or t.assignee_user_id or "",
"status": t.status,
"commentText": t.comment_text or "",
"resolveSource": t.resolve_source or "",
"signMode": t.sign_mode or "",
"createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
"updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
}
_INSTANCE_STATUS_NODE_LABEL = {
"completed": "已完成",
"rejected": "已拒绝",
"returned": "已打回",
"cancelled": "已撤销",
}
def _instance_progress_summary(
inst: OaInstance,
*,
app_name: str,
initiator_user_name: str,
pending_approval_tasks: List[OaTask],
name_map: Dict[str, str],
) -> Dict[str, Any]:
status = (inst.status or "").strip()
apply_time = (
inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else ""
)
if status == "running":
node_names: List[str] = []
seen_nodes = set()
approver_names: List[str] = []
for t in pending_approval_tasks:
nn = (t.node_name or t.node_key or "").strip()
if nn and nn not in seen_nodes:
seen_nodes.add(nn)
node_names.append(nn)
uid = (t.assignee_user_id or "").strip()
if uid:
approver_names.append(name_map.get(uid, uid))
current_node = "、".join(node_names) if node_names else "-"
current_approvers = "、".join(approver_names) if approver_names else "-"
else:
current_node = _INSTANCE_STATUS_NODE_LABEL.get(status, status or "-")
current_approvers = "-"
return {
"appName": app_name or (inst.title or ""),
"initiatorUserId": inst.initiator_user_id or "",
"initiatorUserName": initiator_user_name
or name_map.get(inst.initiator_user_id or "", inst.initiator_user_id or ""),
"applyTime": apply_time,
"instanceStatus": status,
"currentNodeName": current_node,
"currentApproverNames": current_approvers,
}
def _enrich_list_rows(
db: Session,
*,
tenant_no: str,
rows: List[Dict[str, Any]],
instances: List[OaInstance],
) -> List[Dict[str, Any]]:
if not rows:
return []
inst_by_id = {i.id: i for i in instances}
app_ids = list({i.app_id for i in instances if i.app_id})
apps = (
db.query(OaApp)
.filter(
OaApp.id.in_(app_ids),
OaApp.tenant_no == tenant_no,
OaApp.del_flag == 0,
)
.all()
if app_ids
else []
)
app_name_by_id = {a.id: (a.app_name or "") for a in apps}
inst_ids = list(inst_by_id.keys())
pending_tasks = (
db.query(OaTask)
.filter(
OaTask.tenant_no == tenant_no,
OaTask.del_flag == 0,
OaTask.instance_id.in_(inst_ids),
OaTask.task_type == "approval",
OaTask.status == "pending",
)
.order_by(OaTask.create_time.asc())
.all()
if inst_ids
else []
)
pending_by_inst: Dict[str, List[OaTask]] = {}
for t in pending_tasks:
pending_by_inst.setdefault(t.instance_id, []).append(t)
user_ids: List[str] = []
for i in instances:
if i.initiator_user_id:
user_ids.append(i.initiator_user_id)
for t in pending_tasks:
if t.assignee_user_id:
user_ids.append(t.assignee_user_id)
name_map = _user_name_map(db, user_ids)
out: List[Dict[str, Any]] = []
for row in rows:
iid = row.get("instanceId") or row.get("id")
inst = inst_by_id.get(iid)
if not inst:
out.append(row)
continue
summary = _instance_progress_summary(
inst,
app_name=app_name_by_id.get(inst.app_id, ""),
initiator_user_name=name_map.get(inst.initiator_user_id or "", ""),
pending_approval_tasks=pending_by_inst.get(inst.id, []),
name_map=name_map,
)
merged = dict(row)
merged.update(summary)
out.append(merged)
return out
def _user_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
ids = [uid for uid in {str(u or "").strip() for u in user_ids} if uid]
if not ids:
return {}
rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all()
return {r.id: (r.nick_name or r.id) for r in rows}
def list_templates(db: Session, *, tenant_no: str, user_id: str) -> List[Dict[str, Any]]:
apps = (
db.query(OaApp)
.filter(OaApp.tenant_no == tenant_no, OaApp.del_flag == 0, OaApp.status == "1")
.order_by(OaApp.order_num.asc(), OaApp.create_time.desc())
.all()
)
out: List[Dict[str, Any]] = []
for app in apps:
pub = oa_crud._find_latest_published(db, app.id, tenant_no)
if not pub:
continue
try:
_check_initiator_scope(app, user_id)
except OaEngineError:
continue
out.append(
{
"id": app.id,
"appName": app.app_name,
"groupId": app.group_id or "",
"iconUrl": app.icon_url or "",
"description": app.description or "",
"versionId": pub.id,
"versionNo": pub.version_no,
}
)
return out
def get_template_detail(
db: Session, *, tenant_no: str, app_id: str, user_id: str
) -> Dict[str, Any]:
app = oa_crud.get_app(db, app_id, tenant_no)
if not app or app.status != "1":
raise OaEngineError("应用不存在或已停用")
_check_initiator_scope(app, user_id)
pub = oa_crud._find_latest_published(db, app_id, tenant_no)
if not pub:
raise OaEngineError("应用尚未发布")
return {
"id": app.id,
"appName": app.app_name,
"groupId": app.group_id or "",
"iconUrl": app.icon_url or "",
"description": app.description or "",
"versionId": pub.id,
"versionNo": pub.version_no,
"formSchemaJson": _loads(pub.form_schema_json, {}),
"flowTreeJson": _loads(pub.flow_tree_json, {}),
"advancedSettingsJson": _advanced(pub),
}
_ASSIGNEE_TYPE_LABEL = {
"users": "指定成员",
"oa_roles": "工作流角色",
"dept_leader": "发起人部门负责人",
"direct_supervisor": "发起人的主管",
"initiator_select": "发起人自选",
}
def _role_name_map(db: Session, tenant_no: str, role_ids: List[str], codes: List[str]) -> List[str]:
names: List[str] = []
ids = [str(x).strip() for x in (role_ids or []) if str(x).strip()]
code_list = [str(c).strip().upper() for c in (codes or []) if str(c).strip()]
if ids:
rows = (
db.query(OaRole)
.filter(OaRole.tenant_no == tenant_no, OaRole.id.in_(ids), OaRole.del_flag == 0)
.all()
)
names.extend([(r.role_name or r.role_code or r.id) for r in rows])
if code_list:
rows = (
db.query(OaRole)
.filter(
OaRole.tenant_no == tenant_no,
OaRole.role_code.in_(code_list),
OaRole.del_flag == 0,
)
.all()
)
for r in rows:
label = r.role_name or r.role_code or r.id
if label not in names:
names.append(label)
return names
def _assignee_meta(db: Session, tenant_no: str, cfg: Dict[str, Any]) -> str:
assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
base = _ASSIGNEE_TYPE_LABEL.get(assignee_type, assignee_type or "未配置审批人")
if assignee_type == "oa_roles":
names = _role_name_map(
db,
tenant_no,
cfg.get("oaRoleIds") or cfg.get("oa_role_ids") or [],
cfg.get("oaRoleCodes") or cfg.get("oa_role_codes") or [],
)
if names:
return f"{base} · {'/'.join(names)}"
return base
def _preview_people(db: Session, user_ids: List[str]) -> List[Dict[str, str]]:
name_map = _user_name_map(db, user_ids)
out: List[Dict[str, str]] = []
for uid in user_ids:
out.append({"id": uid, "nickName": name_map.get(uid) or uid})
return out
def _walk_preview_nodes(
db: Session,
*,
node: Optional[Dict[str, Any]],
tenant_no: str,
initiator_user_id: str,
initiator_dept_id: Optional[str],
advanced: Dict[str, Any],
initiator_select_user_ids: Optional[List[str]],
steps: List[Dict[str, Any]],
) -> None:
cur = node
while cur and isinstance(cur, dict):
ntype = _node_type(cur)
if ntype == "start":
people = _preview_people(db, [initiator_user_id] if initiator_user_id else [])
steps.append(
{
"id": cur.get("id") or "",
"type": "start",
"name": cur.get("name") or "发起人",
"signMode": "",
"assigneeType": "",
"meta": "",
"people": people,
"resolveError": "",
}
)
cur = cur.get("child")
continue
if ntype == "end":
steps.append(
{
"id": cur.get("id") or "",
"type": "end",
"name": cur.get("name") or "结束",
"signMode": "",
"assigneeType": "",
"meta": "",
"people": [],
"resolveError": "",
}
)
break
if ntype == "condition":
groups = cur.get("conditionGroups") or []
for g in groups:
if not isinstance(g, dict):
continue
steps.append(
{
"id": f"{cur.get('id') or 'cond'}_{(g.get('name') or '分支')}",
"type": "branch",
"name": g.get("name") or "条件分支",
"signMode": "",
"assigneeType": "",
"meta": "条件分支(实际路径以提交时表单为准)",
"people": [],
"resolveError": "",
}
)
child = g.get("child")
if isinstance(child, dict):
_walk_preview_nodes(
db,
node=child,
tenant_no=tenant_no,
initiator_user_id=initiator_user_id,
initiator_dept_id=initiator_dept_id,
advanced=advanced,
initiator_select_user_ids=initiator_select_user_ids,
steps=steps,
)
cur = cur.get("child")
continue
if ntype in ("approval", "cc"):
cfg = cur.get("approverConfig") or {}
assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
sign_mode = (cfg.get("signMode") or cfg.get("sign_mode") or "").strip()
if ntype == "cc":
sign_mode = "cc"
elif not sign_mode and assignee_type != "initiator_select":
sign_mode = "or"
meta = _assignee_meta(db, tenant_no, cfg)
people: List[Dict[str, str]] = []
resolve_error = ""
if assignee_type == "initiator_select":
ids = initiator_select_user_ids or []
people = _preview_people(db, [str(x) for x in ids if str(x).strip()])
if not people:
meta = meta or "发起人自选"
else:
try:
assignees, _meta = resolve_assignees(
db,
tenant_no=tenant_no,
assignee_config=cfg,
initiator_user_id=initiator_user_id,
initiator_dept_id=initiator_dept_id,
advanced=advanced,
initiator_select_user_ids=initiator_select_user_ids,
)
people = _preview_people(db, assignees)
except AssigneeResolveError as e:
resolve_error = str(e)
steps.append(
{
"id": cur.get("id") or "",
"type": ntype,
"name": cur.get("name") or ("抄送人" if ntype == "cc" else "审批人"),
"signMode": sign_mode,
"assigneeType": assignee_type,
"meta": meta,
"people": people,
"resolveError": resolve_error,
}
)
cur = cur.get("child")
continue
cur = cur.get("child")
def preview_start_flow(
db: Session,
*,
tenant_no: str,
app_id: str,
initiator_user_id: str,
initiator_dept_id: Optional[str] = None,
initiator_select_user_ids: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""发起前预览审批链(解析具体人员;条件分支展开展示)。"""
app = oa_crud.get_app(db, app_id, tenant_no)
if not app or app.status != "1":
raise OaEngineError("应用不存在或已停用")
_check_initiator_scope(app, initiator_user_id)
pub = oa_crud._find_latest_published(db, app_id, tenant_no)
if not pub:
raise OaEngineError("应用尚未发布")
initiator = (
db.query(SysUser)
.filter(
SysUser.id == initiator_user_id,
SysUser.tenant_no == tenant_no,
SysUser.del_flag == 0,
)
.first()
)
dept_id = (initiator_dept_id or "").strip() or (
(initiator.dept_id or "") if initiator else ""
)
flow = _loads(pub.flow_tree_json, {}) or {}
root = flow.get("root") if isinstance(flow, dict) else None
if not isinstance(root, dict):
root = flow if isinstance(flow, dict) else {}
steps: List[Dict[str, Any]] = []
_walk_preview_nodes(
db,
node=root,
tenant_no=tenant_no,
initiator_user_id=initiator_user_id,
initiator_dept_id=dept_id or None,
advanced=_advanced(pub),
initiator_select_user_ids=initiator_select_user_ids,
steps=steps,
)
return {"steps": steps}
def _create_tasks_for_node(
db: Session,
*,
inst: OaInstance,
node: Dict[str, Any],
ver: OaAppVersion,
user_id: Optional[str],
initiator_select_user_ids: Optional[List[str]] = None,
order_num_start: Optional[int] = None,
) -> List[OaTask]:
ntype = _node_type(node)
now = datetime.now()
created: List[OaTask] = []
if ntype == "cc":
task_type = "cc"
elif ntype == "approval":
task_type = "approval"
else:
return created
cfg = node.get("approverConfig") or {}
advanced = _advanced(ver)
try:
assignees, meta = resolve_assignees(
db,
tenant_no=inst.tenant_no,
assignee_config=cfg,
initiator_user_id=inst.initiator_user_id,
initiator_dept_id=inst.initiator_dept_id,
advanced=advanced,
initiator_select_user_ids=initiator_select_user_ids,
)
except AssigneeResolveError as e:
raise OaEngineError(str(e)) from e
sign_mode = (cfg.get("signMode") or cfg.get("sign_mode") or "or").strip() or "or"
resolve_source = meta.get("resolve_source")
status = "done" if task_type == "cc" else "pending"
seq = order_num_start if order_num_start is not None else _next_task_order_num(db, inst.id)
for uid in assignees:
t = OaTask(
id=generate_uuid(),
instance_id=inst.id,
node_key=node.get("id") or "",
node_name=node.get("name") or "",
task_type=task_type,
assignee_user_id=uid,
status=status,
comment_text=None,
resolve_source=resolve_source,
sign_mode=sign_mode if task_type == "approval" else None,
round_no=int(getattr(inst, "current_round", None) or 1),
tenant_no=inst.tenant_no,
order_num=seq,
remark=None,
create_by=user_id,
create_dept_id=inst.initiator_dept_id or "",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
seq += 1
db.add(t)
created.append(t)
return created
def _match_condition_group(groups: List[Any], form_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
default_g = None
for g in groups:
if not isinstance(g, dict):
continue
rules = g.get("rules")
when = g.get("when")
# 空规则 = 默认分支
if (not rules and not when) or g.get("isDefault"):
default_g = g
continue
expr = when
if not expr and isinstance(rules, str):
expr = rules
if not expr and isinstance(rules, list) and rules:
# 简化:每条 {field, op, value} 用 AND
parts = []
for r in rules:
if not isinstance(r, dict):
continue
field = r.get("field")
op = r.get("op") or "=="
val = r.get("value")
if field is None:
continue
if isinstance(val, str):
parts.append(f"{{{field}}} {op} '{val}'")
else:
parts.append(f"{{{field}}} {op} {val}")
expr = " and ".join(parts) if parts else None
if not expr:
default_g = g
continue
try:
if eval_formula(str(expr), form_data):
return g
except FormulaError:
continue
return default_g
def advance_from(
db: Session,
*,
inst: OaInstance,
ver: OaAppVersion,
from_node: Optional[Dict[str, Any]],
user_id: Optional[str],
initiator_select_user_ids: Optional[List[str]] = None,
) -> None:
"""从某节点的下一跳开始推进;from_node 为 None 表示从 start 进入。"""
root = _flow_root(_loads(ver.flow_tree_json, {}))
form_data = _loads(inst.form_data_json, {}) or {}
def next_node(node: Dict[str, Any]) -> Optional[Dict[str, Any]]:
ntype = _node_type(node)
if ntype == "condition":
groups = node.get("conditionGroups") or []
g = _match_condition_group(groups, form_data)
if not g or not isinstance(g.get("child"), dict):
raise OaEngineError("条件分支无法匹配出线")
return g["child"]
child = node.get("child")
return child if isinstance(child, dict) else None
# 起点
if from_node is None:
cur = next_node(root) if _node_type(root) == "start" else root
else:
cur = next_node(from_node)
# 同一次推进里可能连续落抄送+审批:共用递增序号,避免 create_time 同秒时乱序
next_order = _next_task_order_num(db, inst.id)
safety = 0
while cur and safety < 50:
safety += 1
ntype = _node_type(cur)
if ntype == "end":
inst.status = "completed"
inst.update_time = datetime.now()
db.add(inst)
return
if ntype == "condition":
cur = next_node(cur)
continue
if ntype in ("approval", "cc"):
tasks = _create_tasks_for_node(
db,
inst=inst,
node=cur,
ver=ver,
user_id=user_id,
initiator_select_user_ids=initiator_select_user_ids,
order_num_start=next_order,
)
next_order += max(len(tasks), 1)
if ntype == "cc":
# 抄送不阻断,继续往下
cur = next_node(cur)
continue
if not tasks:
raise OaEngineError("审批节点未生成任务")
return
# 未知类型跳过子节点
cur = next_node(cur)
# 无结束也无待办:异常结束
if inst.status == "running":
raise OaEngineError("流程推进异常:未到达结束且无待办任务")
def start_instance(
db: Session,
*,
tenant_no: str,
app_id: str,
initiator_user_id: str,
initiator_dept_id: Optional[str] = None,
form_data: Optional[Dict[str, Any]] = None,
title: Optional[str] = None,
initiator_select_user_ids: Optional[List[str]] = None,
related_instance_ids: Optional[List[str]] = None,
source_instance_id: Optional[str] = None,
) -> Dict[str, Any]:
# source_instance_id 保留入参兼容,不再写入关联(复制开新单不挂旧单)
_ = source_instance_id
app = oa_crud.get_app(db, app_id, tenant_no)
if not app or app.status != "1":
raise OaEngineError("应用不存在或已停用")
_check_initiator_scope(app, initiator_user_id)
ver = oa_crud._find_latest_published(db, app_id, tenant_no)
if not ver:
raise OaEngineError("应用尚未发布,无法发起")
schema = _loads(ver.form_schema_json, {})
data = dict(form_data or {})
try:
data = apply_formulas_and_linkages(schema, data)
except FormulaError as e:
raise OaEngineError(f"公式计算失败: {e}") from e
user = (
db.query(SysUser)
.filter(SysUser.id == initiator_user_id, SysUser.del_flag == 0)
.first()
)
dept_id = (initiator_dept_id or (user.dept_id if user else "") or "").strip() or None
now = datetime.now()
title_text = (title or app.app_name or "").strip()
inst = OaInstance(
id=generate_uuid(),
app_id=app_id,
app_version_id=ver.id,
title=title_text[:200],
initiator_user_id=initiator_user_id,
initiator_dept_id=dept_id,
form_data_json=_dumps(data),
status="running",
source_instance_id=None,
current_round=1,
tenant_no=tenant_no,
order_num=0,
remark=None,
create_by=initiator_user_id,
create_dept_id=dept_id or "",
create_time=now,
update_by=initiator_user_id,
update_time=now,
del_flag=0,
)
db.add(inst)
for to_id in related_instance_ids or []:
to_id = (to_id or "").strip()
if not to_id:
continue
db.add(
OaRelation(
id=generate_uuid(),
from_instance_id=inst.id,
to_instance_id=to_id,
field_key=None,
tenant_no=tenant_no,
order_num=0,
remark=None,
create_by=initiator_user_id,
create_dept_id=dept_id or "",
create_time=now,
update_by=initiator_user_id,
update_time=now,
del_flag=0,
)
)
_add_instance_event(
db,
inst=inst,
event_type="submit",
actor_user_id=initiator_user_id,
round_no=1,
)
advance_from(
db,
inst=inst,
ver=ver,
from_node=None,
user_id=initiator_user_id,
initiator_select_user_ids=initiator_select_user_ids,
)
db.commit()
db.refresh(inst)
return get_instance_detail(db, inst.id, tenant_no=tenant_no)
def _get_instance(db: Session, instance_id: str, tenant_no: str) -> OaInstance:
inst = (
db.query(OaInstance)
.filter(OaInstance.id == instance_id, OaInstance.tenant_no == tenant_no, OaInstance.del_flag == 0)
.first()
)
if not inst:
raise OaEngineError("实例不存在")
return inst
def get_instance_detail(db: Session, instance_id: str, *, tenant_no: str) -> Dict[str, Any]:
inst = _get_instance(db, instance_id, tenant_no)
app = oa_crud.get_app(db, inst.app_id, tenant_no)
ver = (
db.query(OaAppVersion)
.filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
.first()
)
tasks = (
db.query(OaTask)
.filter(OaTask.instance_id == instance_id, OaTask.del_flag == 0)
.all()
)
flow_root = None
if ver:
try:
flow_root = _flow_root(_loads(ver.flow_tree_json, {}))
except OaEngineError:
flow_root = None
tasks = _sort_tasks_for_detail(tasks, flow_root=flow_root)
rels = (
db.query(OaRelation)
.filter(OaRelation.from_instance_id == instance_id, OaRelation.del_flag == 0)
.all()
)
name_map = _user_name_map(
db,
[t.assignee_user_id for t in tasks if t.assignee_user_id]
+ ([inst.initiator_user_id] if inst.initiator_user_id else []),
)
data = _instance_to_dict(inst, app_name=app.app_name if app else "")
data["initiatorUserName"] = name_map.get(inst.initiator_user_id or "", inst.initiator_user_id or "")
data["tasks"] = [
_task_to_dict(t, assignee_user_name=name_map.get(t.assignee_user_id or "", ""))
for t in tasks
]
data["relatedInstanceIds"] = [r.to_instance_id for r in rels]
data["formSchemaJson"] = _loads(ver.form_schema_json, {}) if ver else {}
adv = _advanced(ver) if ver else dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
data["advancedSettingsJson"] = adv
data["allowAddSign"] = bool(adv.get("allowAddSign", True))
data["allowReturn"] = bool(adv.get("allowReturn", True))
events = (
db.query(OaInstanceEvent)
.filter(OaInstanceEvent.instance_id == instance_id, OaInstanceEvent.del_flag == 0)
.order_by(OaInstanceEvent.create_time.asc())
.all()
)
event_actors = [e.actor_user_id for e in events if e.actor_user_id]
if event_actors:
name_map.update(_user_name_map(db, event_actors))
data["progress"] = build_progress_items(
inst=inst,
flow_root=flow_root,
tasks=tasks,
name_map=name_map,
events=events,
)
# 不再暴露旧单/重提单关联跳转
data["sourceInstanceId"] = ""
data["resubmittedInstanceId"] = ""
return data
def list_tasks(
db: Session,
*,
tenant_no: str,
user_id: str,
kind: str,
) -> List[Dict[str, Any]]:
kind = (kind or "").strip()
if kind == "pending":
q = db.query(OaTask).filter(
OaTask.tenant_no == tenant_no,
OaTask.del_flag == 0,
OaTask.assignee_user_id == user_id,
OaTask.task_type == "approval",
OaTask.status == "pending",
)
elif kind == "done":
q = db.query(OaTask).filter(
OaTask.tenant_no == tenant_no,
OaTask.del_flag == 0,
OaTask.assignee_user_id == user_id,
OaTask.task_type == "approval",
OaTask.status.in_(("approved", "rejected", "returned")),
)
elif kind == "cc":
q = db.query(OaTask).filter(
OaTask.tenant_no == tenant_no,
OaTask.del_flag == 0,
OaTask.assignee_user_id == user_id,
OaTask.task_type == "cc",
)
elif kind == "mine":
insts = (
db.query(OaInstance)
.filter(
OaInstance.tenant_no == tenant_no,
OaInstance.del_flag == 0,
OaInstance.initiator_user_id == user_id,
)
.order_by(OaInstance.create_time.desc())
.all()
)
rows = [_instance_to_dict(i) for i in insts]
return _enrich_list_rows(db, tenant_no=tenant_no, rows=rows, instances=insts)
else:
raise OaEngineError(f"不支持的列表类型: {kind}")
rows_orm = q.order_by(OaTask.create_time.desc()).all()
rows = [_task_to_dict(t) for t in rows_orm]
inst_ids = list({t.instance_id for t in rows_orm if t.instance_id})
insts = (
db.query(OaInstance)
.filter(
OaInstance.tenant_no == tenant_no,
OaInstance.del_flag == 0,
OaInstance.id.in_(inst_ids),
)
.all()
if inst_ids
else []
)
return _enrich_list_rows(db, tenant_no=tenant_no, rows=rows, instances=insts)
def complete_task(
db: Session,
task_id: str,
*,
tenant_no: str,
user_id: str,
action: str,
comment: Optional[str] = None,
) -> Dict[str, Any]:
action = (action or "").strip()
if action not in ("approve", "reject"):
raise OaEngineError("不支持的操作")
task = (
db.query(OaTask)
.filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0)
.first()
)
if not task:
raise OaEngineError("任务不存在")
if task.assignee_user_id != user_id:
raise OaEngineError("无权处理该任务")
if task.task_type != "approval" or task.status != "pending":
raise OaEngineError("任务状态不可处理")
inst = _get_instance(db, task.instance_id, tenant_no)
if inst.status != "running":
raise OaEngineError("实例已结束")
ver = (
db.query(OaAppVersion)
.filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
.first()
)
if not ver:
raise OaEngineError("流程版本不存在")
advanced = _advanced(ver)
if advanced.get("commentRequired") and not (comment or "").strip():
raise OaEngineError("请填写审批意见")
now = datetime.now()
task.comment_text = (comment or "").strip() or None
task.status = "approved" if action == "approve" else "rejected"
task.update_by = user_id
task.update_time = now
db.add(task)
# Session 默认 autoflush=False:不先 flush 时,后续按 status=pending 的查询仍会命中本任务,
# 或签「取消同节点其余待办」会把本人已同意/已拒绝覆盖成 done,导致「已处理」列表查不到。
db.flush()
if action == "reject":
# 同节点其他待办作废(排除当前任务)
siblings = (
db.query(OaTask)
.filter(
OaTask.instance_id == inst.id,
OaTask.node_key == task.node_key,
OaTask.id != task.id,
OaTask.status == "pending",
OaTask.del_flag == 0,
)
.all()
)
for s in siblings:
s.status = "done"
s.update_by = user_id
s.update_time = now
db.add(s)
inst.status = "rejected"
inst.update_by = user_id
inst.update_time = now
db.add(inst)
db.commit()
return get_instance_detail(db, inst.id, tenant_no=tenant_no)
# 或签:一人通过即可;会签:全部通过
sign_mode = (task.sign_mode or "or").strip() or "or"
pending_same = (
db.query(OaTask)
.filter(
OaTask.instance_id == inst.id,
OaTask.node_key == task.node_key,
OaTask.task_type == "approval",
OaTask.id != task.id,
OaTask.status == "pending",
OaTask.del_flag == 0,
)
.count()
)
if sign_mode == "and" and pending_same > 0:
db.commit()
return get_instance_detail(db, inst.id, tenant_no=tenant_no)
# 或签:取消同节点其余待办(排除当前任务)
if sign_mode != "and":
for s in (
db.query(OaTask)
.filter(
OaTask.instance_id == inst.id,
OaTask.node_key == task.node_key,
OaTask.id != task.id,
OaTask.status == "pending",
OaTask.del_flag == 0,
)
.all()
):
s.status = "done"
s.update_by = user_id
s.update_time = now
db.add(s)
root = _flow_root(_loads(ver.flow_tree_json, {}))
node = _find_node(root, task.node_key or "")
if not node:
raise OaEngineError("找不到对应流程节点")
advance_from(db, inst=inst, ver=ver, from_node=node, user_id=user_id)
db.commit()
return get_instance_detail(db, inst.id, tenant_no=tenant_no)
def add_sign(
db: Session,
task_id: str,
*,
tenant_no: str,
user_id: str,
user_ids: List[str],
comment: Optional[str] = None,
) -> Dict[str, Any]:
"""当前待办处理人加签:为同节点新增待办审批人。"""
task = (
db.query(OaTask)
.filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0)
.first()
)
if not task:
raise OaEngineError("任务不存在")
if task.assignee_user_id != user_id:
raise OaEngineError("无权加签该任务")
if task.task_type != "approval" or task.status != "pending":
raise OaEngineError("仅待处理的审批任务可加签")
inst = _get_instance(db, task.instance_id, tenant_no)
if inst.status != "running":
raise OaEngineError("实例已结束")
ver = (
db.query(OaAppVersion)
.filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
.first()
)
if not ver:
raise OaEngineError("流程版本不存在")
advanced = _advanced(ver)
if not advanced.get("allowAddSign", True):
raise OaEngineError("该应用不允许加签")
ids = []
seen = set()
for uid in user_ids or []:
uid = (uid or "").strip()
if not uid or uid in seen:
continue
seen.add(uid)
ids.append(uid)
if not ids:
raise OaEngineError("请选择加签人")
existing = {
r.assignee_user_id
for r in db.query(OaTask)
.filter(
OaTask.instance_id == inst.id,
OaTask.node_key == task.node_key,
OaTask.del_flag == 0,
OaTask.status == "pending",
)
.all()
}
now = datetime.now()
created = 0
next_order = _next_task_order_num(db, inst.id)
for uid in ids:
if uid in existing:
continue
# 校验用户有效
u = (
db.query(SysUser)
.filter(
SysUser.id == uid,
SysUser.tenant_no == tenant_no,
SysUser.del_flag == 0,
SysUser.status == "1",
)
.first()
)
if not u:
raise OaEngineError(f"加签用户无效:{uid}")
db.add(
OaTask(
id=generate_uuid(),
instance_id=inst.id,
node_key=task.node_key,
node_name=task.node_name,
task_type="approval",
assignee_user_id=uid,
status="pending",
comment_text=None,
resolve_source="add_sign",
sign_mode=task.sign_mode or "or",
round_no=int(getattr(inst, "current_round", None) or getattr(task, "round_no", None) or 1),
tenant_no=tenant_no,
order_num=next_order,
remark=(comment or "").strip() or None,
create_by=user_id,
create_dept_id=inst.initiator_dept_id or "",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
)
next_order += 1
created += 1
if created == 0:
raise OaEngineError("所选用户已在当前节点待办中")
# 加签后默认按会签语义收口更稳妥:保留原 sign_mode;若原或签则一人通过仍可结束
if comment:
task.comment_text = ((task.comment_text or "") + f"\n[加签说明] {comment}").strip()
task.update_by = user_id
task.update_time = now
db.add(task)
db.commit()
return get_instance_detail(db, inst.id, tenant_no=tenant_no)
def return_task(
db: Session,
task_id: str,
*,
tenant_no: str,
user_id: str,
comment: Optional[str] = None,
) -> Dict[str, Any]:
"""审批人打回:本单终态 returned;再办须复制表单开新实例。"""
task = (
db.query(OaTask)
.filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0)
.first()
)
if not task:
raise OaEngineError("任务不存在")
if task.assignee_user_id != user_id:
raise OaEngineError("无权打回该任务")
if task.task_type != "approval" or task.status != "pending":
raise OaEngineError("仅待处理的审批任务可打回")
inst = _get_instance(db, task.instance_id, tenant_no)
if inst.status != "running":
raise OaEngineError("仅进行中的审批可打回")
ver = (
db.query(OaAppVersion)
.filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
.first()
)
if not ver:
raise OaEngineError("流程版本不存在")
advanced = _advanced(ver)
if not advanced.get("allowReturn", True):
raise OaEngineError("该应用不允许打回")
comment_text = (comment or "").strip()
if not comment_text:
raise OaEngineError("打回意见必填")
now = datetime.now()
round_no = int(getattr(inst, "current_round", None) or getattr(task, "round_no", None) or 1)
task.status = "returned"
task.comment_text = comment_text
task.update_by = user_id
task.update_time = now
db.add(task)
for t in (
db.query(OaTask)
.filter(
OaTask.instance_id == inst.id,
OaTask.status == "pending",
OaTask.del_flag == 0,
OaTask.id != task.id,
)
.all()
):
t.status = "cancelled"
t.update_by = user_id
t.update_time = now
db.add(t)
inst.status = "returned"
inst.update_by = user_id
inst.update_time = now
db.add(inst)
_add_instance_event(
db,
inst=inst,
event_type="return",
actor_user_id=user_id,
round_no=round_no,
comment=comment_text,
task_id=task.id,
node_key=task.node_key,
)
db.commit()
return get_instance_detail(db, inst.id, tenant_no=tenant_no)
def resubmit_instance(
db: Session,
instance_id: str,
*,
tenant_no: str,
user_id: str,
form_data: Optional[Dict[str, Any]] = None,
initiator_select_user_ids: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""已废弃:同实例再次提交。请复制表单重新发起新实例。"""
_ = (db, instance_id, tenant_no, user_id, form_data, initiator_select_user_ids)
raise OaEngineError("同实例再次提交已废弃,请复制表单重新发起")
def withdraw_instance(
db: Session, instance_id: str, *, tenant_no: str, user_id: str
) -> Dict[str, Any]:
inst = _get_instance(db, instance_id, tenant_no)
if inst.initiator_user_id != user_id:
raise OaEngineError("仅发起人可撤销")
if inst.status != "running":
raise OaEngineError("仅运行中的实例可撤销")
ver = (
db.query(OaAppVersion)
.filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
.first()
)
advanced = _advanced(ver) if ver else dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
if not advanced.get("allowWithdrawRunning", True):
raise OaEngineError("该应用不允许撤销进行中的审批")
now = datetime.now()
inst.status = "cancelled"
inst.update_by = user_id
inst.update_time = now
db.add(inst)
for t in (
db.query(OaTask)
.filter(OaTask.instance_id == inst.id, OaTask.status == "pending", OaTask.del_flag == 0)
.all()
):
t.status = "cancelled"
t.update_by = user_id
t.update_time = now
db.add(t)
_add_instance_event(
db,
inst=inst,
event_type="withdraw",
actor_user_id=user_id,
round_no=int(getattr(inst, "current_round", None) or 1),
)
db.commit()
return get_instance_detail(db, inst.id, tenant_no=tenant_no)