import json import logging import base64 from datetime import datetime, timezone from typing import Optional import httpx from fastapi import FastAPI, Request, HTTPException from fastapi.responses import PlainTextResponse, HTMLResponse from config import APP_ID, TOKEN, ENCODING_AES_KEY from crypto_utils import aes_decrypt_base64, aes_encrypt_base64, calc_signature logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger(__name__) app = FastAPI(title="WeChat Dialog ThirdAPI Demo") async def handle_business(data: dict) -> str: """ 业务处理钩子： - data 为解密后的请求 JSON（XwBrainThirdApiReqInfo 扩展结构） - 返回值为要展示给用户的短文本答案 后续如果要接数据库、HTTP 接口，只需要改这里的逻辑即可。 """ query = data.get("Query", "") skill_name = data.get("SkillName", "") intent_name = data.get("IntentName", "") user_id = data.get("UserId", "") # 如果想返回普通文本，可以使用下面的示例文案： # parts = [ # f"技能：{skill_name or '未提供'}", # f"意图：{intent_name or '未提供'}", # f"用户ID：{user_id or '未提供'}", # f"用户问题：{query or '空'}", # ] # return "；".join(parts) # 当前示例：返回一个 H5 富文本结构（平台会对 short_answer 做 JSON.parse） h5_msg = { "news": { "articles": [ { "title": "实时更新：新型肺炎疫情最新动态", "description": "腾讯新闻第一时间同步全国新型肺炎疫情动态，欢迎关注、转发", "url": "https://news.qq.com/zt2020/page/feiyan.htm", "picurl": "http://mmbiz.qpic.cn/mmbiz_jpg/W3gQtpV3j8D8kZRqfpTJlfVqubwgFQf47H0GWlGV6leaDF80ZpdtuFhQVsCsM3YKmwkujXzdjR2k6aWfA41ic7Q/0?wx_fmt=jpeg", "type": "h5", } ] } } # 按文档要求，answer/short_answer 是字符串，内部再 JSON.parse 才是对象 return json.dumps(h5_msg, ensure_ascii=False) def _decode_jwt_payload_without_verify(token: str) -> dict: """ 仅用于本地调试分析 JWT 的 payload（不做签名校验）。 """ try: parts = token.split(".") if len(parts) != 3: return {} payload = parts[1] payload += "=" * (-len(payload) % 4) # base64url 补齐 data = base64.urlsafe_b64decode(payload.encode("utf-8")).decode("utf-8") obj = json.loads(data) return obj if isinstance(obj, dict) else {} except Exception: return {} async def query_private_messages( authtoken: str, wxbot_bid: str, page: int = 0, size: int = 30, filter_value: int = 0, request_id: str = "local-debug", ) -> dict: """ 转发请求到微信对话开放平台的私信列表接口（你在 Charles 抓包的接口）。 """ url = "https://chatbot.weixin.qq.com/miniopenai/manualservice/getaccessstautuslist" body = { "bid": wxbot_bid, "filter": filter_value, "order": {"page": page, "size": size}, "base": {"requestid": request_id}, } headers = { "authtoken": authtoken, "wxbot_bid": wxbot_bid, "content-type": "application/json", "charset": "utf-8", } # 调试场景下常见抓包证书/公司代理证书导致校验失败，这里关闭 verify 以保证接口可用。 # 生产环境建议改为 verify=True 并配置受信任证书链。 async with httpx.AsyncClient(timeout=10, verify=False) as client: resp = await client.post(url, json=body, headers=headers) resp.raise_for_status() return resp.json() async def _handle_wechat_request(request: Request, app_id: Optional[str]) -> str: """ 公共处理逻辑：便于同时支持 "/" 和 "/wechat/thirdapi" 两种回调路径。 """ # 1. app_id 校验 # 说明：有些平台配置下不会显式在 URL 带 app_id，这种情况下我们只做日志，不强制拦截，避免 400。 if app_id is None: logger.warning("请求中未携带 app_id（将继续处理，仅做告警）") elif app_id != APP_ID: logger.warning(f"app_id 不匹配，期望={APP_ID}, 实际={app_id}（将继续处理，仅做告警）") # 2. 获取原始 body（加密后的 base64 字符串） raw_body_bytes = await request.body() cipher_b64 = raw_body_bytes.decode("utf-8").strip() if not cipher_b64: raise HTTPException(status_code=400, detail="empty body") # 打印原始 body，便于对照平台文档排查（是否真的是 base64 + AES 加密数据） logger.info(f"收到原始 body（未解密，前 500 字符）：{cipher_b64[:500]}") # 平台当前实际发送的是 {"encrypted":"..."} 这一层 JSON，需要先解析再取字段 if cipher_b64.startswith("{"): try: outer = json.loads(cipher_b64) if isinstance(outer, dict) and "encrypted" in outer: cipher_b64 = str(outer["encrypted"]).replace("\n", "").strip() except Exception: # 如果解析失败，就按原始字符串继续处理（保持兼容） logger.warning("外层 JSON 解析失败，按原始 body 作为密文处理") # 3. 解密 try: decrypted_json_str = aes_decrypt_base64(cipher_b64, ENCODING_AES_KEY) except Exception: # 常见情况：非第三方服务接口的探测/噪音流量，或使用了不同的 AESKey。 logger.error( "AES 解密失败，可能为非本机器人第三方服务请求或密钥不匹配（忽略该请求）。" ) raise HTTPException(status_code=400, detail="decrypt error") # 4. 解析 JSON try: data = json.loads(decrypted_json_str) except json.JSONDecodeError: logger.error(f"解密结果不是合法 JSON: {decrypted_json_str}") raise HTTPException(status_code=400, detail="invalid json in decrypted body") # 5. 打印请求内容到终端日志（分析、显示，带中文注释） logger.info("收到微信第三方服务请求(已解密)：") logger.info( "请求概览：RequestId=%s，用户ID=%s，问题=%s，技能=%s，意图=%s", data.get("RequestId", ""), data.get("UserId", ""), data.get("Query", ""), data.get("SkillName", ""), data.get("IntentName", ""), ) logger.info("完整请求JSON：%s", json.dumps(data, ensure_ascii=False, indent=2)) # 6. （可选）签名校验 try: timestamp = data.get("Timestamp") skill_name = data.get("SkillName", "") intent_name = data.get("IntentName", "") query = data.get("Query", "") provided_sig = data.get("Signature", "") if timestamp is not None and provided_sig: expected_sig = calc_signature(TOKEN, timestamp, skill_name, intent_name, query) if expected_sig != provided_sig: logger.warning( f"签名校验失败，expected={expected_sig}, provided={provided_sig}" ) # 这里直接拒绝请求，你也可以根据需要改成仅告警 raise HTTPException(status_code=400, detail="invalid signature") except HTTPException: raise except Exception: logger.exception("签名校验异常（忽略或按需处理）") # 7. 根据业务逻辑构造应答 try: answer_text = await handle_business(data) except Exception: logger.exception("业务处理异常，将返回兜底文案") answer_text = "系统繁忙，请稍后再试。" resp_obj = { # 文本类型，内容为可被 JSON.parse 的 H5 结构字符串 "answer_type": "text", "text_info": { "short_answer": answer_text } } resp_json_str = json.dumps(resp_obj, ensure_ascii=False) # 8. 加密响应 JSON try: encrypted_resp = aes_encrypt_base64(resp_json_str, ENCODING_AES_KEY) except Exception as e: logger.exception("响应 AES 加密失败") raise HTTPException(status_code=500, detail="encrypt error") from e # 按文档要求：如果配置了加密，这里返回的是加密后的 base64 字符串 return encrypted_resp @app.post("/", response_class=PlainTextResponse) async def wechat_root(request: Request, app_id: Optional[str] = None): """ 有些配置示例直接写根路径，例如： url: http://example.com/ 这种情况下，微信会 POST 到 "/"，所以这里也做同样处理。 """ return await _handle_wechat_request(request, app_id) @app.post("/wechat/thirdapi", response_class=PlainTextResponse) async def wechat_thirdapi(request: Request, app_id: Optional[str] = None): """ 显式的 /wechat/thirdapi 路径，同样处理。 """ return await _handle_wechat_request(request, app_id) @app.get("/health") def health_check(): return {"status": "ok"} @app.get("/debug/echo") async def debug_echo(request: Request, app_id: Optional[str] = None): """ 调试用接口：回显请求头和 app_id，方便排查签名和路由问题。 """ return { "app_id": app_id, "headers": dict(request.headers), } @app.get("/private-message", response_class=HTMLResponse) def private_message_page(): """ 私信查询页面（本地调试）。 """ return """ 私信查询

