shaguabijia-app-server/app/integrations/wx_kf_client.py

"""微信客服(企业微信)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>(消息拉取凭证, 非配置 Token;不传会有严格频控)。
    open_kfid:客服账号 id(回调里的 <OpenKfId>)。
    返回本次新拉到的 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