cykj/zkxw
1
0 Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

feat(oa): resolve initiator_select per node map in engine

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-07-23 16:05:25 +08:00
1 parent cf270630fe
commit 71424e9dd1
4 files changed
+210 -38
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/app/api/oa_approval.py
+7 -7
View File
@@ -550,7 +550,8 @@ async def preview_oa_template_flow(
app_id=app_id,
initiator_user_id=uid,
initiator_dept_id=dept_id,
initiator_select_user_ids=body.initiator_select_user_ids,
# Task 3 将 Body 改为 initiatorSelectByNode；此处先对齐引擎 kwargs
initiator_select_by_node=None,
form_data=body.form_data,
)
return success_response(data)
@@ -576,16 +577,14 @@ async def preview_oa_template_flow_get(
row = db.query(SysUser).filter(SysUser.id == uid, SysUser.del_flag == 0).first()
if row:
dept_id = row.dept_id
select_ids = None
if initiator_select_user_ids:
select_ids = [x.strip() for x in initiator_select_user_ids.split(",") if x.strip()]
_ = initiator_select_user_ids # 扁平 query 已废弃；Task 3 删除
data = preview_start_flow(
db,
tenant_no=tenant_no,
app_id=app_id,
initiator_user_id=uid,
initiator_dept_id=dept_id,
initiator_select_user_ids=select_ids,
initiator_select_by_node=None,
form_data=None,
)
return success_response(data)
@@ -611,7 +610,8 @@ async def start_oa_instance(
initiator_dept_id=body.initiator_dept_id,
form_data=body.form_data,
title=body.title,
initiator_select_user_ids=body.initiator_select_user_ids,
# Task 3 将 Body 改为 initiatorSelectByNode
initiator_select_by_node=None,
related_instance_ids=body.related_instance_ids,
source_instance_id=body.source_instance_id,
)
@@ -788,7 +788,7 @@ async def resubmit_oa_instance(
tenant_no=tenant_no,
user_id=user.get("user_id") or "",
form_data=(body.form_data if body else None),
initiator_select_user_ids=(body.initiator_select_user_ids if body else None),
initiator_select_by_node=None,
)
return success_response(data, message="已再次提交")
except OaEngineError as e:
backend/app/services/oa_engine.py
+93 -29
View File
@@ -44,6 +44,41 @@ 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()
@@ -253,7 +288,7 @@ def _pending_people_for_node(
initiator_user_id: str,
initiator_dept_id: Optional[str],
advanced: Dict[str, Any],
initiator_select_user_ids: Optional[List[str]],
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)
@@ -265,9 +300,11 @@ def _pending_people_for_node(
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":
ids = [str(x).strip() for x in (initiator_select_user_ids or []) if str(x).strip()]
for uid in ids:
for uid in node_select_ids:
people.append(
{
"id": uid,
@@ -287,7 +324,7 @@ def _pending_people_for_node(
initiator_user_id=initiator_user_id,
initiator_dept_id=initiator_dept_id,
advanced=advanced,
initiator_select_user_ids=initiator_select_user_ids,
initiator_select_user_ids=node_select_ids,
)
for uid in assignees:
people.append(
@@ -352,7 +389,7 @@ def _append_pending_nodes_on_path(
initiator_dept_id: Optional[str],
advanced: Dict[str, Any],
form_data: Optional[Dict[str, Any]],
initiator_select_user_ids: Optional[List[str]],
initiator_select_by_node: Optional[Dict[str, List[str]]],
) -> None:
"""按命中路径列出全部后续审批/抄送/结束节点（含人员），对齐发起预览。"""
path = _iter_resolved_path_nodes(flow_root, form_data)
@@ -402,7 +439,7 @@ def _append_pending_nodes_on_path(
initiator_user_id=initiator_user_id,
initiator_dept_id=initiator_dept_id,
advanced=advanced,
initiator_select_user_ids=initiator_select_user_ids,
initiator_select_by_node=initiator_select_by_node,
)
items.append(
{
@@ -471,7 +508,7 @@ def build_progress_items(
events: Optional[List[Any]] = None,
db: Optional[Session] = None,
advanced: Optional[Dict[str, Any]] = None,
initiator_select_user_ids: Optional[List[str]] = None,
initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
) -> List[Dict[str, Any]]:
"""投影审批进度：发起人 → 树序 tasks（同节点多人合并）→ 未到达；终态收尾。
@@ -560,14 +597,19 @@ def build_progress_items(
form_data = _loads(getattr(inst, "form_data_json", None), {}) or {}
if not isinstance(form_data, dict):
form_data = {}
# 自选人：优先入参；否则从已有任务里收集
select_ids = list(initiator_select_user_ids or [])
if not select_ids:
# 自选人：优先入参 → 实例 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()
if src == "initiator_select" and uid and uid not in select_ids:
select_ids.append(uid)
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,
@@ -580,7 +622,7 @@ def build_progress_items(
initiator_dept_id=getattr(inst, "initiator_dept_id", None),
advanced=advanced or dict(oa_crud.DEFAULT_ADVANCED_SETTINGS),
form_data=form_data,
initiator_select_user_ids=select_ids,
initiator_select_by_node=select_by_node,
)
else:
_append_pending_nodes(
@@ -1023,7 +1065,7 @@ def _walk_preview_nodes(
initiator_user_id: str,
initiator_dept_id: Optional[str],
advanced: Dict[str, Any],
initiator_select_user_ids: Optional[List[str]],
initiator_select_by_node: Optional[Dict[str, List[str]]],
form_data: Optional[Dict[str, Any]],
steps: List[Dict[str, Any]],
) -> bool:
@@ -1078,7 +1120,7 @@ def _walk_preview_nodes(
initiator_user_id=initiator_user_id,
initiator_dept_id=initiator_dept_id,
advanced=advanced,
initiator_select_user_ids=initiator_select_user_ids,
initiator_select_by_node=initiator_select_by_node,
form_data=data,
steps=steps,
)
@@ -1098,9 +1140,11 @@ def _walk_preview_nodes(
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":
ids = initiator_select_user_ids or []
people = _preview_people(db, [str(x) for x in ids if str(x).strip()])
people = _preview_people(db, [str(x) for x in node_select_ids if str(x).strip()])
if not people:
meta = meta or "发起人自选"
else:
@@ -1112,7 +1156,7 @@ def _walk_preview_nodes(
initiator_user_id=initiator_user_id,
initiator_dept_id=initiator_dept_id,
advanced=advanced,
initiator_select_user_ids=initiator_select_user_ids,
initiator_select_user_ids=node_select_ids,
)
people = _preview_people(db, assignees)
except AssigneeResolveError as e:
@@ -1143,7 +1187,7 @@ def preview_start_flow(
app_id: str,
initiator_user_id: str,
initiator_dept_id: Optional[str] = None,
initiator_select_user_ids: Optional[List[str]] = None,
initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
form_data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""发起前预览审批链：条件字段齐后只走命中路径并解析人员。"""
@@ -1182,6 +1226,7 @@ def preview_start_flow(
}
steps: List[Dict[str, Any]] = []
select_by_node = _normalize_initiator_select_by_node(initiator_select_by_node)
ok = _walk_preview_nodes(
db,
node=root,
@@ -1189,7 +1234,7 @@ def preview_start_flow(
initiator_user_id=initiator_user_id,
initiator_dept_id=dept_id or None,
advanced=_advanced(pub),
initiator_select_user_ids=initiator_select_user_ids,
initiator_select_by_node=select_by_node,
form_data=data,
steps=steps,
)
@@ -1209,7 +1254,7 @@ def _create_tasks_for_node(
node: Dict[str, Any],
ver: OaAppVersion,
user_id: Optional[str],
initiator_select_user_ids: Optional[List[str]] = None,
initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
order_num_start: Optional[int] = None,
) -> List[OaTask]:
ntype = _node_type(node)
@@ -1223,6 +1268,16 @@ def _create_tasks_for_node(
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(
@@ -1232,7 +1287,7 @@ def _create_tasks_for_node(
initiator_user_id=inst.initiator_user_id,
initiator_dept_id=inst.initiator_dept_id,
advanced=advanced,
initiator_select_user_ids=initiator_select_user_ids,
initiator_select_user_ids=node_select_ids,
)
except AssigneeResolveError as e:
raise OaEngineError(str(e)) from e
@@ -1319,7 +1374,7 @@ def advance_from(
ver: OaAppVersion,
from_node: Optional[Dict[str, Any]],
user_id: Optional[str],
initiator_select_user_ids: Optional[List[str]] = None,
initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
) -> None:
"""从某节点的下一跳开始推进；from_node 为 None 表示从 start 进入。"""
root = _flow_root(_loads(ver.flow_tree_json, {}))
@@ -1367,7 +1422,7 @@ def advance_from(
node=cur,
ver=ver,
user_id=user_id,
initiator_select_user_ids=initiator_select_user_ids,
initiator_select_by_node=initiator_select_by_node,
order_num_start=next_order,
)
next_order += max(len(tasks), 1)
@@ -1395,7 +1450,7 @@ def start_instance(
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,
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]:
@@ -1425,6 +1480,7 @@ def start_instance(
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,
@@ -1433,6 +1489,7 @@ def start_instance(
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,
@@ -1496,7 +1553,7 @@ def start_instance(
ver=ver,
from_node=None,
user_id=initiator_user_id,
initiator_select_user_ids=initiator_select_user_ids,
initiator_select_by_node=normalized_select,
)
db.commit()
db.refresh(inst)
@@ -1760,7 +1817,14 @@ def complete_task(
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)
advance_from(
db,
inst=inst,
ver=ver,
from_node=node,
user_id=user_id,
initiator_select_by_node=_load_initiator_select_by_node(inst),
)
db.commit()
return get_instance_detail(db, inst.id, tenant_no=tenant_no)
@@ -2080,10 +2144,10 @@ def resubmit_instance(
tenant_no: str,
user_id: str,
form_data: Optional[Dict[str, Any]] = None,
initiator_select_user_ids: Optional[List[str]] = 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_user_ids)
_ = (db, instance_id, tenant_no, user_id, form_data, initiator_select_by_node)
raise OaEngineError("同实例再次提交已废弃，请复制表单重新发起")
backend/tests/test_oa_flow_preview.py
+2 -2
View File
@@ -109,7 +109,7 @@ class TestWalkPreviewResolvedPath(unittest.TestCase):
initiator_user_id="init",
initiator_dept_id=None,
advanced={},
initiator_select_user_ids=None,
initiator_select_by_node=None,
form_data={"days": 5},
steps=steps,
)
@@ -133,7 +133,7 @@ class TestWalkPreviewResolvedPath(unittest.TestCase):
initiator_user_id="init",
initiator_dept_id=None,
advanced={},
initiator_select_user_ids=None,
initiator_select_by_node=None,
form_data={"days": 1},
steps=steps,
)
backend/tests/test_oa_initiator_select_by_node.py
+108
View File
@@ -0,0 +1,108 @@
"""OA 发起人自选：按节点 map"""
import unittest
from unittest.mock import MagicMock, patch
from app.services.oa_engine import (
OaEngineError,
_create_tasks_for_node,
_ids_for_initiator_select_node,
_normalize_initiator_select_by_node,
)
class TestNormalizeMap(unittest.TestCase):
def test_normalize_dedupe_preserve_order(self):
raw = {"n1": ["u1", "u1", " u2 ", ""], "n2": "bad"}
out = _normalize_initiator_select_by_node(raw)
self.assertEqual(out["n1"], ["u1", "u2"])
self.assertNotIn("n2", out)
def test_ids_for_node(self):
m = {"a1": ["u9"]}
self.assertEqual(_ids_for_initiator_select_node(m, "a1"), ["u9"])
self.assertEqual(_ids_for_initiator_select_node(m, "missing"), [])
class TestCreateTasksInitiatorSelect(unittest.TestCase):
def _inst(self):
inst = MagicMock()
inst.id = "inst1"
inst.tenant_no = "zkxw"
inst.initiator_user_id = "init"
inst.initiator_dept_id = "d1"
inst.current_round = 1
return inst
def test_cc_empty_skips_without_resolve(self):
db = MagicMock()
node = {
"id": "cc1",
"type": "cc",
"name": "抄送人",
"approverConfig": {"assigneeType": "initiator_select"},
}
ver = MagicMock()
ver.advanced_settings_json = "{}"
with patch("app.services.oa_engine.resolve_assignees") as resolve:
created = _create_tasks_for_node(
db,
inst=self._inst(),
node=node,
ver=ver,
user_id="init",
initiator_select_by_node={},
)
resolve.assert_not_called()
self.assertEqual(created, [])
def test_approval_empty_raises(self):
db = MagicMock()
node = {
"id": "a1",
"type": "approval",
"name": "审批人",
"approverConfig": {"assigneeType": "initiator_select", "signMode": "or"},
}
ver = MagicMock()
ver.advanced_settings_json = "{}"
with self.assertRaises(OaEngineError):
_create_tasks_for_node(
db,
inst=self._inst(),
node=node,
ver=ver,
user_id="init",
initiator_select_by_node={"a1": []},
)
def test_approval_uses_node_ids(self):
db = MagicMock()
node = {
"id": "a1",
"type": "approval",
"name": "审批人",
"approverConfig": {"assigneeType": "initiator_select", "signMode": "or"},
}
ver = MagicMock()
ver.advanced_settings_json = "{}"
with patch(
"app.services.oa_engine.resolve_assignees",
return_value=(["u1"], {"resolve_source": "initiator_select"}),
) as resolve:
with patch("app.services.oa_engine.generate_uuid", return_value="t1"):
with patch("app.services.oa_engine._next_task_order_num", return_value=1):
created = _create_tasks_for_node(
db,
inst=self._inst(),
node=node,
ver=ver,
user_id="init",
initiator_select_by_node={"a1": ["u1"], "other": ["u9"]},
)
self.assertEqual(len(created), 1)
kwargs = resolve.call_args.kwargs
self.assertEqual(kwargs.get("initiator_select_user_ids"), ["u1"])
if __name__ == "__main__":
unittest.main()