From 1c5b46a933fa27f5cce206417ea659e522bbf63a Mon Sep 17 00:00:00 2001 From: eric Date: Thu, 2 Apr 2026 01:04:08 -0500 Subject: [PATCH] 's' --- app/routers/chatbot.py | 321 ++++++++++++------------- app/services/wechat_private_user_pb.py | 232 ++++++++++++++++++ 2 files changed, 386 insertions(+), 167 deletions(-) create mode 100644 app/services/wechat_private_user_pb.py diff --git a/app/routers/chatbot.py b/app/routers/chatbot.py index 197daaa..b01e495 100644 --- a/app/routers/chatbot.py +++ b/app/routers/chatbot.py @@ -4,7 +4,6 @@ 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 @@ -15,14 +14,16 @@ 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 +from ..services.wechat_private_user_pb import ( + contact_avatar_url, + query_private_messages, + safe_sync_wechat_private_user_to_pb, +) router = APIRouter(tags=["chatbot"]) logger = logging.getLogger(__name__) -_processed_callback_cache: dict[str, int] = {} - OPEN_API_PROFILES = { "eric": { @@ -56,11 +57,6 @@ def _resolve_profile_key(profile_key: str) -> str: 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: @@ -121,7 +117,7 @@ class SharedChatbotReply(NamedTuple): """全公众号配置共用的解析结果：要么 H5 图文，要么兜底纯文本（含超链）。""" h5_message: Optional[dict] - """与「发送客服消息」H5 一致：{\"news\": {\"articles\": [...]}}，未命中关键词时为 None。""" + """与「发送客服消息」H5 一致：{\"news\": {\"articles\": [...]}}；关键词为 Query 子串包含命中，未命中时为 None。""" text_fallback: str """未命中关键词时的 short_answer 文案（可含 ）。""" @@ -155,7 +151,11 @@ def _normalize_h5_msg_per_wechat_doc(h5_message: dict) -> dict: def shared_build_h5_keyword_message(query: str, user_id: str, profile_key: str) -> Optional[dict]: - """关键词命中时返回文档约定的 H5 msg（再经 _normalize 输出）。异度(/hackrobot) 群卡片仅认「群」；Eric(/chatbot) 另认「加群」。""" + """关键词命中时返回文档约定的 H5 msg（再经 _normalize 输出）。 + + 命中规则：对用户 Query 首尾 trim 后，**包含**配置词作为连续子串即命中（子串匹配）， + 不要求用户整句与关键词全等。异度(/hackrobot) 群卡片仅认「群」；Eric(/chatbot) 另认「加群」。 + """ q = (query or "").strip() uid = (user_id or "").strip() prof = _resolve_profile_key(profile_key) @@ -206,7 +206,7 @@ def build_fallback_reply_plain_text(user_id: str) -> str: def shared_resolve_reply(data: dict, profile_key: str) -> SharedChatbotReply: - """根据解密后的 data 与 profile 决定 H5 或兜底；无关键词时一律走 build_fallback_reply_plain_text。""" + """根据解密后的 data 与 profile 决定 H5 或兜底；关键词按 Query 子串包含判断，无命中时走 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) @@ -369,151 +369,6 @@ def _parse_decrypted_payload(decrypted_text: str) -> dict: 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], @@ -587,7 +442,7 @@ async def _handle_wechat_request( 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) + background_tasks.add_task(safe_sync_wechat_private_user_to_pb, 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) @@ -746,7 +601,7 @@ async def chatbot_private_message_query(payload: dict): { "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"), + "avatar": contact_avatar_url(c) or None, "lastMsg": (c.get("lastMsg") or {}).get("content"), "lastActiveTime": c.get("lastActiveTime"), } @@ -813,8 +668,16 @@ def _private_message_debug_html(api_prefix: str) -> str: 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; } + #token-countdown-wrap { margin-top: 12px; display: none; } + #token-countdown-wrap.active { display: block; } + .cd-row { display: flex; justify-content: space-between; align-items: center; font-size: 12px; color: #94a3b8; gap: 8px; } + .cd-bar { height: 8px; background: #1e293b; border-radius: 4px; overflow: hidden; margin-top: 8px; } + .cd-bar-fill { height: 100%; width: 100%; border-radius: 4px; transition: width 0.35s linear, background 0.25s; background: #22c55e; } #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; } + .u { border:1px solid #334155; border-radius:8px; padding:8px; background:#0b1220; display:flex; gap:10px; align-items:flex-start; } + .u .ava { width:40px; height:40px; border-radius:8px; object-fit:cover; flex-shrink:0; background:#1e293b; } + .u .ava--ph { display:flex; align-items:center; justify-content:center; font-size:10px; color:#64748b; text-align:center; line-height:1.2; padding:2px; box-sizing:border-box; } + .u .meta { min-width:0; flex:1; } .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; } @@ -824,6 +687,13 @@ def _private_message_debug_html(api_prefix: str) -> str:

Chatbot 私信调试台

准备就绪
+
+
+ Token 剩余 + -- +
+
+