diff --git a/backend/.env.example b/backend/.env.example index dc201f9e..1405e196 100644 --- a/backend/.env.example +++ b/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 diff --git a/backend/app/core/config.py b/backend/app/core/config.py index c1d3c541..2d73fdc6 100644 --- a/backend/app/core/config.py +++ b/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 diff --git a/backend/app/services/wechat_mp_auth.py b/backend/app/services/wechat_mp_auth.py new file mode 100644 index 00000000..626e01b8 --- /dev/null +++ b/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")} diff --git a/backend/app/services/wechat_mp_message.py b/backend/app/services/wechat_mp_message.py new file mode 100644 index 00000000..4332610a --- /dev/null +++ b/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, + ) diff --git a/backend/tests/test_wechat_mp_message.py b/backend/tests/test_wechat_mp_message.py new file mode 100644 index 00000000..47bda341 --- /dev/null +++ b/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"] == "报销"