import asyncio 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 NamedTuple, Optional from urllib.parse import parse_qs, quote import httpx from Crypto.Cipher import AES from fastapi import APIRouter, BackgroundTasks, HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse from ..config import PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL from ..services.mp_recommend import is_mp_recommend_webhook_body, try_persist_from_like_body router = APIRouter(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"] HACKROBOT_PROFILE = "xidu" def _resolve_profile_key(profile_key: str) -> str: k = (profile_key or "").strip().lower() return k if k in OPEN_API_PROFILES else "eric" def _profile_creds(profile_key: str) -> dict: return OPEN_API_PROFILES[_resolve_profile_key(profile_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: if len(decrypted) < 20: raise ValueError("payload too short") content = decrypted[16:] msg_len = struct.unpack("!I", content[:4])[0] msg_start = 4 msg_end = msg_start + msg_len if msg_end > len(content): raise ValueError("msg length out of range") msg = content[msg_start:msg_end] 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") class SharedChatbotReply(NamedTuple): """全公众号配置共用的解析结果：要么 H5 图文，要么兜底纯文本（含超链）。""" h5_message: Optional[dict] """与「发送客服消息」H5 一致：{\"news\": {\"articles\": [...]}}，未命中关键词时为 None。""" text_fallback: str """未命中关键词时的 short_answer 文案（可含 ）。""" def _normalize_h5_msg_per_wechat_doc(h5_message: dict) -> dict: """ 与文档「发送客服消息」H5 一致：msg = { news: { articles: [ { title, description, url, picurl, type } ] } }。 仅保留这五个字段，避免多余键导致客户端不渲染。 """ raw = h5_message.get("news") if isinstance(h5_message, dict) else None arts = raw.get("articles") if isinstance(raw, dict) else None if not isinstance(arts, list): arts = [] norm_articles: list[dict] = [] for a in arts: if not isinstance(a, dict): continue t = str(a.get("type", "h5") or "h5").strip().lower() if t not in ("h5", "mp"): t = "h5" norm_articles.append( { "title": str(a.get("title", "")), "description": str(a.get("description", "")), "url": str(a.get("url", "")), "picurl": str(a.get("picurl", "")), "type": t, } ) return {"news": {"articles": norm_articles}} def shared_build_h5_keyword_message(query: str, user_id: str, profile_key: str) -> Optional[dict]: """关键词命中时返回文档约定的 H5 msg（再经 _normalize 输出）。异度(/hackrobot) 群卡片仅认「群」；Eric(/chatbot) 另认「加群」。""" q = (query or "").strip() uid = (user_id or "").strip() prof = _resolve_profile_key(profile_key) if "电子书" in q: return { "news": { "articles": [ { "title": "数字游民", "description": "地理套利与自动化杠杆", "url": f"https://vip.hackrobot.cn/ebook?userid={uid}", "picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png", "type": "h5", } ] } } if prof == HACKROBOT_PROFILE: # 「加群」含「群」子串，异度仅按「群」触发时需排除「加群」整词 group_hit = ("群" in q) and ("加群" not in q) else: group_hit = ("群" in q) or ("加群" in q) if group_hit: return { "news": { "articles": [ { "title": "异度星球", "description": "电子书、视频教程、私密社群", "url": f"https://vip.hackrobot.cn?userid={uid}", "picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png", "type": "h5", } ] } } return None def build_fallback_reply_plain_text(user_id: str) -> str: """ 兜底回复（单独构建）：未命中「电子书」「群」等关键词时使用。 /chatbot、/hackrobot 共用同一文案与链接逻辑，仅解密密钥因 profile 不同。 """ uid = (user_id or "").strip() href = f"https://vip.hackrobot.cn/?userid={quote(uid, safe='')}" if uid else "https://vip.hackrobot.cn/" return "🤖消息已收到,稍晚看到后回复\n参与交流,👉可加群" def shared_resolve_reply(data: dict, profile_key: str) -> SharedChatbotReply: """根据解密后的 data 与 profile 决定 H5 或兜底；无关键词时一律走 build_fallback_reply_plain_text。""" user_id = str(data.get("UserId", "") or data.get("userId", "") or "").strip() query = str(data.get("Query", "") or "").strip() h5 = shared_build_h5_keyword_message(query, user_id, profile_key) if h5: return SharedChatbotReply(h5, "") return SharedChatbotReply(None, build_fallback_reply_plain_text(user_id)) def shared_build_openapi_text_answer(short_answer: str) -> dict: """开放 API 文档「输出」：文本 / 含超链兜底。""" return {"answer_type": "text", "text_info": {"short_answer": short_answer}} def _xml_channel_h5_envelope(data: dict, h5_message: dict) -> dict: """XML 入站（如 Eric /chatbot）命中关键词：userid 信封 + content.news，与线上一致。""" user_id = str(data.get("UserId", "") or data.get("userId", "") or "").strip() appid = str(data.get("ChannelId", "") or data.get("appid", "")).strip() channel = int(str(data.get("channel", "0") or "0")) return {"userid": user_id, "appid": appid, "content": h5_message, "channel": channel, "from": 1} def _openapi_json_h5_answer_per_sendmsg_doc(msg: dict) -> dict: """ 开放 API JSON 入站命中 H5：与文档「发送客服消息」一致，msg 为 { news: { articles: [...] } }。 第三方回调侧将该 msg 序列化为 JSON 字符串，放入 answer_type=text 的 text_info.short_answer （与客服接口里 msg 对象同源结构，由网关解析为卡片）。 """ compact = json.dumps(msg, ensure_ascii=False, separators=(",", ":")) return {"answer_type": "text", "text_info": {"short_answer": compact}} def _openapi_json_merge_request_echo(data: dict, body: dict) -> dict: """回传 RequestId，便于开放平台把应答与请求关联（避免未命中/空白展示）。""" rid = data.get("RequestId") if rid is None: return body s = str(rid).strip() if not s: return body out = dict(body) out["RequestId"] = rid return out def shared_build_plain_body_for_wechat_callback( *, payload_format: str, reply: SharedChatbotReply, data: dict, ) -> dict: """ 回调加密前的明文 JSON 统一入口（关键词 H5 与兜底共用一套解析结果 reply）。 - /chatbot 多为 XML 入站：H5 用 userid 信封；兜底改用 answer_type=text（信封里 content.msg 对超链异常）。 - /hackrobot 多为 JSON 入站：兜底用 answer_type=text；H5 用同结构的 msg JSON 串作为 short_answer（见发送客服消息文档）。 """ fmt = (payload_format or "").strip().lower() is_xml = fmt == "xml" if reply.h5_message: msg = _normalize_h5_msg_per_wechat_doc(reply.h5_message) if is_xml: return _xml_channel_h5_envelope(data, msg) body = _openapi_json_h5_answer_per_sendmsg_doc(msg) return _openapi_json_merge_request_echo(data, body) body = shared_build_openapi_text_answer(reply.text_fallback) if not is_xml: body = _openapi_json_merge_request_echo(data, body) return body def _ensure_user_id_from_session(data: dict) -> None: """开放 API JSON 可能只带 SessionId（如 0_xxx|openid），补全 UserId 供关键词链接与兜底文案使用。""" uid = str(data.get("UserId", "") or data.get("userId", "") or "").strip() if uid: return sid = str(data.get("SessionId", "") or "").strip() if "|" in sid: tail = sid.rsplit("|", 1)[-1].strip() if tail: data["UserId"] = tail 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 _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, profile_key: str, 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 = _resolve_profile_key(profile_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, profile_key: str) -> 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, profile_key): 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, profile_key: str) -> None: try: await _sync_user_profile_from_b(data, profile_key) except Exception: logger.error("chatbot后台同步失败（已忽略）") async def _handle_wechat_request( request: Request, app_id: Optional[str], background_tasks: BackgroundTasks, *, profile_key: str, ) -> str: creds = _profile_creds(profile_key) expected_app_id = creds["APP_ID"] encoding_key = creds["ENCODING_AES_KEY"] if app_id and app_id != expected_app_id: logger.warning( "chatbot app_id mismatch profile=%s expected=%s actual=%s", profile_key, expected_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_key) logger.info( "chatbot decrypted preview profile=%s app_id=%s preview=%s", profile_key, app_id or "", (decrypted or "")[:200], ) data = _parse_decrypted_payload(decrypted) _ensure_user_id_from_session(data) logger.info( "chatbot payload parsed profile=%s app_id=%s format=%s user=%s query=%s", profile_key, 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 profile=%s app_id=%s content_type=%s body_prefix=%s err=%s", profile_key, app_id or "", request.headers.get("content-type", ""), raw_body[:160], error, ) raise HTTPException(status_code=400, detail="decrypt error") reply = shared_resolve_reply(data, profile_key) background_tasks.add_task(_safe_sync_user_profile_from_b, data, profile_key) fmt = str(data.get("_payload_format", "") or "") plain_obj = shared_build_plain_body_for_wechat_callback(payload_format=fmt, reply=reply, data=data) resp_plain = json.dumps(plain_obj, ensure_ascii=False) logger.info( "chatbot reply_plain profile=%s fmt=%s shape=%s", profile_key, fmt, list(plain_obj.keys()), ) if fmt == "xml": response_app_id = str(data.get("ChannelId", "") or data.get("appid", "")).strip() or expected_app_id else: # 与原始 chatbot 服务保持一致：thirdapi（URL 带 app_id）回包使用纯 AES，不拼 app_id response_app_id = None if app_id else expected_app_id return aes_encrypt_base64(resp_plain, encoding_key, response_app_id) @router.post("/chatbot", response_class=PlainTextResponse) @router.post("/chatbot/", 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, profile_key=ACTIVE_PROFILE) @router.post("/chatbot/wechat/thirdapi", response_class=PlainTextResponse) @router.post("/chatbot/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, profile_key=ACTIVE_PROFILE) @router.get("/chatbot/health") @router.get("/chatbot/health/") def chatbot_health(): return {"status": "ok", "module": "chatbot-clone", "profile": ACTIVE_PROFILE} hackrobot_router = APIRouter(tags=["hackrobot"]) @hackrobot_router.post("/hackrobot", response_class=PlainTextResponse) @hackrobot_router.post("/hackrobot/", response_class=PlainTextResponse) async def hackrobot_root(request: Request, background_tasks: BackgroundTasks, app_id: Optional[str] = None): return await _handle_wechat_request(request, app_id, background_tasks, profile_key=HACKROBOT_PROFILE) @hackrobot_router.post("/hackrobot/wechat/thirdapi", response_class=PlainTextResponse) @hackrobot_router.post("/hackrobot/wechat/thirdapi/", response_class=PlainTextResponse) async def hackrobot_thirdapi(request: Request, background_tasks: BackgroundTasks, app_id: Optional[str] = None): return await _handle_wechat_request(request, app_id, background_tasks, profile_key=HACKROBOT_PROFILE) @hackrobot_router.get("/hackrobot/health") @hackrobot_router.get("/hackrobot/health/") def hackrobot_health(): return {"status": "ok", "module": "chatbot-clone", "profile": HACKROBOT_PROFILE} def _log_like_incoming(request: Request, body: bytes) -> None: """SmsForwarder Webhook 调试：打印完整入参（公众号助手等通知转发）。""" client_host = request.client.host if request.client else None client_port = request.client.port if request.client else None query = {k: v for k, v in request.query_params.multi_items()} headers = {k: v for k, v in request.headers.items()} ct = headers.get("content-type", "") or headers.get("Content-Type", "") logger.info( "[like] ----- incoming POST ----- path=%s full_url=%s client=%s:%s query=%r", request.url.path, str(request.url), client_host, client_port, query, ) logger.info("[like] content_type=%r content_length=%s body_bytes=%s", ct, headers.get("content-length"), len(body)) logger.info("[like] headers=%s", headers) if not body: logger.info("[like] body=(empty)") return text = None try: text = body.decode("utf-8") except UnicodeDecodeError: hex_preview = body[:400].hex() suffix = "..." if len(body) > 400 else "" logger.info("[like] body=(non-utf8, hex prefix) %s%s", hex_preview, suffix) return logger.info("[like] body_text=\n%s", text) low = (ct or "").lower() if "json" in low: try: parsed = json.loads(text) logger.info("[like] body_json=%s", json.dumps(parsed, ensure_ascii=False, indent=2)) except json.JSONDecodeError as e: logger.info("[like] body_json_parse_failed: %s", e) def _client_ip_for_like(request: Request) -> str: xff = request.headers.get("x-forwarded-for") or request.headers.get("X-Forwarded-For") if xff: return xff.split(",")[0].strip() if request.client: return request.client.host or "" return "" @router.post("/chatbot/like") @hackrobot_router.post("/hackrobot/like") async def like_webhook_sink(request: Request): """接收 SmsForwarder 等 Webhook POST，详细打日志；公众号助手通知写入 PocketBase。""" body = await request.body() _log_like_incoming(request, body) ct = request.headers.get("content-type", "") or "" payload: dict = {"ok": True, "received_bytes": len(body)} try: text = body.decode("utf-8") except UnicodeDecodeError: return JSONResponse(status_code=200, content=payload) if is_mp_recommend_webhook_body(text, ct): try: summary = await asyncio.to_thread(try_persist_from_like_body, text, ct, _client_ip_for_like(request)) if summary: payload["mp_recommend"] = summary except Exception: logger.exception("[like] PocketBase 写入 mp_recommend 失败") payload["mp_recommend_error"] = "persist_failed" return JSONResponse(status_code=200, content=payload) @router.post("/chatbot/api/private-message/query") @hackrobot_router.post("/hackrobot/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("/chatbot/api/private-message/analyze") @hackrobot_router.post("/hackrobot/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 解码，不包含签名有效性校验。", }, } def _private_message_debug_html(api_prefix: str) -> str: base = """ Chatbot 私信调试台

Chatbot 私信调试台

准备就绪

用户列表

原始结果

{}
""" return base.replace("__API_PREFIX__", (api_prefix or "").rstrip("/")) @router.get("/chatbot/private-message") @router.get("/chatbot/private-message/") @hackrobot_router.get("/hackrobot/private-message") @hackrobot_router.get("/hackrobot/private-message/") def chatbot_private_message_page(request: Request): api_prefix = "/hackrobot" if str(request.url.path).startswith("/hackrobot") else "/chatbot" return HTMLResponse(_private_message_debug_html(api_prefix))