"""nomadvip 项目支付接口：/nomadvip/*，支持 xorpay / zpay。""" import logging import random import string import threading import time import requests from fastapi import APIRouter, File, HTTPException, Request, UploadFile from fastapi.responses import JSONResponse, RedirectResponse from requests.adapters import HTTPAdapter from ..config import BASE_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL from ..payment import ( get_payment_provider, get_payment_provider_by_name, resolve_provider, resolve_channel, ) from ..payment.zpay import ZPayProvider from ..services import handle_payment_success, processed_orders from ..services.discount_proof import recognize_discount_proof router = APIRouter(prefix="/nomadvip", tags=["nomadvip"]) DEFAULT_HTTP_TIMEOUT = 10 PB_ADMIN_TOKEN_TTL_SECONDS = 60 _http_session = requests.Session() _http_session.mount("http://", HTTPAdapter(pool_connections=32, pool_maxsize=64)) _http_session.mount("https://", HTTPAdapter(pool_connections=32, pool_maxsize=64)) _admin_token_cache_lock = threading.Lock() _admin_token_cache_value: str | None = None _admin_token_cache_expire_at = 0.0 _vip_cache_lock = threading.Lock() _vip_check_cache: dict[str, tuple[float, dict]] = {} _VIP_CACHE_TTL_SECONDS = 20 def _cache_get(key: str): now = time.time() with _vip_cache_lock: item = _vip_check_cache.get(key) if not item: return None expire_at, value = item if expire_at < now: _vip_check_cache.pop(key, None) return None return value def _cache_set(key: str, value: dict, ttl: int = _VIP_CACHE_TTL_SECONDS): with _vip_cache_lock: _vip_check_cache[key] = (time.time() + ttl, value) def _record_get(obj, key: str, default=None): """兼容 PocketBase SDK Record 与 dict 取值。""" if obj is None: return default if isinstance(obj, dict): return obj.get(key, default) if hasattr(obj, key): try: value = getattr(obj, key) return default if value is None else value except Exception: return default try: value = obj[key] # type: ignore[index] return default if value is None else value except Exception: return default def _get_notify_path(provider_name: str) -> str: if provider_name == "xorpay": return f"{BASE_URL.rstrip('/')}/nomadvip/xorpay_notify" return f"{BASE_URL.rstrip('/')}/nomadvip/zpay_notify" @router.post("/payh5") async def nomadvip_payh5(item: dict, request: Request): """创建支付订单。微信内用 xorpay，PC/手机用 zpay，默认微信支付，支付宝暂不提供""" req_provider = (item.get("provider") or "").strip().lower() or None device = (item.get("device") or "pc").lower() ua = (request.headers.get("user-agent") or "").strip() provider = resolve_provider(device, req_provider, ua) channel = resolve_channel(item.get("channel")) logging.info( "[payjsapi] payh5 user_id=%s type=%s provider=%s channel=%s", item.get("user_id", ""), item.get("type", ""), provider.name, channel, ) total_fee = int(item.get("total_fee", 880)) total_fee_yuan = f"{total_fee / 100:.2f}" user_id = item.get("user_id", "unknown") pay_type = (item.get("type") or "vip").lower() random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8)) timestamp = int(time.time()) order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}" type_map = { "book": "购买书籍", "meetup": "数字游民社区报名", "video": "视频解锁", "vip": "VIP会员充值", } name = type_map.get(pay_type, "VIP会员充值") notify_url = _get_notify_path(provider.name) result = provider.create_order( order_id=order_id, name=name, total_fee_yuan=total_fee_yuan, pay_type=channel, notify_url=notify_url, return_url=item.get("return_url"), client_ip=item.get("client_ip"), ) logging.info( "[payjsapi] payh5 created order_id=%s provider=%s notify_url=%s return_url=%s", order_id, provider.name, notify_url, item.get("return_url") or "", ) if provider.name == "xorpay": return { "status": "ok", "order_id": order_id, "xorpay_params": result.get("params", {}), "xorpay_url": result.get("pay_url", ""), "provider": provider.name, } params = result.get("params", {}) params_str = { k: str(v) for k, v in params.items() if v is not None and str(v).strip() != "" } return { "status": "ok", "params": params_str, "pay_url": result.get("pay_url", ""), "provider": provider.name, "submit_method": "POST", } @router.post("/payh5/redirect") async def nomadvip_payh5_redirect(item: dict, request: Request): """ZPAY mapi 接口，PC/手机用 zpay，默认微信支付，支付宝暂不提供""" provider = get_payment_provider_by_name("zpay") channel = resolve_channel(item.get("channel")) logging.info("nomadvip payh5/redirect request: item=%s", item) try: total_fee = float(item.get("total_fee", 880)) except (TypeError, ValueError): logging.warning( "nomadvip payh5/redirect invalid total_fee=%s", item.get("total_fee") ) total_fee = 880.0 total_fee_yuan = f"{total_fee / 100:.2f}" user_id = item.get("user_id", "unknown") pay_type = (item.get("type") or "vip").lower() return_url = item.get("return_url") device = item.get("device", "pc") client_ip = item.get("client_ip") or ( request.client.host if request.client else "127.0.0.1" ) forwarded = request.headers.get("x-forwarded-for") if forwarded: client_ip = forwarded.split(",")[0].strip() random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8)) timestamp = int(time.time()) order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}" # 追加 order_id，供 paid 页在 cookie 丢失时恢复 user_id。 if return_url and "order_id=" not in return_url and "out_trade_no=" not in return_url: sep = "&" if "?" in return_url else "?" return_url = f"{return_url}{sep}order_id={order_id}" type_map = { "book": "购买书籍", "meetup": "数字游民社区报名", "video": "视频解锁", "vip": "VIP会员充值", } name = type_map.get(pay_type, "VIP会员充值") notify_url = f"{BASE_URL.rstrip('/')}/nomadvip/zpay_notify" result = provider.create_order_api( order_id=order_id, name=name, total_fee_yuan=total_fee_yuan, pay_type=channel, notify_url=notify_url, return_url=return_url, client_ip=client_ip, device=device, ) if result.get("status") != "ok": err_msg = result.get("msg", "ZPAY mapi 创建订单失败") logging.error( "nomadvip payh5/redirect ZPAY create failed: %s result=%s", err_msg, result, ) raise HTTPException(status_code=500, detail=err_msg) payurl = result.get("payurl") payurl2 = result.get("payurl2") img = result.get("img") if device == "pc" and img: return JSONResponse( status_code=200, content={ "status": "ok", "qrcode_mode": True, "order_id": order_id, "img": img, "payurl": payurl, "return_url": return_url, "channel": channel, }, ) if device == "wechat" and channel.lower() in ("wxpay", "wx", "wechat"): return JSONResponse( status_code=200, content={"use_form": True, "order_id": order_id} ) headers = {"X-Pay-Order-Id": order_id} if device == "h5" and channel.lower() in ("wxpay", "wx", "wechat") and payurl2: return RedirectResponse(url=payurl2, status_code=302, headers=headers) if payurl: return RedirectResponse(url=payurl, status_code=302, headers=headers) logging.error( "nomadvip payh5/redirect ZPAY missing pay url device=%s channel=%s keys=%s", device, channel, list(result.keys()), ) raise HTTPException(status_code=500, detail="ZPAY 未返回支付链接") @router.get("/zpay_order_status") async def nomadvip_zpay_order_status(out_trade_no: str): """查询 ZPAY 订单支付状态。""" provider = get_payment_provider() if not isinstance(provider, ZPayProvider): raise HTTPException(status_code=400, detail="zpay_order_status 仅支持 zpay") result = provider.query_order_status(out_trade_no, timeout=15) if result.get("error"): logging.warning( "nomadvip ZPAY order status failed: %s", result.get("error") ) return result def _get_linked_email_for_user(pb, user_id: str) -> str | None: """按 user_id 查询 vip_email_link 关联邮箱。""" try: link_rec = pb.collection("vip_email_link").get_list( 1, 1, {"filter": f'anonymous_user_id = "{user_id}"'} ) if link_rec.items: return (_record_get(link_rec.items[0], "email", "") or "").strip().lower() if user_id and user_id.startswith("user") and user_id[4:].isdigit(): link_rec = pb.collection("vip_email_link").get_list( 1, 1, {"filter": f'user_id = "{user_id}"'} ) if link_rec.items: return (_record_get(link_rec.items[0], "email", "") or "").strip().lower() except Exception: pass return None def _uid_has_vip(pb, uid: str, site_id: str = "vip") -> bool: if not uid: return False sv = pb.collection("site_vip").get_list( 1, 1, {"filter": f'user_id = "{uid}" && site_id = "{site_id}"'} ) if sv.items: rec = sv.items[0] expires = (_record_get(rec, "expires_at", "") or "").strip() if not expires: return True from datetime import datetime try: exp_dt = datetime.fromisoformat(expires.replace("Z", "+00:00")) now = datetime.now(exp_dt.tzinfo) if exp_dt.tzinfo else datetime.now() return exp_dt.timestamp() >= now.timestamp() except Exception: return True pm = pb.collection("payments").get_list( 1, 1, {"filter": f'user_id = "{uid}" && type = "vip"', "sort": "-created"}, ) return bool(pm.items) @router.get("/check_vip") async def nomadvip_check_vip(user_id: str = ""): """Return vip status for a user_id.""" user_id = (user_id or "").strip() logging.info("[payjsapi] check_vip user_id=%s", user_id or "(empty)") if not user_id: return {"vip": False, "error": "missing user_id"} cache_key = f"check_vip:{user_id}" cached = _cache_get(cache_key) if cached is not None: return cached try: from ..services.pb import get_pb_client pb = get_pb_client() linked_user_id = "" wx_link_email = "" if not _uid_has_vip(pb, user_id): # 关键补链：真实 wechat userid 可通过 vip_email_link.anonymous_user_id 关联到会员 user_id try: link_rec = pb.collection("vip_email_link").get_list( 1, 1, {"filter": f'anonymous_user_id = "{user_id}"'} ) if link_rec.items: linked_user_id = ( _record_get(link_rec.items[0], "user_id", "") or "" ).strip() except Exception: linked_user_id = "" if not linked_user_id: try: admin_token = _get_pb_admin_token() wx_record = ( _pb_get_wechat_private_user(admin_token, user_id) if admin_token else None ) wx_link = _extract_wechat_vip_link(wx_record) linked_user_id = ( str(wx_link.get("pb_user_id", "")).strip() or str(wx_link.get("anonymous_user_id", "")).strip() ) wx_link_email = str(wx_link.get("email", "")).strip().lower() except Exception: linked_user_id = "" wx_link_email = "" if not linked_user_id or not _uid_has_vip(pb, linked_user_id): logging.info("[payjsapi] check_vip miss user_id=%s vip=False", user_id) result = {"vip": False} _cache_set(cache_key, result) return result email = _get_linked_email_for_user(pb, user_id) if not email and wx_link_email: email = wx_link_email if not email and linked_user_id: try: link_rec = pb.collection("vip_email_link").get_list( 1, 1, {"filter": f'user_id = "{linked_user_id}"'} ) if link_rec.items: email = ( _record_get(link_rec.items[0], "email", "") or "" ).strip().lower() except Exception: pass if not email: admin_token = _get_pb_admin_token() if admin_token and user_id.startswith("user") and user_id[4:].isdigit(): linked_member = _pb_find_linked_member_from_order(admin_token, user_id) if linked_member: linked_user_id, linked_email = linked_member email = linked_email _pb_upsert_email_link(admin_token, linked_email, linked_user_id, user_id) logging.info( "[payjsapi] check_vip hit user_id=%s vip=True email=%s", user_id, email or "(none)", ) result = {"vip": True, "email": email} if linked_user_id: result["linked_user_id"] = linked_user_id _cache_set(cache_key, result) return result except Exception as error: logging.warning("[payjsapi] check_vip failed: %s", error) result = {"vip": False, "error": str(error)} _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 _extract_wechat_vip_link(record: dict | None) -> dict: if not isinstance(record, dict): return {} raw_contact = record.get("raw_contact") if isinstance(raw_contact, str): try: import json as _json raw_contact = _json.loads(raw_contact) except Exception: raw_contact = {} if not isinstance(raw_contact, dict): return {} link = raw_contact.get("_vip_link") return link if isinstance(link, dict) else {} def _pb_bind_wechat_vip_link( admin_token: str, wechat_userid: str, *, email: str | None = None, pb_user_id: str | None = None, anonymous_user_id: str | None = None, ): """ 在 wechat_private_users.raw_contact 中记录 _vip_link 关系。 注意：不改变历史 user_id 语义，anonymous_user_id 仍是历史业务 user_id。 """ record = _pb_get_wechat_private_user(admin_token, wechat_userid) if not isinstance(record, dict): return record_id = str(record.get("id", "")).strip() if not record_id: return raw_contact = record.get("raw_contact") if isinstance(raw_contact, str): try: import json as _json raw_contact = _json.loads(raw_contact) except Exception: raw_contact = {} if not isinstance(raw_contact, dict): raw_contact = {} current = _extract_wechat_vip_link(record) raw_contact["_vip_link"] = { "email": (email or current.get("email") or "").strip().lower() or None, "pb_user_id": (pb_user_id or current.get("pb_user_id") or "").strip() or None, "anonymous_user_id": ( anonymous_user_id or current.get("anonymous_user_id") or "" ).strip() or None, "updated_at": int(time.time()), } base = (PB_URL or "").rstrip("/") _http_session.patch( f"{base}/api/collections/wechat_private_users/records/{record_id}", json={"raw_contact": raw_contact}, headers={ "Authorization": f"Bearer {admin_token}", "Content-Type": "application/json", }, timeout=DEFAULT_HTTP_TIMEOUT, ) 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 has_vip_by_wechat_userid = False if wechat_userid: try: from ..services.pb import get_pb_client pb = get_pb_client() has_vip_by_wechat_userid = _uid_has_vip(pb, wechat_userid) except Exception: has_vip_by_wechat_userid = 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: # 保持历史语义：vip_email_link.anonymous_user_id 仍仅承载历史业务 user_id link_anonymous_id = anonymous_user_id or None _pb_upsert_email_link( admin_token, email, effective_member_user_id, link_anonymous_id, ) if wechat_userid: _pb_bind_wechat_vip_link( admin_token, wechat_userid, email=email, pb_user_id=effective_member_user_id, anonymous_user_id=link_anonymous_id, ) except Exception: pass link_email_items = _pb_list_email_links(admin_token, email) if email else [] link_by_email = link_email_items[0] if link_email_items 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_wechat_userid = ( _pb_find_email_link_by_anonymous_user_id(admin_token, wechat_userid) if wechat_userid 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_wechat_userid, "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_wechat_userid, "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_wechat_userid, "anonymous_user_id", "") or "").strip() or (_record_get(link_by_email, "anonymous_user_id", "") or "").strip() or anonymous_user_id ) has_vip_by_linked_user = False if linked_pb_user_id and linked_pb_user_id != user_id: try: from ..services.pb import get_pb_client pb = get_pb_client() has_vip_by_linked_user = _uid_has_vip(pb, linked_pb_user_id) except Exception: has_vip_by_linked_user = False 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, "has_vip_by_wechat_userid": has_vip_by_wechat_userid, "has_vip_by_linked_user": has_vip_by_linked_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, "has_vip_by_wechat_userid": has_vip_by_wechat_userid, "has_vip_by_linked_user": has_vip_by_linked_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_email_items = _pb_list_email_links(admin_token, email) if email else [] link_by_email = link_email_items[0] if link_email_items 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 now = time.time() with _admin_token_cache_lock: if _admin_token_cache_value and now < _admin_token_cache_expire_at: return _admin_token_cache_value if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD: return None base = (PB_URL or "").rstrip("/") for path in [ "/api/collections/_superusers/auth-with-password", "/api/admins/auth-with-password", ]: try: response = _http_session.post( f"{base}{path}", json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD}, timeout=DEFAULT_HTTP_TIMEOUT, ) if response.status_code == 200: token = response.json().get("token") if token: with _admin_token_cache_lock: _admin_token_cache_value = token _admin_token_cache_expire_at = ( time.time() + PB_ADMIN_TOKEN_TTL_SECONDS ) return token except Exception: pass return None def _pb_escape(value: str) -> str: return str(value).replace("\\", "\\\\").replace('"', '\\"') def _pb_list_email_links(admin_token: str, email: str) -> list[dict]: base = (PB_URL or "").rstrip("/") headers = {"Authorization": f"Bearer {admin_token}"} for candidate in {email, email.lower()}: safe_email = _pb_escape(candidate) response = _http_session.get( f"{base}/api/collections/vip_email_link/records", params={"filter": f'email = "{safe_email}"', "perPage": 1}, headers=headers, timeout=DEFAULT_HTTP_TIMEOUT, ) if response.status_code != 200: continue items = (response.json() or {}).get("items") or [] if items: return items 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, pb_user_id: str, anonymous_user_id: str | None = None, ): base = (PB_URL or "").rstrip("/") headers = { "Authorization": f"Bearer {admin_token}", "Content-Type": "application/json", } items = _pb_list_email_links(admin_token, email) payload = {"email": email.lower(), "user_id": pb_user_id} if anonymous_user_id: payload["anonymous_user_id"] = anonymous_user_id if items: record_id = items[0].get("id") if record_id: _http_session.patch( f"{base}/api/collections/vip_email_link/records/{record_id}", json=payload, headers=headers, timeout=DEFAULT_HTTP_TIMEOUT, ) return _http_session.post( f"{base}/api/collections/vip_email_link/records", json=payload, headers=headers, timeout=DEFAULT_HTTP_TIMEOUT, ) def _pb_move_site_vip( admin_token: str, from_user_id: str, to_user_id: str, site_id: str = "vip", ): if not from_user_id or not to_user_id or from_user_id == to_user_id: return base = (PB_URL or "").rstrip("/") safe_from_user_id = _pb_escape(from_user_id) safe_site_id = _pb_escape(site_id) response = _http_session.get( f"{base}/api/collections/site_vip/records", params={ "filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"', "perPage": 500, }, headers={"Authorization": f"Bearer {admin_token}"}, timeout=DEFAULT_HTTP_TIMEOUT, ) if response.status_code != 200: return headers = { "Authorization": f"Bearer {admin_token}", "Content-Type": "application/json", } items = (response.json() or {}).get("items") or [] for item in items: record_id = item.get("id") if not record_id: continue _http_session.patch( f"{base}/api/collections/site_vip/records/{record_id}", json={"user_id": to_user_id}, headers=headers, timeout=DEFAULT_HTTP_TIMEOUT, ) def _pb_latest_vip_payment(admin_token: str, user_id: str) -> dict | None: base = (PB_URL or "").rstrip("/") safe_user_id = _pb_escape(user_id) 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 def _pb_list_site_vip_by_order( admin_token: str, order_id: str, site_id: str = "vip", ) -> list[dict]: base = (PB_URL or "").rstrip("/") safe_order_id = _pb_escape(order_id) safe_site_id = _pb_escape(site_id) response = _http_session.get( f"{base}/api/collections/site_vip/records", params={ "filter": f'order_id = "{safe_order_id}" && site_id = "{safe_site_id}"', "perPage": 500, "sort": "-created", }, headers={"Authorization": f"Bearer {admin_token}"}, timeout=DEFAULT_HTTP_TIMEOUT, ) if response.status_code != 200: return [] return (response.json() or {}).get("items") or [] def _pb_get_user_email(admin_token: str, user_id: str) -> str | 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 email = (response.json() or {}).get("email") or "" return email.strip().lower() or None def _pb_find_linked_member_from_order( admin_token: str, anonymous_user_id: str, ) -> tuple[str, str] | None: base = (PB_URL or "").rstrip("/") safe_user_id = _pb_escape(anonymous_user_id) response = _http_session.get( f"{base}/api/collections/payments/records", params={ "filter": f'user_id = "{safe_user_id}" && type = "vip"', "perPage": 20, "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 [] seen_orders = set() for item in items: order_id = (item.get("order_id") or "").strip() if not order_id or order_id in seen_orders: continue seen_orders.add(order_id) for site_vip in _pb_list_site_vip_by_order(admin_token, order_id): linked_user_id = (site_vip.get("user_id") or "").strip() if ( not linked_user_id or linked_user_id == anonymous_user_id or (linked_user_id.startswith("user") and linked_user_id[4:].isdigit()) ): continue email = _pb_get_user_email(admin_token, linked_user_id) if email: return linked_user_id, email return None def _pb_ensure_site_vip( admin_token: str, from_user_id: str, to_user_id: str, site_id: str = "vip", ): if not from_user_id or not to_user_id or from_user_id == to_user_id: return base = (PB_URL or "").rstrip("/") safe_from_user_id = _pb_escape(from_user_id) safe_to_user_id = _pb_escape(to_user_id) safe_site_id = _pb_escape(site_id) source_records = _http_session.get( f"{base}/api/collections/site_vip/records", params={ "filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"', "perPage": 500, "sort": "-created", }, headers={"Authorization": f"Bearer {admin_token}"}, timeout=DEFAULT_HTTP_TIMEOUT, ) to_records = _http_session.get( f"{base}/api/collections/site_vip/records", params={ "filter": f'user_id = "{safe_to_user_id}" && site_id = "{safe_site_id}"', "perPage": 500, "sort": "-created", }, headers={"Authorization": f"Bearer {admin_token}"}, timeout=DEFAULT_HTTP_TIMEOUT, ) payment = _pb_latest_vip_payment(admin_token, from_user_id) order_id = (payment or {}).get("order_id") or "" order_records = ( _pb_list_site_vip_by_order(admin_token, order_id, site_id=site_id) if order_id else [] ) source_items = (source_records.json() or {}).get("items") or [] if source_records.status_code == 200 else [] target_items = (to_records.json() or {}).get("items") or [] if to_records.status_code == 200 else [] canonical = (target_items or source_items or order_records or [None])[0] headers = { "Authorization": f"Bearer {admin_token}", "Content-Type": "application/json", } if canonical and canonical.get("id"): patch_body = {"user_id": to_user_id, "site_id": site_id} next_order_id = order_id or (canonical.get("order_id") or "").strip() if next_order_id: patch_body["order_id"] = next_order_id _http_session.patch( f"{base}/api/collections/site_vip/records/{canonical['id']}", json=patch_body, headers=headers, timeout=DEFAULT_HTTP_TIMEOUT, ) duplicate_ids = set() for record in [*target_items, *source_items, *order_records]: record_id = record.get("id") if record_id and record_id != canonical["id"]: duplicate_ids.add(record_id) for record_id in duplicate_ids: _http_session.delete( f"{base}/api/collections/site_vip/records/{record_id}", headers={"Authorization": f"Bearer {admin_token}"}, timeout=DEFAULT_HTTP_TIMEOUT, ) return if not order_id: return _http_session.post( f"{base}/api/collections/site_vip/records", json={"user_id": to_user_id, "site_id": site_id, "order_id": order_id}, headers=headers, timeout=DEFAULT_HTTP_TIMEOUT, ) @router.post("/check_vip_by_email") async def nomadvip_check_vip_by_email(item: dict): """Recover VIP by email/password, optionally with anonymous user_id.""" email = (item.get("email") or "").strip().lower() password = (item.get("password") or "").strip() anonymous_user_id = ( item.get("user_id") or item.get("anonymousUserId") or "" ).strip() wechat_userid = (item.get("wechat_userid") or item.get("userid") or "").strip() logging.info( "[payjsapi] check_vip_by_email email=%s user_id=%s", email or "(empty)", anonymous_user_id or "(empty)", ) if not email or not password: return {"vip": False, "error": "missing email or password"} cache_key = f"check_vip_by_email:{email}:{anonymous_user_id}:{wechat_userid}" cached = _cache_get(cache_key) if cached is not None: return cached base = (PB_URL or "").rstrip("/") login_res = _http_session.post( f"{base}/api/collections/users/auth-with-password", json={"identity": email, "password": password}, timeout=DEFAULT_HTTP_TIMEOUT, ) if login_res.status_code != 200: try: error = login_res.json() or {} except Exception: error = {} logging.info( "[payjsapi] check_vip_by_email login failed status=%s message=%s", login_res.status_code, error.get("message", ""), ) return { "vip": False, "error": error.get("message") or "invalid email or password", } login_data = login_res.json() or {} token = login_data.get("token") record = login_data.get("record") or {} pb_user_id = (record.get("id") or "").strip() if not token or not pb_user_id: return {"vip": False, "error": "login failed"} admin_token = _get_pb_admin_token() if not admin_token: return {"vip": False, "error": "PocketBase admin auth failed"} try: from ..services.pb import get_pb_client pb = get_pb_client() matched_user_id = None vip_check_cache: dict[str, bool] = {} # 保持历史语义：匿名ID只来自历史 user_id，不使用真实 wechat userid 覆盖 recovered_anonymous_user_id = (anonymous_user_id or None) recovery_source = None def has_vip_cached(uid: str) -> bool: if not uid: return False if uid in vip_check_cache: return vip_check_cache[uid] result = _uid_has_vip(pb, uid) vip_check_cache[uid] = result return result if recovered_anonymous_user_id and has_vip_cached(recovered_anonymous_user_id): matched_user_id = recovered_anonymous_user_id recovery_source = "request_user_id" else: items = _pb_list_email_links(admin_token, email) logging.info( "[payjsapi] check_vip_by_email vip_email_link email=%s count=%s", email, len(items), ) for link_record in items: linked_ids = [] linked_pb_user_id = (link_record.get("user_id") or "").strip() linked_anonymous_user_id = ( link_record.get("anonymous_user_id") or "" ).strip() if linked_pb_user_id: linked_ids.append(linked_pb_user_id) if linked_anonymous_user_id and linked_anonymous_user_id not in linked_ids: linked_ids.append(linked_anonymous_user_id) for linked_uid in dict.fromkeys(linked_ids): if not has_vip_cached(linked_uid): continue matched_user_id = linked_uid if linked_anonymous_user_id: recovered_anonymous_user_id = linked_anonymous_user_id elif linked_uid: recovered_anonymous_user_id = linked_uid recovery_source = "vip_email_link" break if matched_user_id: break if not matched_user_id and has_vip_cached(pb_user_id): matched_user_id = pb_user_id recovered_anonymous_user_id = None recovery_source = "pb_user_id" if not matched_user_id: logging.info("[payjsapi] check_vip_by_email no vip match for email=%s", email) result = { "vip": False, "error": "no linked vip record found; provide the paid user_id if this is a historical anonymous order", } _cache_set(cache_key, result, ttl=5) return result if recovered_anonymous_user_id: _pb_ensure_site_vip(admin_token, recovered_anonymous_user_id, pb_user_id) _pb_upsert_email_link( admin_token, email, pb_user_id, recovered_anonymous_user_id, ) else: _pb_upsert_email_link(admin_token, email, pb_user_id) if wechat_userid: _pb_bind_wechat_vip_link( admin_token, wechat_userid, email=email, pb_user_id=pb_user_id, anonymous_user_id=recovered_anonymous_user_id, ) logging.info( "[payjsapi] check_vip_by_email recovered by %s matched_user_id=%s effectiveUserId=%s anonymousUserId=%s", recovery_source or "unknown", matched_user_id, pb_user_id, recovered_anonymous_user_id or "(none)", ) result = { "vip": True, "token": token, "record": record, "effectiveUserId": pb_user_id, "anonymousUserId": recovered_anonymous_user_id, } _cache_set(cache_key, result) return result except Exception as error: logging.warning("[payjsapi] check_vip_by_email failed: %s", error) result = {"vip": False, "error": str(error)} _cache_set(cache_key, result, ttl=5) return result @router.get("/order_status") async def nomadvip_order_status(order_id: str = "", out_trade_no: str = ""): """查询订单支付状态，优先查 ZPAY，失败时回退 PocketBase。""" order_id = (order_id or out_trade_no or "").strip() logging.info("[payjsapi] order_status order_id=%s", order_id or "(空)") if not order_id: return {"paid": False, "error": "缺少 order_id 或 out_trade_no"} try: from ..services.pb import get_pb_client pb = get_pb_client() records = pb.collection("payments").get_list( 1, 1, {"filter": f'order_id = "{order_id}"'} ) paid = len(records.items) > 0 logging.info("[payjsapi] order_status PocketBase paid=%s", paid) if paid: return {"paid": True, "source": "pocketbase"} except Exception as error: logging.warning("[payjsapi] order_status PocketBase failed: %s", error) xorpay = get_payment_provider_by_name("xorpay") xorpay_result = xorpay.query_order_status(order_id, timeout=8) if xorpay_result.get("error"): logging.warning( "[payjsapi] order_status XorPay query failed order_id=%s error=%s", order_id, xorpay_result.get("error"), ) else: logging.info( "[payjsapi] order_status XorPay status=%s paid=%s", xorpay_result.get("status", ""), xorpay_result.get("paid", False), ) if xorpay_result.get("paid"): return { "paid": True, "source": "xorpay", "status": xorpay_result.get("status"), "pay_price": xorpay_result.get("pay_price") or "", } zpay = get_payment_provider_by_name("zpay") if isinstance(zpay, ZPayProvider): result = zpay.query_order_status(order_id, timeout=8) if not result.get("error") and result.get("paid"): logging.info("[payjsapi] order_status ZPAY paid=True") return result return {"paid": False} @router.post("/xorpay_notify") async def nomadvip_xorpay_notify(request: Request): """XorPay 异步通知。""" form_data = await request.form() data = {k: v for k, v in form_data.items()} logging.info("nomadvip XorPay notify: %s", data) order_id = data.get("order_id", "") if order_id in processed_orders: logging.info("nomadvip order already processed, skip: %s", order_id) return "ok" provider = get_payment_provider_by_name("xorpay") if not provider.verify_notify(data): logging.error("nomadvip XorPay verify failed: %s", data) raise HTTPException(status_code=400, detail="sign error") parsed = provider.parse_notify(data) if parsed.get("trade_status") != "success": return "ok" pay_price = parsed.get("pay_price", data.get("pay_price", "")) try: handle_payment_success( order_id, pay_price, "nomadvip-xorpay", use_memos=True, site_id="vip", ) logging.info("nomadvip business done: order_id=%s", order_id) except Exception as error: logging.error( "nomadvip PocketBase/Memos failed: %s", error, exc_info=True ) raise return provider.success_response() @router.get("/zpay_notify") @router.post("/zpay_notify") async def nomadvip_zpay_notify(request: Request): """ZPAY 异步通知。""" provider = get_payment_provider_by_name("zpay") if request.method == "GET": data = dict(request.query_params) else: form_data = await request.form() data = {k: v for k, v in form_data.items()} logging.info("nomadvip ZPAY notify: %s", data) order_id = data.get("out_trade_no", "") if order_id in processed_orders: logging.info("nomadvip order already processed, skip: %s", order_id) return "success" if not provider.verify_notify(data): logging.error("nomadvip ZPAY verify failed: %s", data) raise HTTPException(status_code=400, detail="sign error") parsed = provider.parse_notify(data) trade_status = parsed.get("trade_status") or data.get("trade_status", "") if trade_status not in ("success", "TRADE_SUCCESS"): logging.info("nomadvip non-success trade status, skip: %s", trade_status) return "success" pay_price = parsed.get("pay_price") or data.get("money", "") try: handle_payment_success( order_id, pay_price, "nomadvip-zpay", use_memos=True, site_id="vip", ) logging.info("nomadvip business done: order_id=%s", order_id) except Exception as error: logging.error( "nomadvip PocketBase/Memos failed: %s", error, exc_info=True ) raise return provider.success_response() @router.post("/recognize_discount_proof") async def nomadvip_recognize_discount_proof(image: UploadFile = File(...)): """识别优惠凭证：提取头像区域与昵称（独立接口，不影响现有支付链路）。""" content_type = (image.content_type or "").lower() if not content_type.startswith("image/"): return JSONResponse( status_code=400, content={"ok": False, "error": "仅支持图片文件"} ) raw = await image.read() if not raw: return JSONResponse( status_code=400, content={"ok": False, "error": "图片内容为空"} ) if len(raw) > 12 * 1024 * 1024: return JSONResponse( status_code=400, content={"ok": False, "error": "图片大小超过 12MB"} ) try: result = recognize_discount_proof(raw) status = 200 if result.get("ok") else 400 return JSONResponse(status_code=status, content=result) except Exception as error: logging.warning("[payjsapi] recognize_discount_proof failed: %s", error) return JSONResponse( status_code=500, content={"ok": False, "error": "识别服务异常，请稍后重试"}, )