's'

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"

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 = (
+            "<xml>"
+            f"<appid><![CDATA[{escape(appid)}]]></appid>"
+            f"<openid><![CDATA[{escape(user_id)}]]></openid>"
+            f"<msg><![CDATA[{msg_text}]]></msg>"
+            f"<channel>{channel}</channel>"
+            "</xml>"
+        )
+        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 结构字符串