"""OA 任务创建后服务号通知(commit 后异步)。""" from __future__ import annotations import logging import threading from datetime import datetime from typing import Iterable, List, Optional from urllib.parse import urlencode from app.core.config import settings from app.core.database import SessionLocal from app.models.models import OaApp, OaInstance, OaTask, SysUser from app.services.wechat_mp_message import ( send_oa_applicant_progress_notify, send_oa_approval_notify, send_oa_cc_notify, ) logger = logging.getLogger(__name__) # 微信模板 const4 枚举,必须原样下发 RESULT_PASS = "通过" RESULT_FAIL = "失败" RESULT_TRANSFER = "转办" RESULT_OTHER = "其他" # 模板消息关闭详情后回退的小程序页(与列表入口对齐) RETURN_PENDING = "/pages/oa/pending/index" RETURN_CC = "/pages/oa/cc/index" RETURN_MINE = "/pages/oa/mine/index" RETURN_MENTIONS = "/pages/oa/mentions/index" _ACTION_RESULT = { "approve": RESULT_PASS, "reject": RESULT_FAIL, "transfer": RESULT_TRANSFER, "return": RESULT_OTHER, "add_sign": RESULT_OTHER, } def map_oa_action_to_result(action: str) -> str: return _ACTION_RESULT.get((action or "").strip(), RESULT_OTHER) def build_oa_detail_pagepath( *, instance_id: str, task_id: str = "", mode: str, return_path: str = "", ) -> str: """拼小程序 pagepath(无前导 /);query 做 urlencode,保证 returnPath 含斜杠可用。""" params = { "instanceId": (instance_id or "").strip(), "mode": (mode or "view").strip() or "view", } tid = (task_id or "").strip() if tid and tid != "0": params["taskId"] = tid rp = (return_path or "").strip() if rp: params["returnPath"] = rp return "pages/oa/webview/index?" + urlencode(params) def _build_matter_title(*, applicant: str, process_name: str, fallback: str) -> str: """申请事项/业务名称:申请人 - 流程名称。""" parts = [p for p in [(applicant or "").strip(), (process_name or "").strip()] if p] if parts: return " - ".join(parts) return (fallback or "").strip() or "审批" def _build_mention_title(commenter_name: str) -> str: """抄送模板 thing5:评论人在评论中提到了你。""" name = (commenter_name or "").strip() if name: return f"{name} 在评论中提到了你" return "有人在评论中提到了你" def _resolve_process_and_applicant(db, inst: Optional[OaInstance]): applicant = "" process_name = "" if not inst: return applicant, process_name if inst.initiator_user_id: ini = db.query(SysUser).filter(SysUser.id == inst.initiator_user_id).first() applicant = (ini.nick_name if ini else "") or "" if inst.app_id: app = ( db.query(OaApp) .filter(OaApp.id == inst.app_id, OaApp.del_flag == 0) .first() ) process_name = (app.app_name if app else "") or "" if not process_name: process_name = (inst.title or "").strip() return applicant, process_name 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() ) applicant, process_name = _resolve_process_and_applicant(db, inst) fallback = (inst.title if inst else "") or (task.node_name or "审批") title = _build_matter_title( applicant=applicant, process_name=process_name, fallback=fallback ) 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", return_path=RETURN_PENDING, ) send_oa_approval_notify( mp_openid=mp, pagepath=pagepath, title=title, applicant=applicant, node_name=node_name, time_text=time_text, process_name=process_name, ) elif task.task_type == "cc": pagepath = build_oa_detail_pagepath( instance_id=task.instance_id, task_id=task.id, mode="view", return_path=RETURN_CC, ) send_oa_cc_notify( mp_openid=mp, pagepath=pagepath, title=title, applicant=applicant, node_name=node_name, time_text=time_text, process_name=process_name, ) def _send_applicant_progress( db, *, instance_id: str, actor_user_id: str, result: str, task_id: str = "", ) -> None: iid = (instance_id or "").strip() if not iid: return inst = ( db.query(OaInstance) .filter(OaInstance.id == iid, OaInstance.del_flag == 0) .first() ) if not inst or not (inst.initiator_user_id or "").strip(): return ini = ( db.query(SysUser) .filter(SysUser.id == inst.initiator_user_id, SysUser.del_flag == 0) .first() ) mp = (getattr(ini, "wx_mp_openid", None) or "").strip() if ini else "" if not mp: logger.info( "oa wx applicant progress skip: no mp_openid initiator=%s inst=%s", inst.initiator_user_id, iid, ) return applicant = (ini.nick_name if ini else "") or "" process_name = "" if inst.app_id: app = ( db.query(OaApp) .filter(OaApp.id == inst.app_id, OaApp.del_flag == 0) .first() ) process_name = (app.app_name if app else "") or "" if not process_name: process_name = (inst.title or "").strip() business_name = _build_matter_title( applicant=applicant, process_name=process_name, fallback=(inst.title or "审批"), ) actor = None aid = (actor_user_id or "").strip() if aid: actor = db.query(SysUser).filter(SysUser.id == aid).first() approver_name = (actor.nick_name if actor else "") or "" pagepath = build_oa_detail_pagepath( instance_id=iid, task_id="", mode="view", return_path=RETURN_MINE, ) send_oa_applicant_progress_notify( mp_openid=mp, pagepath=pagepath, business_name=business_name, process_name=process_name, approver_name=approver_name, result=(result or RESULT_OTHER).strip() or RESULT_OTHER, time_text=datetime.now().strftime("%Y-%m-%d %H:%M"), ) def _oa_wx_notify_enabled() -> bool: return bool(getattr(settings, "WX_MP_NOTIFY_ENABLED", False)) def _send_mention_one( db, *, instance_id: str, mentioned_user_id: str, commenter_user_id: str, ) -> None: """评论 @:复用抄送模板;thing2=评论人,thing5=「xxx 在评论中提到了你」。""" uid = (mentioned_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 mention skip: no mp_openid user=%s inst=%s", uid, instance_id ) return iid = (instance_id or "").strip() inst = ( db.query(OaInstance) .filter(OaInstance.id == iid, OaInstance.del_flag == 0) .first() if iid else None ) process_name = "" if inst and inst.app_id: app = ( db.query(OaApp) .filter(OaApp.id == inst.app_id, OaApp.del_flag == 0) .first() ) process_name = (app.app_name if app else "") or "" if not process_name and inst: process_name = (inst.title or "").strip() commenter_name = "" cid = (commenter_user_id or "").strip() if cid: commenter = db.query(SysUser).filter(SysUser.id == cid).first() commenter_name = (commenter.nick_name if commenter else "") or "" pagepath = build_oa_detail_pagepath( instance_id=iid, task_id="", mode="view", return_path=RETURN_MENTIONS, ) send_oa_cc_notify( mp_openid=mp, pagepath=pagepath, title=_build_mention_title(commenter_name), applicant=commenter_name or "未知", node_name="", time_text=datetime.now().strftime("%Y-%m-%d %H:%M"), process_name=process_name or "审批", ) def notify_oa_mentions_sync( *, instance_id: str, commenter_user_id: str, mentioned_user_ids: Iterable[str], ) -> None: if not _oa_wx_notify_enabled(): logger.info("oa wx mention skipped: WX_MP_NOTIFY_ENABLED=false") return iid = (instance_id or "").strip() ids = [str(x).strip() for x in (mentioned_user_ids or []) if str(x).strip()] if not iid or not ids: return db = None try: db = SessionLocal() for uid in ids: try: _send_mention_one( db, instance_id=iid, mentioned_user_id=uid, commenter_user_id=commenter_user_id, ) except Exception: logger.exception( "oa wx mention failed user=%s inst=%s", uid, iid ) except Exception: logger.exception("oa wx mention batch failed inst=%s", iid) finally: if db is not None: db.close() def schedule_oa_mention_notifies( *, instance_id: str, commenter_user_id: str, mentioned_user_ids: List[str], ) -> None: if not _oa_wx_notify_enabled(): logger.info("oa wx mention schedule skipped: WX_MP_NOTIFY_ENABLED=false") return iid = (instance_id or "").strip() ids = [str(x).strip() for x in (mentioned_user_ids or []) if str(x).strip()] if not iid or not ids: return threading.Thread( target=notify_oa_mentions_sync, kwargs={ "instance_id": iid, "commenter_user_id": (commenter_user_id or "").strip(), "mentioned_user_ids": ids, }, daemon=True, name="oa-wx-mention", ).start() def notify_oa_tasks_sync(task_ids: Iterable[str]) -> None: if not _oa_wx_notify_enabled(): logger.info("oa wx notify skipped: WX_MP_NOTIFY_ENABLED=false") return 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 notify_oa_applicant_progress_sync( *, instance_id: str, actor_user_id: str, result: str, task_id: str = "", ) -> None: if not _oa_wx_notify_enabled(): logger.info("oa wx applicant progress skipped: WX_MP_NOTIFY_ENABLED=false") return db = None try: db = SessionLocal() _send_applicant_progress( db, instance_id=instance_id, actor_user_id=actor_user_id, result=result, task_id=task_id, ) except Exception: logger.exception( "oa wx applicant progress failed inst=%s actor=%s", instance_id, actor_user_id, ) finally: if db is not None: db.close() def schedule_oa_task_notifies(task_ids: List[str]) -> None: if not _oa_wx_notify_enabled(): logger.info("oa wx notify schedule skipped: WX_MP_NOTIFY_ENABLED=false") return 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() def schedule_oa_applicant_progress( *, instance_id: str, actor_user_id: str, action: str, task_id: str = "", ) -> None: if not _oa_wx_notify_enabled(): logger.info( "oa wx applicant progress schedule skipped: WX_MP_NOTIFY_ENABLED=false" ) return iid = (instance_id or "").strip() if not iid: return result = map_oa_action_to_result(action) threading.Thread( target=notify_oa_applicant_progress_sync, kwargs={ "instance_id": iid, "actor_user_id": (actor_user_id or "").strip(), "result": result, "task_id": (task_id or "").strip(), }, daemon=True, name="oa-wx-applicant-progress", ).start()