shaguabijia-app-server/app/core/wx_finance_worker.py

"""微信会话内容存档轮询 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)