feat(backend): add WeChat service-account template message sender

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/.env.example

@@ -58,3 +58,14 @@ DATABASE_URL=mysql+pymysql://root:password@localhost:3306/zkxw
 # WXPAY_NOTIFY_URL=https://www.chengyantek.cn/api/payment/notify
 # WXPAY_REFUND_NOTIFY_URL=https://www.chengyantek.cn/api/payment/refund-notify
 # WXPAY_SANDBOX=false
+
+# 微信服务号（OA 审批/抄送模板消息；access_token 用服务号凭证，miniprogram.appid 仍用 WXPAY_MINI_APPID）
+# WX_MP_APPID=wx_service_account_appid
+# WX_MP_SECRET=wx_service_account_secret
+# WX_MP_TEMPLATE_APPROVAL=approval_template_id
+# WX_MP_TEMPLATE_CC=cc_template_id
+# 模板字段名需与微信后台一致（默认 thing1/thing2/thing3/time4）
+# WX_MP_TPL_FIELD_TITLE=thing1
+# WX_MP_TPL_FIELD_APPLICANT=thing2
+# WX_MP_TPL_FIELD_NODE=thing3
+# WX_MP_TPL_FIELD_TIME=time4

backend/app/core/config.py

@@ -100,6 +100,17 @@ class Settings(BaseSettings):
     WX_TEMPLATE_REFUND_RESULT: str = ""  # 退款结果通知模板ID
     WX_TEMPLATE_DELIVERY_WARNING: str = ""  # 配送时效预警模板ID
 
+    # 微信服务号（OA 审批/抄送通知）
+    WX_MP_APPID: str = ""
+    WX_MP_SECRET: str = ""
+    WX_MP_TEMPLATE_APPROVAL: str = ""
+    WX_MP_TEMPLATE_CC: str = ""
+    # 模板 data 字段名（与微信后台申请结果一致，可改 env）
+    WX_MP_TPL_FIELD_TITLE: str = "thing1"
+    WX_MP_TPL_FIELD_APPLICANT: str = "thing2"
+    WX_MP_TPL_FIELD_NODE: str = "thing3"
+    WX_MP_TPL_FIELD_TIME: str = "time4"
+
     # 与 docs/sql/zkxw.sql 基线一致：登录按租户 + sys_user.user_no
     DEFAULT_TENANT_NO: str = "zkxw"
     # 已废弃：无角色用户固定为 guest，不可通过环境变量提升为 super_admin

backend/app/services/wechat_mp_auth.py

