d695f87e33
将A回调链路调整为先返回关键词回复，再以后台任务触发B接口同步，避免B接口耗时或失败影响用户主回复；同时修正微信消息体加解密格式并固定B接口凭据来源。 Made-with: Cursor
1061 lines
36 KiB
Python
1061 lines
36 KiB
Python
import json
|
||
import logging
|
||
import base64
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from typing import Optional
|
||
|
||
import httpx
|
||
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
|
||
from fastapi.responses import PlainTextResponse, HTMLResponse
|
||
|
||
from config import (
|
||
APP_ID,
|
||
TOKEN,
|
||
ENCODING_AES_KEY,
|
||
PB_URL,
|
||
PB_ADMIN_EMAIL,
|
||
PB_ADMIN_PASSWORD,
|
||
PB_COLLECTION,
|
||
PM_SYNC_AUTHTOKEN,
|
||
PM_SYNC_DEFAULT_WXBOT_BID,
|
||
)
|
||
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")
|
||
|
||
_processed_callback_cache: dict[str, int] = {}
|
||
|
||
|
||
def _build_keyword_h5_reply(query: str, user_id: str) -> Optional[str]:
|
||
"""
|
||
按关键词返回 H5 卡片（answer/short_answer 需为 JSON 字符串）。
|
||
"""
|
||
q = (query or "").strip()
|
||
if not q:
|
||
return None
|
||
|
||
if "电子书" in q:
|
||
msg = {
|
||
"news": {
|
||
"articles": [
|
||
{
|
||
"title": "数字游民",
|
||
"description": "地理套利与自动化杠杆",
|
||
"url": f"https://www.hackrobot.cn?userid={user_id}",
|
||
"picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png",
|
||
"type": "h5",
|
||
}
|
||
]
|
||
}
|
||
}
|
||
return json.dumps(msg, ensure_ascii=False)
|
||
|
||
if "群" in q:
|
||
msg = {
|
||
"news": {
|
||
"articles": [
|
||
{
|
||
"title": "异度星球",
|
||
"description": "电子书、视频教程、私密社群",
|
||
"url": f"https://vip.hackrobot.cn?userid={user_id}",
|
||
"picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png",
|
||
"type": "h5",
|
||
}
|
||
]
|
||
}
|
||
}
|
||
return json.dumps(msg, ensure_ascii=False)
|
||
|
||
return None
|
||
|
||
|
||
def _safe_filter_value(value: str) -> str:
|
||
return (value or "").replace("\\", "\\\\").replace('"', '\\"')
|
||
|
||
|
||
def _to_int(value) -> Optional[int]:
|
||
try:
|
||
return int(str(value).strip())
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _normalize_text(value) -> str:
|
||
return str(value or "").strip()
|
||
|
||
|
||
def _build_callback_dedupe_key(data: dict) -> str:
|
||
userid = _normalize_text(data.get("UserId") or data.get("userId"))
|
||
msg_id = _normalize_text(data.get("MsgId") or data.get("MessageId") or data.get("msgId"))
|
||
ts = _normalize_text(data.get("Timestamp") or data.get("MessageTime") or data.get("msgTime"))
|
||
query = _normalize_text(data.get("Query"))
|
||
return "|".join([userid, msg_id, ts, query])
|
||
|
||
|
||
def _is_duplicate_callback(data: dict, ttl_seconds: int = 3600) -> bool:
|
||
"""
|
||
回调幂等（进程内）：同 userid/msg_id/time/query 在 TTL 内仅处理一次同步逻辑。
|
||
不影响主回复。
|
||
"""
|
||
now_ts = int(time.time())
|
||
# 清理过期 key，避免缓存无限增长
|
||
expired = [k for k, exp in _processed_callback_cache.items() if exp <= now_ts]
|
||
for k in expired:
|
||
_processed_callback_cache.pop(k, None)
|
||
|
||
key = _build_callback_dedupe_key(data)
|
||
if not key.strip("|"):
|
||
return False
|
||
if key in _processed_callback_cache:
|
||
return True
|
||
_processed_callback_cache[key] = now_ts + ttl_seconds
|
||
return False
|
||
|
||
|
||
def _extract_b_last_msg(contact: dict) -> dict:
|
||
return (contact.get("lastMsg") or {}) if isinstance(contact, dict) else {}
|
||
|
||
|
||
def _pick_best_contact_match(contacts: list, target_userid: str, source_query: str, source_msg_id: str, source_msg_time) -> Optional[dict]:
|
||
"""
|
||
在同 userid 的候选中做二次匹配：
|
||
- msg_id 一致优先
|
||
- 文本一致次优
|
||
- 时间最接近再次优先
|
||
"""
|
||
if not target_userid:
|
||
return None
|
||
|
||
same_user = [
|
||
c for c in contacts
|
||
if _normalize_text(c.get("userId") or c.get("userid")) == target_userid
|
||
]
|
||
if not same_user:
|
||
return None
|
||
|
||
query_norm = _normalize_text(source_query)
|
||
source_msg_id_norm = _normalize_text(source_msg_id)
|
||
src_ts = _to_int(source_msg_time)
|
||
|
||
def score(contact: dict) -> tuple[int, int, int]:
|
||
last_msg = _extract_b_last_msg(contact)
|
||
last_msg_id = _normalize_text(last_msg.get("msgId") or last_msg.get("id"))
|
||
last_content = _normalize_text(last_msg.get("content"))
|
||
last_active = _to_int(contact.get("lastActiveTime"))
|
||
|
||
msg_id_hit = 1 if source_msg_id_norm and last_msg_id and source_msg_id_norm == last_msg_id else 0
|
||
content_hit = 1 if query_norm and last_content and query_norm == last_content else 0
|
||
# 时间越接近分越高（负值用于降序）
|
||
if src_ts is None or last_active is None:
|
||
time_delta = 10**12
|
||
else:
|
||
time_delta = abs(src_ts - last_active)
|
||
return (msg_id_hit, content_hit, -time_delta)
|
||
|
||
return sorted(same_user, key=score, reverse=True)[0]
|
||
|
||
|
||
async def _pb_upsert_user_profile(record: dict) -> None:
|
||
"""
|
||
使用 PocketBase Admin API upsert 用户资料。
|
||
需要环境变量：
|
||
- PB_URL
|
||
- PB_ADMIN_EMAIL
|
||
- PB_ADMIN_PASSWORD
|
||
可选：
|
||
- PB_COLLECTION（默认 wechat_private_users）
|
||
"""
|
||
pb_url = (PB_URL or "").rstrip("/")
|
||
pb_admin_email = PB_ADMIN_EMAIL or ""
|
||
pb_admin_password = PB_ADMIN_PASSWORD or ""
|
||
pb_collection = PB_COLLECTION or "wechat_private_users"
|
||
if not pb_url or not pb_admin_email or not pb_admin_password:
|
||
logger.warning("PocketBase 配置不完整，跳过资料同步")
|
||
return
|
||
|
||
userid = str(record.get("userid", "")).strip()
|
||
channel_id = str(record.get("channel_id", "")).strip()
|
||
if not userid:
|
||
return
|
||
|
||
async with httpx.AsyncClient(timeout=10, verify=False) as client:
|
||
auth_resp = await client.post(
|
||
f"{pb_url}/api/admins/auth-with-password",
|
||
json={"identity": pb_admin_email, "password": pb_admin_password},
|
||
)
|
||
auth_resp.raise_for_status()
|
||
token = (auth_resp.json() or {}).get("token", "")
|
||
if not token:
|
||
logger.warning("PocketBase 登录无 token，跳过资料同步")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
user_expr = _safe_filter_value(userid)
|
||
chan_expr = _safe_filter_value(channel_id)
|
||
filter_exp = f'userid="{user_expr}" && channel_id="{chan_expr}"' if channel_id else f'userid="{user_expr}"'
|
||
list_resp = await client.get(
|
||
f"{pb_url}/api/collections/{pb_collection}/records",
|
||
params={"page": 1, "perPage": 1, "filter": filter_exp, "sort": "-updated"},
|
||
headers=headers,
|
||
)
|
||
list_resp.raise_for_status()
|
||
items = ((list_resp.json() or {}).get("items")) or []
|
||
|
||
if items:
|
||
rid = items[0].get("id")
|
||
if rid:
|
||
patch_resp = await client.patch(
|
||
f"{pb_url}/api/collections/{pb_collection}/records/{rid}",
|
||
json=record,
|
||
headers=headers,
|
||
)
|
||
patch_resp.raise_for_status()
|
||
return
|
||
|
||
create_resp = await client.post(
|
||
f"{pb_url}/api/collections/{pb_collection}/records",
|
||
json=record,
|
||
headers=headers,
|
||
)
|
||
create_resp.raise_for_status()
|
||
|
||
|
||
async def _sync_user_profile_from_b(data: dict) -> None:
|
||
"""
|
||
A 接口收到消息后，立即调用 B 接口拉取会话并同步用户资料到 PocketBase。
|
||
环境变量：
|
||
- PM_SYNC_AUTHTOKEN（必填）
|
||
- PM_SYNC_DEFAULT_WXBOT_BID（可选）
|
||
- PM_SYNC_BID_BY_CHANNEL_JSON（可选，形如 {"wx5cee...":"4888..."}）
|
||
"""
|
||
authtoken = (PM_SYNC_AUTHTOKEN or "").strip()
|
||
if not authtoken:
|
||
logger.info("未配置 PM_SYNC_AUTHTOKEN，跳过 B 接口同步")
|
||
return
|
||
|
||
channel_id = str(data.get("ChannelId", "") or data.get("channelId", "")).strip()
|
||
# B 接口配置与 A 回调路由解耦：固定使用 B 的默认凭据。
|
||
wxbot_bid = (PM_SYNC_DEFAULT_WXBOT_BID or "").strip()
|
||
if not wxbot_bid:
|
||
logger.warning("未配置可用 wxbot_bid，跳过 B 接口同步")
|
||
return
|
||
|
||
target_userid = str(data.get("UserId", "") or data.get("userId", "")).strip()
|
||
if not target_userid:
|
||
return
|
||
|
||
if _is_duplicate_callback(data):
|
||
logger.info("命中回调幂等缓存，跳过重复同步：userid=%s", target_userid)
|
||
return
|
||
|
||
msg_id = str(data.get("MsgId", "") or data.get("MessageId", "") or data.get("msgId", "")).strip()
|
||
msg_time = data.get("Timestamp", "") or data.get("MessageTime", "") or data.get("msgTime", "")
|
||
query = str(data.get("Query", "")).strip()
|
||
|
||
b_resp = await query_private_messages(
|
||
authtoken=authtoken,
|
||
wxbot_bid=wxbot_bid,
|
||
page=0,
|
||
size=100,
|
||
filter_value=0,
|
||
request_id=f"thirdapi-{target_userid}",
|
||
)
|
||
contacts = b_resp.get("contacts") or []
|
||
target = _pick_best_contact_match(
|
||
contacts=contacts,
|
||
target_userid=target_userid,
|
||
source_query=query,
|
||
source_msg_id=msg_id,
|
||
source_msg_time=msg_time,
|
||
)
|
||
if not target:
|
||
logger.info("B 接口未匹配到用户：%s", target_userid)
|
||
return
|
||
|
||
last_msg = target.get("lastMsg") or {}
|
||
record = {
|
||
"userid": target_userid,
|
||
"channel_id": channel_id,
|
||
"nick": target.get("nick", ""),
|
||
"avatar": target.get("headImg") or target.get("avatar") or target.get("headimgurl") or target.get("headImgUrl") or "",
|
||
"source_query": query,
|
||
"source_message_id": msg_id,
|
||
"source_message_time": str(msg_time),
|
||
"matched_last_msg": last_msg.get("content", ""),
|
||
"matched_last_msg_id": str(last_msg.get("msgId", "") or last_msg.get("id", "")),
|
||
"matched_last_msg_time": str(target.get("lastActiveTime", "")),
|
||
"raw_contact": json.dumps(target, ensure_ascii=False),
|
||
}
|
||
await _pb_upsert_user_profile(record)
|
||
|
||
|
||
@app.post("/debug/sync-test")
|
||
async def debug_sync_test(payload: dict):
|
||
"""
|
||
手动调试 A->B->PocketBase 同步链路。
|
||
body 示例：
|
||
{
|
||
"UserId": "...",
|
||
"ChannelId": "...",
|
||
"Query": "电子书",
|
||
"MsgId": "...",
|
||
"Timestamp": 1774933444
|
||
}
|
||
"""
|
||
await _sync_user_profile_from_b(payload or {})
|
||
return {
|
||
"ok": True,
|
||
"message": "sync triggered",
|
||
"dedupe_key": _build_callback_dedupe_key(payload or {}),
|
||
}
|
||
|
||
|
||
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)
|
||
|
||
keyword_reply = _build_keyword_h5_reply(query, user_id)
|
||
if keyword_reply:
|
||
return keyword_reply
|
||
return "已收到你的消息，输入“电子书”或“群”可获取对应卡片。"
|
||
|
||
|
||
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], background_tasks: BackgroundTasks) -> 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 "encrypted" in cipher_b64:
|
||
try:
|
||
outer = json.loads(cipher_b64)
|
||
if isinstance(outer, dict) and "encrypted" in outer:
|
||
cipher_b64 = "".join(str(outer["encrypted"]).split()).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 = "系统繁忙，请稍后再试。"
|
||
|
||
# 7.1 主回复确定后再后台触发 B 同步，避免影响回复链路耗时
|
||
background_tasks.add_task(_sync_user_profile_from_b, data)
|
||
|
||
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, APP_ID)
|
||
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, background_tasks: BackgroundTasks, app_id: Optional[str] = None):
|
||
"""
|
||
有些配置示例直接写根路径，例如：
|
||
url: http://example.com/
|
||
这种情况下，微信会 POST 到 "/"，所以这里也做同样处理。
|
||
"""
|
||
return await _handle_wechat_request(request, app_id, background_tasks)
|
||
|
||
|
||
@app.post("/wechat/thirdapi", response_class=PlainTextResponse)
|
||
async def wechat_thirdapi(request: Request, background_tasks: BackgroundTasks, app_id: Optional[str] = None):
|
||
"""
|
||
显式的 /wechat/thirdapi 路径，同样处理。
|
||
"""
|
||
return await _handle_wechat_request(request, app_id, background_tasks)
|
||
|
||
|
||
@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 """
|
||
<!doctype html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="utf-8" />
|
||
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||
<title>私信查询</title>
|
||
<style>
|
||
:root {
|
||
--bg: #0b1020;
|
||
--card: rgba(17, 24, 39, 0.78);
|
||
--card-border: rgba(255, 255, 255, 0.1);
|
||
--text: #e5e7eb;
|
||
--muted: #9ca3af;
|
||
--primary: #60a5fa;
|
||
--primary-2: #3b82f6;
|
||
--ok: #34d399;
|
||
--warn: #f59e0b;
|
||
}
|
||
* { box-sizing: border-box; }
|
||
body {
|
||
margin: 0;
|
||
color: var(--text);
|
||
font-family: "PingFang SC", "Microsoft YaHei", system-ui, -apple-system, Segoe UI, Roboto, sans-serif;
|
||
background:
|
||
radial-gradient(1200px 800px at 0% 0%, #1e3a8a 0%, rgba(30,58,138,0) 40%),
|
||
radial-gradient(1000px 600px at 100% 20%, #4338ca 0%, rgba(67,56,202,0) 35%),
|
||
var(--bg);
|
||
min-height: 100vh;
|
||
padding: 28px 14px 40px;
|
||
}
|
||
.container {
|
||
max-width: 1060px;
|
||
margin: 0 auto;
|
||
display: grid;
|
||
gap: 14px;
|
||
}
|
||
.card {
|
||
background: var(--card);
|
||
border: 1px solid var(--card-border);
|
||
border-radius: 14px;
|
||
backdrop-filter: blur(10px);
|
||
box-shadow: 0 10px 30px rgba(0, 0, 0, 0.25);
|
||
padding: 16px;
|
||
}
|
||
h2 {
|
||
margin: 0 0 6px;
|
||
font-size: 22px;
|
||
}
|
||
.sub {
|
||
color: var(--muted);
|
||
margin: 0 0 10px;
|
||
line-height: 1.5;
|
||
font-size: 14px;
|
||
}
|
||
label {
|
||
display: block;
|
||
margin: 8px 0 6px;
|
||
color: #cbd5e1;
|
||
font-size: 13px;
|
||
}
|
||
textarea, input, select {
|
||
width: 100%;
|
||
border: 1px solid rgba(148, 163, 184, 0.35);
|
||
background: rgba(15, 23, 42, 0.75);
|
||
color: var(--text);
|
||
border-radius: 10px;
|
||
margin: 0 0 10px;
|
||
padding: 10px 12px;
|
||
outline: none;
|
||
transition: border-color .2s, box-shadow .2s;
|
||
}
|
||
textarea:focus, input:focus, select:focus {
|
||
border-color: var(--primary);
|
||
box-shadow: 0 0 0 3px rgba(96, 165, 250, 0.15);
|
||
}
|
||
.row { display: grid; grid-template-columns: 1fr 1fr; gap: 12px; }
|
||
.actions {
|
||
display: flex;
|
||
flex-wrap: wrap;
|
||
gap: 10px;
|
||
margin-top: 6px;
|
||
}
|
||
button {
|
||
border: 0;
|
||
color: white;
|
||
border-radius: 10px;
|
||
padding: 10px 16px;
|
||
cursor: pointer;
|
||
font-weight: 600;
|
||
letter-spacing: .2px;
|
||
background: linear-gradient(135deg, var(--primary), var(--primary-2));
|
||
transition: transform .08s ease, opacity .2s;
|
||
}
|
||
button.secondary {
|
||
background: linear-gradient(135deg, #6366f1, #4f46e5);
|
||
}
|
||
button:active { transform: translateY(1px); }
|
||
.hint {
|
||
margin-top: 10px;
|
||
color: var(--muted);
|
||
font-size: 12px;
|
||
}
|
||
.status {
|
||
margin: 0 0 8px;
|
||
font-size: 13px;
|
||
color: var(--ok);
|
||
}
|
||
.status.warn { color: var(--warn); }
|
||
.panel-title {
|
||
margin: 0 0 8px;
|
||
font-size: 15px;
|
||
color: #dbeafe;
|
||
}
|
||
.token-grid {
|
||
display: grid;
|
||
grid-template-columns: 1fr 1fr;
|
||
gap: 12px;
|
||
}
|
||
.token-kpi {
|
||
border: 1px solid rgba(148, 163, 184, 0.25);
|
||
border-radius: 12px;
|
||
background: rgba(15, 23, 42, 0.55);
|
||
padding: 10px 12px;
|
||
}
|
||
.token-kpi .label {
|
||
color: var(--muted);
|
||
font-size: 12px;
|
||
margin-bottom: 4px;
|
||
}
|
||
.token-kpi .value {
|
||
font-size: 20px;
|
||
font-weight: 700;
|
||
color: #bfdbfe;
|
||
}
|
||
.user-list {
|
||
display: grid;
|
||
grid-template-columns: repeat(auto-fill, minmax(250px, 1fr));
|
||
gap: 10px;
|
||
margin-top: 6px;
|
||
}
|
||
.user-item {
|
||
display: flex;
|
||
align-items: center;
|
||
gap: 10px;
|
||
border: 1px solid rgba(148, 163, 184, 0.2);
|
||
border-radius: 12px;
|
||
background: rgba(15, 23, 42, 0.55);
|
||
padding: 10px;
|
||
}
|
||
.avatar {
|
||
width: 36px;
|
||
height: 36px;
|
||
border-radius: 999px;
|
||
object-fit: cover;
|
||
background: rgba(148, 163, 184, 0.2);
|
||
border: 1px solid rgba(148, 163, 184, 0.35);
|
||
}
|
||
.avatar-placeholder {
|
||
width: 36px;
|
||
height: 36px;
|
||
border-radius: 999px;
|
||
display: inline-flex;
|
||
align-items: center;
|
||
justify-content: center;
|
||
background: rgba(99, 102, 241, 0.35);
|
||
color: #e0e7ff;
|
||
font-size: 14px;
|
||
font-weight: 700;
|
||
border: 1px solid rgba(148, 163, 184, 0.35);
|
||
}
|
||
.user-meta {
|
||
min-width: 0;
|
||
display: grid;
|
||
gap: 2px;
|
||
}
|
||
.user-nick {
|
||
font-size: 14px;
|
||
color: #f1f5f9;
|
||
white-space: nowrap;
|
||
overflow: hidden;
|
||
text-overflow: ellipsis;
|
||
}
|
||
.user-id {
|
||
font-size: 12px;
|
||
color: #94a3b8;
|
||
white-space: nowrap;
|
||
overflow: hidden;
|
||
text-overflow: ellipsis;
|
||
}
|
||
pre {
|
||
margin: 0;
|
||
background: #090d18;
|
||
color: #d1fae5;
|
||
border: 1px solid rgba(16, 185, 129, 0.2);
|
||
padding: 12px;
|
||
border-radius: 10px;
|
||
overflow: auto;
|
||
max-height: 520px;
|
||
line-height: 1.45;
|
||
}
|
||
@media (max-width: 768px) {
|
||
.row { grid-template-columns: 1fr; }
|
||
.token-grid { grid-template-columns: 1fr; }
|
||
}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<section class="card">
|
||
<h2>私信查询调试台</h2>
|
||
<p class="sub">用于调试微信对话开放平台私信列表接口，支持一键查询、Token 过期分析与结果可视化。</p>
|
||
<label>默认配置</label>
|
||
<select id="presetSelect"></select>
|
||
<label>authtoken</label>
|
||
<textarea id="authtoken" rows="4" placeholder="粘贴 Charles 抓包中的 authtoken"></textarea>
|
||
<label>wxbot_bid</label>
|
||
<input id="wxbot_bid" placeholder="例如：75021053883a9090c67d5becde7f4b0a" />
|
||
<div class="row">
|
||
<div>
|
||
<label>page</label>
|
||
<input id="page" value="0" />
|
||
</div>
|
||
<div>
|
||
<label>size</label>
|
||
<input id="size" value="30" />
|
||
</div>
|
||
</div>
|
||
<div class="actions">
|
||
<button id="queryBtn" onclick="queryPm()">查询私信</button>
|
||
<button id="analyzeBtn" class="secondary" onclick="analyzeToken()">分析 Token 过期时间</button>
|
||
</div>
|
||
<div class="hint">提示：查询失败时，先检查 authtoken 是否过期、wxbot_bid 是否与当前机器人一致。</div>
|
||
</section>
|
||
|
||
<section class="card">
|
||
<p id="status" class="status">等待操作...</p>
|
||
<h3 class="panel-title">Token 过期信息</h3>
|
||
<div id="tokenReport" class="token-grid">
|
||
<div class="token-kpi">
|
||
<div class="label">状态</div>
|
||
<div id="tokenStatus" class="value">-</div>
|
||
</div>
|
||
<div class="token-kpi">
|
||
<div class="label">剩余时间</div>
|
||
<div id="tokenRemain" class="value">-</div>
|
||
</div>
|
||
</div>
|
||
<h3 class="panel-title" style="margin-top: 12px;">用户列表（userid / 头像 / 昵称）</h3>
|
||
<div id="userList" class="user-list"></div>
|
||
<h3 class="panel-title" style="margin-top: 12px;">原始返回</h3>
|
||
<pre id="output">{
|
||
"message": "点击上方按钮开始查询"
|
||
}</pre>
|
||
</section>
|
||
</div>
|
||
|
||
<script>
|
||
const defaultPresets = [
|
||
{
|
||
id: 'eric-travel',
|
||
name: 'Eric在旅行',
|
||
authtoken: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyaWQiOjE5MzE1MjcsIm9wZW5pZCI6Im85VS04NW1DSXZ0eDZ0a0pteTQyMkkzcUNHVWciLCJzaWduZXRpbWUiOjE3NzQ5MzM0NDQ4MTgsImlhdCI6MTc3NDkzMzQ0NCwiZXhwIjoxNzc3NTI1NDQ0fQ.iC9vdjBqa2LGlqA6F-Dg4GcfDyv3Rp8JwpBaO3j_eE4',
|
||
wxbot_bid: '75021053883a9090c67d5becde7f4b0a',
|
||
},
|
||
{
|
||
id: 'xidu-shijie',
|
||
name: '异度世界',
|
||
authtoken: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyaWQiOjE5MzE1MjcsIm9wZW5pZCI6Im85VS04NW1DSXZ0eDZ0a0pteTQyMkkzcUNHVWciLCJzaWduZXRpbWUiOjE3NzQ5MzM0NDQ4MTgsImlhdCI6MTc3NDkzMzQ0NCwiZXhwIjoxNzc3NTI1NDQ0fQ.iC9vdjBqa2LGlqA6F-Dg4GcfDyv3Rp8JwpBaO3j_eE4',
|
||
wxbot_bid: '4888c6ab5cb70dcc3adea1e2d2ff0201',
|
||
},
|
||
];
|
||
|
||
function initPresetSelector() {
|
||
const select = document.getElementById('presetSelect');
|
||
select.innerHTML = defaultPresets
|
||
.map((item) => `<option value="${item.id}">${item.name}</option>`)
|
||
.join('');
|
||
select.addEventListener('change', () => {
|
||
const preset = defaultPresets.find((x) => x.id === select.value);
|
||
if (!preset) return;
|
||
document.getElementById('authtoken').value = preset.authtoken;
|
||
document.getElementById('wxbot_bid').value = preset.wxbot_bid;
|
||
});
|
||
if (defaultPresets.length > 0) {
|
||
select.value = defaultPresets[0].id;
|
||
select.dispatchEvent(new Event('change'));
|
||
}
|
||
}
|
||
|
||
function updatePresetAuthtoken() {
|
||
const select = document.getElementById('presetSelect');
|
||
const selectedId = select.value;
|
||
const idx = defaultPresets.findIndex((x) => x.id === selectedId);
|
||
if (idx >= 0) {
|
||
defaultPresets[idx].authtoken = document.getElementById('authtoken').value.trim();
|
||
}
|
||
}
|
||
|
||
function setLoading(loading, text) {
|
||
document.getElementById('queryBtn').disabled = loading;
|
||
document.getElementById('analyzeBtn').disabled = loading;
|
||
const status = document.getElementById('status');
|
||
status.classList.remove('warn');
|
||
status.textContent = text || (loading ? '请求中...' : '完成');
|
||
if (loading) status.classList.add('warn');
|
||
}
|
||
|
||
function render(data) {
|
||
document.getElementById('output').textContent = JSON.stringify(data, null, 2);
|
||
}
|
||
|
||
function formatRemain(seconds) {
|
||
if (typeof seconds !== 'number') return '-';
|
||
if (seconds <= 0) return '已过期';
|
||
const days = Math.floor(seconds / 86400);
|
||
const hours = Math.floor((seconds % 86400) / 3600);
|
||
const mins = Math.floor((seconds % 3600) / 60);
|
||
return `${days}天 ${hours}小时 ${mins}分钟`;
|
||
}
|
||
|
||
function renderTokenReport(report) {
|
||
const statusEl = document.getElementById('tokenStatus');
|
||
const remainEl = document.getElementById('tokenRemain');
|
||
const statusText = report?.status || '-';
|
||
const remainText = formatRemain(report?.analysis?.expires_in_seconds);
|
||
statusEl.textContent = statusText;
|
||
remainEl.textContent = remainText;
|
||
statusEl.style.color = statusText.includes('过期') ? '#fca5a5' : '#86efac';
|
||
remainEl.style.color = remainText === '已过期' ? '#fca5a5' : '#bfdbfe';
|
||
}
|
||
|
||
function renderUsers(users) {
|
||
const list = document.getElementById('userList');
|
||
if (!Array.isArray(users) || users.length === 0) {
|
||
list.innerHTML = '<div class="hint">当前暂无用户数据，先点击“查询私信”。</div>';
|
||
return;
|
||
}
|
||
list.innerHTML = users.map((u) => {
|
||
const nick = u.nick || '未命名用户';
|
||
const uid = u.userId || '未知 userid';
|
||
const avatar = u.avatar || '';
|
||
const safeNick = String(nick).replace(/</g, '<').replace(/>/g, '>');
|
||
const safeUid = String(uid).replace(/</g, '<').replace(/>/g, '>');
|
||
const avatarHtml = avatar
|
||
? `<img class="avatar" src="${avatar}" alt="${safeNick}" referrerpolicy="no-referrer" />`
|
||
: `<div class="avatar-placeholder">${safeNick.slice(0, 1)}</div>`;
|
||
return `<div class="user-item">${avatarHtml}<div class="user-meta"><div class="user-nick">${safeNick}</div><div class="user-id">${safeUid}</div></div></div>`;
|
||
}).join('');
|
||
}
|
||
|
||
async function queryPm() {
|
||
updatePresetAuthtoken();
|
||
const payload = {
|
||
authtoken: document.getElementById('authtoken').value.trim(),
|
||
wxbot_bid: document.getElementById('wxbot_bid').value.trim(),
|
||
page: Number(document.getElementById('page').value || 0),
|
||
size: Number(document.getElementById('size').value || 30),
|
||
};
|
||
try {
|
||
setLoading(true, '正在查询私信...');
|
||
const res = await fetch('/api/private-message/query', {
|
||
method: 'POST',
|
||
headers: { 'Content-Type': 'application/json' },
|
||
body: JSON.stringify(payload)
|
||
});
|
||
const data = await res.json();
|
||
if (!res.ok) {
|
||
document.getElementById('status').classList.add('warn');
|
||
document.getElementById('status').textContent = '查询失败';
|
||
} else {
|
||
document.getElementById('status').textContent = '查询成功';
|
||
}
|
||
renderUsers(data.users);
|
||
render(data);
|
||
} catch (err) {
|
||
document.getElementById('status').classList.add('warn');
|
||
document.getElementById('status').textContent = '请求异常';
|
||
render({ error: String(err) });
|
||
} finally {
|
||
setLoading(false);
|
||
}
|
||
}
|
||
|
||
async function analyzeToken() {
|
||
updatePresetAuthtoken();
|
||
const payload = {
|
||
authtoken: document.getElementById('authtoken').value.trim(),
|
||
};
|
||
try {
|
||
setLoading(true, '正在分析 Token...');
|
||
const res = await fetch('/api/private-message/analyze', {
|
||
method: 'POST',
|
||
headers: { 'Content-Type': 'application/json' },
|
||
body: JSON.stringify(payload)
|
||
});
|
||
const data = await res.json();
|
||
if (!res.ok) {
|
||
document.getElementById('status').classList.add('warn');
|
||
document.getElementById('status').textContent = '分析失败';
|
||
} else {
|
||
document.getElementById('status').textContent = '分析成功';
|
||
}
|
||
renderTokenReport(data);
|
||
render(data);
|
||
} catch (err) {
|
||
document.getElementById('status').classList.add('warn');
|
||
document.getElementById('status').textContent = '请求异常';
|
||
render({ error: String(err) });
|
||
} finally {
|
||
setLoading(false);
|
||
}
|
||
}
|
||
|
||
initPresetSelector();
|
||
renderUsers([]);
|
||
</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
@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"),
|
||
"users": [
|
||
{
|
||
"nick": c.get("nick"),
|
||
"userId": c.get("userId") or c.get("userid"),
|
||
"avatar": c.get("avatar") or c.get("headImg") or c.get("headimgurl") or c.get("headImgUrl"),
|
||
"lastMsg": (c.get("lastMsg") or {}).get("content"),
|
||
"lastActiveTime": c.get("lastActiveTime"),
|
||
}
|
||
for c in contacts
|
||
],
|
||
"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
|
||
|