From 8c3483c36058629a81df24002a4d629dd52f1081 Mon Sep 17 00:00:00 2001 From: eric Date: Thu, 2 Apr 2026 00:26:00 -0500 Subject: [PATCH] 's' --- app/main.py | 2 + app/routers/__init__.py | 3 +- app/routers/chatbot.py | 314 ++++++++++++++++++++++++++++++++-------- 3 files changed, 259 insertions(+), 60 deletions(-) diff --git a/app/main.py b/app/main.py index 4e87577..4272cdf 100755 --- a/app/main.py +++ b/app/main.py @@ -31,6 +31,7 @@ from .routers import ( common_router, meetup_router, chatbot_router, + hackrobot_router, ) from .routers.salon_meetup import router as salon_meetup_router @@ -177,6 +178,7 @@ app.include_router(common_router) app.include_router(meetup_router) app.include_router(salon_meetup_router) app.include_router(chatbot_router) +app.include_router(hackrobot_router) @app.get("/") diff --git a/app/routers/__init__.py b/app/routers/__init__.py index 63171f7..a8d990d 100644 --- a/app/routers/__init__.py +++ b/app/routers/__init__.py @@ -13,7 +13,7 @@ from .nomadlms import router as nomadlms_router from .legacy import router as legacy_router from .common import router as common_router from .meetup import router as meetup_router -from .chatbot import router as chatbot_router +from .chatbot import hackrobot_router, router as chatbot_router __all__ = [ "nomadvip_router", @@ -24,4 +24,5 @@ __all__ = [ "common_router", "meetup_router", "chatbot_router", + "hackrobot_router", ] diff --git a/app/routers/chatbot.py b/app/routers/chatbot.py index 566feb8..197daaa 100644 --- a/app/routers/chatbot.py +++ b/app/routers/chatbot.py @@ -7,8 +7,8 @@ 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 +from typing import NamedTuple, Optional +from urllib.parse import parse_qs, quote import httpx from Crypto.Cipher import AES @@ -18,7 +18,7 @@ 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(prefix="/chatbot", tags=["chatbot"]) +router = APIRouter(tags=["chatbot"]) logger = logging.getLogger(__name__) _processed_callback_cache: dict[str, int] = {} @@ -45,6 +45,17 @@ 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" @@ -106,8 +117,48 @@ def aes_encrypt_base64(plaintext: str, encoding_aes_key: str, app_id: str | None return base64.b64encode(cipher.encrypt(padded)).decode("utf-8") -def _build_keyword_h5_obj(query: str, user_id: str) -> Optional[dict]: +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": { @@ -115,21 +166,26 @@ def _build_keyword_h5_obj(query: str, user_id: str) -> Optional[dict]: { "title": "数字游民", "description": "地理套利与自动化杠杆", - "url": f"https://vip.hackrobot.cn/ebook?userid={user_id}", + "url": f"https://vip.hackrobot.cn/ebook?userid={uid}", "picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png", "type": "h5", } ] } } - if "群" in q: + 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={user_id}", + "url": f"https://vip.hackrobot.cn?userid={uid}", "picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png", "type": "h5", } @@ -139,6 +195,102 @@ def _build_keyword_h5_obj(query: str, user_id: str) -> Optional[dict]: 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 / 控制字符前缀 @@ -217,15 +369,6 @@ def _parse_decrypted_payload(decrypted_text: str) -> dict: 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('"', '\\"') @@ -241,12 +384,12 @@ def _build_callback_dedupe_key(data: dict) -> str: ) -def _is_duplicate_callback(data: dict, ttl_seconds: int = 3600) -> bool: +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 = _build_callback_dedupe_key(data) + key = _resolve_profile_key(profile_key) + "|" + _build_callback_dedupe_key(data) if not key.strip("|"): return False if key in _processed_callback_cache: @@ -326,13 +469,13 @@ async def _pb_upsert_user_profile(record: dict) -> None: 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: +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): + if not target_userid or _is_duplicate_callback(data, profile_key): return result = await query_private_messages( authtoken=authtoken, @@ -364,25 +507,30 @@ async def _sync_user_profile_from_b(data: dict) -> None: ) -async def _safe_sync_user_profile_from_b(data: dict) -> None: +async def _safe_sync_user_profile_from_b(data: dict, profile_key: str) -> None: try: - await _sync_user_profile_from_b(data) + await _sync_user_profile_from_b(data, profile_key) 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 "" - - -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) +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") @@ -410,15 +558,18 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str], backgr cipher_b64 = "".join(str(cipher_b64).split()).strip() try: - decrypted = aes_decrypt_base64(cipher_b64, ENCODING_AES_KEY) + decrypted = aes_decrypt_base64(cipher_b64, encoding_key) logger.info( - "chatbot decrypted preview app_id=%s preview=%s", + "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 app_id=%s format=%s user=%s query=%s", + "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], @@ -426,7 +577,8 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str], backgr ) except Exception as error: logger.error( - "chatbot decrypt failed app_id=%s content_type=%s body_prefix=%s err=%s", + "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], @@ -434,35 +586,67 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str], backgr ) 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) + reply = shared_resolve_reply(data, profile_key) + background_tasks.add_task(_safe_sync_user_profile_from_b, data, profile_key) - 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 + 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: - resp_plain = json.dumps({"answer_type": "text", "text_info": {"short_answer": answer_text}}, ensure_ascii=False) # 与原始 chatbot 服务保持一致：thirdapi（URL 带 app_id）回包使用纯 AES，不拼 app_id - response_app_id = None if app_id else APP_ID + response_app_id = None if app_id else expected_app_id - return aes_encrypt_base64(resp_plain, ENCODING_AES_KEY, response_app_id) + return aes_encrypt_base64(resp_plain, encoding_key, response_app_id) -@router.post("/", response_class=PlainTextResponse) +@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) + return await _handle_wechat_request(request, app_id, background_tasks, profile_key=ACTIVE_PROFILE) -@router.post("/wechat/thirdapi", response_class=PlainTextResponse) +@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) + return await _handle_wechat_request(request, app_id, background_tasks, profile_key=ACTIVE_PROFILE) -@router.get("/health") +@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 @@ -514,7 +698,8 @@ def _client_ip_for_like(request: Request) -> str: return "" -@router.post("/like") +@router.post("/chatbot/like") +@hackrobot_router.post("/hackrobot/like") async def like_webhook_sink(request: Request): """接收 SmsForwarder 等 Webhook POST，详细打日志；公众号助手通知写入 PocketBase。""" body = await request.body() @@ -538,7 +723,8 @@ async def like_webhook_sink(request: Request): return JSONResponse(status_code=200, content=payload) -@router.post("/api/private-message/query") +@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() @@ -570,7 +756,8 @@ async def chatbot_private_message_query(payload: dict): } -@router.post("/api/private-message/analyze") +@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: @@ -605,9 +792,8 @@ async def chatbot_private_message_analyze(payload: dict): } -@router.get("/private-message", response_class=HTMLResponse) -def chatbot_private_message_page(): - return """ +def _private_message_debug_html(api_prefix: str) -> str: + base = """ @@ -718,7 +904,7 @@ def chatbot_private_message_page(): page: Number($("page").value || 0), size: Number($("size").value || 30), }; - const res = await fetch("/chatbot/api/private-message/query", { + const res = await fetch("__API_PREFIX__/api/private-message/query", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(payload), @@ -737,7 +923,7 @@ def chatbot_private_message_page(): try{ setStatus("分析中..."); const payload = { authtoken: $("authtoken").value.trim() }; - const res = await fetch("/chatbot/api/private-message/analyze", { + const res = await fetch("__API_PREFIX__/api/private-message/analyze", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(payload), @@ -757,4 +943,14 @@ def chatbot_private_message_page(): """ + 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))