's'

app/main.py

@@ -30,6 +30,7 @@ from .routers import (
     legacy_router,
     common_router,
     meetup_router,
+    chatbot_router,
 )
 from .routers.salon_meetup import router as salon_meetup_router
 
@@ -175,6 +176,7 @@ app.include_router(legacy_router)
 app.include_router(common_router)
 app.include_router(meetup_router)
 app.include_router(salon_meetup_router)
+app.include_router(chatbot_router)
 
 
 @app.get("/")

app/routers/__init__.py

@@ -13,6 +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
 
 __all__ = [
     "nomadvip_router",
@@ -22,4 +23,5 @@ __all__ = [
     "legacy_router",
     "common_router",
     "meetup_router",
+    "chatbot_router",
 ]

app/routers/chatbot.py

@@ -0,0 +1,448 @@
+import base64
+import json
+import logging
+import os
+import struct
+import time
+import xml.etree.ElementTree as ET
+from datetime import datetime, timezone
+from typing import Optional
+
+import httpx
+from Crypto.Cipher import AES
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Request
+from fastapi.responses import HTMLResponse, PlainTextResponse
+
+from ..config import PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL
+
+router = APIRouter(prefix="/chatbot", tags=["chatbot"])
+
+logger = logging.getLogger(__name__)
+_processed_callback_cache: dict[str, int] = {}
+
+
+OPEN_API_PROFILES = {
+    "eric": {
+        "name": "Eric在旅行",
+        "APP_ID": "fXYWvGfq04uMScP",
+        "TOKEN": "EWRotyMHfTLAlYmCHpdH8AZcuZPBHn",
+        "ENCODING_AES_KEY": "1wJZiJm8URfRjkHu1QfIVDeNFc2jbzzeCZqvhQecsCx",
+    },
+    "xidu": {
+        "name": "异度世界",
+        "APP_ID": "LTKfSFeK0h8TRLM",
+        "TOKEN": "64DbvAd5m7YOJ1XslyMgJ9V42jDIUH",
+        "ENCODING_AES_KEY": "KCZGtPDjqFZnBuApvG3xqhvHUMnqFHbC5cnD23HKACX",
+    },
+}
+ACTIVE_PROFILE = (os.getenv("CHATBOT_ACTIVE_PROFILE") or "eric").strip().lower()
+if ACTIVE_PROFILE not in OPEN_API_PROFILES:
+    ACTIVE_PROFILE = "eric"
+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"]
+
+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:
+        return data
+    return data[:-pad]
+
+
+def pkcs5_padding(data: bytes) -> bytes:
+    amount_to_pad = 32 - (len(data) % 32)
+    return data + bytes([amount_to_pad]) * amount_to_pad
+
+
+def get_aes_key_and_iv(encoding_aes_key: str) -> tuple[bytes, bytes]:
+    aes_key = base64.b64decode(encoding_aes_key + "=")
+    if len(aes_key) != 32:
+        raise ValueError("encodingAESKey invalid")
+    return aes_key, aes_key[:16]
+
+
+def aes_decrypt_base64(cipher_b64: str, encoding_aes_key: str) -> str:
+    aes_key, iv = get_aes_key_and_iv(encoding_aes_key)
+    compact = "".join(str(cipher_b64 or "").split())
+    compact += "=" * (-len(compact) % 4)
+    cipher_bytes = base64.b64decode(compact)
+    cipher = AES.new(aes_key, AES.MODE_CBC, iv)
+    decrypted = pkcs5_unpadding(cipher.decrypt(cipher_bytes))
+    try:
+        content = decrypted[16:]
+        msg_len = struct.unpack("!I", content[:4])[0]
+        msg = content[4 : 4 + msg_len]
+        text = msg.decode("utf-8")
+        if text:
+            return text
+    except Exception:
+        pass
+    return decrypted.decode("utf-8")
+
+
+def aes_encrypt_base64(plaintext: str, encoding_aes_key: str, app_id: str | None = None) -> str:
+    aes_key, iv = get_aes_key_and_iv(encoding_aes_key)
+    raw = plaintext.encode("utf-8")
+    if app_id:
+        data = os.urandom(16) + struct.pack("!I", len(raw)) + raw + app_id.encode("utf-8")
+    else:
+        data = raw
+    padded = pkcs5_padding(data)
+    cipher = AES.new(aes_key, AES.MODE_CBC, iv)
+    return base64.b64encode(cipher.encrypt(padded)).decode("utf-8")
+
+
+def _build_keyword_h5_obj(query: str, user_id: str) -> Optional[dict]:
+    q = (query or "").strip()
+    if "电子书" in q:
+        return {
+            "news": {
+                "articles": [
+                    {
+                        "title": "数字游民",
+                        "description": "地理套利与自动化杠杆",
+                        "url": f"https://vip.hackrobot.cn/ebook?userid={user_id}",
+                        "picurl": "https://www.hackrobot.cn/static/images/images/dgnomad.png",
+                        "type": "h5",
+                    }
+                ]
+            }
+        }
+    if "群" in q:
+        return {
+            "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 None
+
+
+def _parse_decrypted_payload(decrypted_text: str) -> dict:
+    text = (decrypted_text or "").strip()
+    if text.startswith("{"):
+        obj = json.loads(text)
+        obj["_payload_format"] = "json"
+        return obj
+    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 = ""
+        if content_node is not None:
+            msg_node = content_node.find("msg")
+            if msg_node is not None and msg_node.text:
+                query = msg_node.text.strip()
+        return {
+            "UserId": get_text("userid", ""),
+            "Query": query,
+            "Timestamp": get_text("createtime", ""),
+            "MsgId": get_text("msgid", ""),
+            "ChannelId": get_text("appid", ""),
+            "appid": get_text("appid", ""),
+            "channel": get_text("channel", "0"),
+            "from": get_text("from", "0"),
+            "_payload_format": "xml",
+        }
+    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('"', '\\"')
+
+
+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, 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)
+    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) -> 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):
+        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) -> None:
+    try:
+        await _sync_user_profile_from_b(data)
+    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 "1111111"
+
+
+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)
+    raw_body = (await request.body()).decode("utf-8").strip()
+    if not raw_body:
+        raise HTTPException(status_code=400, detail="empty body")
+    cipher_b64 = raw_body
+    if "encrypted" in raw_body:
+        try:
+            outer = json.loads(raw_body)
+            if isinstance(outer, dict) and "encrypted" in outer:
+                cipher_b64 = "".join(str(outer["encrypted"]).split()).strip()
+        except Exception:
+            pass
+    try:
+        decrypted = aes_decrypt_base64(cipher_b64, ENCODING_AES_KEY)
+        data = _parse_decrypted_payload(decrypted)
+    except Exception:
+        logger.error("chatbot decrypt failed")
+        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)
+
+    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
+    else:
+        resp_plain = json.dumps({"answer_type": "text", "text_info": {"short_answer": answer_text}}, ensure_ascii=False)
+        response_app_id = None if app_id else APP_ID
+
+    return aes_encrypt_base64(resp_plain, ENCODING_AES_KEY, response_app_id)
+
+
+@router.post("/", 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)
+
+
+@router.post("/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)
+
+
+@router.get("/health")
+def chatbot_health():
+    return {"status": "ok", "module": "chatbot-clone", "profile": ACTIVE_PROFILE}
+
+
+@router.post("/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()
+    if not authtoken or not wxbot_bid:
+        raise HTTPException(status_code=400, detail="authtoken 和 wxbot_bid 不能为空")
+    result = await query_private_messages(
+        authtoken=authtoken,
+        wxbot_bid=wxbot_bid,
+        page=int(payload.get("page", 0)),
+        size=int(payload.get("size", 30)),
+        filter_value=int(payload.get("filter", 0)),
+        request_id=str(payload.get("requestid", "local-debug")).strip() or "local-debug",
+    )
+    contacts = result.get("contacts") or []
+    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
+        ],
+        "raw": result,
+    }
+
+
+@router.post("/api/private-message/analyze")
+async def chatbot_private_message_analyze(payload: dict):
+    authtoken = str(payload.get("authtoken", "")).strip()
+    if not authtoken:
+        raise HTTPException(status_code=400, detail="authtoken 不能为空")
+    try:
+        parts = authtoken.split(".")
+        payload_part = parts[1] if len(parts) == 3 else ""
+        payload_part += "=" * (-len(payload_part) % 4)
+        claims = json.loads(base64.urlsafe_b64decode(payload_part.encode("utf-8")).decode("utf-8")) if payload_part else {}
+    except Exception:
+        claims = {}
+    now_ts = int(datetime.now(timezone.utc).timestamp())
+    exp = int(claims.get("exp", 0)) if str(claims.get("exp", "")).isdigit() else 0
+    expires_in_seconds = exp - now_ts if exp else None
+    status = "无法解析过期时间"
+    if isinstance(expires_in_seconds, int):
+        if expires_in_seconds <= 0:
+            status = "已过期"
+        elif expires_in_seconds < 3600:
+            status = "即将过期（1小时内）"
+        else:
+            status = "有效"
+    return {
+        "status": status,
+        "now_utc": datetime.now(timezone.utc).isoformat(),
+        "claims": claims,
+        "analysis": {
+            "expires_in_seconds": expires_in_seconds,
+            "expires_in_hours": round(expires_in_seconds / 3600, 2) if isinstance(expires_in_seconds, int) else None,
+            "summary": "该分析基于 JWT payload 解码，不包含签名有效性校验。",
+        },
+    }
+
+
+@router.get("/private-message", response_class=HTMLResponse)
+def chatbot_private_message_page():
+    return "<html><body><h3>chatbot clone ready</h3><p>请使用 /chatbot/api/private-message/query 调试接口</p></body></html>"
+

