From e0cfd41b1fcb14101d477994e12016b187a816ff Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Thu, 13 Aug 2026 18:08:35 +0800 Subject: [PATCH] feat(oa): schedule WeChat MP notifies after approval/cc tasks create Co-authored-by: Cursor --- backend/app/services/oa_engine.py | 66 ++++++++----- backend/app/services/oa_wechat_notify.py | 101 ++++++++++++++++++++ backend/tests/test_oa_wechat_notify_hook.py | 81 ++++++++++++++++ 3 files changed, 225 insertions(+), 23 deletions(-) create mode 100644 backend/app/services/oa_wechat_notify.py create mode 100644 backend/tests/test_oa_wechat_notify_hook.py diff --git a/backend/app/services/oa_engine.py b/backend/app/services/oa_engine.py index 86eb2680..67406eee 100644 --- a/backend/app/services/oa_engine.py +++ b/backend/app/services/oa_engine.py @@ -1673,10 +1673,14 @@ def advance_from( 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 进入。""" +) -> List[str]: + """从某节点的下一跳开始推进；from_node 为 None 表示从 start 进入。 + + 返回本次推进新创建的任务 id 列表（含抄送与嵌套自动通过产生的后续任务）。 + """ root = _flow_root(_loads(ver.flow_tree_json, {})) form_data = _loads(inst.form_data_json, {}) or {} + created_ids: List[str] = [] def next_node(node: Dict[str, Any]) -> Optional[Dict[str, Any]]: ntype = _node_type(node) @@ -1723,7 +1727,7 @@ def advance_from( logger.exception( "resignation kit user sync failed instance=%s", getattr(inst, "id", None) ) - return + return created_ids if ntype == "condition": cur = next_node(cur) continue @@ -1737,6 +1741,7 @@ def advance_from( initiator_select_by_node=initiator_select_by_node, order_num_start=next_order, ) + created_ids.extend([t.id for t in tasks if getattr(t, "id", None)]) next_order += max(len(tasks), 1) if ntype == "cc": # 抄送不阻断，继续往下 @@ -1744,20 +1749,22 @@ def advance_from( continue if not tasks: raise OaEngineError("审批节点未生成任务") - _maybe_auto_pass_initiator( + nested = _maybe_auto_pass_initiator( db, inst=inst, ver=ver, tasks=tasks, user_id=user_id, ) - return + created_ids.extend(nested or []) + return created_ids # 未知类型跳过子节点 cur = next_node(cur) # 无结束也无待办：异常结束 if inst.status == "running": raise OaEngineError("流程推进异常：未到达结束且无待办任务") + return created_ids def start_instance( @@ -1933,7 +1940,7 @@ def start_instance( actor_user_id=initiator_user_id, round_no=1, ) - advance_from( + created = advance_from( db, inst=inst, ver=ver, @@ -1942,6 +1949,9 @@ def start_instance( initiator_select_by_node=normalized_select, ) db.commit() + from app.services.oa_wechat_notify import schedule_oa_task_notifies + + schedule_oa_task_notifies(created or []) db.refresh(inst) return get_instance_detail(db, inst.id, tenant_no=tenant_no) @@ -2363,10 +2373,10 @@ def _approve_task_core( ver: OaAppVersion, user_id: str, comment: Optional[str], -) -> bool: +) -> List[str]: """将 pending 审批任务置为 approved 并按或签/会签决定是否推进。 - 返回 True 表示已调用 advance_from 推进；False 表示会签未齐、未推进。 + 返回 advance_from 新创建的任务 id 列表；会签未齐、未推进时返回空列表。 调用方负责 commit（若需要）。 """ now = datetime.now() @@ -2393,7 +2403,7 @@ def _approve_task_core( .count() ) if sign_mode == "and" and pending_same > 0: - return False + return [] if sign_mode != "and": for s in ( @@ -2416,7 +2426,7 @@ def _approve_task_core( node = _find_node(root, task.node_key or "") if not node: raise OaEngineError("找不到对应流程节点") - advance_from( + return advance_from( db, inst=inst, ver=ver, @@ -2424,7 +2434,6 @@ def _approve_task_core( user_id=user_id, initiator_select_by_node=_load_initiator_select_by_node(inst), ) - return True def _maybe_auto_pass_initiator( @@ -2434,16 +2443,16 @@ def _maybe_auto_pass_initiator( ver: OaAppVersion, tasks: List[OaTask], user_id: Optional[str], -) -> None: +) -> List[str]: _ = user_id advanced = _advanced(ver) if not advanced.get("autoPassWhenApproverIsInitiator"): - return + return [] if inst.status != "running": - return + return [] initiator = (inst.initiator_user_id or "").strip() if not initiator: - return + return [] self_task = None for t in tasks: if ( @@ -2454,8 +2463,8 @@ def _maybe_auto_pass_initiator( self_task = t break if not self_task: - return - _approve_task_core( + return [] + return _approve_task_core( db, task=self_task, inst=inst, @@ -2537,7 +2546,7 @@ def complete_task( db.commit() return get_instance_detail(db, inst.id, tenant_no=tenant_no) - _approve_task_core( + created = _approve_task_core( db, task=task, inst=inst, @@ -2546,6 +2555,9 @@ def complete_task( comment=comment, ) db.commit() + from app.services.oa_wechat_notify import schedule_oa_task_notifies + + schedule_oa_task_notifies(created or []) return get_instance_detail(db, inst.id, tenant_no=tenant_no) @@ -2609,7 +2621,7 @@ def add_sign( .all() } now = datetime.now() - created = 0 + created_ids: List[str] = [] next_order = _next_task_order_num(db, inst.id) for uid in ids: if uid in existing: @@ -2627,9 +2639,10 @@ def add_sign( ) if not u: raise OaEngineError(f"加签用户无效：{uid}") + new_id = generate_uuid() db.add( OaTask( - id=generate_uuid(), + id=new_id, instance_id=inst.id, node_key=task.node_key, node_name=task.node_name, @@ -2652,8 +2665,8 @@ def add_sign( ) ) next_order += 1 - created += 1 - if created == 0: + created_ids.append(new_id) + if not created_ids: raise OaEngineError("所选用户已在当前节点待办中") # 加签后默认按会签语义收口更稳妥：保留原 sign_mode；若原或签则一人通过仍可结束 if comment: @@ -2662,6 +2675,9 @@ def add_sign( task.update_time = now db.add(task) db.commit() + from app.services.oa_wechat_notify import schedule_oa_task_notifies + + schedule_oa_task_notifies(created_ids) return get_instance_detail(db, inst.id, tenant_no=tenant_no) @@ -2735,9 +2751,10 @@ def transfer_task( db.add(task) next_order = _next_task_order_num(db, inst.id) + new_task_id = generate_uuid() db.add( OaTask( - id=generate_uuid(), + id=new_task_id, instance_id=inst.id, node_key=task.node_key, node_name=task.node_name, @@ -2773,6 +2790,9 @@ def transfer_task( ) db.commit() + from app.services.oa_wechat_notify import schedule_oa_task_notifies + + schedule_oa_task_notifies([new_task_id]) return get_instance_detail(db, inst.id, tenant_no=tenant_no) diff --git a/backend/app/services/oa_wechat_notify.py b/backend/app/services/oa_wechat_notify.py new file mode 100644 index 00000000..010b2516 --- /dev/null +++ b/backend/app/services/oa_wechat_notify.py @@ -0,0 +1,101 @@ +"""OA 任务创建后服务号通知（commit 后异步）。""" +from __future__ import annotations + +import logging +import threading +from datetime import datetime +from typing import Iterable, List + +from app.core.database import SessionLocal +from app.models.models import OaInstance, OaTask, SysUser +from app.services.wechat_mp_message import send_oa_approval_notify, send_oa_cc_notify + +logger = logging.getLogger(__name__) + + +def build_oa_detail_pagepath(*, instance_id: str, task_id: str, mode: str) -> str: + # 无前导 / + return ( + f"pages/oa/webview/index?instanceId={instance_id}" + f"&taskId={task_id}&mode={mode}" + ) + + +def _send_one(db, task: OaTask) -> None: + uid = (task.assignee_user_id or "").strip() + if not uid: + return + user = db.query(SysUser).filter(SysUser.id == uid, SysUser.del_flag == 0).first() + mp = (getattr(user, "wx_mp_openid", None) or "").strip() if user else "" + if not mp: + logger.info("oa wx notify skip: no mp_openid user=%s task=%s", uid, task.id) + return + inst = ( + db.query(OaInstance) + .filter(OaInstance.id == task.instance_id, OaInstance.del_flag == 0) + .first() + ) + title = (inst.title if inst else "") or (task.node_name or "审批") + applicant = "" + if inst and inst.initiator_user_id: + ini = db.query(SysUser).filter(SysUser.id == inst.initiator_user_id).first() + applicant = (ini.nick_name if ini else "") or "" + time_text = datetime.now().strftime("%Y-%m-%d %H:%M") + node_name = task.node_name or "" + if task.task_type == "approval" and (task.status or "") == "pending": + pagepath = build_oa_detail_pagepath( + instance_id=task.instance_id, task_id=task.id, mode="approve" + ) + send_oa_approval_notify( + mp_openid=mp, + pagepath=pagepath, + title=title, + applicant=applicant, + node_name=node_name, + time_text=time_text, + ) + elif task.task_type == "cc": + pagepath = build_oa_detail_pagepath( + instance_id=task.instance_id, task_id=task.id, mode="view" + ) + send_oa_cc_notify( + mp_openid=mp, + pagepath=pagepath, + title=title, + applicant=applicant, + node_name=node_name, + time_text=time_text, + ) + + +def notify_oa_tasks_sync(task_ids: Iterable[str]) -> None: + ids = [str(x).strip() for x in task_ids if str(x).strip()] + if not ids: + return + db = None + try: + db = SessionLocal() + tasks = ( + db.query(OaTask) + .filter(OaTask.id.in_(ids), OaTask.del_flag == 0) + .all() + ) + for t in tasks: + try: + _send_one(db, t) + except Exception: + logger.exception("oa wx notify failed task=%s", t.id) + except Exception: + logger.exception("oa wx notify batch failed ids=%s", ids) + finally: + if db is not None: + db.close() + + +def schedule_oa_task_notifies(task_ids: List[str]) -> None: + ids = [str(x).strip() for x in (task_ids or []) if str(x).strip()] + if not ids: + return + threading.Thread( + target=notify_oa_tasks_sync, args=(ids,), daemon=True, name="oa-wx-notify" + ).start() diff --git a/backend/tests/test_oa_wechat_notify_hook.py b/backend/tests/test_oa_wechat_notify_hook.py new file mode 100644 index 00000000..35fa6e6b --- /dev/null +++ b/backend/tests/test_oa_wechat_notify_hook.py @@ -0,0 +1,81 @@ +"""OA 微信服务号通知钩子单测。""" +from unittest.mock import MagicMock, patch + +from app.services.oa_wechat_notify import _send_one, build_oa_detail_pagepath + + +def test_build_pagepath_no_leading_slash(): + p = build_oa_detail_pagepath(instance_id="i1", task_id="t1", mode="approve") + assert p.startswith("pages/") + assert "mode=approve" in p + assert "instanceId=i1" in p + assert "taskId=t1" in p + + +def test_send_one_routes_approval_vs_cc(): + db = MagicMock() + + user = MagicMock() + user.wx_mp_openid = "mp-openid-1" + user.nick_name = "张三" + + inst = MagicMock() + inst.title = "请假" + inst.initiator_user_id = "u-ini" + + ini = MagicMock() + ini.nick_name = "李四" + + def query_side_effect(model): + q = MagicMock() + name = getattr(model, "__name__", "") or str(model) + if "SysUser" in name: + # first call: assignee; second: initiator (approval path may call twice) + q.filter.return_value.first.side_effect = [user, ini] + else: + q.filter.return_value.first.return_value = inst + return q + + db.query.side_effect = query_side_effect + + approval_task = MagicMock() + approval_task.id = "t-appr" + approval_task.instance_id = "i1" + approval_task.assignee_user_id = "u1" + approval_task.task_type = "approval" + approval_task.status = "pending" + approval_task.node_name = "主管审批" + + with patch( + "app.services.oa_wechat_notify.send_oa_approval_notify" + ) as send_appr, patch( + "app.services.oa_wechat_notify.send_oa_cc_notify" + ) as send_cc: + _send_one(db, approval_task) + send_appr.assert_called_once() + send_cc.assert_not_called() + kwargs = send_appr.call_args.kwargs + assert kwargs["mp_openid"] == "mp-openid-1" + assert "mode=approve" in kwargs["pagepath"] + + # reset query for cc path + db.query.side_effect = query_side_effect + + cc_task = MagicMock() + cc_task.id = "t-cc" + cc_task.instance_id = "i1" + cc_task.assignee_user_id = "u1" + cc_task.task_type = "cc" + cc_task.status = "done" + cc_task.node_name = "抄送" + + with patch( + "app.services.oa_wechat_notify.send_oa_approval_notify" + ) as send_appr, patch( + "app.services.oa_wechat_notify.send_oa_cc_notify" + ) as send_cc: + _send_one(db, cc_task) + send_cc.assert_called_once() + send_appr.assert_not_called() + kwargs = send_cc.call_args.kwargs + assert "mode=view" in kwargs["pagepath"]