258 lines
9.6 KiB
Python
258 lines
9.6 KiB
Python
"""支付成功后的统一业务处理（基于 SDK）"""
|
||
import logging
|
||
import threading
|
||
import time
|
||
|
||
import requests
|
||
from cachetools import TTLCache
|
||
|
||
from ..config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD
|
||
from .pb import get_pb_client
|
||
from .memos import create_memos_user
|
||
|
||
# 已处理订单（防重复），24h TTL，最大 10000 条，避免内存泄漏
|
||
_processed_orders: TTLCache[str, bool] = TTLCache(maxsize=10000, ttl=86400)
|
||
_processed_orders_lock = threading.Lock()
|
||
|
||
# 兼容旧导入
|
||
processed_orders = _processed_orders
|
||
|
||
DEFAULT_PASSWORD = "12345678"
|
||
|
||
# Admin token 缓存（5 分钟），与 meetup/salon 一致
|
||
_ADMIN_TOKEN_CACHE: tuple[str, float] | None = None
|
||
_ADMIN_TOKEN_CACHE_TTL = 300
|
||
|
||
|
||
def try_add_processed_order(order_id: str) -> bool:
|
||
"""原子性标记订单已处理。返回 True 表示首次处理，False 表示已处理过（跳过）。"""
|
||
with _processed_orders_lock:
|
||
if order_id in _processed_orders:
|
||
return False
|
||
_processed_orders[order_id] = True
|
||
return True
|
||
|
||
|
||
def remove_processed_order(order_id: str) -> None:
|
||
"""处理失败时移除标记，允许支付方重试。"""
|
||
with _processed_orders_lock:
|
||
_processed_orders.pop(order_id, None)
|
||
|
||
|
||
def parse_order_id(order_id: str) -> tuple[str, str]:
|
||
"""从 order_id 解析 user_id 和 pay_type"""
|
||
if "_order_" not in order_id:
|
||
return "", "unknown"
|
||
prefix = order_id.split("_order_")[0]
|
||
if "_" in prefix:
|
||
parts = prefix.rsplit("_", 1)
|
||
return parts[0], parts[1].lower()
|
||
return prefix, "unknown"
|
||
|
||
|
||
def _decode_pending_email(user_id: str) -> str | None:
|
||
"""从 pending_hex 格式解析邮箱"""
|
||
if not user_id.startswith("pending_"):
|
||
return None
|
||
hex_part = user_id[8:] # 去掉 "pending_"
|
||
try:
|
||
return bytes.fromhex(hex_part).decode("utf-8")
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _create_user_by_email(email: str) -> str | None:
|
||
"""用管理员创建用户，返回 user_id。若已存在则返回已有用户 id"""
|
||
token, _ = _get_admin_token()
|
||
if not token:
|
||
return None
|
||
base = (PB_URL or "").rstrip("/")
|
||
r = requests.post(
|
||
f"{base}/api/collections/users/records",
|
||
json={
|
||
"email": email,
|
||
"password": DEFAULT_PASSWORD,
|
||
"passwordConfirm": DEFAULT_PASSWORD,
|
||
},
|
||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
|
||
timeout=10,
|
||
)
|
||
if r.status_code == 200:
|
||
return r.json().get("id")
|
||
if r.status_code == 400:
|
||
try:
|
||
err = r.json()
|
||
if "already" in str(err.get("data", {}).get("email", {}).get("message", "")).lower():
|
||
# 用户已存在，查询获取 id
|
||
get_r = requests.get(
|
||
f"{base}/api/collections/users/records",
|
||
params={"filter": f'email="{email}"', "perPage": 1},
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
timeout=10,
|
||
)
|
||
if get_r.status_code == 200:
|
||
items = get_r.json().get("items", [])
|
||
if items:
|
||
return items[0].get("id")
|
||
except Exception:
|
||
pass
|
||
logging.warning(f"create user failed: {r.status_code} {r.text[:200]}")
|
||
return None
|
||
|
||
|
||
def _get_admin_token() -> tuple[str | None, str | None]:
|
||
"""获取 PocketBase 管理员 token，带 5 分钟缓存"""
|
||
global _ADMIN_TOKEN_CACHE
|
||
now = time.time()
|
||
if _ADMIN_TOKEN_CACHE and _ADMIN_TOKEN_CACHE[1] > now:
|
||
return _ADMIN_TOKEN_CACHE[0], None
|
||
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:
|
||
r = requests.post(
|
||
f"{base}{path}",
|
||
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
|
||
timeout=10,
|
||
)
|
||
if r.status_code == 200:
|
||
token = r.json().get("token")
|
||
if token:
|
||
_ADMIN_TOKEN_CACHE = (token, now + _ADMIN_TOKEN_CACHE_TTL)
|
||
return token, None
|
||
if r.status_code == 404:
|
||
continue
|
||
except Exception:
|
||
pass
|
||
return None, "认证失败"
|
||
|
||
|
||
def handle_payment_success(
|
||
order_id: str,
|
||
pay_price: str,
|
||
provider_name: str,
|
||
*,
|
||
use_memos: bool = False,
|
||
site_id: str | None = None,
|
||
):
|
||
"""
|
||
支付成功后的统一业务处理
|
||
:param use_memos: 是否创建 Memos 用户，仅 nomadvip 为 True
|
||
:param site_id: 站点标识（digital/meetup/vip），用于写入 site_vip
|
||
"""
|
||
user_id, pay_type = parse_order_id(order_id)
|
||
if not user_id:
|
||
return
|
||
|
||
# meetup/salon 新用户：支付成功后才创建用户。solanRed 记录创建时用的是 pending_xxx，查找需用原值
|
||
solan_lookup_user_id = user_id
|
||
if pay_type in ("meetup", "salon") and user_id.startswith("pending_"):
|
||
email = _decode_pending_email(user_id)
|
||
if email:
|
||
real_id = _create_user_by_email(email)
|
||
if real_id:
|
||
user_id = real_id
|
||
logging.info(f"meetup 支付成功，已创建用户: {email}")
|
||
else:
|
||
logging.error(f"meetup 支付成功但创建用户失败: {email}")
|
||
return
|
||
|
||
logging.info("[payjsapi] 支付成功 provider=%s user_id=%s pay_type=%s site_id=%s order_id=%s",
|
||
provider_name, user_id, pay_type, site_id, order_id)
|
||
|
||
if pay_type == "vip" and use_memos:
|
||
create_memos_user(user_id, password="12345678")
|
||
|
||
if pay_type in ("meetup", "salon"):
|
||
logging.info(f"{pay_type} 类型支付成功，不创建 memos 用户 - user_id: {user_id}")
|
||
|
||
valid_types = ["vip", "book", "meetup", "video", "salon"]
|
||
if pay_type not in valid_types:
|
||
pay_type = "vip"
|
||
|
||
amount_cents = int(float(pay_price or 0) * 100)
|
||
|
||
try:
|
||
pb_client = get_pb_client()
|
||
if pay_type == "meetup":
|
||
solan = pb_client.collection("solan")
|
||
solan_collection = "solan"
|
||
elif pay_type == "salon":
|
||
solan = pb_client.collection("solanRed")
|
||
solan_collection = "solanRed"
|
||
else:
|
||
solan = None
|
||
solan_collection = None
|
||
|
||
if solan is not None:
|
||
update_payload = {
|
||
"order_id": order_id,
|
||
"amount": amount_cents,
|
||
}
|
||
if pay_type == "salon":
|
||
update_payload["vip"] = True
|
||
elif pay_type == "meetup":
|
||
update_payload["type"] = pay_type
|
||
existing = solan.get_list(1, 1, {
|
||
"filter": f'user_id = "{solan_lookup_user_id}"',
|
||
"sort": "-created",
|
||
})
|
||
if existing.items:
|
||
record_id = existing.items[0].id
|
||
patch = dict(update_payload)
|
||
if solan_lookup_user_id != user_id:
|
||
patch["user_id"] = user_id
|
||
solan.update(record_id, patch)
|
||
logging.info(f"{solan_collection} 支付字段补齐成功 - user_id: {user_id}, record_id: {record_id}")
|
||
else:
|
||
create_payload = {"user_id": user_id, **update_payload}
|
||
if pay_type == "salon":
|
||
create_payload["type"] = ""
|
||
solan.create(create_payload)
|
||
logging.info(f"{solan_collection} 完整记录创建（兜底） - user_id: {user_id}")
|
||
elif pay_type not in ("meetup", "salon"):
|
||
existing = pb_client.collection("payments").get_list(1, 1, {
|
||
"filter": f'order_id = "{order_id}"',
|
||
})
|
||
if not existing.items:
|
||
payload = {
|
||
"user_id": user_id,
|
||
"type": pay_type,
|
||
"order_id": order_id,
|
||
"amount": amount_cents,
|
||
"total_fee": amount_cents,
|
||
"channel": "wxpay",
|
||
}
|
||
pb_client.collection("payments").create(payload)
|
||
|
||
# 写入 site_vip（站点独立 VIP）。meetup 写 site_id=meetup，salon 写 site_id=salon
|
||
effective_site_id = "meetup" if pay_type == "meetup" else ("salon" if pay_type == "salon" else site_id)
|
||
if effective_site_id:
|
||
try:
|
||
site_vip = pb_client.collection("site_vip")
|
||
existing_vip = site_vip.get_list(1, 1, {
|
||
"filter": f'user_id = "{user_id}" && site_id = "{effective_site_id}"',
|
||
})
|
||
if existing_vip.items:
|
||
site_vip.update(existing_vip.items[0].id, {
|
||
"order_id": order_id,
|
||
"expires_at": "",
|
||
})
|
||
logging.info(f"site_vip 更新成功 - user_id: {user_id}, site_id: {effective_site_id}")
|
||
else:
|
||
site_vip.create({
|
||
"user_id": user_id,
|
||
"site_id": effective_site_id,
|
||
"order_id": order_id,
|
||
"expires_at": "",
|
||
})
|
||
logging.info(f"site_vip 创建成功 - user_id: {user_id}, site_id: {effective_site_id}")
|
||
except Exception as e:
|
||
logging.warning(f"site_vip 写入失败（不影响主流程）: {e}")
|
||
|
||
# 由回调方在验签后、调用前通过 try_add_processed_order 原子性标记，此处不再 add
|
||
except Exception as e:
|
||
logging.error("PocketBase 操作失败 order_id=%s error=%s", order_id, e)
|
||
raise
|