From 99816764b52d9bb6ff175e77097194031d8e524d Mon Sep 17 00:00:00 2001 From: eric Date: Tue, 31 Mar 2026 12:45:04 -0500 Subject: [PATCH] 's' --- app/main.py | 2 + app/routers/__init__.py | 2 + app/routers/chatbot.py | 448 ++++++++++++++++++++++++++++++++++++++++ app/routers/nomadvip.py | 395 +++++++++++++++++++++++++++++++++++ requirements.txt | 3 +- 5 files changed, 849 insertions(+), 1 deletion(-) create mode 100644 app/routers/chatbot.py diff --git a/app/main.py b/app/main.py index effbac1..4e87577 100755 --- a/app/main.py +++ b/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("/") diff --git a/app/routers/__init__.py b/app/routers/__init__.py index 99afa37..63171f7 100644 --- a/app/routers/__init__.py +++ b/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", ] diff --git a/app/routers/chatbot.py b/app/routers/chatbot.py new file mode 100644 index 0000000..4bb2e88 --- /dev/null +++ b/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 "

chatbot clone ready

请使用 /chatbot/api/private-message/query 调试接口

" + diff --git a/app/routers/nomadvip.py b/app/routers/nomadvip.py index f5a2870..7c48235 100644 --- a/app/routers/nomadvip.py +++ b/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, diff --git a/requirements.txt b/requirements.txt index 79d43d6..6aa16bb 100755 --- a/requirements.txt +++ b/requirements.txt @@ -8,4 +8,5 @@ pocketbase requests minio opencv-python-headless -rapidocr-onnxruntime \ No newline at end of file +rapidocr-onnxruntime +pycryptodome \ No newline at end of file