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()
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",
}
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)
data = _parse_decrypted_payload(decrypted)
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 "chatbot clone ready
请使用 /chatbot/api/private-message/query 调试接口
"