app/routers/nomadvip.py

@@ -363,6 +363,361 @@ async def nomadvip_check_vip(user_id: str = ""):
         _cache_set(cache_key, result, ttl=5)
         return result
 
+
+def _pb_get_user_by_email(admin_token: str, email: str) -> dict | None:
+    base = (PB_URL or "").rstrip("/")
+    safe_email = _pb_escape(email.strip().lower())
+    response = _http_session.get(
+        f"{base}/api/collections/users/records",
+        params={"filter": f'email = "{safe_email}"', "perPage": 1},
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=DEFAULT_HTTP_TIMEOUT,
+    )
+    if response.status_code != 200:
+        return None
+    items = (response.json() or {}).get("items") or []
+    return items[0] if items else None
+
+
+def _pb_get_wechat_private_user(admin_token: str, wechat_userid: str) -> dict | None:
+    base = (PB_URL or "").rstrip("/")
+    safe_uid = _pb_escape(wechat_userid.strip())
+    response = _http_session.get(
+        f"{base}/api/collections/wechat_private_users/records",
+        params={
+            "filter": f'userid = "{safe_uid}"',
+            "perPage": 1,
+            "sort": "-updated",
+        },
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=DEFAULT_HTTP_TIMEOUT,
+    )
+    if response.status_code != 200:
+        return None
+    items = (response.json() or {}).get("items") or []
+    return items[0] if items else None
+
+
+def _pb_get_user_by_id(admin_token: str, user_id: str) -> dict | None:
+    base = (PB_URL or "").rstrip("/")
+    response = _http_session.get(
+        f"{base}/api/collections/users/records/{user_id}",
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=DEFAULT_HTTP_TIMEOUT,
+    )
+    if response.status_code != 200:
+        return None
+    return response.json() or None
+
+
+def _pb_latest_payment_by_user_id(admin_token: str, user_id: str) -> dict | None:
+    base = (PB_URL or "").rstrip("/")
+    safe_user_id = _pb_escape(user_id.strip())
+    response = _http_session.get(
+        f"{base}/api/collections/payments/records",
+        params={
+            "filter": f'user_id = "{safe_user_id}" && type = "vip"',
+            "perPage": 1,
+            "sort": "-created",
+        },
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=DEFAULT_HTTP_TIMEOUT,
+    )
+    if response.status_code != 200:
+        return None
+    items = (response.json() or {}).get("items") or []
+    return items[0] if items else None
+
+
+@router.post("/wechat_profile")
+async def nomadvip_wechat_profile(item: dict):
+    """
+    按 user_id / email / wechat_userid 做关联查询，返回首页可展示的微信头像昵称。
+    """
+    user_id = str(item.get("user_id", "") or "").strip()
+    email = str(item.get("email", "") or "").strip().lower()
+    wechat_userid = str(
+        item.get("wechat_userid", "") or item.get("userid", "") or ""
+    ).strip()
+
+    if not user_id and not email and not wechat_userid:
+        return {
+            "ok": False,
+            "found": False,
+            "error": "missing user_id/email/wechat_userid",
+        }
+
+    admin_token = _get_pb_admin_token()
+    if not admin_token:
+        return {"ok": False, "found": False, "error": "PocketBase admin auth failed"}
+
+    pb_user = _pb_get_user_by_email(admin_token, email) if email else None
+    pb_user_id = (pb_user or {}).get("id", "") if isinstance(pb_user, dict) else ""
+    anonymous_user_id = (
+        user_id if user_id.startswith("user") and user_id[4:].isdigit() else ""
+    )
+
+    has_vip_by_user_id = False
+    if user_id:
+        try:
+            from ..services.pb import get_pb_client
+
+            pb = get_pb_client()
+            has_vip_by_user_id = _uid_has_vip(pb, user_id)
+        except Exception:
+            has_vip_by_user_id = False
+
+    has_vip_by_pb_user = False
+    if pb_user_id:
+        try:
+            from ..services.pb import get_pb_client
+
+            pb = get_pb_client()
+            has_vip_by_pb_user = _uid_has_vip(pb, pb_user_id)
+        except Exception:
+            has_vip_by_pb_user = False
+
+    # 关联补链：当 email + 会员 user_id 能确定时，固化 vip_email_link，打通 payments/users 关系
+    effective_member_user_id = pb_user_id or (
+        user_id if user_id and not anonymous_user_id else ""
+    )
+    if email and effective_member_user_id:
+        try:
+            _pb_upsert_email_link(
+                admin_token,
+                email,
+                effective_member_user_id,
+                anonymous_user_id or None,
+            )
+        except Exception:
+            pass
+
+    link_by_email = _pb_list_email_links(admin_token, email)[0] if email else None
+    link_by_user_id = (
+        _pb_find_email_link_by_user_id(admin_token, user_id)
+        if user_id and not anonymous_user_id
+        else None
+    )
+    link_by_anonymous = (
+        _pb_find_email_link_by_anonymous_user_id(admin_token, anonymous_user_id)
+        if anonymous_user_id
+        else None
+    )
+    linked_email = (
+        (_record_get(link_by_email, "email", "") or "").strip().lower()
+        or (_record_get(link_by_user_id, "email", "") or "").strip().lower()
+        or (_record_get(link_by_anonymous, "email", "") or "").strip().lower()
+        or email
+    )
+    linked_pb_user_id = (
+        (_record_get(link_by_email, "user_id", "") or "").strip()
+        or (_record_get(link_by_user_id, "user_id", "") or "").strip()
+        or (_record_get(link_by_anonymous, "user_id", "") or "").strip()
+        or pb_user_id
+    )
+    linked_anonymous_user_id = (
+        (_record_get(link_by_anonymous, "anonymous_user_id", "") or "").strip()
+        or (_record_get(link_by_email, "anonymous_user_id", "") or "").strip()
+        or anonymous_user_id
+    )
+
+    profile_record = _pb_get_wechat_private_user(admin_token, wechat_userid) if wechat_userid else None
+    if not profile_record:
+        return {
+            "ok": True,
+            "found": False,
+            "relations": {
+                "user_id": user_id or None,
+                "email": email or None,
+                "linked_email": linked_email or None,
+                "pb_user_id": pb_user_id or None,
+                "linked_pb_user_id": linked_pb_user_id or None,
+                "anonymous_user_id": anonymous_user_id or None,
+                "linked_anonymous_user_id": linked_anonymous_user_id or None,
+                "has_vip_by_user_id": has_vip_by_user_id,
+                "has_vip_by_pb_user": has_vip_by_pb_user,
+            },
+        }
+
+    nickname = (
+        (profile_record.get("nick") or "").strip()
+        if isinstance(profile_record, dict)
+        else ""
+    )
+    avatar = (
+        (profile_record.get("avatar") or "").strip()
+        if isinstance(profile_record, dict)
+        else ""
+    )
+    return {
+        "ok": True,
+        "found": True,
+        "wechat_userid": wechat_userid,
+        "profile": {
+            "nickname": nickname or "微信用户",
+            "avatar": avatar or "",
+        },
+        "relations": {
+            "user_id": user_id or None,
+            "wechat_userid": wechat_userid or None,
+            "email": email or None,
+            "linked_email": linked_email or None,
+            "pb_user_id": pb_user_id or None,
+            "linked_pb_user_id": linked_pb_user_id or None,
+            "anonymous_user_id": anonymous_user_id or None,
+            "linked_anonymous_user_id": linked_anonymous_user_id or None,
+            "has_vip_by_user_id": has_vip_by_user_id,
+            "has_vip_by_pb_user": has_vip_by_pb_user,
+        },
+    }
+
+
+@router.post("/identity_diagnose")
+async def nomadvip_identity_diagnose(item: dict):
+    """
+    用 user_id / email / wechat_userid 任一输入，诊断三表关联：
+    payments.user_id <-> users.email <-> wechat_private_users.userid
+    """
+    user_id = str(item.get("user_id", "") or item.get("userId", "") or "").strip()
+    email = str(item.get("email", "") or "").strip().lower()
+    wechat_userid = str(
+        item.get("wechat_userid", "") or item.get("userid", "") or ""
+    ).strip()
+
+    if not user_id and not email and not wechat_userid:
+        return {
+            "ok": False,
+            "error": "missing user_id/email/wechat_userid",
+        }
+
+    admin_token = _get_pb_admin_token()
+    if not admin_token:
+        return {"ok": False, "error": "PocketBase admin auth failed"}
+
+    pb_user_by_email = _pb_get_user_by_email(admin_token, email) if email else None
+    pb_user_id_by_email = (
+        (pb_user_by_email or {}).get("id", "") if isinstance(pb_user_by_email, dict) else ""
+    )
+    wechat_record = (
+        _pb_get_wechat_private_user(admin_token, wechat_userid) if wechat_userid else None
+    )
+    link_by_email = _pb_list_email_links(admin_token, email)[0] if email else None
+    link_by_user_id = _pb_find_email_link_by_user_id(admin_token, user_id) if user_id else None
+    link_by_anonymous = (
+        _pb_find_email_link_by_anonymous_user_id(admin_token, user_id)
+        if user_id.startswith("user") and user_id[4:].isdigit()
+        else None
+    )
+
+    linked_user_id = (
+        (_record_get(link_by_email, "user_id", "") or "").strip()
+        or (_record_get(link_by_user_id, "user_id", "") or "").strip()
+        or (_record_get(link_by_anonymous, "user_id", "") or "").strip()
+        or pb_user_id_by_email
+    )
+    linked_anonymous_user_id = (
+        (_record_get(link_by_anonymous, "anonymous_user_id", "") or "").strip()
+        or (_record_get(link_by_email, "anonymous_user_id", "") or "").strip()
+        or (user_id if user_id.startswith("user") and user_id[4:].isdigit() else "")
+    )
+    linked_email = (
+        (_record_get(link_by_email, "email", "") or "").strip().lower()
+        or (_record_get(link_by_user_id, "email", "") or "").strip().lower()
+        or (_record_get(link_by_anonymous, "email", "") or "").strip().lower()
+        or email
+    )
+
+    linked_user = _pb_get_user_by_id(admin_token, linked_user_id) if linked_user_id else None
+    payment_by_user_id = (
+        _pb_latest_payment_by_user_id(admin_token, user_id) if user_id else None
+    )
+    payment_by_linked_user_id = (
+        _pb_latest_payment_by_user_id(admin_token, linked_user_id)
+        if linked_user_id and linked_user_id != user_id
+        else None
+    )
+
+    has_vip_by_user_id = False
+    has_vip_by_linked_user_id = False
+    try:
+        from ..services.pb import get_pb_client
+
+        pb = get_pb_client()
+        if user_id:
+            has_vip_by_user_id = _uid_has_vip(pb, user_id)
+        if linked_user_id and linked_user_id != user_id:
+            has_vip_by_linked_user_id = _uid_has_vip(pb, linked_user_id)
+    except Exception:
+        pass
+
+    return {
+        "ok": True,
+        "input": {
+            "user_id": user_id or None,
+            "email": email or None,
+            "wechat_userid": wechat_userid or None,
+        },
+        "resolved": {
+            "linked_email": linked_email or None,
+            "linked_user_id": linked_user_id or None,
+            "linked_anonymous_user_id": linked_anonymous_user_id or None,
+        },
+        "wechat_private_user": {
+            "found": bool(wechat_record),
+            "userid": (_record_get(wechat_record, "userid", "") or "").strip() or None,
+            "nick": (_record_get(wechat_record, "nick", "") or "").strip() or None,
+            "avatar": (_record_get(wechat_record, "avatar", "") or "").strip() or None,
+            "record_id": (_record_get(wechat_record, "id", "") or "").strip() or None,
+        },
+        "users": {
+            "by_email": {
+                "id": pb_user_id_by_email or None,
+                "email": (_record_get(pb_user_by_email, "email", "") or "").strip().lower() or None,
+            },
+            "linked_user": {
+                "id": (_record_get(linked_user, "id", "") or "").strip() or None,
+                "email": (_record_get(linked_user, "email", "") or "").strip().lower() or None,
+            },
+        },
+        "vip_email_link": {
+            "by_email": {
+                "id": (_record_get(link_by_email, "id", "") or "").strip() or None,
+                "user_id": (_record_get(link_by_email, "user_id", "") or "").strip() or None,
+                "anonymous_user_id": (_record_get(link_by_email, "anonymous_user_id", "") or "").strip() or None,
+                "email": (_record_get(link_by_email, "email", "") or "").strip().lower() or None,
+            },
+            "by_user_id": {
+                "id": (_record_get(link_by_user_id, "id", "") or "").strip() or None,
+                "user_id": (_record_get(link_by_user_id, "user_id", "") or "").strip() or None,
+                "anonymous_user_id": (_record_get(link_by_user_id, "anonymous_user_id", "") or "").strip() or None,
+                "email": (_record_get(link_by_user_id, "email", "") or "").strip().lower() or None,
+            },
+            "by_anonymous_user_id": {
+                "id": (_record_get(link_by_anonymous, "id", "") or "").strip() or None,
+                "user_id": (_record_get(link_by_anonymous, "user_id", "") or "").strip() or None,
+                "anonymous_user_id": (_record_get(link_by_anonymous, "anonymous_user_id", "") or "").strip() or None,
+                "email": (_record_get(link_by_anonymous, "email", "") or "").strip().lower() or None,
+            },
+        },
+        "payments": {
+            "by_user_id": {
+                "record_id": (_record_get(payment_by_user_id, "id", "") or "").strip() or None,
+                "user_id": (_record_get(payment_by_user_id, "user_id", "") or "").strip() or None,
+                "order_id": (_record_get(payment_by_user_id, "order_id", "") or "").strip() or None,
+            },
+            "by_linked_user_id": {
+                "record_id": (_record_get(payment_by_linked_user_id, "id", "") or "").strip() or None,
+                "user_id": (_record_get(payment_by_linked_user_id, "user_id", "") or "").strip() or None,
+                "order_id": (_record_get(payment_by_linked_user_id, "order_id", "") or "").strip() or None,
+            },
+        },
+        "vip_flags": {
+            "has_vip_by_user_id": has_vip_by_user_id,
+            "has_vip_by_linked_user_id": has_vip_by_linked_user_id,
+        },
+    }
+
+
 def _get_pb_admin_token() -> str | None:
     """获取 PocketBase 管理员 token。"""
     global _admin_token_cache_value, _admin_token_cache_expire_at
