653 lines
25 KiB
Python
653 lines
25 KiB
Python
import base64
|
||
import json
|
||
import logging
|
||
import os
|
||
import struct
|
||
import time
|
||
import xml.etree.ElementTree as ET
|
||
from datetime import datetime, timezone
|
||
from typing import Optional
|
||
from urllib.parse import parse_qs
|
||
|
||
import httpx
|
||
from Crypto.Cipher import AES
|
||
from fastapi import APIRouter, BackgroundTasks, HTTPException, Request
|
||
from fastapi.responses import HTMLResponse, PlainTextResponse
|
||
|
||
from ..config import PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL
|
||
|
||
router = APIRouter(prefix="/chatbot", tags=["chatbot"])
|
||
|
||
logger = logging.getLogger(__name__)
|
||
_processed_callback_cache: dict[str, int] = {}
|
||
|
||
|
||
OPEN_API_PROFILES = {
|
||
"eric": {
|
||
"name": "Eric在旅行",
|
||
"APP_ID": "fXYWvGfq04uMScP",
|
||
"TOKEN": "EWRotyMHfTLAlYmCHpdH8AZcuZPBHn",
|
||
"ENCODING_AES_KEY": "1wJZiJm8URfRjkHu1QfIVDeNFc2jbzzeCZqvhQecsCx",
|
||
},
|
||
"xidu": {
|
||
"name": "异度世界",
|
||
"APP_ID": "LTKfSFeK0h8TRLM",
|
||
"TOKEN": "64DbvAd5m7YOJ1XslyMgJ9V42jDIUH",
|
||
"ENCODING_AES_KEY": "KCZGtPDjqFZnBuApvG3xqhvHUMnqFHbC5cnD23HKACX",
|
||
},
|
||
}
|
||
ACTIVE_PROFILE = (os.getenv("CHATBOT_ACTIVE_PROFILE") or "eric").strip().lower()
|
||
if ACTIVE_PROFILE not in OPEN_API_PROFILES:
|
||
ACTIVE_PROFILE = "eric"
|
||
APP_ID = OPEN_API_PROFILES[ACTIVE_PROFILE]["APP_ID"]
|
||
TOKEN = OPEN_API_PROFILES[ACTIVE_PROFILE]["TOKEN"]
|
||
ENCODING_AES_KEY = OPEN_API_PROFILES[ACTIVE_PROFILE]["ENCODING_AES_KEY"]
|
||
|
||
PM_SYNC_AUTHTOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyaWQiOjE5MzE1MjcsIm9wZW5pZCI6Im85VS04NW1DSXZ0eDZ0a0pteTQyMkkzcUNHVWciLCJzaWduZXRpbWUiOjE3NzQ5MzM0NDQ4MTgsImlhdCI6MTc3NDkzMzQ0NCwiZXhwIjoxNzc3NTI1NDQ0fQ.iC9vdjBqa2LGlqA6F-Dg4GcfDyv3Rp8JwpBaO3j_eE4"
|
||
PM_SYNC_DEFAULT_WXBOT_BID = "75021053883a9090c67d5becde7f4b0a"
|
||
PB_COLLECTION = "wechat_private_users"
|
||
|
||
|
||
def pkcs5_unpadding(data: bytes) -> bytes:
|
||
pad = data[-1]
|
||
if pad < 1 or pad > 32:
|
||
return data
|
||
return data[:-pad]
|
||
|
||
|
||
def pkcs5_padding(data: bytes) -> bytes:
|
||
amount_to_pad = 32 - (len(data) % 32)
|
||
return data + bytes([amount_to_pad]) * amount_to_pad
|
||
|
||
|
||
def get_aes_key_and_iv(encoding_aes_key: str) -> tuple[bytes, bytes]:
|
||
aes_key = base64.b64decode(encoding_aes_key + "=")
|
||
if len(aes_key) != 32:
|
||
raise ValueError("encodingAESKey invalid")
|
||
return aes_key, aes_key[:16]
|
||
|
||
|
||
def aes_decrypt_base64(cipher_b64: str, encoding_aes_key: str) -> str:
|
||
aes_key, iv = get_aes_key_and_iv(encoding_aes_key)
|
||
compact = "".join(str(cipher_b64 or "").split())
|
||
compact += "=" * (-len(compact) % 4)
|
||
cipher_bytes = base64.b64decode(compact)
|
||
cipher = AES.new(aes_key, AES.MODE_CBC, iv)
|
||
decrypted = pkcs5_unpadding(cipher.decrypt(cipher_bytes))
|
||
try:
|
||
content = decrypted[16:]
|
||
msg_len = struct.unpack("!I", content[:4])[0]
|
||
msg = content[4 : 4 + msg_len]
|
||
text = msg.decode("utf-8")
|
||
if text:
|
||
return text
|
||
except Exception:
|
||
pass
|
||
return decrypted.decode("utf-8")
|
||
|
||
|
||
def aes_encrypt_base64(plaintext: str, encoding_aes_key: str, app_id: str | None = None) -> str:
|
||
aes_key, iv = get_aes_key_and_iv(encoding_aes_key)
|
||
raw = plaintext.encode("utf-8")
|
||
if app_id:
|
||
data = os.urandom(16) + struct.pack("!I", len(raw)) + raw + app_id.encode("utf-8")
|
||
else:
|
||
data = raw
|
||
padded = pkcs5_padding(data)
|
||
cipher = AES.new(aes_key, AES.MODE_CBC, iv)
|
||
return base64.b64encode(cipher.encrypt(padded)).decode("utf-8")
|
||
|
||
|
||
def _build_keyword_h5_obj(query: str, user_id: str) -> Optional[dict]:
|
||
q = (query or "").strip()
|
||
if "电子书" in q:
|
||
return {
|
||
"news": {
|
||
"articles": [
|
||
{
|
||
"title": "数字游民",
|
||
"description": "地理套利与自动化杠杆",
|
||
"url": f"https://vip.hackrobot.cn/ebook?userid={user_id}",
|
||
"picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png",
|
||
"type": "h5",
|
||
}
|
||
]
|
||
}
|
||
}
|
||
if "群" in q:
|
||
return {
|
||
"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 None
|
||
|
||
|
||
def _parse_decrypted_payload(decrypted_text: str) -> dict:
|
||
text = (decrypted_text or "").strip()
|
||
# 兼容 BOM / NUL / 控制字符前缀
|
||
text = text.lstrip("\ufeff\x00\r\n\t ")
|
||
if not text:
|
||
raise ValueError("empty decrypted payload")
|
||
|
||
# 某些上游会把 JSON 再包一层字符串："{\"UserId\":\"...\"}"
|
||
if text.startswith('"') and text.endswith('"'):
|
||
try:
|
||
unwrapped = json.loads(text)
|
||
if isinstance(unwrapped, str):
|
||
text = unwrapped.strip().lstrip("\ufeff\x00\r\n\t ")
|
||
except Exception:
|
||
pass
|
||
|
||
# 若前面夹杂少量杂字符，尝试定位第一个结构起始位
|
||
first_json = text.find("{")
|
||
first_xml = text.find("<")
|
||
starts = [x for x in [first_json, first_xml] if x >= 0]
|
||
if starts:
|
||
first = min(starts)
|
||
if first > 0:
|
||
text = text[first:]
|
||
|
||
if text.startswith("{"):
|
||
obj = json.loads(text)
|
||
obj["_payload_format"] = "json"
|
||
return obj
|
||
if text.startswith("<"):
|
||
root = ET.fromstring(text)
|
||
|
||
def get_text(tag: str, default: str = "") -> str:
|
||
node = root.find(tag)
|
||
return (node.text or "").strip() if node is not None and node.text is not None else default
|
||
|
||
content_node = root.find("content")
|
||
query = ""
|
||
if content_node is not None:
|
||
msg_node = content_node.find("msg")
|
||
if msg_node is not None and msg_node.text:
|
||
query = msg_node.text.strip()
|
||
return {
|
||
"UserId": get_text("userid", ""),
|
||
"Query": query,
|
||
"Timestamp": get_text("createtime", ""),
|
||
"MsgId": get_text("msgid", ""),
|
||
"ChannelId": get_text("appid", ""),
|
||
"appid": get_text("appid", ""),
|
||
"channel": get_text("channel", "0"),
|
||
"from": get_text("from", "0"),
|
||
"_payload_format": "xml",
|
||
}
|
||
# 兼容 querystring 明文：UserId=xxx&Query=你好&appid=...
|
||
if "=" in text and "&" in text:
|
||
qs = parse_qs(text, keep_blank_values=True)
|
||
|
||
def qv(*keys: str) -> str:
|
||
for key in keys:
|
||
vals = qs.get(key) or qs.get(key.lower()) or qs.get(key.upper()) or []
|
||
if vals and str(vals[0]).strip():
|
||
return str(vals[0]).strip()
|
||
return ""
|
||
|
||
return {
|
||
"UserId": qv("UserId", "userid", "userId", "from_user"),
|
||
"Query": qv("Query", "query", "msg", "content"),
|
||
"Timestamp": qv("Timestamp", "timestamp", "createTime", "msg_time"),
|
||
"MsgId": qv("MsgId", "msgId", "msgid"),
|
||
"ChannelId": qv("ChannelId", "appid", "appId"),
|
||
"appid": qv("appid", "appId"),
|
||
"channel": qv("channel"),
|
||
"from": qv("from"),
|
||
"_payload_format": "querystring",
|
||
}
|
||
raise ValueError("unsupported payload format")
|
||
|
||
|
||
def _build_xml_callback_reply(data: dict, answer_text: str) -> dict:
|
||
user_id = str(data.get("UserId", "")).strip()
|
||
appid = str(data.get("ChannelId", "") or data.get("appid", "")).strip()
|
||
channel = int(str(data.get("channel", "0") or "0"))
|
||
msg = _build_keyword_h5_obj(str(data.get("Query", "")), user_id)
|
||
content = msg if msg else {"msg": answer_text}
|
||
return {"userid": user_id, "appid": appid, "content": content, "channel": channel, "from": 1}
|
||
|
||
|
||
def _safe_filter_value(value: str) -> str:
|
||
return (value or "").replace("\\", "\\\\").replace('"', '\\"')
|
||
|
||
|
||
def _build_callback_dedupe_key(data: dict) -> str:
|
||
return "|".join(
|
||
[
|
||
str(data.get("UserId", "")).strip(),
|
||
str(data.get("MsgId", "")).strip(),
|
||
str(data.get("Timestamp", "")).strip(),
|
||
str(data.get("Query", "")).strip(),
|
||
]
|
||
)
|
||
|
||
|
||
def _is_duplicate_callback(data: dict, ttl_seconds: int = 3600) -> bool:
|
||
now_ts = int(time.time())
|
||
for k, exp in list(_processed_callback_cache.items()):
|
||
if exp <= now_ts:
|
||
_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
|
||
|
||
|
||
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:
|
||
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",
|
||
}
|
||
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 _pb_upsert_user_profile(record: dict) -> None:
|
||
pb_url = (PB_URL or "").rstrip("/")
|
||
if not pb_url or not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
|
||
return
|
||
userid = str(record.get("userid", "")).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},
|
||
)
|
||
if auth_resp.status_code == 404:
|
||
auth_resp = await client.post(
|
||
f"{pb_url}/api/collections/_superusers/auth-with-password",
|
||
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
|
||
)
|
||
if auth_resp.status_code != 200:
|
||
return
|
||
token = (auth_resp.json() or {}).get("token", "")
|
||
if not token:
|
||
return
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
user_expr = _safe_filter_value(userid)
|
||
# 业务要求：wechat_private_users.userid 全局唯一（不按 channel_id 分桶）
|
||
filter_exp = f'userid="{user_expr}"'
|
||
list_resp = await client.get(
|
||
f"{pb_url}/api/collections/{PB_COLLECTION}/records",
|
||
params={"page": 1, "perPage": 200, "filter": filter_exp, "sort": "-updated"},
|
||
headers=headers,
|
||
)
|
||
items = ((list_resp.json() or {}).get("items")) or [] if list_resp.status_code == 200 else []
|
||
if items:
|
||
rid = items[0].get("id")
|
||
if rid:
|
||
await client.patch(f"{pb_url}/api/collections/{PB_COLLECTION}/records/{rid}", json=record, headers=headers)
|
||
# 清理历史重复数据，确保 userid 唯一
|
||
for extra in items[1:]:
|
||
extra_id = str(extra.get("id", "")).strip()
|
||
if not extra_id:
|
||
continue
|
||
await client.delete(
|
||
f"{pb_url}/api/collections/{PB_COLLECTION}/records/{extra_id}",
|
||
headers=headers,
|
||
)
|
||
return
|
||
await client.post(f"{pb_url}/api/collections/{PB_COLLECTION}/records", json=record, headers=headers)
|
||
|
||
|
||
async def _sync_user_profile_from_b(data: dict) -> None:
|
||
authtoken = (PM_SYNC_AUTHTOKEN or "").strip()
|
||
wxbot_bid = (PM_SYNC_DEFAULT_WXBOT_BID or "").strip()
|
||
if not authtoken or not wxbot_bid:
|
||
return
|
||
target_userid = str(data.get("UserId", "")).strip()
|
||
if not target_userid or _is_duplicate_callback(data):
|
||
return
|
||
result = await query_private_messages(
|
||
authtoken=authtoken,
|
||
wxbot_bid=wxbot_bid,
|
||
page=0,
|
||
size=100,
|
||
filter_value=0,
|
||
request_id=f"thirdapi-{target_userid}",
|
||
)
|
||
contacts = result.get("contacts") or []
|
||
target = next((c for c in contacts if str(c.get("userId", "") or c.get("userid", "")).strip() == target_userid), None)
|
||
if not target:
|
||
return
|
||
last_msg = target.get("lastMsg") or {}
|
||
await _pb_upsert_user_profile(
|
||
{
|
||
"userid": target_userid,
|
||
"channel_id": str(data.get("ChannelId", "")).strip(),
|
||
"nick": target.get("nick", ""),
|
||
"avatar": target.get("headImg") or target.get("avatar") or target.get("headimgurl") or target.get("headImgUrl") or "",
|
||
"source_query": str(data.get("Query", "")).strip(),
|
||
"source_message_id": str(data.get("MsgId", "")).strip(),
|
||
"source_message_time": str(data.get("Timestamp", "")),
|
||
"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),
|
||
}
|
||
)
|
||
|
||
|
||
async def _safe_sync_user_profile_from_b(data: dict) -> None:
|
||
try:
|
||
await _sync_user_profile_from_b(data)
|
||
except Exception:
|
||
logger.error("chatbot后台同步失败（已忽略）")
|
||
|
||
|
||
async def handle_business(data: dict) -> str:
|
||
user_id = str(data.get("UserId", "")).strip()
|
||
query = str(data.get("Query", "")).strip()
|
||
msg = _build_keyword_h5_obj(query, user_id)
|
||
if msg:
|
||
return json.dumps(msg, ensure_ascii=False)
|
||
return "1111111"
|
||
|
||
|
||
async def _handle_wechat_request(request: Request, app_id: Optional[str], background_tasks: BackgroundTasks) -> str:
|
||
if app_id and app_id != APP_ID:
|
||
logger.warning("chatbot app_id mismatch expected=%s actual=%s", APP_ID, app_id)
|
||
raw_body = (await request.body()).decode("utf-8", errors="ignore").strip()
|
||
if not raw_body:
|
||
raise HTTPException(status_code=400, detail="empty body")
|
||
|
||
cipher_b64 = raw_body
|
||
# 兼容三种请求体：
|
||
# 1) 纯密文 base64
|
||
# 2) JSON: {"encrypted":"..."} / {"encrypt":"..."} / {"Encrypt":"..."}
|
||
# 3) x-www-form-urlencoded: encrypted=... / encrypt=...
|
||
try:
|
||
outer = json.loads(raw_body)
|
||
if isinstance(outer, dict):
|
||
for key in ("encrypted", "encrypt", "Encrypt"):
|
||
val = outer.get(key)
|
||
if isinstance(val, str) and val.strip():
|
||
cipher_b64 = val
|
||
break
|
||
except Exception:
|
||
form = parse_qs(raw_body, keep_blank_values=True)
|
||
for key in ("encrypted", "encrypt", "Encrypt"):
|
||
vals = form.get(key) or []
|
||
if vals and str(vals[0]).strip():
|
||
cipher_b64 = str(vals[0])
|
||
break
|
||
|
||
cipher_b64 = "".join(str(cipher_b64).split()).strip()
|
||
try:
|
||
decrypted = aes_decrypt_base64(cipher_b64, ENCODING_AES_KEY)
|
||
logger.info(
|
||
"chatbot decrypted preview app_id=%s preview=%s",
|
||
app_id or "",
|
||
(decrypted or "")[:200],
|
||
)
|
||
data = _parse_decrypted_payload(decrypted)
|
||
logger.info(
|
||
"chatbot payload parsed app_id=%s format=%s user=%s query=%s",
|
||
app_id or "",
|
||
data.get("_payload_format", ""),
|
||
str(data.get("UserId", "") or data.get("userId", ""))[:80],
|
||
str(data.get("Query", ""))[:80],
|
||
)
|
||
except Exception as error:
|
||
logger.error(
|
||
"chatbot decrypt failed app_id=%s content_type=%s body_prefix=%s err=%s",
|
||
app_id or "",
|
||
request.headers.get("content-type", ""),
|
||
raw_body[:160],
|
||
error,
|
||
)
|
||
raise HTTPException(status_code=400, detail="decrypt error")
|
||
|
||
answer_text = await handle_business(data)
|
||
background_tasks.add_task(_safe_sync_user_profile_from_b, data)
|
||
|
||
if str(data.get("_payload_format", "")) == "xml":
|
||
resp_plain = json.dumps(_build_xml_callback_reply(data, answer_text), ensure_ascii=False)
|
||
response_app_id = str(data.get("ChannelId", "") or data.get("appid", "")).strip() or APP_ID
|
||
else:
|
||
resp_plain = json.dumps({"answer_type": "text", "text_info": {"short_answer": answer_text}}, ensure_ascii=False)
|
||
# thirdapi 场景也要携带 app_id 参与加密，否则平台可能判定回包无效
|
||
response_app_id = (app_id or APP_ID).strip() if isinstance(app_id, str) else APP_ID
|
||
|
||
return aes_encrypt_base64(resp_plain, ENCODING_AES_KEY, response_app_id)
|
||
|
||
|
||
@router.post("/", response_class=PlainTextResponse)
|
||
async def chatbot_root(request: Request, background_tasks: BackgroundTasks, app_id: Optional[str] = None):
|
||
return await _handle_wechat_request(request, app_id, background_tasks)
|
||
|
||
|
||
@router.post("/wechat/thirdapi", response_class=PlainTextResponse)
|
||
async def chatbot_thirdapi(request: Request, background_tasks: BackgroundTasks, app_id: Optional[str] = None):
|
||
return await _handle_wechat_request(request, app_id, background_tasks)
|
||
|
||
|
||
@router.get("/health")
|
||
def chatbot_health():
|
||
return {"status": "ok", "module": "chatbot-clone", "profile": ACTIVE_PROFILE}
|
||
|
||
|
||
@router.post("/api/private-message/query")
|
||
async def chatbot_private_message_query(payload: dict):
|
||
authtoken = str(payload.get("authtoken", "")).strip()
|
||
wxbot_bid = str(payload.get("wxbot_bid", "")).strip()
|
||
if not authtoken or not wxbot_bid:
|
||
raise HTTPException(status_code=400, detail="authtoken 和 wxbot_bid 不能为空")
|
||
result = await query_private_messages(
|
||
authtoken=authtoken,
|
||
wxbot_bid=wxbot_bid,
|
||
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",
|
||
)
|
||
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
|
||
],
|
||
"raw": result,
|
||
}
|
||
|
||
|
||
@router.post("/api/private-message/analyze")
|
||
async def chatbot_private_message_analyze(payload: dict):
|
||
authtoken = str(payload.get("authtoken", "")).strip()
|
||
if not authtoken:
|
||
raise HTTPException(status_code=400, detail="authtoken 不能为空")
|
||
try:
|
||
parts = authtoken.split(".")
|
||
payload_part = parts[1] if len(parts) == 3 else ""
|
||
payload_part += "=" * (-len(payload_part) % 4)
|
||
claims = json.loads(base64.urlsafe_b64decode(payload_part.encode("utf-8")).decode("utf-8")) if payload_part else {}
|
||
except Exception:
|
||
claims = {}
|
||
now_ts = int(datetime.now(timezone.utc).timestamp())
|
||
exp = int(claims.get("exp", 0)) if str(claims.get("exp", "")).isdigit() else 0
|
||
expires_in_seconds = exp - now_ts if exp else None
|
||
status = "无法解析过期时间"
|
||
if isinstance(expires_in_seconds, int):
|
||
if expires_in_seconds <= 0:
|
||
status = "已过期"
|
||
elif expires_in_seconds < 3600:
|
||
status = "即将过期（1小时内）"
|
||
else:
|
||
status = "有效"
|
||
return {
|
||
"status": status,
|
||
"now_utc": datetime.now(timezone.utc).isoformat(),
|
||
"claims": claims,
|
||
"analysis": {
|
||
"expires_in_seconds": expires_in_seconds,
|
||
"expires_in_hours": round(expires_in_seconds / 3600, 2) if isinstance(expires_in_seconds, int) else None,
|
||
"summary": "该分析基于 JWT payload 解码，不包含签名有效性校验。",
|
||
},
|
||
}
|
||
|
||
|
||
@router.get("/private-message", response_class=HTMLResponse)
|
||
def chatbot_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>Chatbot 私信调试台</title>
|
||
<style>
|
||
body { font-family: system-ui, -apple-system, "Microsoft YaHei", sans-serif; margin: 0; padding: 20px; background: #0f172a; color: #e2e8f0; }
|
||
.wrap { max-width: 980px; margin: 0 auto; display: grid; gap: 12px; }
|
||
.card { background: #111827; border: 1px solid #334155; border-radius: 10px; padding: 14px; }
|
||
h2 { margin: 0 0 8px; }
|
||
label { display:block; margin: 8px 0 6px; font-size: 13px; color: #cbd5e1; }
|
||
input, textarea { width: 100%; border-radius: 8px; border: 1px solid #475569; background: #0b1220; color: #e2e8f0; padding: 10px; box-sizing: border-box; }
|
||
textarea { min-height: 88px; }
|
||
.row { display:grid; grid-template-columns: 1fr 1fr; gap: 10px; }
|
||
.btns { display:flex; gap:10px; margin-top: 10px; }
|
||
button { border: 0; background: #2563eb; color: white; border-radius: 8px; padding: 10px 14px; cursor: pointer; }
|
||
button.alt { background: #4f46e5; }
|
||
#status { font-size: 13px; color: #93c5fd; }
|
||
#users { display:grid; grid-template-columns: repeat(auto-fill,minmax(250px,1fr)); gap: 8px; }
|
||
.u { border:1px solid #334155; border-radius:8px; padding:8px; background:#0b1220; }
|
||
.uid { color:#93c5fd; font-size:12px; word-break:break-all; }
|
||
pre { background:#020617; border:1px solid #334155; border-radius:8px; padding:10px; overflow:auto; max-height:420px; }
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="wrap">
|
||
<section class="card">
|
||
<h2>Chatbot 私信调试台</h2>
|
||
<div id="status">准备就绪</div>
|
||
<label>authtoken</label>
|
||
<textarea id="authtoken" placeholder="粘贴抓包 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="btns">
|
||
<button onclick="queryPm()">查询私信</button>
|
||
<button class="alt" onclick="analyzeToken()">分析 Token</button>
|
||
</div>
|
||
</section>
|
||
|
||
<section class="card">
|
||
<h3 style="margin:0 0 8px;">用户列表</h3>
|
||
<div id="users"></div>
|
||
</section>
|
||
|
||
<section class="card">
|
||
<h3 style="margin:0 0 8px;">原始结果</h3>
|
||
<pre id="out">{}</pre>
|
||
</section>
|
||
</div>
|
||
|
||
<script>
|
||
const $ = (id) => document.getElementById(id);
|
||
function setStatus(s){ $("status").textContent = s; }
|
||
function renderJson(v){ $("out").textContent = JSON.stringify(v, null, 2); }
|
||
function renderUsers(users){
|
||
const box = $("users");
|
||
if(!Array.isArray(users) || !users.length){
|
||
box.innerHTML = '<div class="u">暂无数据</div>';
|
||
return;
|
||
}
|
||
box.innerHTML = users.map(u => {
|
||
const nick = (u.nick || "未命名").replace(/</g,"<").replace(/>/g,">");
|
||
const uid = (u.userId || "").replace(/</g,"<").replace(/>/g,">");
|
||
const last = (u.lastMsg || "").replace(/</g,"<").replace(/>/g,">");
|
||
return `<div class="u"><div>${nick}</div><div class="uid">${uid}</div><div style="font-size:12px;color:#94a3b8;margin-top:4px;">${last}</div></div>`;
|
||
}).join("");
|
||
}
|
||
|
||
async function queryPm(){
|
||
try{
|
||
setStatus("查询中...");
|
||
const payload = {
|
||
authtoken: $("authtoken").value.trim(),
|
||
wxbot_bid: $("wxbot_bid").value.trim(),
|
||
page: Number($("page").value || 0),
|
||
size: Number($("size").value || 30),
|
||
};
|
||
const res = await fetch("/chatbot/api/private-message/query", {
|
||
method: "POST",
|
||
headers: { "Content-Type": "application/json" },
|
||
body: JSON.stringify(payload),
|
||
});
|
||
const data = await res.json().catch(() => ({}));
|
||
setStatus(res.ok ? "查询成功" : "查询失败");
|
||
renderUsers(data.users || []);
|
||
renderJson(data);
|
||
}catch(e){
|
||
setStatus("请求异常");
|
||
renderJson({ error: String(e) });
|
||
}
|
||
}
|
||
|
||
async function analyzeToken(){
|
||
try{
|
||
setStatus("分析中...");
|
||
const payload = { authtoken: $("authtoken").value.trim() };
|
||
const res = await fetch("/chatbot/api/private-message/analyze", {
|
||
method: "POST",
|
||
headers: { "Content-Type": "application/json" },
|
||
body: JSON.stringify(payload),
|
||
});
|
||
const data = await res.json().catch(() => ({}));
|
||
setStatus(res.ok ? "分析成功" : "分析失败");
|
||
renderJson(data);
|
||
}catch(e){
|
||
setStatus("请求异常");
|
||
renderJson({ error: String(e) });
|
||
}
|
||
}
|
||
</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|