2424 lines
78 KiB
Python
2424 lines
78 KiB
Python
"""OA 审批运行时引擎（钉钉式竖向树）"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from collections import OrderedDict
|
||
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,
|
||
SysDept,
|
||
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):
|
||
"""引擎业务错误"""
|
||
|
||
|
||
_AUTO_PASS_COMMENT = "系统自动通过"
|
||
|
||
INITIATOR_CONDITION_FIELD = "__initiator__"
|
||
|
||
|
||
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 _normalize_initiator_select_by_node(raw: Any) -> Dict[str, List[str]]:
|
||
if not isinstance(raw, dict):
|
||
return {}
|
||
out: Dict[str, List[str]] = {}
|
||
for k, v in raw.items():
|
||
nid = str(k or "").strip()
|
||
if not nid or not isinstance(v, (list, tuple)):
|
||
continue
|
||
ids: List[str] = []
|
||
seen = set()
|
||
for x in v:
|
||
uid = str(x or "").strip()
|
||
if uid and uid not in seen:
|
||
seen.add(uid)
|
||
ids.append(uid)
|
||
out[nid] = ids
|
||
return out
|
||
|
||
|
||
def _ids_for_initiator_select_node(
|
||
select_by_node: Optional[Dict[str, List[str]]], node_id: str
|
||
) -> List[str]:
|
||
m = select_by_node or {}
|
||
nid = str(node_id or "").strip()
|
||
if not nid:
|
||
return []
|
||
return list(m.get(nid) or [])
|
||
|
||
|
||
def _load_initiator_select_by_node(inst: Any) -> Dict[str, List[str]]:
|
||
return _normalize_initiator_select_by_node(
|
||
_loads(getattr(inst, "initiator_select_json", None), {}) or {}
|
||
)
|
||
|
||
|
||
def _node_type(node: Dict[str, Any]) -> str:
|
||
return (node.get("type") or "").strip()
|
||
|
||
|
||
def _subtree_contains(node: Optional[Dict[str, Any]], node_id: str) -> bool:
|
||
"""节点子树（含自身、child、条件分支）是否包含指定 id。"""
|
||
if not isinstance(node, dict):
|
||
return False
|
||
if (node.get("id") or "") == node_id:
|
||
return True
|
||
child = node.get("child")
|
||
if isinstance(child, dict) and _subtree_contains(child, node_id):
|
||
return True
|
||
for g in node.get("conditionGroups") or []:
|
||
if isinstance(g, dict) and _subtree_contains(g.get("child"), node_id):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _merge_after_branch_end(
|
||
root: Dict[str, Any], end_id: str
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""若 end 落在某条件分支组内，且该条件节点有汇合后 child，则返回汇合节点。"""
|
||
|
||
def walk(node: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
|
||
if not isinstance(node, dict):
|
||
return None
|
||
if _node_type(node) == "condition":
|
||
for g in node.get("conditionGroups") or []:
|
||
if not isinstance(g, dict):
|
||
continue
|
||
if _subtree_contains(g.get("child"), end_id):
|
||
merge = node.get("child")
|
||
return merge if isinstance(merge, dict) else None
|
||
merge_child = node.get("child")
|
||
if isinstance(merge_child, dict):
|
||
return walk(merge_child)
|
||
return None
|
||
child = node.get("child")
|
||
if isinstance(child, dict):
|
||
return walk(child)
|
||
return None
|
||
|
||
return walk(root)
|
||
|
||
|
||
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", "end") or nid in reached_keys:
|
||
continue
|
||
if nt == "end":
|
||
title = node.get("name") or "结束"
|
||
elif nt == "cc":
|
||
title = node.get("name") or "抄送人"
|
||
else:
|
||
title = node.get("name") or "审批人"
|
||
items.append(
|
||
{
|
||
"kind": "pending_node",
|
||
"roundNo": round_no,
|
||
"title": title,
|
||
"status": "not_reached",
|
||
"assigneeUserId": "",
|
||
"assigneeUserName": "",
|
||
"timestamp": "",
|
||
"commentText": "",
|
||
"nodeKey": nid,
|
||
"taskType": nt,
|
||
"signMode": "",
|
||
"people": [],
|
||
}
|
||
)
|
||
|
||
|
||
def _pending_people_for_node(
|
||
db: Session,
|
||
*,
|
||
node: Dict[str, Any],
|
||
tenant_no: str,
|
||
initiator_user_id: str,
|
||
initiator_dept_id: Optional[str],
|
||
advanced: Dict[str, Any],
|
||
initiator_select_by_node: Optional[Dict[str, List[str]]],
|
||
) -> Tuple[List[Dict[str, Any]], str, str]:
|
||
"""解析未到达节点的展示人员与签批模式。返回 (people, sign_mode, resolve_source)。"""
|
||
ntype = _node_type(node)
|
||
cfg = node.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"
|
||
people: List[Dict[str, Any]] = []
|
||
node_select_ids = _ids_for_initiator_select_node(
|
||
initiator_select_by_node, node.get("id") or ""
|
||
)
|
||
if assignee_type == "initiator_select":
|
||
for uid in node_select_ids:
|
||
people.append(
|
||
{
|
||
"id": uid,
|
||
"nickName": "", # 下方统一补名
|
||
"status": "not_reached",
|
||
"timestamp": "",
|
||
"commentText": "",
|
||
"taskId": "",
|
||
}
|
||
)
|
||
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=node_select_ids,
|
||
)
|
||
for uid in assignees:
|
||
people.append(
|
||
{
|
||
"id": uid,
|
||
"nickName": "",
|
||
"status": "not_reached",
|
||
"timestamp": "",
|
||
"commentText": "",
|
||
"taskId": "",
|
||
}
|
||
)
|
||
except AssigneeResolveError:
|
||
people = []
|
||
if people:
|
||
name_map = _user_name_map(db, [p["id"] for p in people if p.get("id")])
|
||
for p in people:
|
||
p["nickName"] = name_map.get(p["id"], "") or p["id"]
|
||
return people, sign_mode, assignee_type
|
||
|
||
|
||
def _iter_resolved_path_nodes(
|
||
root: Optional[Dict[str, Any]],
|
||
form_data: Optional[Dict[str, Any]],
|
||
*,
|
||
initiator_user_id: Optional[str] = None,
|
||
) -> Optional[List[Dict[str, Any]]]:
|
||
"""沿条件命中路径收集节点；无法匹配时返回 None。"""
|
||
if not isinstance(root, dict):
|
||
return []
|
||
data = form_data or {}
|
||
out: List[Dict[str, Any]] = []
|
||
cur: Optional[Dict[str, Any]] = root
|
||
while cur and isinstance(cur, dict):
|
||
ntype = _node_type(cur)
|
||
if ntype == "end":
|
||
out.append(cur)
|
||
break
|
||
if ntype == "condition":
|
||
g = _match_condition_group(
|
||
cur.get("conditionGroups") or [], data, initiator_user_id=initiator_user_id
|
||
)
|
||
if not g:
|
||
return None
|
||
child = g.get("child")
|
||
if isinstance(child, dict):
|
||
sub = _iter_resolved_path_nodes(
|
||
child, data, initiator_user_id=initiator_user_id
|
||
)
|
||
if sub is None:
|
||
return None
|
||
out.extend(sub)
|
||
cur = cur.get("child")
|
||
continue
|
||
out.append(cur)
|
||
cur = cur.get("child")
|
||
return out
|
||
|
||
|
||
def _append_pending_nodes_on_path(
|
||
items: List[Dict[str, Any]],
|
||
*,
|
||
db: Session,
|
||
flow_root: Dict[str, Any],
|
||
reached_keys: set,
|
||
round_no: int,
|
||
tenant_no: str,
|
||
initiator_user_id: str,
|
||
initiator_dept_id: Optional[str],
|
||
advanced: Dict[str, Any],
|
||
form_data: Optional[Dict[str, Any]],
|
||
initiator_select_by_node: Optional[Dict[str, List[str]]],
|
||
) -> None:
|
||
"""按命中路径列出全部后续审批/抄送/结束节点（含人员），对齐发起预览。"""
|
||
path = _iter_resolved_path_nodes(
|
||
flow_root, form_data, initiator_user_id=initiator_user_id
|
||
)
|
||
if path is None:
|
||
already = {
|
||
(x.get("nodeKey") or "").strip()
|
||
for x in items
|
||
if x.get("kind") in ("task", "pending_node") and (x.get("nodeKey") or "").strip()
|
||
}
|
||
_append_pending_nodes(
|
||
items,
|
||
flow_root=flow_root,
|
||
reached_keys=reached_keys | already,
|
||
round_no=round_no,
|
||
)
|
||
return
|
||
|
||
for node in path:
|
||
ntype = _node_type(node)
|
||
if ntype not in ("approval", "cc", "end"):
|
||
continue
|
||
nid = (node.get("id") or "").strip()
|
||
if not nid or nid in reached_keys:
|
||
continue
|
||
if ntype == "end":
|
||
items.append(
|
||
{
|
||
"kind": "pending_node",
|
||
"roundNo": round_no,
|
||
"title": node.get("name") or "结束",
|
||
"status": "not_reached",
|
||
"assigneeUserId": "",
|
||
"assigneeUserName": "",
|
||
"timestamp": "",
|
||
"commentText": "",
|
||
"nodeKey": nid,
|
||
"taskType": "end",
|
||
"signMode": "",
|
||
"people": [],
|
||
}
|
||
)
|
||
continue
|
||
people, sign_mode, _src = _pending_people_for_node(
|
||
db,
|
||
node=node,
|
||
tenant_no=tenant_no,
|
||
initiator_user_id=initiator_user_id,
|
||
initiator_dept_id=initiator_dept_id,
|
||
advanced=advanced,
|
||
initiator_select_by_node=initiator_select_by_node,
|
||
)
|
||
items.append(
|
||
{
|
||
"kind": "pending_node",
|
||
"roundNo": round_no,
|
||
"title": node.get("name") or ("抄送人" if ntype == "cc" else "审批人"),
|
||
"status": "not_reached",
|
||
"assigneeUserId": "",
|
||
"assigneeUserName": "",
|
||
"timestamp": "",
|
||
"commentText": "",
|
||
"nodeKey": nid,
|
||
"taskType": ntype,
|
||
"signMode": sign_mode,
|
||
"people": people,
|
||
}
|
||
)
|
||
|
||
|
||
def _progress_person(t: Any, name_map: Dict[str, str]) -> Dict[str, Any]:
|
||
uid = (getattr(t, "assignee_user_id", None) or "").strip()
|
||
return {
|
||
"id": uid,
|
||
"nickName": name_map.get(uid, "") or uid,
|
||
"status": getattr(t, "status", None) or "",
|
||
"timestamp": _fmt_dt(getattr(t, "update_time", None) or getattr(t, "create_time", None)),
|
||
"commentText": getattr(t, "comment_text", None) or "",
|
||
"taskId": getattr(t, "id", None) or "",
|
||
"transferToUserId": "",
|
||
"transferToUserName": "",
|
||
}
|
||
|
||
|
||
def _enrich_transfer_targets(
|
||
people: List[Dict[str, Any]],
|
||
group: List[Any],
|
||
name_map: Dict[str, str],
|
||
) -> None:
|
||
"""为已转交人员补齐 transferTo*：同节点 resolve_source=transfer 且 create_by=转交人 的新任务处理人。"""
|
||
by_creator: Dict[str, List[Any]] = {}
|
||
for t in group:
|
||
if (getattr(t, "resolve_source", None) or "").strip() != "transfer":
|
||
continue
|
||
creator = (getattr(t, "create_by", None) or "").strip()
|
||
if not creator:
|
||
continue
|
||
by_creator.setdefault(creator, []).append(t)
|
||
|
||
for p in people:
|
||
if (p.get("status") or "") != "transferred":
|
||
continue
|
||
uid = (p.get("id") or "").strip()
|
||
if not uid:
|
||
continue
|
||
candidates = by_creator.get(uid) or []
|
||
if not candidates:
|
||
continue
|
||
# 连续转交时同一人可能只对应一次；若有多条取创建时间最接近转交时间的
|
||
p_ts = (p.get("timestamp") or "").strip()
|
||
|
||
def _score(t: Any) -> tuple:
|
||
cts = _fmt_dt(getattr(t, "create_time", None))
|
||
# 优先同秒/最近；无时间时靠后
|
||
if p_ts and cts:
|
||
return (0, abs(_dt_sort_key(cts) - _dt_sort_key(p_ts)), cts)
|
||
return (1, 0, cts or "")
|
||
|
||
target_task = sorted(candidates, key=_score)[0]
|
||
tid = (getattr(target_task, "assignee_user_id", None) or "").strip()
|
||
if not tid:
|
||
continue
|
||
p["transferToUserId"] = tid
|
||
p["transferToUserName"] = name_map.get(tid, "") or tid
|
||
|
||
|
||
def _dt_sort_key(s: str) -> int:
|
||
"""将 'YYYY-MM-DD HH:MM:SS' 粗转为可比较整数；非法则 0。"""
|
||
raw = (s or "").strip().replace("-", "").replace(":", "").replace(" ", "")
|
||
try:
|
||
return int(raw[:14]) if raw else 0
|
||
except ValueError:
|
||
return 0
|
||
|
||
|
||
def _aggregate_node_status(statuses: List[str]) -> str:
|
||
"""同节点多人任务收成卡片状态：拒绝/打回优先，其次待办，再已同意。"""
|
||
sset = [str(s or "").strip() for s in statuses]
|
||
if any(s == "rejected" for s in sset):
|
||
return "rejected"
|
||
if any(s == "returned" for s in sset):
|
||
return "returned"
|
||
if any(s in ("pending", "running") for s in sset):
|
||
return "pending"
|
||
if any(s == "approved" for s in sset):
|
||
return "approved"
|
||
if any(s == "completed" for s in sset):
|
||
return "completed"
|
||
if sset and all(s == "done" for s in sset):
|
||
return "done"
|
||
return sset[0] if sset else "pending"
|
||
|
||
|
||
def _group_tasks_by_node(sorted_tasks: List[Any]) -> List[List[Any]]:
|
||
"""按 node_key 合并同节点多人（或签/会签/抄送），无 key 的任务各自独立。"""
|
||
buckets: OrderedDict[str, List[Any]] = OrderedDict()
|
||
for t in sorted_tasks:
|
||
nk = (getattr(t, "node_key", None) or "").strip()
|
||
key = nk if nk else f"__task_{(getattr(t, 'id', None) or id(t))}"
|
||
buckets.setdefault(key, []).append(t)
|
||
return list(buckets.values())
|
||
|
||
|
||
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,
|
||
db: Optional[Session] = None,
|
||
advanced: Optional[Dict[str, Any]] = None,
|
||
initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""投影审批进度：发起人 → 树序 tasks（同节点多人合并）→ 未到达；终态收尾。
|
||
|
||
钉钉对齐：打回结果落在审批任务节点上，不另插 kind=return / terminal 终态「已打回」行；不做同单多轮分段。
|
||
或签/会签：同一 nodeKey 只出一张卡片，people 列出各审批人及状态。
|
||
进行中：按命中路径把后续审批/抄送节点全部挂为未到达（尽量解析人员）。
|
||
"""
|
||
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": "",
|
||
"signMode": "",
|
||
"people": [],
|
||
}
|
||
]
|
||
reached_keys = {
|
||
(t.node_key or "").strip() for t in sorted_tasks if (t.node_key or "").strip()
|
||
}
|
||
for group in _group_tasks_by_node(sorted_tasks):
|
||
head = group[0]
|
||
people = [_progress_person(t, name_map) for t in group]
|
||
_enrich_transfer_targets(people, group, name_map)
|
||
statuses = [p["status"] for p in people]
|
||
node_status = _aggregate_node_status(statuses)
|
||
# 时间：优先取已操作人最新时间，否则取组内最新
|
||
acted = [
|
||
p
|
||
for p in people
|
||
if p["status"] in ("approved", "rejected", "returned", "completed", "transferred")
|
||
]
|
||
ts_src = acted or people
|
||
timestamp = ""
|
||
for p in ts_src:
|
||
if p.get("timestamp") and (not timestamp or p["timestamp"] > timestamp):
|
||
timestamp = p["timestamp"]
|
||
comments = [
|
||
f"{p['nickName']}：{p['commentText']}"
|
||
if len(people) > 1 and p.get("nickName")
|
||
else p["commentText"]
|
||
for p in people
|
||
if (p.get("commentText") or "").strip()
|
||
]
|
||
sign_mode = (getattr(head, "sign_mode", None) or "").strip()
|
||
primary = next(
|
||
(p for p in people if p["status"] == node_status),
|
||
people[0] if people else None,
|
||
)
|
||
items.append(
|
||
{
|
||
"kind": "task",
|
||
"roundNo": rn,
|
||
"title": head.node_name or head.node_key or "-",
|
||
"status": node_status,
|
||
"assigneeUserId": (primary or {}).get("id", "") if len(people) == 1 else "",
|
||
"assigneeUserName": (primary or {}).get("nickName", "")
|
||
if len(people) == 1
|
||
else "",
|
||
"timestamp": timestamp,
|
||
"commentText": "；".join(comments),
|
||
"nodeKey": head.node_key or "",
|
||
"taskType": head.task_type or "",
|
||
"resolveSource": getattr(head, "resolve_source", None) or "",
|
||
"signMode": sign_mode,
|
||
"taskId": getattr(head, "id", None) or "",
|
||
"people": people,
|
||
}
|
||
)
|
||
if inst.status == "running" and isinstance(flow_root, dict):
|
||
form_data = _loads(getattr(inst, "form_data_json", None), {}) or {}
|
||
if not isinstance(form_data, dict):
|
||
form_data = {}
|
||
# 自选人：优先入参 → 实例 JSON → 旧数据从任务按 node_key 聚合
|
||
select_by_node = _normalize_initiator_select_by_node(initiator_select_by_node)
|
||
if not select_by_node:
|
||
select_by_node = _load_initiator_select_by_node(inst)
|
||
if not select_by_node:
|
||
for t in sorted_tasks:
|
||
src = (getattr(t, "resolve_source", None) or "").strip()
|
||
uid = (getattr(t, "assignee_user_id", None) or "").strip()
|
||
nk = (getattr(t, "node_key", None) or "").strip()
|
||
if src == "initiator_select" and uid and nk:
|
||
bucket = select_by_node.setdefault(nk, [])
|
||
if uid not in bucket:
|
||
bucket.append(uid)
|
||
if db is not None:
|
||
_append_pending_nodes_on_path(
|
||
items,
|
||
db=db,
|
||
flow_root=flow_root,
|
||
reached_keys=reached_keys,
|
||
round_no=rn,
|
||
tenant_no=getattr(inst, "tenant_no", None) or "",
|
||
initiator_user_id=inst.initiator_user_id or "",
|
||
initiator_dept_id=getattr(inst, "initiator_dept_id", None),
|
||
advanced=advanced or dict(oa_crud.DEFAULT_ADVANCED_SETTINGS),
|
||
form_data=form_data,
|
||
initiator_select_by_node=select_by_node,
|
||
)
|
||
else:
|
||
_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))
|
||
# returned：打回结果只落在审批任务节点上，不再追加独立终态「已打回」行
|
||
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 _dept_name(db: Session, dept_id: Optional[str], *, tenant_no: str) -> str:
|
||
did = (dept_id or "").strip()
|
||
if not did:
|
||
return ""
|
||
row = (
|
||
db.query(SysDept.dept_name)
|
||
.filter(SysDept.id == did, SysDept.tenant_no == tenant_no, SysDept.del_flag == 0)
|
||
.first()
|
||
)
|
||
if not row:
|
||
return ""
|
||
return (row[0] or "").strip()
|
||
|
||
|
||
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
|
||
|
||
|
||
_FLOW_PREVIEW_PENDING_MSG = "必填信息填写后，流程将自动显示"
|
||
_FLOW_PREVIEW_UNMATCHED_MSG = "流程条件无法匹配，请检查表单或联系管理员"
|
||
_FORMULA_FIELD_RE = re.compile(r"\{([^{}]+)\}")
|
||
|
||
|
||
def _form_value_present(form_data: Optional[Dict[str, Any]], field: str) -> bool:
|
||
"""条件字段是否已有有效值（0 / false 视为已填）。"""
|
||
if not field:
|
||
return True
|
||
data = form_data or {}
|
||
if field not in data:
|
||
return False
|
||
val = data.get(field)
|
||
if val is None:
|
||
return False
|
||
if isinstance(val, str) and not val.strip():
|
||
return False
|
||
if isinstance(val, (list, tuple, set, dict)) and len(val) == 0:
|
||
return False
|
||
return True
|
||
|
||
|
||
def _collect_condition_fields(node: Optional[Dict[str, Any]]) -> List[str]:
|
||
"""收集流程树中非默认条件分支引用的表单字段（去重保序）。"""
|
||
fields: List[str] = []
|
||
seen = set()
|
||
|
||
def add(name: Any) -> None:
|
||
key = str(name or "").strip()
|
||
if not key or key in seen:
|
||
return
|
||
seen.add(key)
|
||
fields.append(key)
|
||
|
||
def walk(n: Any) -> None:
|
||
if not isinstance(n, dict):
|
||
return
|
||
if _node_type(n) == "condition":
|
||
for g in n.get("conditionGroups") or []:
|
||
if not isinstance(g, dict):
|
||
continue
|
||
rules = g.get("rules")
|
||
when = g.get("when")
|
||
if g.get("isDefault") or (not rules and not when):
|
||
child = g.get("child")
|
||
if isinstance(child, dict):
|
||
walk(child)
|
||
continue
|
||
if isinstance(rules, list) and rules:
|
||
for r in rules:
|
||
if isinstance(r, dict) and r.get("field") is not None:
|
||
add(r.get("field"))
|
||
elif isinstance(rules, str) and rules.strip():
|
||
for m in _FORMULA_FIELD_RE.findall(rules):
|
||
add(m)
|
||
elif when:
|
||
for m in _FORMULA_FIELD_RE.findall(str(when)):
|
||
add(m)
|
||
child = g.get("child")
|
||
if isinstance(child, dict):
|
||
walk(child)
|
||
join_child = n.get("child")
|
||
if isinstance(join_child, dict):
|
||
walk(join_child)
|
||
return
|
||
child = n.get("child")
|
||
if isinstance(child, dict):
|
||
walk(child)
|
||
for g in n.get("conditionGroups") or []:
|
||
if isinstance(g, dict) and isinstance(g.get("child"), dict):
|
||
walk(g.get("child"))
|
||
|
||
walk(node)
|
||
return fields
|
||
|
||
|
||
def _condition_fields_ready(
|
||
root: Optional[Dict[str, Any]], form_data: Optional[Dict[str, Any]]
|
||
) -> bool:
|
||
fields = _collect_condition_fields(root)
|
||
if not fields:
|
||
return True
|
||
data = form_data or {}
|
||
return all(_form_value_present(data, f) for f in fields)
|
||
|
||
|
||
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_by_node: Optional[Dict[str, List[str]]],
|
||
form_data: Optional[Dict[str, Any]],
|
||
steps: List[Dict[str, Any]],
|
||
) -> bool:
|
||
"""沿最终路径写入 steps。条件无法匹配时返回 False。"""
|
||
cur = node
|
||
data = form_data or {}
|
||
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 []
|
||
g = _match_condition_group(
|
||
groups, data, initiator_user_id=initiator_user_id
|
||
)
|
||
if not g:
|
||
return False
|
||
child = g.get("child")
|
||
if isinstance(child, dict):
|
||
ok = _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_by_node=initiator_select_by_node,
|
||
form_data=data,
|
||
steps=steps,
|
||
)
|
||
if not ok:
|
||
return False
|
||
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 = ""
|
||
node_select_ids = _ids_for_initiator_select_node(
|
||
initiator_select_by_node, cur.get("id") or ""
|
||
)
|
||
if assignee_type == "initiator_select":
|
||
people = _preview_people(db, [str(x) for x in node_select_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=node_select_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")
|
||
return True
|
||
|
||
|
||
def preview_start_flow(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
app_id: str,
|
||
initiator_user_id: str,
|
||
initiator_dept_id: Optional[str] = None,
|
||
initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
|
||
form_data: Optional[Dict[str, Any]] = 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 {}
|
||
|
||
data = dict(form_data or {})
|
||
if not _condition_fields_ready(root, data):
|
||
return {
|
||
"resolved": False,
|
||
"message": _FLOW_PREVIEW_PENDING_MSG,
|
||
"steps": [],
|
||
}
|
||
|
||
steps: List[Dict[str, Any]] = []
|
||
select_by_node = _normalize_initiator_select_by_node(initiator_select_by_node)
|
||
ok = _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_by_node=select_by_node,
|
||
form_data=data,
|
||
steps=steps,
|
||
)
|
||
if not ok:
|
||
return {
|
||
"resolved": False,
|
||
"message": _FLOW_PREVIEW_UNMATCHED_MSG,
|
||
"steps": [],
|
||
}
|
||
return {"resolved": True, "message": "", "steps": steps}
|
||
|
||
|
||
def _create_tasks_for_node(
|
||
db: Session,
|
||
*,
|
||
inst: OaInstance,
|
||
node: Dict[str, Any],
|
||
ver: OaAppVersion,
|
||
user_id: Optional[str],
|
||
initiator_select_by_node: Optional[Dict[str, 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 {}
|
||
assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
|
||
node_select_ids = _ids_for_initiator_select_node(
|
||
initiator_select_by_node, node.get("id") or ""
|
||
)
|
||
if assignee_type == "initiator_select":
|
||
if ntype == "cc" and not node_select_ids:
|
||
return created
|
||
if ntype == "approval" and not node_select_ids:
|
||
raise OaEngineError("发起人未选择审批人")
|
||
|
||
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=node_select_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 _eval_condition_rules(
|
||
rules: List[Any],
|
||
form_data: Dict[str, Any],
|
||
*,
|
||
initiator_user_id: Optional[str] = None,
|
||
) -> bool:
|
||
"""AND 多条 {field, op, value}；contains 走人员集合交集，其余走公式。"""
|
||
from app.services.oa_person_select import person_contains_match
|
||
|
||
for r in rules:
|
||
if not isinstance(r, dict):
|
||
return False
|
||
field = r.get("field")
|
||
op = r.get("op") or "=="
|
||
val = r.get("value")
|
||
if field is None:
|
||
return False
|
||
if op == "contains":
|
||
if field == INITIATOR_CONDITION_FIELD:
|
||
uid = (initiator_user_id or "").strip()
|
||
if isinstance(val, list):
|
||
allowed = {str(x or "").strip() for x in val if str(x or "").strip()}
|
||
elif isinstance(val, str) and val.strip():
|
||
allowed = {val.strip()}
|
||
else:
|
||
allowed = set()
|
||
if not uid or uid not in allowed:
|
||
return False
|
||
continue
|
||
if not person_contains_match(form_data.get(field), val):
|
||
return False
|
||
continue
|
||
if isinstance(val, str):
|
||
expr = f"{{{field}}} {op} '{val}'"
|
||
else:
|
||
expr = f"{{{field}}} {op} {val}"
|
||
try:
|
||
if not eval_formula(expr, form_data):
|
||
return False
|
||
except FormulaError:
|
||
return False
|
||
return True
|
||
|
||
|
||
def _match_condition_group(
|
||
groups: List[Any],
|
||
form_data: Dict[str, Any],
|
||
*,
|
||
initiator_user_id: Optional[str] = None,
|
||
) -> 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
|
||
if when:
|
||
try:
|
||
if eval_formula(str(when), form_data):
|
||
return g
|
||
except FormulaError:
|
||
continue
|
||
continue
|
||
if isinstance(rules, str):
|
||
try:
|
||
if eval_formula(str(rules), form_data):
|
||
return g
|
||
except FormulaError:
|
||
continue
|
||
continue
|
||
if isinstance(rules, list) and rules:
|
||
try:
|
||
if _eval_condition_rules(
|
||
rules, form_data, initiator_user_id=initiator_user_id
|
||
):
|
||
return g
|
||
except FormulaError:
|
||
continue
|
||
continue
|
||
default_g = g
|
||
return default_g
|
||
|
||
|
||
def advance_from(
|
||
db: Session,
|
||
*,
|
||
inst: OaInstance,
|
||
ver: OaAppVersion,
|
||
from_node: Optional[Dict[str, Any]],
|
||
user_id: Optional[str],
|
||
initiator_select_by_node: Optional[Dict[str, 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, initiator_user_id=inst.initiator_user_id
|
||
)
|
||
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":
|
||
merge = _merge_after_branch_end(root, cur.get("id") or "")
|
||
if merge is not None:
|
||
cur = merge
|
||
continue
|
||
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_by_node=initiator_select_by_node,
|
||
order_num_start=next_order,
|
||
)
|
||
next_order += max(len(tasks), 1)
|
||
if ntype == "cc":
|
||
# 抄送不阻断，继续往下
|
||
cur = next_node(cur)
|
||
continue
|
||
if not tasks:
|
||
raise OaEngineError("审批节点未生成任务")
|
||
_maybe_auto_pass_initiator(
|
||
db,
|
||
inst=inst,
|
||
ver=ver,
|
||
tasks=tasks,
|
||
user_id=user_id,
|
||
)
|
||
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_by_node: Optional[Dict[str, 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
|
||
|
||
from app.services.oa_dept_select import validate_dept_form_values
|
||
from app.services.oa_money_field import validate_money_form_values
|
||
from app.services.oa_person_select import validate_person_form_values
|
||
from app.services.oa_receive_account import validate_receive_account_form_values
|
||
|
||
dept_errs = validate_dept_form_values(schema, data)
|
||
if dept_errs:
|
||
raise OaEngineError(dept_errs[0])
|
||
|
||
money_errs = validate_money_form_values(schema, data)
|
||
if money_errs:
|
||
raise OaEngineError(money_errs[0])
|
||
|
||
person_errs = validate_person_form_values(schema, data)
|
||
if person_errs:
|
||
raise OaEngineError(person_errs[0])
|
||
|
||
recv_errs = validate_receive_account_form_values(schema, data)
|
||
if recv_errs:
|
||
raise OaEngineError(recv_errs[0])
|
||
|
||
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()
|
||
normalized_select = _normalize_initiator_select_by_node(initiator_select_by_node)
|
||
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),
|
||
initiator_select_json=_dumps(normalized_select),
|
||
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)
|
||
|
||
from app.services.oa_related_approval import (
|
||
assert_linkable_instance,
|
||
merge_relation_targets,
|
||
)
|
||
|
||
targets = merge_relation_targets(schema, data, related_instance_ids)
|
||
for field_key, to_id in targets:
|
||
target = (
|
||
db.query(OaInstance)
|
||
.filter(
|
||
OaInstance.id == to_id,
|
||
OaInstance.tenant_no == tenant_no,
|
||
OaInstance.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
assert_linkable_instance(target, user_id=initiator_user_id, tenant_no=tenant_no)
|
||
db.add(
|
||
OaRelation(
|
||
id=generate_uuid(),
|
||
from_instance_id=inst.id,
|
||
to_instance_id=to_id,
|
||
field_key=field_key,
|
||
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_by_node=normalized_select,
|
||
)
|
||
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["initiatorDeptName"] = _dept_name(
|
||
db, inst.initiator_dept_id, tenant_no=tenant_no
|
||
)
|
||
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,
|
||
db=db,
|
||
advanced=adv,
|
||
)
|
||
# 不再暴露旧单/重提单关联跳转
|
||
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", "transferred")),
|
||
)
|
||
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 _approve_task_core(
|
||
db: Session,
|
||
*,
|
||
task: OaTask,
|
||
inst: OaInstance,
|
||
ver: OaAppVersion,
|
||
user_id: str,
|
||
comment: Optional[str],
|
||
) -> bool:
|
||
"""将 pending 审批任务置为 approved 并按或签/会签决定是否推进。
|
||
|
||
返回 True 表示已调用 advance_from 推进；False 表示会签未齐、未推进。
|
||
调用方负责 commit（若需要）。
|
||
"""
|
||
now = datetime.now()
|
||
task.comment_text = (comment or "").strip() or None
|
||
task.status = "approved"
|
||
task.update_by = user_id
|
||
task.update_time = now
|
||
db.add(task)
|
||
# Session 默认 autoflush=False：不先 flush 时，后续按 status=pending 的查询仍会命中本任务，
|
||
# 或签「取消同节点其余待办」会把本人已同意覆盖成 done。
|
||
db.flush()
|
||
|
||
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:
|
||
return False
|
||
|
||
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,
|
||
initiator_select_by_node=_load_initiator_select_by_node(inst),
|
||
)
|
||
return True
|
||
|
||
|
||
def _maybe_auto_pass_initiator(
|
||
db: Session,
|
||
*,
|
||
inst: OaInstance,
|
||
ver: OaAppVersion,
|
||
tasks: List[OaTask],
|
||
user_id: Optional[str],
|
||
) -> None:
|
||
_ = user_id
|
||
advanced = _advanced(ver)
|
||
if not advanced.get("autoPassWhenApproverIsInitiator"):
|
||
return
|
||
if inst.status != "running":
|
||
return
|
||
initiator = (inst.initiator_user_id or "").strip()
|
||
if not initiator:
|
||
return
|
||
self_task = None
|
||
for t in tasks:
|
||
if (
|
||
t.task_type == "approval"
|
||
and t.status == "pending"
|
||
and (t.assignee_user_id or "") == initiator
|
||
):
|
||
self_task = t
|
||
break
|
||
if not self_task:
|
||
return
|
||
_approve_task_core(
|
||
db,
|
||
task=self_task,
|
||
inst=inst,
|
||
ver=ver,
|
||
user_id=initiator,
|
||
comment=_AUTO_PASS_COMMENT,
|
||
)
|
||
|
||
|
||
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("请填写审批意见")
|
||
|
||
if action == "reject":
|
||
now = datetime.now()
|
||
task.comment_text = (comment or "").strip() or None
|
||
task.status = "rejected"
|
||
task.update_by = user_id
|
||
task.update_time = now
|
||
db.add(task)
|
||
db.flush()
|
||
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)
|
||
|
||
_approve_task_core(
|
||
db,
|
||
task=task,
|
||
inst=inst,
|
||
ver=ver,
|
||
user_id=user_id,
|
||
comment=comment,
|
||
)
|
||
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 transfer_task(
|
||
db: Session,
|
||
task_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
user_id: str,
|
||
target_user_id: str,
|
||
comment: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""当前待办处理人转交：结束本人待办并为同节点目标用户新建 pending 任务。"""
|
||
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("实例已结束")
|
||
|
||
target = (target_user_id or "").strip()
|
||
if not target:
|
||
raise OaEngineError("请选择转交人")
|
||
if target == user_id:
|
||
raise OaEngineError("不能转交给自己")
|
||
|
||
u = (
|
||
db.query(SysUser)
|
||
.filter(
|
||
SysUser.id == target,
|
||
SysUser.tenant_no == tenant_no,
|
||
SysUser.del_flag == 0,
|
||
SysUser.status == "1",
|
||
)
|
||
.first()
|
||
)
|
||
if not u:
|
||
raise OaEngineError(f"转交用户无效：{target}")
|
||
|
||
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()
|
||
}
|
||
if target in existing:
|
||
raise OaEngineError("该用户已在当前节点待办中")
|
||
|
||
now = datetime.now()
|
||
comment_text = (comment or "").strip() or None
|
||
round_no = int(getattr(inst, "current_round", None) or getattr(task, "round_no", None) or 1)
|
||
|
||
task.status = "transferred"
|
||
task.comment_text = comment_text
|
||
task.update_by = user_id
|
||
task.update_time = now
|
||
db.add(task)
|
||
|
||
next_order = _next_task_order_num(db, inst.id)
|
||
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=target,
|
||
status="pending",
|
||
comment_text=None,
|
||
resolve_source="transfer",
|
||
sign_mode=task.sign_mode or "or",
|
||
round_no=round_no,
|
||
tenant_no=tenant_no,
|
||
order_num=next_order,
|
||
remark=comment_text,
|
||
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,
|
||
)
|
||
)
|
||
|
||
event_comment = comment_text or f"转交给 {target}"
|
||
_add_instance_event(
|
||
db,
|
||
inst=inst,
|
||
event_type="transfer",
|
||
actor_user_id=user_id,
|
||
round_no=round_no,
|
||
comment=event_comment,
|
||
task_id=task.id,
|
||
node_key=task.node_key,
|
||
)
|
||
|
||
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_by_node: Optional[Dict[str, List[str]]] = None,
|
||
) -> Dict[str, Any]:
|
||
"""已废弃：同实例再次提交。请复制表单重新发起新实例。"""
|
||
_ = (db, instance_id, tenant_no, user_id, form_data, initiator_select_by_node)
|
||
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)
|