"""支付成功后的统一业务处理（基于 SDK）""" import logging import requests from ..config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD from .pb import get_pb_client from .memos import create_memos_user # 已处理订单（防重复） processed_orders: set[str] = set() DEFAULT_PASSWORD = "12345678" 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""" 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: return r.json().get("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}") processed_orders.add(order_id) except Exception as e: logging.error("PocketBase 操作失败 order_id=%s error=%s", order_id, e) raise