diff --git a/config.py b/config.py index a0780ab..7138423 100644 --- a/config.py +++ b/config.py @@ -1,3 +1,16 @@ APP_ID = "fXYWvGfq04uMScP" TOKEN = "EWRotyMHfTLAlYmCHpdH8AZcuZPBHn" ENCODING_AES_KEY = "1wJZiJm8URfRjkHu1QfIVDeNFc2jbzzeCZqvhQecsCx" + +# PocketBase（默认直连本机；与 payjsapi 项目保持同风格） +PB_URL = "http://127.0.0.1:8090" +PB_ADMIN_EMAIL = "xiaoshuang.eric@gmail.com" +PB_ADMIN_PASSWORD = "Xiao4669805@" +PB_COLLECTION = "wechat_private_users" + +# A 回调触发 B 接口同步的默认配置 +PM_SYNC_AUTHTOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyaWQiOjE5MzE1MjcsIm9wZW5pZCI6Im85VS04NW1DSXZ0eDZ0a0pteTQyMkkzcUNHVWciLCJzaWduZXRpbWUiOjE3NzQ5MzM0NDQ4MTgsImlhdCI6MTc3NDkzMzQ0NCwiZXhwIjoxNzc3NTI1NDQ0fQ.iC9vdjBqa2LGlqA6F-Dg4GcfDyv3Rp8JwpBaO3j_eE4" +PM_SYNC_DEFAULT_WXBOT_BID = "4888c6ab5cb70dcc3adea1e2d2ff0201" +PM_SYNC_BID_BY_CHANNEL = { + "wx5cee92c2c93fb480": "4888c6ab5cb70dcc3adea1e2d2ff0201", +} diff --git a/main.py b/main.py index 597f8c4..f070e8b 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,7 @@ import json import logging import base64 +import time from datetime import datetime, timezone from typing import Optional @@ -8,7 +9,18 @@ import httpx from fastapi import FastAPI, Request, HTTPException from fastapi.responses import PlainTextResponse, HTMLResponse -from config import APP_ID, TOKEN, ENCODING_AES_KEY +from config import ( + APP_ID, + TOKEN, + ENCODING_AES_KEY, + PB_URL, + PB_ADMIN_EMAIL, + PB_ADMIN_PASSWORD, + PB_COLLECTION, + PM_SYNC_AUTHTOKEN, + PM_SYNC_DEFAULT_WXBOT_BID, + PM_SYNC_BID_BY_CHANNEL, +) from crypto_utils import aes_decrypt_base64, aes_encrypt_base64, calc_signature @@ -17,6 +29,294 @@ logger = logging.getLogger(__name__) app = FastAPI(title="WeChat Dialog ThirdAPI Demo") +_processed_callback_cache: dict[str, int] = {} + + +def _build_keyword_h5_reply(query: str, user_id: str) -> Optional[str]: + """ + 按关键词返回 H5 卡片（answer/short_answer 需为 JSON 字符串）。 + """ + q = (query or "").strip() + if not q: + return None + + if "电子书" in q: + msg = { + "news": { + "articles": [ + { + "title": "数字游民", + "description": "地理套利与自动化杠杆", + "url": f"https://www.hackrobot.cn?userid={user_id}", + "picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png", + "type": "h5", + } + ] + } + } + return json.dumps(msg, ensure_ascii=False) + + if "群" in q: + msg = { + "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 json.dumps(msg, ensure_ascii=False) + + return None + + +def _safe_filter_value(value: str) -> str: + return (value or "").replace("\\", "\\\\").replace('"', '\\"') + + +def _to_int(value) -> Optional[int]: + try: + return int(str(value).strip()) + except Exception: + return None + + +def _normalize_text(value) -> str: + return str(value or "").strip() + + +def _build_callback_dedupe_key(data: dict) -> str: + userid = _normalize_text(data.get("UserId") or data.get("userId")) + msg_id = _normalize_text(data.get("MsgId") or data.get("MessageId") or data.get("msgId")) + ts = _normalize_text(data.get("Timestamp") or data.get("MessageTime") or data.get("msgTime")) + query = _normalize_text(data.get("Query")) + return "|".join([userid, msg_id, ts, query]) + + +def _is_duplicate_callback(data: dict, ttl_seconds: int = 3600) -> bool: + """ + 回调幂等（进程内）：同 userid/msg_id/time/query 在 TTL 内仅处理一次同步逻辑。 + 不影响主回复。 + """ + now_ts = int(time.time()) + # 清理过期 key，避免缓存无限增长 + expired = [k for k, exp in _processed_callback_cache.items() if exp <= now_ts] + for k in expired: + _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 + + +def _extract_b_last_msg(contact: dict) -> dict: + return (contact.get("lastMsg") or {}) if isinstance(contact, dict) else {} + + +def _pick_best_contact_match(contacts: list, target_userid: str, source_query: str, source_msg_id: str, source_msg_time) -> Optional[dict]: + """ + 在同 userid 的候选中做二次匹配： + - msg_id 一致优先 + - 文本一致次优 + - 时间最接近再次优先 + """ + if not target_userid: + return None + + same_user = [ + c for c in contacts + if _normalize_text(c.get("userId") or c.get("userid")) == target_userid + ] + if not same_user: + return None + + query_norm = _normalize_text(source_query) + source_msg_id_norm = _normalize_text(source_msg_id) + src_ts = _to_int(source_msg_time) + + def score(contact: dict) -> tuple[int, int, int]: + last_msg = _extract_b_last_msg(contact) + last_msg_id = _normalize_text(last_msg.get("msgId") or last_msg.get("id")) + last_content = _normalize_text(last_msg.get("content")) + last_active = _to_int(contact.get("lastActiveTime")) + + msg_id_hit = 1 if source_msg_id_norm and last_msg_id and source_msg_id_norm == last_msg_id else 0 + content_hit = 1 if query_norm and last_content and query_norm == last_content else 0 + # 时间越接近分越高（负值用于降序） + if src_ts is None or last_active is None: + time_delta = 10**12 + else: + time_delta = abs(src_ts - last_active) + return (msg_id_hit, content_hit, -time_delta) + + return sorted(same_user, key=score, reverse=True)[0] + + +async def _pb_upsert_user_profile(record: dict) -> None: + """ + 使用 PocketBase Admin API upsert 用户资料。 + 需要环境变量： + - PB_URL + - PB_ADMIN_EMAIL + - PB_ADMIN_PASSWORD + 可选： + - PB_COLLECTION（默认 wechat_private_users） + """ + pb_url = (PB_URL or "").rstrip("/") + pb_admin_email = PB_ADMIN_EMAIL or "" + pb_admin_password = PB_ADMIN_PASSWORD or "" + pb_collection = PB_COLLECTION or "wechat_private_users" + if not pb_url or not pb_admin_email or not pb_admin_password: + logger.warning("PocketBase 配置不完整，跳过资料同步") + return + + userid = str(record.get("userid", "")).strip() + channel_id = str(record.get("channel_id", "")).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}, + ) + auth_resp.raise_for_status() + token = (auth_resp.json() or {}).get("token", "") + if not token: + logger.warning("PocketBase 登录无 token，跳过资料同步") + return + + headers = {"Authorization": f"Bearer {token}"} + user_expr = _safe_filter_value(userid) + chan_expr = _safe_filter_value(channel_id) + filter_exp = f'userid="{user_expr}" && channel_id="{chan_expr}"' if channel_id else f'userid="{user_expr}"' + list_resp = await client.get( + f"{pb_url}/api/collections/{pb_collection}/records", + params={"page": 1, "perPage": 1, "filter": filter_exp, "sort": "-updated"}, + headers=headers, + ) + list_resp.raise_for_status() + items = ((list_resp.json() or {}).get("items")) or [] + + if items: + rid = items[0].get("id") + if rid: + patch_resp = await client.patch( + f"{pb_url}/api/collections/{pb_collection}/records/{rid}", + json=record, + headers=headers, + ) + patch_resp.raise_for_status() + return + + create_resp = await client.post( + f"{pb_url}/api/collections/{pb_collection}/records", + json=record, + headers=headers, + ) + create_resp.raise_for_status() + + +async def _sync_user_profile_from_b(data: dict) -> None: + """ + A 接口收到消息后，立即调用 B 接口拉取会话并同步用户资料到 PocketBase。 + 环境变量： + - PM_SYNC_AUTHTOKEN（必填） + - PM_SYNC_DEFAULT_WXBOT_BID（可选） + - PM_SYNC_BID_BY_CHANNEL_JSON（可选，形如 {"wx5cee...":"4888..."}） + """ + authtoken = (PM_SYNC_AUTHTOKEN or "").strip() + if not authtoken: + logger.info("未配置 PM_SYNC_AUTHTOKEN，跳过 B 接口同步") + return + + channel_id = str(data.get("ChannelId", "") or data.get("channelId", "")).strip() + bid_map = PM_SYNC_BID_BY_CHANNEL if isinstance(PM_SYNC_BID_BY_CHANNEL, dict) else {} + bid_map = {str(k): str(v) for k, v in bid_map.items()} + + wxbot_bid = bid_map.get(channel_id) or (PM_SYNC_DEFAULT_WXBOT_BID or "").strip() + if not wxbot_bid: + logger.warning("未配置可用 wxbot_bid，跳过 B 接口同步") + return + + target_userid = str(data.get("UserId", "") or data.get("userId", "")).strip() + if not target_userid: + return + + if _is_duplicate_callback(data): + logger.info("命中回调幂等缓存，跳过重复同步：userid=%s", target_userid) + return + + msg_id = str(data.get("MsgId", "") or data.get("MessageId", "") or data.get("msgId", "")).strip() + msg_time = data.get("Timestamp", "") or data.get("MessageTime", "") or data.get("msgTime", "") + query = str(data.get("Query", "")).strip() + + b_resp = await query_private_messages( + authtoken=authtoken, + wxbot_bid=wxbot_bid, + page=0, + size=100, + filter_value=0, + request_id=f"thirdapi-{target_userid}", + ) + contacts = b_resp.get("contacts") or [] + target = _pick_best_contact_match( + contacts=contacts, + target_userid=target_userid, + source_query=query, + source_msg_id=msg_id, + source_msg_time=msg_time, + ) + if not target: + logger.info("B 接口未匹配到用户：%s", target_userid) + return + + last_msg = target.get("lastMsg") or {} + record = { + "userid": target_userid, + "channel_id": channel_id, + "nick": target.get("nick", ""), + "avatar": target.get("headImg") or target.get("avatar") or target.get("headimgurl") or target.get("headImgUrl") or "", + "source_query": query, + "source_message_id": msg_id, + "source_message_time": str(msg_time), + "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), + } + await _pb_upsert_user_profile(record) + + +@app.post("/debug/sync-test") +async def debug_sync_test(payload: dict): + """ + 手动调试 A->B->PocketBase 同步链路。 + body 示例： + { + "UserId": "...", + "ChannelId": "...", + "Query": "电子书", + "MsgId": "...", + "Timestamp": 1774933444 + } + """ + await _sync_user_profile_from_b(payload or {}) + return { + "ok": True, + "message": "sync triggered", + "dedupe_key": _build_callback_dedupe_key(payload or {}), + } + async def handle_business(data: dict) -> str: """ @@ -39,22 +339,10 @@ async def handle_business(data: dict) -> str: # ] # return "；".join(parts) - # 当前示例：返回一个 H5 富文本结构（平台会对 short_answer 做 JSON.parse） - h5_msg = { - "news": { - "articles": [ - { - "title": "实时更新：新型肺炎疫情最新动态", - "description": "腾讯新闻第一时间同步全国新型肺炎疫情动态，欢迎关注、转发", - "url": "https://news.qq.com/zt2020/page/feiyan.htm", - "picurl": "http://mmbiz.qpic.cn/mmbiz_jpg/W3gQtpV3j8D8kZRqfpTJlfVqubwgFQf47H0GWlGV6leaDF80ZpdtuFhQVsCsM3YKmwkujXzdjR2k6aWfA41ic7Q/0?wx_fmt=jpeg", - "type": "h5", - } - ] - } - } - # 按文档要求，answer/short_answer 是字符串，内部再 JSON.parse 才是对象 - return json.dumps(h5_msg, ensure_ascii=False) + keyword_reply = _build_keyword_h5_reply(query, user_id) + if keyword_reply: + return keyword_reply + return "已收到你的消息，输入“电子书”或“群”可获取对应卡片。" def _decode_jwt_payload_without_verify(token: str) -> dict: @@ -187,6 +475,13 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str]) -> str logger.exception("签名校验异常（忽略或按需处理）") # 7. 根据业务逻辑构造应答 + # 7.1 先做用户资料同步（异常不影响主回复） + try: + await _sync_user_profile_from_b(data) + except Exception: + logger.exception("触发 B 接口同步到 PocketBase 失败（不影响回复）") + + # 7.2 再生成业务回复 try: answer_text = await handle_business(data) except Exception: @@ -314,7 +609,7 @@ def private_message_page(): color: #cbd5e1; font-size: 13px; } - textarea, input { + textarea, input, select { width: 100%; border: 1px solid rgba(148, 163, 184, 0.35); background: rgba(15, 23, 42, 0.75); @@ -325,7 +620,7 @@ def private_message_page(): outline: none; transition: border-color .2s, box-shadow .2s; } - textarea:focus, input:focus { + textarea:focus, input:focus, select:focus { border-color: var(--primary); box-shadow: 0 0 0 3px rgba(96, 165, 250, 0.15); } @@ -362,6 +657,87 @@ def private_message_page(): color: var(--ok); } .status.warn { color: var(--warn); } + .panel-title { + margin: 0 0 8px; + font-size: 15px; + color: #dbeafe; + } + .token-grid { + display: grid; + grid-template-columns: 1fr 1fr; + gap: 12px; + } + .token-kpi { + border: 1px solid rgba(148, 163, 184, 0.25); + border-radius: 12px; + background: rgba(15, 23, 42, 0.55); + padding: 10px 12px; + } + .token-kpi .label { + color: var(--muted); + font-size: 12px; + margin-bottom: 4px; + } + .token-kpi .value { + font-size: 20px; + font-weight: 700; + color: #bfdbfe; + } + .user-list { + display: grid; + grid-template-columns: repeat(auto-fill, minmax(250px, 1fr)); + gap: 10px; + margin-top: 6px; + } + .user-item { + display: flex; + align-items: center; + gap: 10px; + border: 1px solid rgba(148, 163, 184, 0.2); + border-radius: 12px; + background: rgba(15, 23, 42, 0.55); + padding: 10px; + } + .avatar { + width: 36px; + height: 36px; + border-radius: 999px; + object-fit: cover; + background: rgba(148, 163, 184, 0.2); + border: 1px solid rgba(148, 163, 184, 0.35); + } + .avatar-placeholder { + width: 36px; + height: 36px; + border-radius: 999px; + display: inline-flex; + align-items: center; + justify-content: center; + background: rgba(99, 102, 241, 0.35); + color: #e0e7ff; + font-size: 14px; + font-weight: 700; + border: 1px solid rgba(148, 163, 184, 0.35); + } + .user-meta { + min-width: 0; + display: grid; + gap: 2px; + } + .user-nick { + font-size: 14px; + color: #f1f5f9; + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; + } + .user-id { + font-size: 12px; + color: #94a3b8; + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; + } pre { margin: 0; background: #090d18; @@ -375,6 +751,7 @@ def private_message_page(): } @media (max-width: 768px) { .row { grid-template-columns: 1fr; } + .token-grid { grid-template-columns: 1fr; } } @@ -383,6 +760,8 @@ def private_message_page():

私信查询调试台

用于调试微信对话开放平台私信列表接口，支持一键查询、Token 过期分析与结果可视化。

+ + @@ -406,6 +785,20 @@ def private_message_page():

等待操作...

+

Token 过期信息

+
+
+
状态
+
-
+
+
+
剩余时间
+
-
+
+
+

用户列表（userid / 头像 / 昵称）

+
+

原始返回

{
   "message": "点击上方按钮开始查询"
 }
@@ -413,6 +806,47 @@ def private_message_page(): @@ -522,6 +1002,16 @@ async def api_private_message_query(payload: dict): 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 + ], "contacts_preview": [ { "nick": c.get("nick"),