From 9eaa987a8170b028df6f28e9e9125e782c591afe Mon Sep 17 00:00:00 2001 From: zzhyyyyy <2685922758@qq.com> Date: Fri, 17 Jul 2026 21:54:05 +0800 Subject: [PATCH] =?UTF-8?q?feat(wx-finance):=20=E4=BC=9A=E8=AF=9D=E5=86=85?= =?UTF-8?q?=E5=AE=B9=E5=AD=98=E6=A1=A3=E6=AF=94=E4=BB=B7=E4=B8=8A=E6=B8=B8?= =?UTF-8?q?=E5=85=A5=E5=8F=A3(=E5=8F=91=E7=BB=99=E6=88=90=E5=91=98?= =?UTF-8?q?=E9=82=A3=E6=9D=A1=E8=B7=AF)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 企业微信「会话内容存档」侧新增上游入口:用户把美团小程序卡片/截图发给成员(非客服) → 轮询 GetChatData 拉取 → RSA+AES 解密 → 识别 weapp/image → set_pending。下游(心跳→ 弹窗→比价)与 wx_kf/服务号版完全复用。补微信客服收不了第三方小程序卡片分享的短板。 - wx_finance_sdk: WeWorkFinanceSdk C 库 ctypes 绑定(NewSdk/Init/GetChatData/DecryptData) + RSA(PKCS1v15)解 encrypt_random_key。.so 实例化时才加载, 无 .so 机器可安全 import。 - wx_finance_worker: 独立线程轮询(阻塞 C 调用不入 asyncio)+ seq 游标文件持久化; 只处理外部用户(from≠接收成员)发来的 weapp/image。 - scripts/wx_finance_probe.py: 上 worker 前的连通性探针(云上验 .so/Init/GetChatData/解密)。 - config 增 WX_FINANCE_*(ENABLED 默认关 → 未配好零影响)+ wx_finance_configured; main lifespan 起停 worker;corpid 复用 WX_KF_CORP_ID。无新依赖、无 migration。 Co-Authored-By: Claude Opus 4.8 --- app/core/config.py | 18 +++++ app/core/wx_finance_worker.py | 116 +++++++++++++++++++++++++++++ app/integrations/wx_finance_sdk.py | 100 +++++++++++++++++++++++++ app/main.py | 7 ++ scripts/wx_finance_probe.py | 45 +++++++++++ 5 files changed, 286 insertions(+) create mode 100644 app/core/wx_finance_worker.py create mode 100644 app/integrations/wx_finance_sdk.py create mode 100644 scripts/wx_finance_probe.py diff --git a/app/core/config.py b/app/core/config.py index 4fdae08..3c583ff 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -156,6 +156,19 @@ class Settings(BaseSettings): WX_KF_TOKEN: str = "" # 客服「接收消息」回调的 Token(URL 验签 + 消息验签) WX_KF_AES_KEY: str = "" # 客服「接收消息」回调的 EncodingAESKey(AES-256-CBC 解密) + # ===== 微信会话内容存档(企业微信;"发给成员"那条路的截图/美团卡片比价上游入口) ===== + # 用户把美团小程序卡片/截图发给企业微信【成员】(非客服) → 会话存档 GetChatData 轮询拉取 → + # RSA+AES 解密 → 识别 weapp/image → 给测试设备打比价信号。下游(心跳→弹窗→比价)复用。 + # 需后台开通「会话内容存档」+ 上传 RSA 公钥 + 配可信IP + 拿存档 Secret;.so 放服务器。 + # corpid 复用上面的 WX_KF_CORP_ID(同一企业)。ENABLED 默认关 → 未配好前零影响。 + WX_FINANCE_ENABLED: bool = False # 总开关(worker 是否启动) + WX_FINANCE_SECRET: str = "" # 会话存档【专用】Secret(≠微信客服/自建应用) + WX_FINANCE_SDK_PATH: str = "./libWeWorkFinanceSdk_C.so" # WeWorkFinanceSdk C 库(Linux .so) + WX_FINANCE_PRIVATE_KEY_PATH: str = "./secrets/wx_finance_private.pem" # RSA 私钥(解 encrypt_random_key) + WX_FINANCE_POLL_INTERVAL_SEC: int = 3 # GetChatData 轮询间隔(无回调, 自己定频) + WX_FINANCE_SEQ_FILE: str = "./data/wx_finance_seq.txt" # seq 游标持久化(防丢/重复) + WX_FINANCE_RECEIVER_USERID: str = "" # 接收成员 userid(万朗杰);只处理【非他发】的消息, 空=不过滤 + @property def wx_mp_configured(self) -> bool: """服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。""" @@ -181,6 +194,11 @@ class Settings(BaseSettings): and self.WX_KF_SECRET ) + @property + def wx_finance_configured(self) -> bool: + """会话存档 worker 是否该启动 = 总开关开 且 corpid + 存档 Secret 齐全。""" + return bool(self.WX_FINANCE_ENABLED and self.WX_KF_CORP_ID and self.WX_FINANCE_SECRET) + # ===== 微信支付(商家转账到零钱 / 提现)===== # 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于 # 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。 diff --git a/app/core/wx_finance_worker.py b/app/core/wx_finance_worker.py new file mode 100644 index 0000000..7909bfa --- /dev/null +++ b/app/core/wx_finance_worker.py @@ -0,0 +1,116 @@ +"""微信会话内容存档轮询 worker:定时 GetChatData → 解密 → 识别【外部用户发给成员】的 weapp/image +→ 打比价信号(复用 wx_poc_signal)。下游(心跳 → 弹窗 → 比价)完全复用, 与 wx_kf / 服务号版一致。 + +在【独立线程】跑:GetChatData 是阻塞式 C 调用, 不放进 asyncio 事件循环。gate 在 +settings.wx_finance_configured(总开关 + corpid + secret)。seq 游标持久化到文件(防重启重拉 / 丢消息)。 +""" +from __future__ import annotations + +import logging +import threading +from pathlib import Path + +from app.core import wx_poc_signal +from app.core.config import settings +from app.integrations.wx_finance_sdk import WxFinanceError, WxFinanceSdk + +logger = logging.getLogger("shagua.wx_finance") + +_stop = threading.Event() +_thread: threading.Thread | None = None + + +def _load_seq() -> int: + try: + return int(Path(settings.WX_FINANCE_SEQ_FILE).read_text().strip() or "0") + except Exception: + return 0 + + +def _save_seq(seq: int) -> None: + try: + p = Path(settings.WX_FINANCE_SEQ_FILE) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(str(seq)) + except Exception: + logger.exception("wx_finance seq 持久化失败") + + +def _handle_msg(msg: dict) -> None: + """明文消息:外部用户(from≠接收成员)发来的 weapp(美团卡片)/ image(截图)→ 打比价信号。 + + 只处理"发给成员"的:from==接收成员 = 成员自己发的, 跳过(避免成员回消息也触发)。 + 源平台/订单识别留后续(可据 weapp.username=appid 判平台), PoC 写死 meituan。 + """ + msgtype = msg.get("msgtype") or "" + frm = msg.get("from") or "" + receiver = settings.WX_FINANCE_RECEIVER_USERID + if receiver and frm == receiver: + return # 成员自己发的, 不触发 + if msgtype not in ("weapp", "image"): + return # 只关心美团卡片 / 截图 + logger.info("wx_finance 命中触发 from=%s type=%s", frm, msgtype) + poc_dev = settings.WX_POC_TEST_DEVICE_ID + if poc_dev: + wx_poc_signal.set_pending(poc_dev, "meituan") + logger.info( + "wx_finance PoC: 已给测试设备 %s 打比价信号 source=meituan (触发=%s)", poc_dev, msgtype + ) + + +def _run() -> None: + try: + sdk = WxFinanceSdk( + settings.WX_FINANCE_SDK_PATH, + settings.WX_KF_CORP_ID, + settings.WX_FINANCE_SECRET, + Path(settings.WX_FINANCE_PRIVATE_KEY_PATH).read_bytes(), + ) + except Exception: + logger.exception( + "wx_finance SDK 初始化失败, worker 退出(检查 .so / corpid / 存档Secret / 私钥 / 可信IP)" + ) + return + seq = _load_seq() + logger.info("wx_finance worker 启动, 从 seq=%d 开始轮询, 间隔 %ds", seq, settings.WX_FINANCE_POLL_INTERVAL_SEC) + while not _stop.is_set(): + try: + msgs = sdk.get_chat_data(seq, limit=1000) + if msgs: + logger.info("wx_finance 拉到 %d 条(seq>%d)", len(msgs), seq) + for item in msgs: + try: + plain = sdk.decrypt(item["encrypt_random_key"], item["encrypt_chat_msg"]) + _handle_msg(plain) + except Exception: + logger.exception("wx_finance 解密/处理单条失败 seq=%s", item.get("seq")) + seq = max(seq, int(item.get("seq", seq))) + if msgs: + _save_seq(seq) + except WxFinanceError: + logger.exception("wx_finance GetChatData 出错") + except Exception: + logger.exception("wx_finance 轮询异常") + _stop.wait(settings.WX_FINANCE_POLL_INTERVAL_SEC) + try: + sdk.close() + except Exception: + pass + logger.info("wx_finance worker 已停止") + + +def start_wx_finance_worker() -> None: + """gate 在 wx_finance_configured;未开启则直接返回(零影响)。""" + global _thread + if not settings.wx_finance_configured: + return + _stop.clear() + _thread = threading.Thread(target=_run, name="wx-finance-worker", daemon=True) + _thread.start() + logger.info("wx_finance worker 线程已启动") + + +def stop_wx_finance_worker() -> None: + _stop.set() + if _thread is not None: + _thread.join(timeout=5) diff --git a/app/integrations/wx_finance_sdk.py b/app/integrations/wx_finance_sdk.py new file mode 100644 index 0000000..b6e92a3 --- /dev/null +++ b/app/integrations/wx_finance_sdk.py @@ -0,0 +1,100 @@ +"""微信会话内容存档 SDK 封装(WeWorkFinanceSdk C 库的 ctypes 绑定 + 消息解密)。 + +企业微信「会话内容存档」是拿"成员↔外部用户私聊消息正文"(含美团小程序卡片)的官方途径: +主动轮询 GetChatData(seq) 拉取。每条消息两段密文: + - encrypt_random_key:用【企业自持 RSA 私钥】解出 AES 密钥 + - encrypt_chat_msg :SDK DecryptData 用该密钥解出明文消息 JSON +本模块封装:加载 .so → Init → GetChatData → RSA 解密随机密钥 → DecryptData。纯外部依赖, +不含业务逻辑(识别 weapp/image → 打信号在 core/wx_finance_worker.py)。 + +.so 需放服务器(Linux `libWeWorkFinanceSdk_C.so`),路径 settings.WX_FINANCE_SDK_PATH。 +⚠️ CDLL 在实例化时才加载(非 import 期),故本模块在无 .so 的机器上也能安全 import。 +""" +from __future__ import annotations + +import base64 +import ctypes +import json +import logging +from typing import Any + +from cryptography.hazmat.primitives.asymmetric import padding +from cryptography.hazmat.primitives.serialization import load_pem_private_key + +logger = logging.getLogger("shagua.wx_finance") + + +class WxFinanceError(RuntimeError): + """SDK 调用返回非 0 / 业务 errcode 时抛。""" + + +class WxFinanceSdk: + """WeWorkFinanceSdk 薄封装。实例化即加载 .so + Init(需 corpid/secret 正确 + 服务器 IP 在可信IP)。""" + + def __init__(self, sdk_path: str, corpid: str, secret: str, private_key_pem: bytes) -> None: + self._priv = load_pem_private_key(private_key_pem, password=None) + self._lib = ctypes.CDLL(sdk_path) # 缺 .so / 缺 libssl 依赖会在此抛 OSError + self._bind() + self._sdk = self._lib.NewSdk() + ret = self._lib.Init(self._sdk, corpid.encode(), secret.encode()) + if ret != 0: + raise WxFinanceError(f"Init 失败 ret={ret}(检查 corpid / 会话存档 Secret / 可信IP)") + + def _bind(self) -> None: + lib = self._lib + lib.NewSdk.restype = ctypes.c_void_p + lib.Init.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p] + lib.Init.restype = ctypes.c_int + lib.GetChatData.argtypes = [ + ctypes.c_void_p, ctypes.c_ulonglong, ctypes.c_uint, + ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p, + ] + lib.GetChatData.restype = ctypes.c_int + lib.DecryptData.argtypes = [ + ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_void_p, + ] + lib.DecryptData.restype = ctypes.c_int + lib.NewSlice.restype = ctypes.c_void_p + lib.FreeSlice.argtypes = [ctypes.c_void_p] + lib.GetContentFromSlice.argtypes = [ctypes.c_void_p] + lib.GetContentFromSlice.restype = ctypes.c_void_p + lib.GetSliceLen.argtypes = [ctypes.c_void_p] + lib.GetSliceLen.restype = ctypes.c_int + lib.DestroySdk.argtypes = [ctypes.c_void_p] + + def _slice_bytes(self, slc: int) -> bytes: + ptr = self._lib.GetContentFromSlice(slc) + length = self._lib.GetSliceLen(slc) + return ctypes.string_at(ptr, length) + + def get_chat_data(self, seq: int, limit: int = 1000, timeout: int = 10) -> list[dict[str, Any]]: + """拉 seq 之后的消息(返回从 seq+1 起)。返回 chatdata 列表(每项含 seq/encrypt_random_key/encrypt_chat_msg)。""" + slc = self._lib.NewSlice() + try: + ret = self._lib.GetChatData(self._sdk, int(seq), int(limit), None, None, int(timeout), slc) + if ret != 0: + raise WxFinanceError(f"GetChatData 失败 ret={ret}") + data = json.loads(self._slice_bytes(slc).decode("utf-8")) + finally: + self._lib.FreeSlice(slc) + if data.get("errcode"): + raise WxFinanceError(f"GetChatData errcode={data.get('errcode')} {data.get('errmsg')}") + return data.get("chatdata", []) + + def decrypt(self, encrypt_random_key_b64: str, encrypt_chat_msg: str) -> dict[str, Any]: + """RSA 私钥解 encrypt_random_key → 得 AES 密钥 → DecryptData 解 encrypt_chat_msg → 明文 dict。""" + # 企业微信用你上传的 RSA 公钥(PKCS1)加密随机密钥, 这里用对应私钥解出, 原样交给 DecryptData。 + aes_key = self._priv.decrypt(base64.b64decode(encrypt_random_key_b64), padding.PKCS1v15()) + slc = self._lib.NewSlice() + try: + ret = self._lib.DecryptData(self._sdk, aes_key, encrypt_chat_msg.encode(), slc) + if ret != 0: + raise WxFinanceError(f"DecryptData 失败 ret={ret}") + return json.loads(self._slice_bytes(slc).decode("utf-8")) + finally: + self._lib.FreeSlice(slc) + + def close(self) -> None: + if getattr(self, "_sdk", None): + self._lib.DestroySdk(self._sdk) + self._sdk = None diff --git a/app/main.py b/app/main.py index 7dcd4fe..c8c3502 100644 --- a/app/main.py +++ b/app/main.py @@ -51,6 +51,10 @@ from app.core.heartbeat_monitor_worker import ( start_heartbeat_monitor, stop_heartbeat_monitor, ) +from app.core.wx_finance_worker import ( + start_wx_finance_worker, + stop_wx_finance_worker, +) from app.core.logging import setup_logging from app.core.pricebot_client import aclose_pricebot_client, get_pricebot_client from app.core.withdraw_reconcile_worker import ( @@ -82,9 +86,12 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: reconcile_task = start_withdraw_reconcile_worker() heartbeat_task = start_heartbeat_monitor() daily_exchange_task = start_daily_exchange_worker() + # 会话存档轮询 worker(独立线程;gate 在 wx_finance_configured, 未开启则 no-op、不加载 .so) + start_wx_finance_worker() try: yield finally: + stop_wx_finance_worker() await stop_heartbeat_monitor(heartbeat_task) await stop_withdraw_reconcile_worker(reconcile_task) await stop_daily_exchange_worker(daily_exchange_task) diff --git a/scripts/wx_finance_probe.py b/scripts/wx_finance_probe.py new file mode 100644 index 0000000..365d52e --- /dev/null +++ b/scripts/wx_finance_probe.py @@ -0,0 +1,45 @@ +"""会话存档 SDK 连通性探针 —— 上 worker 前,在【云服务器】单独验证整条链路是否打通。 + + 在 /opt/shaguabijia-app-server 下跑: .venv/bin/python scripts/wx_finance_probe.py + +依次验证:加载 .so → NewSdk/Init(corpid+存档Secret+可信IP)→ GetChatData(0)→ 试解第一条。 +只读、不写任何比价信号。任一步报错都会明确指出卡在哪(便于逐项排:.so 缺依赖 / Init 失败 / +可信IP 没放行 / 私钥不匹配)。全绿了再把 WX_FINANCE_ENABLED 置 true 上 worker。 +""" +from __future__ import annotations + +from pathlib import Path + +from app.core.config import settings +from app.integrations.wx_finance_sdk import WxFinanceSdk + + +def main() -> None: + print(f"[cfg] sdk_path = {settings.WX_FINANCE_SDK_PATH}") + print(f"[cfg] corpid = {settings.WX_KF_CORP_ID}") + print(f"[cfg] secret_set = {bool(settings.WX_FINANCE_SECRET)}") + print(f"[cfg] privkey = {settings.WX_FINANCE_PRIVATE_KEY_PATH}") + + sdk = WxFinanceSdk( + settings.WX_FINANCE_SDK_PATH, + settings.WX_KF_CORP_ID, + settings.WX_FINANCE_SECRET, + Path(settings.WX_FINANCE_PRIVATE_KEY_PATH).read_bytes(), + ) + print("[1] 加载 .so + Init 成功") + + msgs = sdk.get_chat_data(0, limit=100) + print(f"[2] GetChatData(0) 成功, 拉到 {len(msgs)} 条") + if msgs: + first = msgs[0] + print(f" 第一条 seq={first.get('seq')} publickey_ver={first.get('publickey_ver')}") + plain = sdk.decrypt(first["encrypt_random_key"], first["encrypt_chat_msg"]) + print(f"[3] 解密成功, msgtype={plain.get('msgtype')} from={plain.get('from')}") + else: + print("[3] 暂无历史消息(先让用户加接收成员好友、发条消息, 再跑一次)") + sdk.close() + print("探针完成:链路全通 ✓") + + +if __name__ == "__main__": + main()