@@ -0,0 +1,78 @@
+"""微信服务号：access_token、oauth code 换 openid。"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timedelta
+from typing import Any, Dict, Optional
+
+import httpx
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+_token: Optional[str] = None
+_token_expire: Optional[datetime] = None
+
+
+class WechatMpAuthError(Exception):
+    pass
+
+
+def _require_mp_credentials() -> tuple:
+    appid = (settings.WX_MP_APPID or "").strip()
+    secret = (settings.WX_MP_SECRET or "").strip()
+    if not appid or not secret:
+        raise WechatMpAuthError("未配置服务号凭证（WX_MP_APPID / WX_MP_SECRET）")
+    return appid, secret
+
+
+def get_mp_access_token() -> str:
+    global _token, _token_expire
+    if _token and _token_expire and datetime.now() < _token_expire:
+        return _token
+    appid, secret = _require_mp_credentials()
+    with httpx.Client(timeout=15.0) as client:
+        r = client.get(
+            "https://api.weixin.qq.com/cgi-bin/token",
+            params={
+                "grant_type": "client_credential",
+                "appid": appid,
+                "secret": secret,
+            },
+        )
+        r.raise_for_status()
+        data = r.json()
+    token = data.get("access_token")
+    if not token:
+        logger.warning("mp access_token failed: %s", data)
+        raise WechatMpAuthError("获取服务号 access_token 失败")
+    _token = token
+    _token_expire = datetime.now() + timedelta(seconds=int(data.get("expires_in", 7200)) - 300)
+    return token
+
+
+def oauth_code_to_openid(code: str) -> Dict[str, Any]:
+    """snsapi_base：code → openid / unionid。"""
+    c = (code or "").strip()
+    if not c:
+        raise WechatMpAuthError("code 不能为空")
+    appid, secret = _require_mp_credentials()
+    with httpx.Client(timeout=15.0) as client:
+        r = client.get(
+            "https://api.weixin.qq.com/sns/oauth2/access_token",
+            params={
+                "appid": appid,
+                "secret": secret,
+                "code": c,
+                "grant_type": "authorization_code",
+            },
+        )
+        r.raise_for_status()
+        data = r.json()
+    if data.get("errcode"):
+        raise WechatMpAuthError(data.get("errmsg") or "服务号授权失败")
+    openid = data.get("openid")
+    if not openid:
+        raise WechatMpAuthError("服务号未返回 openid")
+    return {"openid": openid, "unionid": data.get("unionid")}

backend/app/services/wechat_mp_message.py

@@ -0,0 +1,146 @@
+"""微信服务号模板消息：OA 审批/抄送通知。"""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict
+
+import httpx
+
+from app.core.config import settings
+from app.services.wechat_mp_auth import WechatMpAuthError, get_mp_access_token
+
+logger = logging.getLogger(__name__)
+
+_WX_TEMPLATE_SEND = "https://api.weixin.qq.com/cgi-bin/message/template/send"
+
+# 微信 thing 类字段建议约 20 汉字；按字符保守截断
+_THING_MAX_CHARS = 20
+
+
+def _truncate_thing(value: str, max_chars: int = _THING_MAX_CHARS) -> str:
+    s = (value or "").strip()
+    if len(s) <= max_chars:
+        return s
+    return s[:max_chars]
+
+
+def _build_oa_data(
+    *,
+    title: str,
+    applicant: str,
+    node_name: str,
+    time_text: str,
+) -> Dict[str, Dict[str, str]]:
+    return {
+        settings.WX_MP_TPL_FIELD_TITLE: {"value": _truncate_thing(title)},
+        settings.WX_MP_TPL_FIELD_APPLICANT: {"value": _truncate_thing(applicant)},
+        settings.WX_MP_TPL_FIELD_NODE: {"value": _truncate_thing(node_name)},
+        settings.WX_MP_TPL_FIELD_TIME: {"value": (time_text or "").strip()},
+    }
+
+
+def _send_template(
+    *,
+    mp_openid: str,
+    template_id: str,
+    data: Dict[str, Any],
+    pagepath: str,
+) -> Dict[str, Any]:
+    oid = (mp_openid or "").strip()
+    if not oid:
+        return {"success": False, "error": "missing_mp_openid"}
+
+    tid = (template_id or "").strip()
+    if not tid:
+        return {"success": False, "error": "missing_template_id"}
+
+    mini_appid = (settings.WXPAY_MINI_APPID or "").strip()
+    if not mini_appid:
+        return {"success": False, "error": "missing_mini_appid"}
+
+    try:
+        access_token = get_mp_access_token()
+    except WechatMpAuthError as exc:
+        logger.warning("mp template send aborted: %s", exc)
+        return {"success": False, "error": str(exc)}
+
+    payload = {
+        "touser": oid,
+        "template_id": tid,
+        "data": data,
+        "miniprogram": {
+            "appid": mini_appid,
+            "pagepath": (pagepath or "").lstrip("/"),
+        },
+    }
+
+    try:
+        with httpx.Client(timeout=15.0) as client:
+            r = client.post(
+                f"{_WX_TEMPLATE_SEND}?access_token={access_token}",
+                json=payload,
+            )
+            r.raise_for_status()
+            result = r.json()
+    except Exception as exc:
+        logger.exception("mp template send request failed")
+        return {"success": False, "error": str(exc)}
+
+    if result.get("errcode") in (0, None):
+        logger.info(
+            "mp template send ok: openid=%s msgid=%s",
+            oid,
+            result.get("msgid"),
+        )
+        return {"success": True, "msgid": result.get("msgid")}
+
+    logger.error("mp template send failed: %s", result)
+    return {
+        "success": False,
+        "error": result.get("errmsg") or "send_failed",
+        "errcode": result.get("errcode"),
+    }
+
+
+def send_oa_approval_notify(
+    *,
+    mp_openid: str,
+    pagepath: str,
+    title: str,
+    applicant: str,
+    node_name: str,
+    time_text: str,
+) -> dict:
+    return _send_template(
+        mp_openid=mp_openid,
+        template_id=settings.WX_MP_TEMPLATE_APPROVAL,
+        data=_build_oa_data(
+            title=title,
+            applicant=applicant,
+            node_name=node_name,
+            time_text=time_text,
+        ),
+        pagepath=pagepath,
+    )
+
+
+def send_oa_cc_notify(
+    *,
+    mp_openid: str,
+    pagepath: str,
+    title: str,
+    applicant: str,
+    node_name: str,
+    time_text: str,
+) -> dict:
+    return _send_template(
+        mp_openid=mp_openid,
+        template_id=settings.WX_MP_TEMPLATE_CC,
+        data=_build_oa_data(
+            title=title,
+            applicant=applicant,
+            node_name=node_name,
+            time_text=time_text,
+        ),
+        pagepath=pagepath,
+    )

backend/tests/test_wechat_mp_message.py

@@ -0,0 +1,121 @@
+from unittest.mock import MagicMock, patch
+
+from app.core.config import settings
+
+
+def test_send_approval_requires_mp_openid():
+    from app.services.wechat_mp_message import send_oa_approval_notify
+
+    r = send_oa_approval_notify(
+        mp_openid="",
+        pagepath="pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
+        title="请假",
+        applicant="张三",
+        node_name="主管审批",
+        time_text="2026-08-13 12:00",
+    )
+    assert r.get("success") is False
+    assert r.get("error") == "missing_mp_openid"
+
+
+def test_send_approval_requires_template_id():
+    from app.services.wechat_mp_message import send_oa_approval_notify
+
+    with patch.object(settings, "WX_MP_TEMPLATE_APPROVAL", ""):
+        r = send_oa_approval_notify(
+            mp_openid="mp_openid_1",
+            pagepath="pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
+            title="请假",
+            applicant="张三",
+            node_name="主管审批",
+            time_text="2026-08-13 12:00",
+        )
+    assert r.get("success") is False
+    assert r.get("error") == "missing_template_id"
+
+
+@patch("app.services.wechat_mp_message.get_mp_access_token", return_value="tok_mp")
+@patch("app.services.wechat_mp_message.httpx.Client")
+def test_send_approval_success_payload_shape(mock_client_cls, _mock_token):
+    from app.services.wechat_mp_message import send_oa_approval_notify
+
+    mock_resp = MagicMock()
+    mock_resp.raise_for_status = MagicMock()
+    mock_resp.json.return_value = {"errcode": 0, "msgid": 1001}
+    mock_client = MagicMock()
+    mock_client.__enter__.return_value = mock_client
+    mock_client.__exit__.return_value = False
+    mock_client.post.return_value = mock_resp
+    mock_client_cls.return_value = mock_client
+
+    long_title = "这是一个非常非常长的审批标题需要截断处理一下才行"
+
+    with patch.object(settings, "WX_MP_TEMPLATE_APPROVAL", "tpl_approval"), patch.object(
+        settings, "WXPAY_MINI_APPID", "wx_mini_appid"
+    ), patch.object(settings, "WX_MP_TPL_FIELD_TITLE", "thing1"), patch.object(
+        settings, "WX_MP_TPL_FIELD_APPLICANT", "thing2"
+    ), patch.object(settings, "WX_MP_TPL_FIELD_NODE", "thing3"), patch.object(
+        settings, "WX_MP_TPL_FIELD_TIME", "time4"
+    ):
+        r = send_oa_approval_notify(
+            mp_openid="mp_openid_1",
+            pagepath="/pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
+            title=long_title,
+            applicant="张三",
+            node_name="主管审批",
+            time_text="2026-08-13 12:00",
+        )
+
+    assert r == {"success": True, "msgid": 1001}
+    assert mock_client.post.call_count == 1
+    args, kwargs = mock_client.post.call_args
+    assert args[0].startswith(
+        "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token="
+    )
+    payload = kwargs["json"]
+    assert payload["touser"] == "mp_openid_1"
+    assert payload["template_id"] == "tpl_approval"
+    assert payload["miniprogram"] == {
+        "appid": "wx_mini_appid",
+        "pagepath": "pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
+    }
+    assert payload["data"]["thing1"]["value"] == long_title[:20]
+    assert payload["data"]["thing2"]["value"] == "张三"
+    assert payload["data"]["thing3"]["value"] == "主管审批"
+    assert payload["data"]["time4"]["value"] == "2026-08-13 12:00"
+
+
+@patch("app.services.wechat_mp_message.get_mp_access_token", return_value="tok_mp")
+@patch("app.services.wechat_mp_message.httpx.Client")
+def test_send_cc_success_payload_shape(mock_client_cls, _mock_token):
+    from app.services.wechat_mp_message import send_oa_cc_notify
+
+    mock_resp = MagicMock()
+    mock_resp.raise_for_status = MagicMock()
+    mock_resp.json.return_value = {"errcode": 0, "msgid": 2002}
+    mock_client = MagicMock()
+    mock_client.__enter__.return_value = mock_client
+    mock_client.__exit__.return_value = False
+    mock_client.post.return_value = mock_resp
+    mock_client_cls.return_value = mock_client
+
+    with patch.object(settings, "WX_MP_TEMPLATE_CC", "tpl_cc"), patch.object(
+        settings, "WXPAY_MINI_APPID", "wx_mini_appid"
+    ):
+        r = send_oa_cc_notify(
+            mp_openid="mp_openid_2",
+            pagepath="pages/oa/webview/index?instanceId=i2&taskId=t2&mode=view",
+            title="报销",
+            applicant="李四",
+            node_name="抄送知会",
+            time_text="2026-08-13 15:30",
+        )
+
+    assert r == {"success": True, "msgid": 2002}
+    payload = mock_client.post.call_args.kwargs["json"]
+    assert payload["touser"] == "mp_openid_2"
+    assert payload["template_id"] == "tpl_cc"
+    assert payload["miniprogram"]["appid"] == "wx_mini_appid"
+    assert payload["miniprogram"]["pagepath"].endswith("mode=view")
+    assert "thing1" in payload["data"]
+    assert payload["data"]["thing1"]["value"] == "报销"