433 lines
15 KiB
Python
433 lines
15 KiB
Python
"""支付订单业务逻辑。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import secrets
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import func, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from auth.account_quota import default_stop_worker, sync_user_account_quota
|
|
from auth.roles import is_admin
|
|
from auth.system_settings import SystemSettingsData, load_settings
|
|
from models.models import PaymentOrder, User
|
|
from . import alipay, wechat
|
|
|
|
logger = logging.getLogger("payments.service")
|
|
|
|
ORDER_TTL_MINUTES = 30
|
|
ORDER_STATUSES = ("pending", "paid", "expired", "cancelled", "refunded")
|
|
|
|
|
|
def generate_order_no() -> str:
|
|
return f"AC{datetime.utcnow().strftime('%Y%m%d%H%M%S')}{secrets.token_hex(4).upper()}"
|
|
|
|
|
|
def yuan_to_fen(yuan: float) -> int:
|
|
return int(round(yuan * 100))
|
|
|
|
|
|
def fen_to_yuan(fen: int) -> float:
|
|
return round(fen / 100, 2)
|
|
|
|
|
|
def order_to_dict(order: PaymentOrder, demo_mode: bool = False) -> dict[str, Any]:
|
|
return {
|
|
"order_no": order.order_no,
|
|
"channel": order.channel,
|
|
"slots": order.slots,
|
|
"amount_fen": order.amount_fen,
|
|
"amount_yuan": fen_to_yuan(order.amount_fen),
|
|
"status": order.status,
|
|
"qr_code": order.qr_code,
|
|
"pay_url": order.pay_url,
|
|
"paid_at": order.paid_at.isoformat() if order.paid_at else None,
|
|
"expires_at": order.expires_at.isoformat() if order.expires_at else None,
|
|
"created_at": order.created_at.isoformat() if order.created_at else None,
|
|
"demo_mode": demo_mode,
|
|
}
|
|
|
|
|
|
def payment_config_payload(settings: SystemSettingsData) -> dict[str, Any]:
|
|
channels = settings.available_payment_channels()
|
|
demo = settings.payment_demo_mode and settings.payment_enabled
|
|
return {
|
|
"payment_enabled": settings.payment_enabled,
|
|
"demo_mode": demo,
|
|
"unit_price": settings.account_slot_unit_price,
|
|
"min_slots": settings.account_slot_purchase_min,
|
|
"max_slots": settings.account_slot_purchase_max,
|
|
"channels": channels,
|
|
"wechat_available": "wechat" in channels,
|
|
"alipay_available": "alipay" in channels,
|
|
}
|
|
|
|
|
|
def _validate_purchase_request(settings: SystemSettingsData, slots: int, channel: str) -> None:
|
|
if not settings.payment_enabled:
|
|
raise HTTPException(status_code=400, detail="在线购买功能未开启")
|
|
if slots < settings.account_slot_purchase_min or slots > settings.account_slot_purchase_max:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"购买数量需在 {settings.account_slot_purchase_min}~{settings.account_slot_purchase_max} 之间",
|
|
)
|
|
if channel not in ("wechat", "alipay"):
|
|
raise HTTPException(status_code=400, detail="不支持的支付方式")
|
|
if channel == "wechat" and not settings.wechat_pay_enabled:
|
|
raise HTTPException(status_code=400, detail="微信支付未开启")
|
|
if channel == "alipay" and not settings.alipay_pay_enabled:
|
|
raise HTTPException(status_code=400, detail="支付宝支付未开启")
|
|
if not settings.payment_channel_selectable(channel):
|
|
name = "微信" if channel == "wechat" else "支付宝"
|
|
raise HTTPException(status_code=400, detail=f"{name}支付暂不可用，请联系管理员")
|
|
if not settings.payment_channel_available(channel):
|
|
if not settings.payment_demo_mode:
|
|
name = "微信" if channel == "wechat" else "支付宝"
|
|
raise HTTPException(status_code=400, detail=f"{name}支付尚未配置完成，请联系管理员")
|
|
|
|
|
|
async def create_order(
|
|
db: AsyncSession,
|
|
user: User,
|
|
slots: int,
|
|
channel: str,
|
|
) -> tuple[PaymentOrder, bool]:
|
|
if is_admin(user.role):
|
|
raise HTTPException(status_code=400, detail="管理员账号无需购买额度")
|
|
settings = await load_settings(db)
|
|
_validate_purchase_request(settings, slots, channel)
|
|
|
|
amount_fen = yuan_to_fen(settings.account_slot_unit_price * slots)
|
|
if amount_fen < 1:
|
|
raise HTTPException(status_code=400, detail="订单金额无效")
|
|
|
|
order = PaymentOrder(
|
|
order_no=generate_order_no(),
|
|
user_id=user.id,
|
|
channel=channel,
|
|
slots=slots,
|
|
amount_fen=amount_fen,
|
|
status="pending",
|
|
expires_at=datetime.utcnow() + timedelta(minutes=ORDER_TTL_MINUTES),
|
|
)
|
|
db.add(order)
|
|
await db.flush()
|
|
|
|
demo_mode = False
|
|
notify_base = settings.app_url_normalized()
|
|
description = f"抖音账号额度 x{slots}"
|
|
|
|
try:
|
|
if channel == "wechat" and settings.wechat_pay_enabled and settings.wechat_pay_configured():
|
|
notify_url = f"{notify_base}/api/payments/notify/wechat"
|
|
order.qr_code = wechat.create_native_order(
|
|
settings,
|
|
order.order_no,
|
|
description,
|
|
amount_fen,
|
|
notify_url,
|
|
)
|
|
elif channel == "alipay" and settings.alipay_pay_enabled and settings.alipay_configured():
|
|
notify_url = f"{notify_base}/api/payments/notify/alipay"
|
|
order.qr_code = alipay.create_precreate_order(
|
|
settings,
|
|
order.order_no,
|
|
description,
|
|
f"{fen_to_yuan(amount_fen):.2f}",
|
|
notify_url,
|
|
)
|
|
elif settings.payment_demo_mode and settings.payment_channel_selectable(channel):
|
|
demo_mode = True
|
|
order.qr_code = f"DEMO-{order.order_no}"
|
|
else:
|
|
raise HTTPException(status_code=400, detail="支付渠道未配置")
|
|
except HTTPException:
|
|
await db.rollback()
|
|
raise
|
|
except Exception as exc:
|
|
await db.rollback()
|
|
logger.exception("Create payment order failed")
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
await db.commit()
|
|
await db.refresh(order)
|
|
return order, demo_mode
|
|
|
|
|
|
async def get_user_order(db: AsyncSession, user: User, order_no: str) -> PaymentOrder:
|
|
result = await db.execute(
|
|
select(PaymentOrder).where(
|
|
PaymentOrder.order_no == order_no,
|
|
PaymentOrder.user_id == user.id,
|
|
)
|
|
)
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
raise HTTPException(status_code=404, detail="订单不存在")
|
|
if order.status == "pending" and order.expires_at < datetime.utcnow():
|
|
order.status = "expired"
|
|
await db.commit()
|
|
await db.refresh(order)
|
|
return order
|
|
|
|
|
|
async def _get_order_user(db: AsyncSession, order: PaymentOrder) -> User:
|
|
result = await db.execute(select(User).where(User.id == order.user_id))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
return user
|
|
|
|
|
|
async def _apply_order_slots(db: AsyncSession, order: PaymentOrder, user: User) -> None:
|
|
if order.slots_applied:
|
|
return
|
|
if is_admin(user.role) or (user.max_accounts is not None and user.max_accounts < 0):
|
|
order.slots_applied = True
|
|
return
|
|
base = int(user.max_accounts) if user.max_accounts is not None else 0
|
|
user.max_accounts = base + int(order.slots)
|
|
order.slots_applied = True
|
|
await sync_user_account_quota(db, user, stop_worker=default_stop_worker)
|
|
|
|
|
|
async def _revoke_order_slots(db: AsyncSession, order: PaymentOrder, user: User) -> None:
|
|
if not order.slots_applied:
|
|
return
|
|
if is_admin(user.role) or (user.max_accounts is not None and user.max_accounts < 0):
|
|
order.slots_applied = False
|
|
return
|
|
base = int(user.max_accounts) if user.max_accounts is not None else 0
|
|
new_limit = max(0, base - int(order.slots))
|
|
user.max_accounts = new_limit
|
|
order.slots_applied = False
|
|
await sync_user_account_quota(db, user, stop_worker=default_stop_worker)
|
|
|
|
|
|
async def _get_order_by_no(db: AsyncSession, order_no: str) -> PaymentOrder:
|
|
result = await db.execute(select(PaymentOrder).where(PaymentOrder.order_no == order_no))
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
raise HTTPException(status_code=404, detail="订单不存在")
|
|
return order
|
|
|
|
|
|
async def fulfill_order(db: AsyncSession, order: PaymentOrder, trade_no: str | None = None) -> None:
|
|
if order.status == "paid" and order.slots_applied:
|
|
return
|
|
if order.status != "pending":
|
|
raise HTTPException(status_code=400, detail="订单状态不可支付")
|
|
|
|
user = await _get_order_user(db, order)
|
|
await _apply_order_slots(db, order, user)
|
|
|
|
order.status = "paid"
|
|
order.paid_at = datetime.utcnow()
|
|
if trade_no:
|
|
order.trade_no = trade_no
|
|
await db.commit()
|
|
|
|
|
|
async def simulate_pay(db: AsyncSession, user: User, order_no: str) -> PaymentOrder:
|
|
settings = await load_settings(db)
|
|
if not settings.payment_demo_mode:
|
|
raise HTTPException(status_code=403, detail="演示支付未开启")
|
|
order = await get_user_order(db, user, order_no)
|
|
if order.status != "pending":
|
|
raise HTTPException(status_code=400, detail="订单不可支付")
|
|
await fulfill_order(db, order, trade_no=f"DEMO-{order.order_no}")
|
|
await db.refresh(order)
|
|
return order
|
|
|
|
|
|
async def handle_wechat_notify(db: AsyncSession, body: bytes, headers: dict[str, str]) -> bool:
|
|
settings = await load_settings(db)
|
|
if not settings.wechat_pay_configured():
|
|
return False
|
|
if not wechat.verify_notify_signature(
|
|
settings,
|
|
body,
|
|
headers.get("wechatpay-timestamp", ""),
|
|
headers.get("wechatpay-nonce", ""),
|
|
headers.get("wechatpay-signature", ""),
|
|
headers.get("wechatpay-serial", ""),
|
|
):
|
|
return False
|
|
|
|
import json
|
|
|
|
payload = json.loads(body.decode("utf-8"))
|
|
resource = payload.get("resource") or {}
|
|
data = wechat.decrypt_notify_resource(settings.wechat_api_v3_key.strip(), resource)
|
|
if data.get("trade_state") != "SUCCESS":
|
|
return True
|
|
|
|
order_no = data.get("out_trade_no")
|
|
trade_no = data.get("transaction_id")
|
|
result = await db.execute(select(PaymentOrder).where(PaymentOrder.order_no == order_no))
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
return False
|
|
order.notify_payload = body.decode("utf-8", errors="replace")
|
|
await fulfill_order(db, order, trade_no=trade_no)
|
|
return True
|
|
|
|
|
|
async def handle_alipay_notify(db: AsyncSession, params: dict[str, Any]) -> bool:
|
|
settings = await load_settings(db)
|
|
if not settings.alipay_configured():
|
|
return False
|
|
if not alipay.verify_notify(params, settings):
|
|
return False
|
|
if params.get("trade_status") not in ("TRADE_SUCCESS", "TRADE_FINISHED"):
|
|
return True
|
|
|
|
order_no = params.get("out_trade_no")
|
|
trade_no = params.get("trade_no")
|
|
result = await db.execute(select(PaymentOrder).where(PaymentOrder.order_no == order_no))
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
return False
|
|
order.notify_payload = str(params)
|
|
await fulfill_order(db, order, trade_no=trade_no)
|
|
return True
|
|
|
|
|
|
async def refresh_expired_orders(db: AsyncSession) -> None:
|
|
now = datetime.utcnow()
|
|
await db.execute(
|
|
update(PaymentOrder)
|
|
.where(PaymentOrder.status == "pending", PaymentOrder.expires_at < now)
|
|
.values(status="expired")
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
def _order_is_demo(order: PaymentOrder) -> bool:
|
|
trade_no = (order.trade_no or "").strip()
|
|
return trade_no.startswith("DEMO-")
|
|
|
|
|
|
def order_list_item(order: PaymentOrder, user: User | None = None) -> dict[str, Any]:
|
|
payload = {
|
|
"id": order.id,
|
|
"order_no": order.order_no,
|
|
"user_id": order.user_id,
|
|
"username": user.username if user else None,
|
|
"display_name": user.display_name if user else None,
|
|
"channel": order.channel,
|
|
"slots": order.slots,
|
|
"amount_fen": order.amount_fen,
|
|
"amount_yuan": fen_to_yuan(order.amount_fen),
|
|
"status": order.status,
|
|
"trade_no": order.trade_no,
|
|
"is_demo": _order_is_demo(order),
|
|
"paid_at": order.paid_at,
|
|
"created_at": order.created_at,
|
|
}
|
|
return payload
|
|
|
|
|
|
async def list_orders(
|
|
db: AsyncSession,
|
|
current_user: User,
|
|
*,
|
|
status: str | None = None,
|
|
channel: str | None = None,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
) -> dict[str, Any]:
|
|
await refresh_expired_orders(db)
|
|
|
|
page = max(1, page)
|
|
page_size = max(1, min(100, page_size))
|
|
|
|
filters = []
|
|
if not is_admin(current_user.role):
|
|
filters.append(PaymentOrder.user_id == current_user.id)
|
|
if status:
|
|
filters.append(PaymentOrder.status == status)
|
|
if channel in ("wechat", "alipay"):
|
|
filters.append(PaymentOrder.channel == channel)
|
|
|
|
count_stmt = select(func.count()).select_from(PaymentOrder)
|
|
for clause in filters:
|
|
count_stmt = count_stmt.where(clause)
|
|
total = int((await db.execute(count_stmt)).scalar() or 0)
|
|
|
|
paid_filters = list(filters) + [PaymentOrder.status == "paid"]
|
|
paid_count_stmt = select(func.count()).select_from(PaymentOrder)
|
|
for clause in paid_filters:
|
|
paid_count_stmt = paid_count_stmt.where(clause)
|
|
paid_count = int((await db.execute(paid_count_stmt)).scalar() or 0)
|
|
|
|
paid_amount_stmt = select(func.coalesce(func.sum(PaymentOrder.amount_fen), 0)).select_from(PaymentOrder)
|
|
for clause in paid_filters:
|
|
paid_amount_stmt = paid_amount_stmt.where(clause)
|
|
paid_amount_fen = int((await db.execute(paid_amount_stmt)).scalar() or 0)
|
|
|
|
stmt = select(PaymentOrder, User).join(User, User.id == PaymentOrder.user_id)
|
|
for clause in filters:
|
|
stmt = stmt.where(clause)
|
|
stmt = (
|
|
stmt.order_by(PaymentOrder.created_at.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
)
|
|
rows = (await db.execute(stmt)).all()
|
|
|
|
items = [order_list_item(order, user) for order, user in rows]
|
|
return {
|
|
"items": items,
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"paid_count": paid_count,
|
|
"paid_amount_yuan": fen_to_yuan(paid_amount_fen),
|
|
}
|
|
|
|
|
|
async def admin_update_order_status(
|
|
db: AsyncSession,
|
|
order_no: str,
|
|
new_status: str,
|
|
) -> dict[str, Any]:
|
|
if new_status not in ORDER_STATUSES:
|
|
raise HTTPException(status_code=400, detail="无效的订单状态")
|
|
|
|
order = await _get_order_by_no(db, order_no)
|
|
user = await _get_order_user(db, order)
|
|
old_status = order.status
|
|
|
|
if new_status == old_status:
|
|
return order_list_item(order, user)
|
|
|
|
if new_status == "paid":
|
|
await _apply_order_slots(db, order, user)
|
|
order.status = "paid"
|
|
if not order.paid_at:
|
|
order.paid_at = datetime.utcnow()
|
|
else:
|
|
if order.slots_applied:
|
|
await _revoke_order_slots(db, order, user)
|
|
order.status = new_status
|
|
if new_status not in ("paid", "refunded"):
|
|
order.paid_at = None
|
|
|
|
await db.commit()
|
|
await db.refresh(order)
|
|
return order_list_item(order, user)
|
|
|
|
|
|
async def admin_delete_order(db: AsyncSession, order_no: str) -> None:
|
|
order = await _get_order_by_no(db, order_no)
|
|
if order.slots_applied:
|
|
user = await _get_order_user(db, order)
|
|
await _revoke_order_slots(db, order, user)
|
|
await db.delete(order)
|
|
await db.commit()
|