【💰】有人用过微信小程序的文本内容安全识别（wxa/msg_sec_check）接口吗

client = WeChatMiniProgramSecCheck(appid=appid, secret=secret)
openid = get_user_openid(user)

if require_openid and not openid:
    raise ValidationError({"message": "未获取到 openid，无法完成内容审核", "sec_check": {"suggest": "no_openid"}})

parts_reports: List[dict] = []

for part in _split_text(text, max_len=2000):
    res = client.msg_sec_check(content=part, openid=openid, scene=int(scene), version=2)
    report = _normalize_wx_resp(res.raw)
    report.update({"suggest": res.suggest, "label": res.label})
    parts_reports.append(report)

    logger.info("wx sec_check: uid=%s scene=%s suggest=%s label=%s trace=%s",
                getattr(user, "id", None), scene, res.suggest, res.label, report.get("trace_id"))

    if res.ok:
        continue

    # review/risky：明确不通过
    if res.suggest in ("review", "risky"):
        raise ValidationError({"message": "内容可能包含敏感信息，请修改后再提交", "sec_check": report})

    # 其它异常：strict=True 直接拒绝（并把微信返回带出去）
    if strict:
        raise ValidationError({"message": "内容审核服务异常，请稍后重试", "sec_check": report})

    # strict=False：按你原来策略（这里默认还是拒绝更安全）
    raise ValidationError({"message": "内容审核失败，请稍后重试", "sec_check": report})

# 多段合并返回：给前端 parts，必要时也能取最后一次
return {
    "scene": int(scene),
    "openid": (openid or "")[:8] + "..." if openid else "",
    "parts": parts_reports,
    "result": parts_reports[-1].get("result") if parts_reports else {},
    "detail": sum((p.get("detail", []) for p in parts_reports), []),
}

TOKEN_INVALID_CODES = {40001, 40014, 42001}

def __init__(self, appid: str, secret: str):
    self.appid = appid
    self.secret = secret

def _cache_key(self) -> str:
    return f"wx:mp:access_token:{self.appid}"

def _get_access_token(self) -> str:
    ck = self._cache_key()
    cached = cache.get(ck)
    if cached:
        return cached

    resp = requests.get(
        self.TOKEN_URL,
        params={
            "grant_type": "client_credential",
            "appid": self.appid,
            "secret": self.secret,
        },
        timeout=6,
    )
    data = resp.json()
    if "access_token" not in data:
        raise RuntimeError(f"get_access_token failed: {data}")

    token = data["access_token"]
    expires_in = int(data.get("expires_in", 7200))
    cache.set(ck, token, timeout=max(expires_in - 300, 60))
    return token

def _invalidate_access_token(self) -> None:
    cache.delete(self._cache_key())

def msg_sec_check(self, *, content: str, openid: Optional[str] = None, scene: int = 2,
                  version: int = 2) -> WxSecCheckResult:
    content = (content or "").strip()
    if not content:
        return WxSecCheckResult(ok=False, suggest="empty")

    if not getattr(settings, "WECHAT_SEC_CHECK_ENABLED", True):
        return WxSecCheckResult(ok=True, suggest="pass (disabled)")

    if not openid:
        return WxSecCheckResult(ok=False, suggest="no_openid", raw={"errmsg": "missing openid"})

    def _do_call(token: str) -> dict:
        payload: dict[str, Any] = {
            "content": content,
            "openid": openid,
            "scene": int(scene),
            "version": int(version),
        }
        r = requests.post(self.MSG_SEC_CHECK_URL, params={"access_token": token}, json=payload, timeout=6)
        return r.json()

    token = self._get_access_token()
    data = _do_call(token)

    if int(data.get("errcode") or 0) in self.TOKEN_INVALID_CODES:
        self._invalidate_access_token()
        token = self._get_access_token()
        data = _do_call(token)

    errcode = int(data.get("errcode") or 0)
    if errcode != 0:
        logger.warning("wx msg_sec_check error: %s", data)
        return WxSecCheckResult(ok=False, suggest="api_error", raw=data)

    if "result" not in data:
        logger.warning("wx msg_sec_check bad response (no result): %s", data)
        return WxSecCheckResult(ok=False, suggest="bad_response", raw=data)

    result = data.get("result") or {}
    suggest = result.get("suggest") or "pass"
    label = result.get("label")
    return WxSecCheckResult(ok=(suggest == "pass"), suggest=suggest, label=label, raw=data)

info = getattr(user, "info", None)
if info:
    for key in ("openid", "wx_openid", "wechat_openid"):
        v = getattr(info, key, None)
        if v:
            return v
return None

buf = []
cur = ""
for ch in s:
    if len(cur) >= max_len:
        buf.append(cur)
        cur = ""
    cur += ch
if cur:
    buf.append(cur)
return buf

client = WeChatMiniProgramSecCheck(appid=appid, secret=secret)
openid = get_user_openid(user)

if require_openid and not openid:
    raise ValidationError("未获取到 openid，无法完成内容审核，请退出重进小程序或重新登录后再试")

for part in _split_text(text, max_len=2000):
    res = client.msg_sec_check(
        content=part,
        openid=openid,
        scene=int(scene),
        version=2,
    )

    if res.ok:
        continue

    logger.warning("wx sec_check block: suggest=%s label=%s raw=%s", res.suggest, res.label, res.raw)

    if res.suggest in ("review", "risky"):
        raise ValidationError("内容可能包含敏感信息，请修改后再提交")
    if res.suggest == "empty":
        raise ValidationError("内容不能为空")

    # api_error：严格模式一律拒绝；非严格按 fail-open 配置决定
    if res.suggest == "api_error":
        if strict:
            raise ValidationError("内容审核服务异常，请稍后重试")
        if getattr(settings, "WECHAT_SEC_CHECK_FAIL_OPEN", False):
            continue
        raise ValidationError("内容审核失败，请稍后重试")

    raise ValidationError("内容审核失败，请稍后重试")