私信查询调试台

用于调试微信对话开放平台私信列表接口，支持一键查询、Token 过期分析与结果可视化。

提示：查询失败时，先检查 authtoken 是否过期、wxbot_bid 是否与当前机器人一致。

等待操作...

{
  "message": "点击上方按钮开始查询"
}
""" @app.post("/api/private-message/query") async def api_private_message_query(payload: dict): """ 查询私信列表接口（调试版）。 """ authtoken = str(payload.get("authtoken", "")).strip() wxbot_bid = str(payload.get("wxbot_bid", "")).strip() page = int(payload.get("page", 0)) size = int(payload.get("size", 30)) filter_value = int(payload.get("filter", 0)) request_id = str(payload.get("requestid", "local-debug")).strip() or "local-debug" if not authtoken or not wxbot_bid: raise HTTPException(status_code=400, detail="authtoken 和 wxbot_bid 不能为空") try: result = await query_private_messages( authtoken=authtoken, wxbot_bid=wxbot_bid, page=page, size=size, filter_value=filter_value, request_id=request_id, ) except Exception as e: logger.error("私信查询失败：%s", str(e)) raise HTTPException(status_code=502, detail=f"private-message query failed: {e}") contacts = result.get("contacts") or [] return { "ok": True, "count": result.get("count"), "contacts_preview": [ { "nick": c.get("nick"), "userId": c.get("userId"), "lastMsg": (c.get("lastMsg") or {}).get("content"), "lastActiveTime": c.get("lastActiveTime"), } for c in contacts ], "raw": result, } @app.post("/api/private-message/analyze") async def api_private_message_analyze(payload: dict): """ 分析 authtoken 的过期时间和基本信息（基于 JWT payload，不校验签名）。 """ authtoken = str(payload.get("authtoken", "")).strip() if not authtoken: raise HTTPException(status_code=400, detail="authtoken 不能为空") claims = _decode_jwt_payload_without_verify(authtoken) now_ts = int(datetime.now(timezone.utc).timestamp()) exp = int(claims.get("exp", 0)) if str(claims.get("exp", "")).isdigit() else 0 iat = int(claims.get("iat", 0)) if str(claims.get("iat", "")).isdigit() else 0 sign_time_ms = claims.get("signetime") expires_in_seconds = exp - now_ts if exp else None if expires_in_seconds is None: status = "无法解析过期时间" elif expires_in_seconds <= 0: status = "已过期" elif expires_in_seconds < 3600: status = "即将过期（1小时内）" else: status = "有效" report = { "status": status, "now_utc": datetime.now(timezone.utc).isoformat(), "claims": claims, "analysis": { "iat_utc": datetime.fromtimestamp(iat, tz=timezone.utc).isoformat() if iat else None, "exp_utc": datetime.fromtimestamp(exp, tz=timezone.utc).isoformat() if exp else None, "expires_in_seconds": expires_in_seconds, "expires_in_hours": round(expires_in_seconds / 3600, 2) if isinstance(expires_in_seconds, int) else None, "signetime_ms": sign_time_ms, "summary": "该分析基于 JWT payload 解码，不包含签名有效性校验。", }, } return report