shaguabijia-app-server/scripts/wx_finance_poller.py

"""会话存档【独立进程】轮询 poller —— 单独 systemd 服务跑, 与 app-server 隔离。

为什么独立进程:WeWorkFinanceSdk 是 Go 编译的 c-shared 库, 自带 Go runtime。嵌进 app-server(uvicorn)
后台线程里跑会 segfault、把主进程带崩(2026-07-17 实测 core-dump 每 5s 循环)。独立进程让 Go runtime
独占进程 + 主线程, 崩了 systemd 只重启本 poller、不动 app-server。拉到美团卡片(weapp)/截图(image)
→ POST app-server 内部端点 /internal/wx-finance/pending 打比价信号(下游心跳→弹窗→比价 完全复用)。

  手动跑:  cd /opt/shaguabijia-app-server && .venv/bin/python scripts/wx_finance_poller.py
  生产:    deploy/wx-finance-poller.service 常驻(systemctl enable --now wx-finance-poller)
"""
from __future__ import annotations

import json
import logging
import signal
import time
from pathlib import Path

import httpx

from app.core.config import settings
from app.core.logging import setup_logging
from app.integrations.wx_finance_sdk import WxFinanceError, WxFinanceSdk

logger = logging.getLogger("shagua.wx_finance_poller")

_stop = False


def _on_signal(_signum: int, _frame: object) -> None:
    global _stop
    _stop = True


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("seq 持久化失败")


def _notify(client: httpx.Client, source: str, kind: str) -> None:
    """POST 内部端点让 app-server 打信号(set_pending 在 app-server 进程内存)。"""
    url = f"{settings.WX_FINANCE_INTERNAL_URL.rstrip('/')}/internal/wx-finance/pending"
    try:
        resp = client.post(
            url,
            json={"source": source, "kind": kind},
            headers={"X-Internal-Secret": settings.INTERNAL_API_SECRET},
            timeout=10,
        )
        logger.info("notify app-server (%s) → %s", kind, resp.status_code)
    except Exception:
        logger.exception("notify app-server 失败")


def _resolve_source(msg: dict) -> str:
    """从 weapp 小程序卡判源平台代号(= app-server _SOURCE_PACKAGES 的 key)。

    会话存档 weapp 结构里 username(gh_id)/appid/pagepath/title 任一含平台特征即判定:
    先按关键词启发式(覆盖大多数);认不出 → 默认 meituan(过渡期不阻断)+ WARNING,
    并把整个 weapp 结构打进日志,便于照真实 username/appid 精确补规则。
    """
    weapp = msg.get("weapp") or {}
    blob = json.dumps(weapp, ensure_ascii=False).lower()
    # 排障日志:打全 weapp 结构(顺带 msg 顶层 key,防结构不在 weapp 下),便于补/改判定规则
    logger.info("weapp 判源: msg_keys=%s weapp=%s", list(msg.keys()), blob[:1500])
    if any(k in blob for k in ("jingdong", "jd.com", "jddj", "京东")):
        return "jd"
    if any(k in blob for k in ("meituan", "sankuai", "美团")):
        return "meituan"
    logger.warning("weapp 源平台未识别, 暂默认 meituan;见上一行结构日志补 _resolve_source 规则")
    return "meituan"


def _handle(client: httpx.Client, msg: dict) -> None:
    """明文消息:外部用户(from≠接收成员)发来的 weapp/image → 通知 app-server 打信号。"""
    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
    # weapp 卡片按小程序判源平台;image(截图)判不出平台,沿用 meituan。
    source = _resolve_source(msg) if msgtype == "weapp" else "meituan"
    logger.info("命中触发 from=%s type=%s source=%s", frm, msgtype, source)
    _notify(client, source, msgtype)


def main() -> None:
    setup_logging(debug=settings.APP_DEBUG)
    if not settings.wx_finance_configured:
        logger.warning("WX_FINANCE 未配齐(ENABLED/corpid/存档Secret), poller 退出")
        return
    if not settings.INTERNAL_API_SECRET:
        logger.error("INTERNAL_API_SECRET 未配置, poller 无法通知 app-server, 退出")
        return
    signal.signal(signal.SIGTERM, _on_signal)
    signal.signal(signal.SIGINT, _on_signal)

    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(),
    )
    seq = _load_seq()
    logger.info(
        "wx_finance poller 启动, 从 seq=%d 轮询, 间隔 %ss", seq, settings.WX_FINANCE_POLL_INTERVAL_SEC
    )
    with httpx.Client() as client:
        while not _stop:
            try:
                msgs = sdk.get_chat_data(seq, limit=1000)
                if msgs:
                    logger.info("拉到 %d 条(seq>%d)", len(msgs), seq)
                for item in msgs:
                    try:
                        plain = sdk.decrypt(item["encrypt_random_key"], item["encrypt_chat_msg"])
                        _handle(client, plain)
                    except Exception:
                        logger.exception("解密/处理单条失败 seq=%s", item.get("seq"))
                    seq = max(seq, int(item.get("seq", seq)))
                if msgs:
                    _save_seq(seq)
            except WxFinanceError:
                logger.exception("GetChatData 出错")
            except Exception:
                logger.exception("轮询异常")
            # 可打断的 sleep:按 ≤0.5s 分片, 支持亚秒间隔(如 0.5s), 收到 SIGTERM 尽快退
            interval = max(0.1, settings.WX_FINANCE_POLL_INTERVAL_SEC)
            step = min(0.5, interval)
            slept = 0.0
            while slept < interval and not _stop:
                time.sleep(step)
                slept += step
    sdk.close()
    logger.info("wx_finance poller 已停止")


if __name__ == "__main__":
    main()