eric/gitlab-instance-0a899031_payjsapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

's'

Browse Source
This commit is contained in:
eric
2026-03-26 03:15:23 -05:00
parent a1b50b6da9
commit 00b7ee27db
1 changed files with 75 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
app/routers/nomadvip.py
View File

@@ -3,11 +3,13 @@
import logging import logging
import random import random
import string import string
import threading
import time import time
import requests import requests
from fastapi import APIRouter, HTTPException, Request from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import JSONResponse, RedirectResponse 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 ..config import BASE_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL
from ..payment import ( from ..payment import (
@@ -21,6 +23,17 @@ from ..services import handle_payment_success, processed_orders
router = APIRouter(prefix="/nomadvip", tags=["nomadvip"]) 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
def _get_notify_path(provider_name: str) -> str: def _get_notify_path(provider_name: str) -> str:
if provider_name == "xorpay": if provider_name == "xorpay":
@@ -301,6 +314,13 @@ async def nomadvip_check_vip(user_id: str = ""):
def _get_pb_admin_token() -> str | None: def _get_pb_admin_token() -> str | None:
"""获取 PocketBase 管理员 token。""" """获取 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: if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
return None return None
@@ -310,13 +330,20 @@ def _get_pb_admin_token() -> str | None:
"/api/admins/auth-with-password", "/api/admins/auth-with-password",
]: ]:
try: try:
response = requests.post( response = _http_session.post(
f"{base}{path}", f"{base}{path}",
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD}, json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
if response.status_code == 200: if response.status_code == 200:
return response.json().get("token") 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: except Exception:
pass pass
return None return None
@@ -330,13 +357,13 @@ def _pb_list_email_links(admin_token: str, email: str) -> list[dict]:
base = (PB_URL or "").rstrip("/") base = (PB_URL or "").rstrip("/")
headers = {"Authorization": f"Bearer {admin_token}"} headers = {"Authorization": f"Bearer {admin_token}"}
for candidate in [email, email.lower()]: for candidate in {email, email.lower()}:
safe_email = _pb_escape(candidate) safe_email = _pb_escape(candidate)
response = requests.get( response = _http_session.get(
f"{base}/api/collections/vip_email_link/records", f"{base}/api/collections/vip_email_link/records",
params={"filter": f'email = "{safe_email}"', "perPage": 1}, params={"filter": f'email = "{safe_email}"', "perPage": 1},
headers=headers, headers=headers,
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
if response.status_code != 200: if response.status_code != 200:
continue continue
@@ -367,19 +394,19 @@ def _pb_upsert_email_link(
if items: if items:
record_id = items[0].get("id") record_id = items[0].get("id")
if record_id: if record_id:
requests.patch( _http_session.patch(
f"{base}/api/collections/vip_email_link/records/{record_id}", f"{base}/api/collections/vip_email_link/records/{record_id}",
json=payload, json=payload,
headers=headers, headers=headers,
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
return return
requests.post( _http_session.post(
f"{base}/api/collections/vip_email_link/records", f"{base}/api/collections/vip_email_link/records",
json=payload, json=payload,
headers=headers, headers=headers,
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
@@ -395,14 +422,14 @@ def _pb_move_site_vip(
base = (PB_URL or "").rstrip("/") base = (PB_URL or "").rstrip("/")
safe_from_user_id = _pb_escape(from_user_id) safe_from_user_id = _pb_escape(from_user_id)
safe_site_id = _pb_escape(site_id) safe_site_id = _pb_escape(site_id)
response = requests.get( response = _http_session.get(
f"{base}/api/collections/site_vip/records", f"{base}/api/collections/site_vip/records",
params={ params={
"filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"', "filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"',
"perPage": 500, "perPage": 500,
}, },
headers={"Authorization": f"Bearer {admin_token}"}, headers={"Authorization": f"Bearer {admin_token}"},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
if response.status_code != 200: if response.status_code != 200:
return return
@@ -416,18 +443,18 @@ def _pb_move_site_vip(
record_id = item.get("id") record_id = item.get("id")
if not record_id: if not record_id:
continue continue
requests.patch( _http_session.patch(
f"{base}/api/collections/site_vip/records/{record_id}", f"{base}/api/collections/site_vip/records/{record_id}",
json={"user_id": to_user_id}, json={"user_id": to_user_id},
headers=headers, headers=headers,
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
def _pb_latest_vip_payment(admin_token: str, user_id: str) -> dict | None: def _pb_latest_vip_payment(admin_token: str, user_id: str) -> dict | None:
base = (PB_URL or "").rstrip("/") base = (PB_URL or "").rstrip("/")
safe_user_id = _pb_escape(user_id) safe_user_id = _pb_escape(user_id)
response = requests.get( response = _http_session.get(
f"{base}/api/collections/payments/records", f"{base}/api/collections/payments/records",
params={ params={
"filter": f'user_id = "{safe_user_id}" && type = "vip"', "filter": f'user_id = "{safe_user_id}" && type = "vip"',
@@ -435,7 +462,7 @@ def _pb_latest_vip_payment(admin_token: str, user_id: str) -> dict | None:
"sort": "-created", "sort": "-created",
}, },
headers={"Authorization": f"Bearer {admin_token}"}, headers={"Authorization": f"Bearer {admin_token}"},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
if response.status_code != 200: if response.status_code != 200:
return None return None
@@ -452,7 +479,7 @@ def _pb_list_site_vip_by_order(
base = (PB_URL or "").rstrip("/") base = (PB_URL or "").rstrip("/")
safe_order_id = _pb_escape(order_id) safe_order_id = _pb_escape(order_id)
safe_site_id = _pb_escape(site_id) safe_site_id = _pb_escape(site_id)
response = requests.get( response = _http_session.get(
f"{base}/api/collections/site_vip/records", f"{base}/api/collections/site_vip/records",
params={ params={
"filter": f'order_id = "{safe_order_id}" && site_id = "{safe_site_id}"', "filter": f'order_id = "{safe_order_id}" && site_id = "{safe_site_id}"',
@@ -460,7 +487,7 @@ def _pb_list_site_vip_by_order(
"sort": "-created", "sort": "-created",
}, },
headers={"Authorization": f"Bearer {admin_token}"}, headers={"Authorization": f"Bearer {admin_token}"},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
if response.status_code != 200: if response.status_code != 200:
return [] return []
@@ -470,10 +497,10 @@ def _pb_list_site_vip_by_order(
def _pb_get_user_email(admin_token: str, user_id: str) -> str | None: def _pb_get_user_email(admin_token: str, user_id: str) -> str | None:
base = (PB_URL or "").rstrip("/") base = (PB_URL or "").rstrip("/")
response = requests.get( response = _http_session.get(
f"{base}/api/collections/users/records/{user_id}", f"{base}/api/collections/users/records/{user_id}",
headers={"Authorization": f"Bearer {admin_token}"}, headers={"Authorization": f"Bearer {admin_token}"},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
if response.status_code != 200: if response.status_code != 200:
return None return None
@@ -488,7 +515,7 @@ def _pb_find_linked_member_from_order(
) -> tuple[str, str] | None: ) -> tuple[str, str] | None:
base = (PB_URL or "").rstrip("/") base = (PB_URL or "").rstrip("/")
safe_user_id = _pb_escape(anonymous_user_id) safe_user_id = _pb_escape(anonymous_user_id)
response = requests.get( response = _http_session.get(
f"{base}/api/collections/payments/records", f"{base}/api/collections/payments/records",
params={ params={
"filter": f'user_id = "{safe_user_id}" && type = "vip"', "filter": f'user_id = "{safe_user_id}" && type = "vip"',
@@ -496,7 +523,7 @@ def _pb_find_linked_member_from_order(
"sort": "-created", "sort": "-created",
}, },
headers={"Authorization": f"Bearer {admin_token}"}, headers={"Authorization": f"Bearer {admin_token}"},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
if response.status_code != 200: if response.status_code != 200:
return None return None
@@ -538,7 +565,7 @@ def _pb_ensure_site_vip(
safe_from_user_id = _pb_escape(from_user_id) safe_from_user_id = _pb_escape(from_user_id)
safe_to_user_id = _pb_escape(to_user_id) safe_to_user_id = _pb_escape(to_user_id)
safe_site_id = _pb_escape(site_id) safe_site_id = _pb_escape(site_id)
source_records = requests.get( source_records = _http_session.get(
f"{base}/api/collections/site_vip/records", f"{base}/api/collections/site_vip/records",
params={ params={
"filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"', "filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"',
@@ -546,9 +573,9 @@ def _pb_ensure_site_vip(
"sort": "-created", "sort": "-created",
}, },
headers={"Authorization": f"Bearer {admin_token}"}, headers={"Authorization": f"Bearer {admin_token}"},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
to_records = requests.get( to_records = _http_session.get(
f"{base}/api/collections/site_vip/records", f"{base}/api/collections/site_vip/records",
params={ params={
"filter": f'user_id = "{safe_to_user_id}" && site_id = "{safe_site_id}"', "filter": f'user_id = "{safe_to_user_id}" && site_id = "{safe_site_id}"',
@@ -556,7 +583,7 @@ def _pb_ensure_site_vip(
"sort": "-created", "sort": "-created",
}, },
headers={"Authorization": f"Bearer {admin_token}"}, headers={"Authorization": f"Bearer {admin_token}"},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
payment = _pb_latest_vip_payment(admin_token, from_user_id) payment = _pb_latest_vip_payment(admin_token, from_user_id)
order_id = (payment or {}).get("order_id") or "" order_id = (payment or {}).get("order_id") or ""
@@ -578,11 +605,11 @@ def _pb_ensure_site_vip(
next_order_id = order_id or (canonical.get("order_id") or "").strip() next_order_id = order_id or (canonical.get("order_id") or "").strip()
if next_order_id: if next_order_id:
patch_body["order_id"] = next_order_id patch_body["order_id"] = next_order_id
requests.patch( _http_session.patch(
f"{base}/api/collections/site_vip/records/{canonical['id']}", f"{base}/api/collections/site_vip/records/{canonical['id']}",
json=patch_body, json=patch_body,
headers=headers, headers=headers,
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
duplicate_ids = set() duplicate_ids = set()
@@ -591,21 +618,21 @@ def _pb_ensure_site_vip(
if record_id and record_id != canonical["id"]: if record_id and record_id != canonical["id"]:
duplicate_ids.add(record_id) duplicate_ids.add(record_id)
for record_id in duplicate_ids: for record_id in duplicate_ids:
requests.delete( _http_session.delete(
f"{base}/api/collections/site_vip/records/{record_id}", f"{base}/api/collections/site_vip/records/{record_id}",
headers={"Authorization": f"Bearer {admin_token}"}, headers={"Authorization": f"Bearer {admin_token}"},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
return return
if not order_id: if not order_id:
return return
requests.post( _http_session.post(
f"{base}/api/collections/site_vip/records", f"{base}/api/collections/site_vip/records",
json={"user_id": to_user_id, "site_id": site_id, "order_id": order_id}, json={"user_id": to_user_id, "site_id": site_id, "order_id": order_id},
headers=headers, headers=headers,
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
@@ -627,10 +654,10 @@ async def nomadvip_check_vip_by_email(item: dict):
return {"vip": False, "error": "missing email or password"} return {"vip": False, "error": "missing email or password"}
base = (PB_URL or "").rstrip("/") base = (PB_URL or "").rstrip("/")
login_res = requests.post( login_res = _http_session.post(
f"{base}/api/collections/users/auth-with-password", f"{base}/api/collections/users/auth-with-password",
json={"identity": email, "password": password}, json={"identity": email, "password": password},
timeout=10, timeout=DEFAULT_HTTP_TIMEOUT,
) )
if login_res.status_code != 200: if login_res.status_code != 200:
try: try:
@@ -663,6 +690,7 @@ async def nomadvip_check_vip_by_email(item: dict):
pb = get_pb_client() pb = get_pb_client()
matched_user_id = None matched_user_id = None
vip_check_cache: dict[str, bool] = {}
recovered_anonymous_user_id = ( recovered_anonymous_user_id = (
anonymous_user_id anonymous_user_id
if anonymous_user_id.startswith("user") and anonymous_user_id[4:].isdigit() if anonymous_user_id.startswith("user") and anonymous_user_id[4:].isdigit()
@@ -670,7 +698,16 @@ async def nomadvip_check_vip_by_email(item: dict):
) )
recovery_source = None recovery_source = None
if recovered_anonymous_user_id and _uid_has_vip(pb, recovered_anonymous_user_id): 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 matched_user_id = recovered_anonymous_user_id
recovery_source = "request_user_id" recovery_source = "request_user_id"
else: else:
@@ -691,8 +728,8 @@ async def nomadvip_check_vip_by_email(item: dict):
if linked_anonymous_user_id and linked_anonymous_user_id not in linked_ids: if linked_anonymous_user_id and linked_anonymous_user_id not in linked_ids:
linked_ids.append(linked_anonymous_user_id) linked_ids.append(linked_anonymous_user_id)
for linked_uid in linked_ids: for linked_uid in dict.fromkeys(linked_ids):
if not _uid_has_vip(pb, linked_uid): if not has_vip_cached(linked_uid):
continue continue
matched_user_id = linked_uid matched_user_id = linked_uid
if linked_anonymous_user_id.startswith("user") and linked_anonymous_user_id[4:].isdigit(): if linked_anonymous_user_id.startswith("user") and linked_anonymous_user_id[4:].isdigit():
@@ -705,7 +742,7 @@ async def nomadvip_check_vip_by_email(item: dict):
if matched_user_id: if matched_user_id:
break break
if not matched_user_id and _uid_has_vip(pb, pb_user_id): if not matched_user_id and has_vip_cached(pb_user_id):
matched_user_id = pb_user_id matched_user_id = pb_user_id
recovered_anonymous_user_id = None recovered_anonymous_user_id = None
recovery_source = "pb_user_id" recovery_source = "pb_user_id"