diff --git a/__pycache__/config.cpython-312.pyc b/__pycache__/config.cpython-312.pyc index a9ec32d..4305abd 100644 Binary files a/__pycache__/config.cpython-312.pyc and b/__pycache__/config.cpython-312.pyc differ diff --git a/__pycache__/crypto_utils.cpython-312.pyc b/__pycache__/crypto_utils.cpython-312.pyc index cc4d9d8..aa20361 100644 Binary files a/__pycache__/crypto_utils.cpython-312.pyc and b/__pycache__/crypto_utils.cpython-312.pyc differ diff --git a/__pycache__/main.cpython-312.pyc b/__pycache__/main.cpython-312.pyc index 55bbde4..c730073 100644 Binary files a/__pycache__/main.cpython-312.pyc and b/__pycache__/main.cpython-312.pyc differ diff --git a/config.py b/config.py index 23f72ca..a9ea42d 100644 --- a/config.py +++ b/config.py @@ -2,8 +2,22 @@ APP_ID = "fXYWvGfq04uMScP" TOKEN = "EWRotyMHfTLAlYmCHpdH8AZcuZPBHn" ENCODING_AES_KEY = "1wJZiJm8URfRjkHu1QfIVDeNFc2jbzzeCZqvhQecsCx" -# PocketBase（默认直连本机；与 payjsapi 项目保持同风格） -PB_URL = "http://127.0.0.1:8090" +import os + +# PocketBase 自动切换策略： +# - 本地调试（Windows）优先使用数据库域名 +# - 线上（非 Windows）优先使用内网 IP+端口 +# 你也可以通过环境变量 PB_URL 强制覆盖。 +PB_DOMAIN_URL = "https://pocketbase.hackrobot.cn" +PB_INTERNAL_URL = "http://127.0.0.1:8090" +_pb_force = (os.getenv("PB_URL") or "").strip() +if _pb_force: + PB_URL = _pb_force +elif os.name == "nt": + PB_URL = PB_DOMAIN_URL +else: + PB_URL = PB_INTERNAL_URL + PB_ADMIN_EMAIL = "xiaoshuang.eric@gmail.com" PB_ADMIN_PASSWORD = "Xiao4669805@" PB_COLLECTION = "wechat_private_users" diff --git a/main.py b/main.py index 0ea0396..e34e391 100644 --- a/main.py +++ b/main.py @@ -2,6 +2,8 @@ import json import logging import base64 import time +from html import escape +import xml.etree.ElementTree as ET from datetime import datetime, timezone from typing import Optional @@ -31,16 +33,12 @@ 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 字符串）。 - """ +def _build_keyword_h5_obj(query: str, user_id: str) -> Optional[dict]: q = (query or "").strip() if not q: return None - if "电子书" in q: - msg = { + return { "news": { "articles": [ { @@ -53,10 +51,8 @@ def _build_keyword_h5_reply(query: str, user_id: str) -> Optional[str]: ] } } - return json.dumps(msg, ensure_ascii=False) - if "群" in q: - msg = { + return { "news": { "articles": [ { @@ -69,11 +65,128 @@ def _build_keyword_h5_reply(query: str, user_id: str) -> Optional[str]: ] } } - return json.dumps(msg, ensure_ascii=False) - return None +def _parse_decrypted_payload(decrypted_text: str) -> dict: + """ + 同时兼容 JSON 与 XML 的回调明文，统一成业务侧可读字段。 + """ + text = (decrypted_text or "").strip() + if not text: + raise ValueError("empty decrypted payload") + + # 1) 优先 JSON + if text.startswith("{"): + obj = json.loads(text) + if not isinstance(obj, dict): + raise ValueError("invalid json payload") + obj["_payload_format"] = "json" + return obj + + # 2) XML（第三方客服回调常见） + 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 = "" + msg_type = get_text("msgtype", "") + if content_node is not None: + msg_node = content_node.find("msg") + type_node = content_node.find("msgtype") + if msg_node is not None and msg_node.text: + query = msg_node.text.strip() + if type_node is not None and type_node.text: + msg_type = type_node.text.strip() + + return { + # 统一成原业务逻辑使用的字段 + "UserId": get_text("userid", ""), + "Query": query, + "Timestamp": get_text("createtime", ""), + "MsgId": get_text("msgid", ""), + "ChannelId": get_text("appid", ""), + "MessageType": msg_type, + # 保留原始字段便于日志与后续扩展 + "appid": get_text("appid", ""), + "channel": get_text("channel", ""), + "from": get_text("from", ""), + "status": get_text("status", ""), + "kfstate": get_text("kfstate", ""), + "_payload_format": "xml", + } + + raise ValueError("unsupported decrypted payload format") + + +def _build_keyword_h5_reply(query: str, user_id: str) -> Optional[str]: + """ + 按关键词返回 H5 卡片（answer/short_answer 需为 JSON 字符串）。 + """ + msg = _build_keyword_h5_obj(query, user_id) + return json.dumps(msg, ensure_ascii=False) if msg else None + + +async def _send_kefu_msg_for_xml_callback(data: dict) -> None: + """ + XML 回调渠道下，主动调用 sendmsg 下发消息（回调响应体不会直接展示给用户）。 + """ + query = str(data.get("Query", "")).strip() + 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(query, user_id) + if not msg: + return + + url = f"https://chatbot.weixin.qq.com/openapi/sendmsg/{TOKEN}" + body = { + "appid": appid, + "openid": user_id, + "msg": msg, + "channel": channel, + } + async with httpx.AsyncClient(timeout=10, verify=False) as client: + # 1) 先尝试明文 JSON 发送（多数场景可直接工作） + resp = await client.post(url, json=body, headers={"content-type": "application/json"}) + resp.raise_for_status() + payload = resp.json() if resp.content else {} + errcode = int(payload.get("errcode", -1)) if isinstance(payload, dict) else -1 + if errcode == 0: + logger.info("关键词命中，sendmsg 明文发送成功") + return + + logger.warning("sendmsg 明文返回非成功，准备加密重试：%s", payload) + + # 2) 按文档加密格式重试：body = {"encrypt": "..."} + msg_text = json.dumps(msg, ensure_ascii=False) + xml_data = ( + "" + f"" + f"" + f"" + f"{channel}" + "" + ) + encrypted = aes_encrypt_base64(xml_data, ENCODING_AES_KEY, APP_ID) + enc_resp = await client.post( + url, + json={"encrypt": encrypted}, + headers={"content-type": "application/json"}, + ) + enc_resp.raise_for_status() + enc_payload = enc_resp.json() if enc_resp.content else {} + enc_errcode = int(enc_payload.get("errcode", -1)) if isinstance(enc_payload, dict) else -1 + if enc_errcode == 0: + logger.info("关键词命中，sendmsg 加密发送成功") + return + logger.error("sendmsg 加密发送失败：%s", enc_payload) + + def _safe_filter_value(value: str) -> str: return (value or "").replace("\\", "\\\\").replace('"', '\\"') @@ -183,46 +296,55 @@ async def _pb_upsert_user_profile(record: dict) -> None: 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, + try: + 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}, + ) + # 兼容新旧 PocketBase：旧版本 admins，新版本 _superusers + 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}, ) - patch_resp.raise_for_status() + auth_resp.raise_for_status() + token = (auth_resp.json() or {}).get("token", "") + if not token: + logger.warning("PocketBase 登录无 token，跳过资料同步") return - create_resp = await client.post( - f"{pb_url}/api/collections/{pb_collection}/records", - json=record, - headers=headers, - ) - create_resp.raise_for_status() + 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() + except httpx.HTTPError as e: + logger.error("PocketBase 同步失败：%s", str(e)) async def _sync_user_profile_from_b(data: dict) -> None: @@ -294,6 +416,16 @@ async def _sync_user_profile_from_b(data: dict) -> None: await _pb_upsert_user_profile(record) +async def _safe_sync_user_profile_from_b(data: dict) -> None: + """ + 后台任务安全包装：B/PocketBase 失败只记日志，不抛异常影响 ASGI。 + """ + try: + await _sync_user_profile_from_b(data) + except Exception: + logger.error("后台同步失败（已忽略，不影响A接口主回复）") + + @app.post("/debug/sync-test") async def debug_sync_test(payload: dict): """ @@ -431,12 +563,12 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str], backgr ) raise HTTPException(status_code=400, detail="decrypt error") - # 4. 解析 JSON + # 4. 解析明文（兼容 JSON/XML） try: - data = json.loads(decrypted_json_str) - except json.JSONDecodeError: - logger.error(f"解密结果不是合法 JSON: {decrypted_json_str}") - raise HTTPException(status_code=400, detail="invalid json in decrypted body") + data = _parse_decrypted_payload(decrypted_json_str) + except Exception: + logger.error(f"解密结果无法解析为 JSON/XML: {decrypted_json_str}") + raise HTTPException(status_code=400, detail="invalid decrypted payload") # 5. 打印请求内容到终端日志（分析、显示，带中文注释） logger.info("收到微信第三方服务请求(已解密)：") @@ -478,8 +610,15 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str], backgr logger.exception("业务处理异常，将返回兜底文案") answer_text = "系统繁忙，请稍后再试。" + # XML 回调场景需要主动 sendmsg 才能给用户展示消息 + if str(data.get("_payload_format", "")) == "xml": + try: + await _send_kefu_msg_for_xml_callback(data) + except Exception: + logger.error("sendmsg 下发失败（不影响回调200）") + # 7.1 主回复确定后再后台触发 B 同步，避免影响回复链路耗时 - background_tasks.add_task(_sync_user_profile_from_b, data) + background_tasks.add_task(_safe_sync_user_profile_from_b, data) resp_obj = { # 文本类型，内容为可被 JSON.parse 的 H5 结构字符串