@@ -426,6 +781,46 @@ def _pb_list_email_links(admin_token: str, email: str) -> list[dict]:
     return []
 
 
+def _pb_find_email_link_by_user_id(admin_token: str, user_id: str) -> dict | None:
+    base = (PB_URL or "").rstrip("/")
+    safe_user_id = _pb_escape(user_id.strip())
+    response = _http_session.get(
+        f"{base}/api/collections/vip_email_link/records",
+        params={
+            "filter": f'user_id = "{safe_user_id}"',
+            "perPage": 1,
+            "sort": "-updated",
+        },
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=DEFAULT_HTTP_TIMEOUT,
+    )
+    if response.status_code != 200:
+        return None
+    items = (response.json() or {}).get("items") or []
+    return items[0] if items else None
+
+
+def _pb_find_email_link_by_anonymous_user_id(
+    admin_token: str, anonymous_user_id: str
+) -> dict | None:
+    base = (PB_URL or "").rstrip("/")
+    safe_uid = _pb_escape(anonymous_user_id.strip())
+    response = _http_session.get(
+        f"{base}/api/collections/vip_email_link/records",
+        params={
+            "filter": f'anonymous_user_id = "{safe_uid}"',
+            "perPage": 1,
+            "sort": "-updated",
+        },
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=DEFAULT_HTTP_TIMEOUT,
+    )
+    if response.status_code != 200:
+        return None
+    items = (response.json() or {}).get("items") or []
+    return items[0] if items else None
+
+
 def _pb_upsert_email_link(
     admin_token: str,
     email: str,

requirements.txt

@@ -8,4 +8,5 @@ pocketbase
 requests
 minio
 opencv-python-headless
-rapidocr-onnxruntime
+rapidocr-onnxruntime
+pycryptodome