diff --git a/app/api/v1/wx_kf.py b/app/api/v1/wx_kf.py new file mode 100644 index 0000000..b12305c --- /dev/null +++ b/app/api/v1/wx_kf.py @@ -0,0 +1,159 @@ +"""微信客服(企业微信)消息接收回调 /wx/kf/callback(裸路径, 非 /api/v1)。 + +对应企业微信「微信客服」→ 客服账号 → 接收消息回调里填的 URL。是服务号版(wx_mp.py)的 +平行实现:下游(set_pending → 心跳下发 → 前端弹窗)完全一致, 只有上游"如何拿到用户消息"不同 —— +服务号是微信把消息直接 POST 给你;微信客服是 POST 一个事件通知, 你再用 主动 sync_msg 拉。 + + GET : URL 接入验证 → msg_signature 验签(含密文 echostr) → 解密 echostr → 返回明文。 + ⚠️ 与公众号明文模式不同:企业微信 echostr 是密文, 必须解密后返回。 + POST : 收事件(密文) → msg_signature 验签 → AES 解密 → 解析 XML。 + Event=kf_msg_or_event → 取 / → sync_msg 增量拉消息 → + 客户(origin=3)发来的图片(image) → set_pending(测试设备, "meituan") + 下载落盘。 + 其余消息/事件记日志忽略。一律返回 "success"(异常也吞掉, 不让企业微信重试轰炸)。 + +回调加密与公众号「安全模式」同源, 直接复用 integrations/wx_mp_crypto(验签 + AES-256-CBC 解密), +差别只在 receiveid 用 corpid(公众号是 appid)。MVP 只做"收图打信号 + 落盘证明链路通"。 +""" +from __future__ import annotations + +import logging +import xml.etree.ElementTree as ET +from pathlib import Path +from typing import Any + +from fastapi import APIRouter, Request +from fastapi.responses import PlainTextResponse + +from app.core import wx_poc_signal +from app.core.config import settings +from app.integrations import wx_kf_client, wx_mp_crypto + +logger = logging.getLogger("shagua.wx_kf") + +router = APIRouter(prefix="/wx/kf", tags=["wx-kf"]) + +# 收到的截图暂存目录(MVP 验证用;接上识别后可改为不落盘或定期清理) +_INBOX = Path(settings.MEDIA_ROOT) / "wx_kf_inbox" + + +@router.get("/callback", include_in_schema=False) +async def verify( + msg_signature: str = "", timestamp: str = "", nonce: str = "", echostr: str = "" +) -> PlainTextResponse: + """企业微信 URL 接入验证:验签(密文 echostr 参与)→ 解密 echostr → 返回明文。""" + if not settings.wx_kf_callback_configured: + logger.warning("wx_kf verify: 回调凭证未配齐(CorpId/Token/AESKey/Secret)") + return PlainTextResponse("", status_code=503) + if not wx_mp_crypto.verify_msg_signature( + settings.WX_KF_TOKEN, timestamp, nonce, echostr, msg_signature + ): + logger.warning("wx_kf verify: signature 校验失败 ts=%s nonce=%s", timestamp, nonce) + return PlainTextResponse("invalid signature", status_code=403) + try: + plain = wx_mp_crypto.decrypt_message( + settings.WX_KF_AES_KEY, settings.WX_KF_CORP_ID, echostr + ) + except Exception: + logger.exception("wx_kf verify: echostr 解密失败") + return PlainTextResponse("decrypt failed", status_code=403) + return PlainTextResponse(plain) + + +@router.post("/callback", include_in_schema=False) +async def receive(request: Request) -> PlainTextResponse: + """收事件通知(安全模式)。任何异常都吞掉返回 success, 不让企业微信重试轰炸, 靠日志排查。""" + if not settings.wx_kf_callback_configured: + logger.warning("wx_kf receive: 回调凭证未配齐") + return PlainTextResponse("success") + try: + body = (await request.body()).decode("utf-8") + qp = request.query_params + # 安全模式外层 XML 只有 ToUserName + Encrypt。来源=企业微信服务器(HTTPS)+ 下面验签, + # 且 stdlib ET 不扩展外部实体, XXE 不适用。 + encrypt = ET.fromstring(body).findtext("Encrypt") or "" + if not wx_mp_crypto.verify_msg_signature( + settings.WX_KF_TOKEN, + qp.get("timestamp", ""), + qp.get("nonce", ""), + encrypt, + qp.get("msg_signature", ""), + ): + logger.warning("wx_kf receive: msg_signature 校验失败") + return PlainTextResponse("success") + xml = wx_mp_crypto.decrypt_message( + settings.WX_KF_AES_KEY, settings.WX_KF_CORP_ID, encrypt + ) + await _handle_event(ET.fromstring(xml)) + except Exception: + logger.exception("wx_kf receive: 处理异常") + return PlainTextResponse("success") + + +async def _handle_event(root: ET.Element) -> None: + """客服事件:MsgType=event 且 Event=kf_msg_or_event → 用 Token 增量拉消息。""" + event = root.findtext("Event") or "" + if event != "kf_msg_or_event": + logger.info("wx_kf 收到事件(暂忽略): event=%s", event) + return + token = root.findtext("Token") or "" + open_kfid = root.findtext("OpenKfId") or "" + msgs = await wx_kf_client.sync_messages(token, open_kfid) + logger.info("wx_kf sync_msg 拉到 %d 条 open_kfid=%s", len(msgs), open_kfid) + for msg in msgs: + await _handle_message(msg) + + +async def _handle_message(msg: dict[str, Any]) -> None: + """处理单条客服消息。只关心【客户发来的】(origin=3)图片 / 小程序卡片:打比价信号(+ 落盘)。 + + origin:3=客户发的 / 4=系统推送 / 5=接待人员发的。接待人员或系统消息一律不触发比价。 + 两种触发物对应场景:image=用户发结算页截图;miniprogram=用户分享美团小程序卡片(见需求截图)。 + """ + if msg.get("origin") != 3: + return + msgtype = msg.get("msgtype") or "" + external_userid = msg.get("external_userid") or "" + if msgtype == "image": + media_id = (msg.get("image") or {}).get("media_id", "") + logger.info( + "wx_kf 收到图片: external_userid=%s media_id=%s", external_userid, media_id + ) + _fire_pending("image") + await _save_media(media_id, external_userid) + elif msgtype == "miniprogram": + mp = msg.get("miniprogram") or {} + logger.info( + "wx_kf 收到小程序卡片: external_userid=%s title=%s appid=%s", + external_userid, + mp.get("title"), + mp.get("appid"), + ) + # 卡片无可下载媒体, 只打信号;源平台/订单识别(可据 appid 判平台)留后续, 现写死美团。 + _fire_pending("miniprogram") + else: + logger.info( + "wx_kf 收到消息(暂忽略): external_userid=%s type=%s", external_userid, msgtype + ) + + +def _fire_pending(kind: str) -> None: + """PoC:给写死的测试设备打"从美团比价"信号(下次心跳带回前端弹窗, 与服务号版完全一致)。""" + poc_dev = settings.WX_POC_TEST_DEVICE_ID + if not poc_dev: + return + wx_poc_signal.set_pending(poc_dev, "meituan") + logger.info("wx_kf PoC: 已给测试设备 %s 打比价信号 source=meituan (触发=%s)", poc_dev, kind) + + +async def _save_media(media_id: str, external_userid: str) -> None: + """图片消息只给 media_id, 走临时素材下载接口拉回落盘(PoC 验证用)。""" + content = await wx_kf_client.download_media(media_id) + if not content: + return + try: + _INBOX.mkdir(parents=True, exist_ok=True) + dest = _INBOX / f"{external_userid[:16]}_{media_id[:16]}.jpg" + dest.write_bytes(content) + logger.info("wx_kf 图片已落盘: %s (%d bytes)", dest, len(content)) + except Exception: + logger.exception("wx_kf 图片落盘失败 media_id=%s", media_id) diff --git a/app/core/config.py b/app/core/config.py index 88777ab..4fdae08 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -147,6 +147,15 @@ class Settings(BaseSettings): # 收图 → 给此 device_id 打 pending → 该设备下次心跳带回 → 前端弹选平台窗。 WX_POC_TEST_DEVICE_ID: str = "" + # ===== 微信客服(企业微信;截图比价上游入口 —— 服务号版的平行实现) ===== + # 用户在「微信客服」会话发消息 → 企业微信 POST 事件到 /wx/kf/callback → 后端 sync_msg 拉消息。 + # 回调加密与公众号「安全模式」同源(复用 wx_mp_crypto), 但 receiveid 用 corpid 且 echostr 需解密。 + # 拉消息 / 下载图片走 qyapi, 需 access_token(corpid + secret 换取)。全空=回调端点拒收, 零风险。 + WX_KF_CORP_ID: str = "" # 企业微信 corpid(回调解密 receiveid + 换 access_token) + WX_KF_SECRET: str = "" # 「微信客服」Secret(换 access_token) + WX_KF_TOKEN: str = "" # 客服「接收消息」回调的 Token(URL 验签 + 消息验签) + WX_KF_AES_KEY: str = "" # 客服「接收消息」回调的 EncodingAESKey(AES-256-CBC 解密) + @property def wx_mp_configured(self) -> bool: """服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。""" @@ -162,6 +171,16 @@ class Settings(BaseSettings): """消息接收回调凭证齐全(Token + AESKey + AppID)。缺则回调端点拒绝处理消息。""" return bool(self.WX_MP_TOKEN and self.WX_MP_AES_KEY and self.WX_MP_APPID) + @property + def wx_kf_callback_configured(self) -> bool: + """微信客服回调凭证齐全(CorpId + Token + AESKey + Secret)。缺则回调端点拒绝处理。""" + return bool( + self.WX_KF_CORP_ID + and self.WX_KF_TOKEN + and self.WX_KF_AES_KEY + and self.WX_KF_SECRET + ) + # ===== 微信支付(商家转账到零钱 / 提现)===== # 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于 # 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。 diff --git a/app/integrations/wx_kf_client.py b/app/integrations/wx_kf_client.py new file mode 100644 index 0000000..761279a --- /dev/null +++ b/app/integrations/wx_kf_client.py @@ -0,0 +1,126 @@ +"""微信客服(企业微信)API 客户端:access_token 缓存 + sync_msg 拉消息 + 临时素材下载。 + +对应 api/v1/wx_kf.py:收到 `kf_msg_or_event` 回调后, 用这里的函数增量拉取用户消息。 + - access_token(7200s)进程内缓存, 提前 300s 刷新(单 worker 够用, 重启即失效)。 + - sync_msg 游标 next_cursor 按 open_kfid 进程内存续(PoC 重启后从最近 3 天重拉, 可接受)。 +凭证来自 settings(WX_KF_CORP_ID / WX_KF_SECRET);纯外部 HTTP, 不含业务逻辑。 +""" +from __future__ import annotations + +import asyncio +import logging +import time +from typing import Any + +import httpx + +from app.core.config import settings + +logger = logging.getLogger("shagua.wx_kf") + +_QYAPI = "https://qyapi.weixin.qq.com/cgi-bin" + +# access_token 进程内缓存(单 worker)。token 空 / 未到刷新点直接用。 +_token_lock = asyncio.Lock() +_token_cache: dict[str, Any] = {"token": "", "expire_at": 0.0} + +# sync_msg 游标:open_kfid -> next_cursor(PoC 进程内存;重启从最近 3 天重拉) +_cursors: dict[str, str] = {} + + +async def _get_access_token() -> str: + """取企业微信 access_token, 进程内缓存, 提前 300s 过期刷新。失败返回空串(调用方降级)。""" + now = time.time() + if _token_cache["token"] and _token_cache["expire_at"] - 300 > now: + return _token_cache["token"] + async with _token_lock: + now = time.time() # 拿锁后复检, 避免并发重复刷新 + if _token_cache["token"] and _token_cache["expire_at"] - 300 > now: + return _token_cache["token"] + try: + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.get( + f"{_QYAPI}/gettoken", + params={ + "corpid": settings.WX_KF_CORP_ID, + "corpsecret": settings.WX_KF_SECRET, + }, + ) + data = resp.json() + except Exception: + logger.exception("wx_kf gettoken 请求失败") + return "" + if data.get("errcode"): + logger.warning( + "wx_kf gettoken errcode=%s errmsg=%s", data.get("errcode"), data.get("errmsg") + ) + return "" + _token_cache["token"] = data.get("access_token", "") + _token_cache["expire_at"] = now + int(data.get("expires_in", 7200)) + return _token_cache["token"] + + +async def sync_messages(token: str, open_kfid: str) -> list[dict[str, Any]]: + """收到 kf_msg_or_event 回调后调用:用回调带的一次性 token 增量拉消息, has_more 循环拉完。 + + token: 回调事件里的 (消息拉取凭证, 非配置 Token;不传会有严格频控)。 + open_kfid:客服账号 id(回调里的 )。 + 返回本次新拉到的 msg_list(已合并多页);游标按 open_kfid 进程内存续。 + """ + access_token = await _get_access_token() + if not access_token: + return [] + out: list[dict[str, Any]] = [] + cursor = _cursors.get(open_kfid, "") + try: + async with httpx.AsyncClient(timeout=15) as client: + for _ in range(20): # 最多 20 页护栏, 防异常时死循环 + body: dict[str, Any] = {"cursor": cursor, "token": token, "limit": 1000} + if open_kfid: + body["open_kfid"] = open_kfid + resp = await client.post( + f"{_QYAPI}/kf/sync_msg", + params={"access_token": access_token}, + json=body, + ) + data = resp.json() + if data.get("errcode"): + logger.warning( + "wx_kf sync_msg errcode=%s errmsg=%s", + data.get("errcode"), + data.get("errmsg"), + ) + break + out.extend(data.get("msg_list", [])) + cursor = data.get("next_cursor", cursor) + _cursors[open_kfid] = cursor + if not data.get("has_more"): + break + except Exception: + logger.exception("wx_kf sync_msg 请求失败 open_kfid=%s", open_kfid) + return out + + +async def download_media(media_id: str) -> bytes | None: + """临时素材下载(图片消息只给 media_id)。成功返回二进制, 失败返回 None。""" + if not media_id: + return None + access_token = await _get_access_token() + if not access_token: + return None + try: + async with httpx.AsyncClient(timeout=20) as client: + resp = await client.get( + f"{_QYAPI}/media/get", + params={"access_token": access_token, "media_id": media_id}, + ) + resp.raise_for_status() + # 成功=二进制文件流;失败=JSON(errcode)。据 Content-Type 区分。 + ctype = resp.headers.get("Content-Type", "") + if "application/json" in ctype or "text/plain" in ctype: + logger.warning("wx_kf media/get 非文件响应: %s", resp.text[:200]) + return None + return resp.content + except Exception: + logger.exception("wx_kf media/get 下载失败 media_id=%s", media_id) + return None diff --git a/app/main.py b/app/main.py index 5c31f9f..7dcd4fe 100644 --- a/app/main.py +++ b/app/main.py @@ -39,6 +39,7 @@ from app.api.v1.signin import router as signin_router from app.api.v1.tasks import router as tasks_router from app.api.v1.user import router as user_router from app.api.v1.wallet import router as wallet_router +from app.api.v1.wx_kf import router as wx_kf_router from app.api.v1.wx_mp import router as wx_mp_router from app.api.v1.wxpay import router as wxpay_router from app.core.config import settings @@ -143,6 +144,8 @@ app.include_router(platform_router) app.include_router(cps_redirect_router) # 微信服务号消息接收回调 /wx/mp/callback(公网无鉴权:微信服务器验签, 截图比价入口) app.include_router(wx_mp_router) +# 微信客服(企业微信)消息接收回调 /wx/kf/callback(服务号版的平行实现, 收事件→sync_msg 拉图) +app.include_router(wx_kf_router) # 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。 _media_root = Path(settings.MEDIA_ROOT)