"""nomadvip 项目支付接口：/nomadvip/*，支持 xorpay / zpay。""" import logging import random import string import time import requests from fastapi import APIRouter, HTTPException, Request from fastapi.responses import JSONResponse, RedirectResponse 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 router = APIRouter(prefix="/nomadvip", tags=["nomadvip"]) 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 link_rec.items[0].get("email") or "" 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 link_rec.items[0].get("email") or "" 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 = rec.get("expires_at") or "" 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"} try: from ..services.pb import get_pb_client pb = get_pb_client() if not _uid_has_vip(pb, user_id): logging.info("[payjsapi] check_vip miss user_id=%s vip=False", user_id) return {"vip": False} email = _get_linked_email_for_user(pb, user_id) 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)", ) return {"vip": True, "email": email} except Exception as error: logging.warning("[payjsapi] check_vip failed: %s", error) return {"vip": False, "error": str(error)} def _get_pb_admin_token() -> str | None: """获取 PocketBase 管理员 token。""" 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 = requests.post( f"{base}{path}", json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD}, timeout=10, ) if response.status_code == 200: return response.json().get("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 = requests.get( f"{base}/api/collections/vip_email_link/records", params={"filter": f'email = "{safe_email}"', "perPage": 1}, headers=headers, timeout=10, ) if response.status_code != 200: continue items = (response.json() or {}).get("items") or [] if items: return items return [] 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: requests.patch( f"{base}/api/collections/vip_email_link/records/{record_id}", json=payload, headers=headers, timeout=10, ) return requests.post( f"{base}/api/collections/vip_email_link/records", json=payload, headers=headers, timeout=10, ) 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 = requests.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=10, ) 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 requests.patch( f"{base}/api/collections/site_vip/records/{record_id}", json={"user_id": to_user_id}, headers=headers, timeout=10, ) 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 = requests.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=10, ) 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 = requests.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=10, ) 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 = requests.get( f"{base}/api/collections/users/records/{user_id}", headers={"Authorization": f"Bearer {admin_token}"}, timeout=10, ) 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 = requests.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=10, ) 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 = requests.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=10, ) to_records = requests.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=10, ) 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 requests.patch( f"{base}/api/collections/site_vip/records/{canonical['id']}", json=patch_body, headers=headers, timeout=10, ) 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: requests.delete( f"{base}/api/collections/site_vip/records/{record_id}", headers={"Authorization": f"Bearer {admin_token}"}, timeout=10, ) return if not order_id: return requests.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=10, ) @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() 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"} base = (PB_URL or "").rstrip("/") login_res = requests.post( f"{base}/api/collections/users/auth-with-password", json={"identity": email, "password": password}, timeout=10, ) 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 recovered_anonymous_user_id = ( anonymous_user_id if anonymous_user_id.startswith("user") and anonymous_user_id[4:].isdigit() else None ) recovery_source = None if recovered_anonymous_user_id and _uid_has_vip(pb, 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 linked_ids: if not _uid_has_vip(pb, linked_uid): continue matched_user_id = linked_uid if linked_anonymous_user_id.startswith("user") and linked_anonymous_user_id[4:].isdigit(): recovered_anonymous_user_id = linked_anonymous_user_id elif linked_uid.startswith("user") and linked_uid[4:].isdigit(): recovered_anonymous_user_id = linked_uid recovery_source = "vip_email_link" break if matched_user_id: break if not matched_user_id and _uid_has_vip(pb, 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) return { "vip": False, "error": "no linked vip record found; provide the paid user_id if this is a historical anonymous order", } 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) 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)", ) return { "vip": True, "token": token, "record": record, "effectiveUserId": pb_user_id, "anonymousUserId": recovered_anonymous_user_id, } except Exception as error: logging.warning("[payjsapi] check_vip_by_email failed: %s", error) return {"vip": False, "error": str(error)} @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()