mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

团队积分V1

Browse Source
This commit is contained in:
GinHa
2026-08-14 15:23:46 +08:00
parent 516106c045
commit fac54b5667
70 changed files with 6670 additions and 4649 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/services/payment.py
+170 -180
View File
@@ -30,9 +30,12 @@ from app.enums.credit_record import (
)
from app.services.credit_record_meta_service import CreditRecordMeta, build_recharge_meta
from app.services.credits import add_credits, deduct_credits
from app.services.credit.product_service import product_to_dict
from app.services.credit.subscription_service import fulfill_payment_product, revoke_payment_order_credits
from app.services.credit.upgrade_service import quote_and_reserve_product_purchase, release_upgrade_reservation
from app.enums.common import PaymentOrderSourceEnum
from app.enums.credit_product import CreditProductType
from app.models.team import Team
from app.services.credit.locking import acquire_team_business_lock, acquire_user_credit_lock
from app.services.credit.product_service import ensure_repeat_purchase_allowed, quote_product, resolve_team_purchase_context
from app.services.credit.subscription_service import fulfill_payment_product
from app.services.credit.utils import to_credit_decimal, utc_now
from app.utils.id_gen import generate_id, generate_order_no
@@ -259,8 +262,6 @@ async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool
return False
order.status = "cancelled"
if order.upgrade_period_ids_json:
await release_upgrade_reservation(db, order=order, released_at=utc_now())
await db.flush()
logger.info(
f"ORDER_EXPIRED order_no={order.order_no} user={order.user_id} "
@@ -334,10 +335,6 @@ async def expire_all_pending_orders(db: AsyncSession) -> int:
log_user_id = str(order.user_id)
log_amount = order.amount
order.status = "cancelled"
if order.upgrade_period_ids_json:
await release_upgrade_reservation(
db, order=order, released_at=utc_now()
)
await db.commit()
expired_count += 1
@@ -420,9 +417,14 @@ async def create_recharge_order(
method: str = "wechat",
*,
product_id: str | None = None,
quantity: int = 1,
request_time: datetime | None = None,
) -> PaymentOrder:
"""创建支付订单；支付渠道流程保持原样，仅增加积分商品快照和订阅升级预留。"""
"""创建线上支付订单。
商品订单在创建时冻结商品、定价和数量快照；之后商品改价、下架或软删除均不改变订单合同。
个人/团队订阅存在未完成订单时全局阻止再次创建订阅订单。
"""
db_configs = await _get_payment_configs(db)
mock_mode = _is_mock_mode(db_configs)
if not mock_mode:
@@ -434,120 +436,184 @@ async def create_recharge_order(
raise ValueError("支付宝支付未完成配置，请联系管理员")
elif method == "wechat":
required_configs = [
"payment_wechat_appid",
"payment_wechat_mch_id",
"payment_wechat_private_key",
"payment_wechat_cert_serial_no",
"payment_wechat_api_v3_key",
"payment_wechat_appid", "payment_wechat_mch_id", "payment_wechat_private_key",
"payment_wechat_cert_serial_no", "payment_wechat_api_v3_key",
]
missing_configs = [key for key in required_configs if not db_configs.get(key)]
if missing_configs:
raise ValueError(f"微信支付未完成配置，缺少: {', '.join(missing_configs)}，请联系管理员")
else:
raise ValueError("不支持的线上支付方式")
checked_at = request_time or utc_now()
await acquire_user_credit_lock(db, user_id)
user_result = await db.execute(select(User).where(User.id == user_id).limit(1))
user = user_result.scalar_one_or_none()
if user is None:
raise ValueError("用户不存在")
order_id = generate_id()
order_no = generate_order_no()
product: CreditProduct | None = None
quote = None
final_price = to_credit_decimal(price)
total_credits = to_credit_decimal(float(credits or 0) + float(bonus_credits or 0))
snapshot = None
product_type = None
team_id_snapshot = None
quoted_unit = final_price
quoted_amount = final_price
actual_unit = final_price
purchase_scene = "legacy_recharge"
price_type = "regular"
if product_id:
product_result = await db.execute(
select(CreditProduct)
.where(
CreditProduct.id == product_id,
CreditProduct.deleted_at.is_(None),
CreditProduct.is_active.is_(True),
)
.limit(1)
)
product = product_result.scalar_one_or_none()
if product is None:
raise ValueError("积分商品不存在、已下架或已删除")
product_type = product.product_type
if product_type == CreditProductType.TEAM_SUBSCRIPTION.value:
if not 2 <= int(quantity) <= 20:
raise ValueError("客户端团队套餐单次购买数量必须在2到20之间")
team, available, reason = await resolve_team_purchase_context(db, user=user)
if not available:
raise ValueError(reason or "当前不能购买团队订阅套餐")
if team is not None:
await acquire_team_business_lock(db, team.id)
team, available, reason = await resolve_team_purchase_context(db, user=user)
if not available or team is None:
raise ValueError(reason or "当前不能购买团队订阅套餐")
team_id_snapshot = team.id if team else None
first_purchase = bool(team is None or team.first_subscription_paid_at is None)
else:
if int(quantity) != 1:
raise ValueError("个人订阅和积分增值包购买数量固定为1")
quantity = 1
first_purchase = user.first_membership_paid_at is None if product_type == CreditProductType.SUBSCRIPTION.value else False
if product_type in {CreditProductType.SUBSCRIPTION.value, CreditProductType.TEAM_SUBSCRIPTION.value}:
ensure_repeat_purchase_allowed(product, first_purchase=first_purchase)
incomplete_result = await db.execute(
select(PaymentOrder.id).where(
PaymentOrder.user_id == user_id,
PaymentOrder.product_type.in_([
CreditProductType.SUBSCRIPTION.value,
CreditProductType.TEAM_SUBSCRIPTION.value,
]),
(
(PaymentOrder.status == "pending")
| ((PaymentOrder.status == "paid") & (PaymentOrder.fulfillment_status != "fulfilled"))
),
).limit(1)
)
if incomplete_result.scalar_one_or_none():
raise ValueError("存在未完成的订阅订单，请先前往订单记录完成支付或处理原订单")
quote = quote_product(
product,
first_purchase=first_purchase,
request_time=checked_at,
quantity=int(quantity),
)
quoted_unit = quote.quoted_unit_price
quoted_amount = quote.quoted_amount
final_price = quoted_amount
actual_unit = quoted_unit
purchase_scene = quote.purchase_scene
price_type = quote.price_type
label = product.name
if product_type == CreditProductType.TEAM_SUBSCRIPTION.value:
total_credits = to_credit_decimal((product.monthly_grant_credits or 0) * int(quantity))
elif product_type == CreditProductType.SUBSCRIPTION.value:
total_credits = to_credit_decimal(product.monthly_grant_credits or 0)
else:
total_credits = to_credit_decimal(product.grant_credits or 0)
snapshot = {
"id": product.id,
"product_code": product.product_code,
"product_type": product.product_type,
"name": product.name,
"description": product.description,
"features": product.features_json or [],
"tier_code": product.tier_code,
"tier_rank": product.tier_rank,
"billing_cycle": product.billing_cycle,
"monthly_grant_credits": float(product.monthly_grant_credits or 0),
"first_purchase_price": float(product.first_purchase_price or 0),
"regular_price": float(product.regular_price or 0),
"activity_price": float(product.activity_price) if product.activity_price is not None else None,
"activity_start_at": product.activity_start_at.isoformat() if product.activity_start_at else None,
"activity_end_at": product.activity_end_at.isoformat() if product.activity_end_at else None,
"renewal_enabled": bool(product.renewal_enabled),
"grant_credits": float(product.grant_credits or 0),
"validity_months": product.validity_months,
"credit_level": product.credit_level,
"currency": product.currency,
}
# 固定锁序：user advisory 已获取；团队订单已先获取 Team advisory，最后才锁 User 行。
locked_user_result = await db.execute(select(User).where(User.id == user_id).limit(1).with_for_update())
locked_user = locked_user_result.scalar_one_or_none()
if locked_user is None:
raise ValueError("用户不存在")
if locked_user.team_id != user.team_id:
raise ValueError("用户团队关系已发生变化，请刷新后重试")
user = locked_user
# 先持久化订单主记录，再做升级周期预留。订阅周期的 upgrade_order_id
# 有外键约束，若先更新周期后插入订单，flush 顺序可能触发外键异常。
order = PaymentOrder(
id=order_id,
user_id=user_id,
order_no=order_no,
amount=price,
credits=round(float(credits or 0) + float(bonus_credits or 0), 2),
amount=final_price,
credits=total_credits,
payment_method=method,
order_source=PaymentOrderSourceEnum.ONLINE_PAYMENT.value,
status="pending",
purchase_scene="legacy_recharge",
price_type="regular",
product_id=product.id if product else None,
product_type=product_type,
purchase_scene=purchase_scene,
price_type=price_type,
product_code_snapshot=product.product_code if product else None,
product_name_snapshot=label,
target_price_snapshot=price,
deduction_amount_snapshot=0,
payable_amount_snapshot=price,
product_snapshot_json=snapshot,
quantity=int(quantity),
quoted_unit_price_snapshot=quoted_unit,
quoted_amount_snapshot=quoted_amount,
actual_unit_price_snapshot=actual_unit,
team_id_snapshot=team_id_snapshot,
fulfillment_status="pending" if product else None,
)
db.add(order)
await db.flush()
if product_id:
product_result = await db.execute(
select(CreditProduct).where(CreditProduct.id == product_id, CreditProduct.is_active.is_(True)).limit(1)
)
product = product_result.scalar_one_or_none()
if product is None:
raise ValueError("积分商品不存在或已下架")
# 订阅报价服务内部先获取用户级 advisory lock，再按统一顺序锁订阅周期。
# 此处不预先锁 users 行，避免与支付履约（advisory -> users）形成反向锁序。
user_result = await db.execute(select(User).where(User.id == user_id).limit(1))
user = user_result.scalar_one_or_none()
if user is None:
raise ValueError("用户不存在")
quote = await quote_and_reserve_product_purchase(
db, user=user, product=product, order_id=order.id, request_time=checked_at
)
price = float(quote.payable_amount)
label = product.name
credits = float(product.grant_credits or product.monthly_grant_credits or 0)
bonus_credits = 0.0
order.amount = quote.payable_amount
order.credits = round(float(credits or 0), 2)
order.product_id = product.id
order.product_type = product.product_type
order.purchase_scene = quote.purchase_scene
order.price_type = quote.price_type
order.product_code_snapshot = product.product_code
order.product_name_snapshot = product.name
product_snapshot = product_to_dict(product)
for time_key in ("activity_start_at", "activity_end_at"):
value = product_snapshot.get(time_key)
if value is not None:
product_snapshot[time_key] = value.isoformat()
order.product_snapshot_json = product_snapshot
order.source_subscription_id = quote.source_subscription_id
order.upgrade_period_ids_json = list(quote.upgrade_period_ids) or None
order.target_price_snapshot = quote.target_price
order.deduction_amount_snapshot = quote.deduction_amount
order.payable_amount_snapshot = quote.payable_amount
order.fulfillment_status = "pending"
total_credits = round(float(credits or 0) + float(bonus_credits or 0), 2)
await db.flush()
logger.info(
f"ORDER_CREATED order_no={order.order_no} user={user_id} amount={price} "
f"credits={total_credits} method={method} product_id={product_id} mock={mock_mode}"
f"ORDER_CREATED order_no={order.order_no} user={user_id} amount={order.amount} "
f"credits={order.credits} method={method} product_id={product_id} quantity={quantity} mock={mock_mode}"
)
if mock_mode:
order.status = "paid"
order.paid_at = checked_at
desc = f"充值{label}({total_credits}积分)"
if bonus_credits > 0:
desc += f"(含赠送{bonus_credits}积分)"
await _fulfill_paid_order(
db,
order=order,
fulfilled_at=checked_at,
legacy_description=desc,
)
desc = f"充值{label}({order.credits}积分)"
await _fulfill_paid_order(db, order=order, fulfilled_at=checked_at, legacy_description=desc)
await db.flush()
else:
if method == "wechat":
qr_code_content = _create_wechat_order(order, db_configs)
if qr_code_content:
order.qr_url = qr_code_content # type: ignore[attr-defined]
else:
if quote and quote.upgrade_period_ids:
await release_upgrade_reservation(db, order=order, released_at=checked_at)
raise ValueError("微信支付预下单失败，请检查配置或稍后重试")
elif method == "alipay":
qr_url = _create_alipay_order(order, db_configs)
if qr_url:
order.qr_url = qr_url # type: ignore[attr-defined]
else:
if quote and quote.upgrade_period_ids:
await release_upgrade_reservation(db, order=order, released_at=checked_at)
raise ValueError("支付宝预下单失败，请检查配置或稍后重试")
elif method == "wechat":
qr_code_content = _create_wechat_order(order, db_configs)
if not qr_code_content:
raise ValueError("微信支付预下单失败，请检查配置或稍后重试")
order.qr_url = qr_code_content # type: ignore[attr-defined]
elif method == "alipay":
qr_url = _create_alipay_order(order, db_configs)
if not qr_url:
raise ValueError("支付宝预下单失败，请检查配置或稍后重试")
order.qr_url = qr_url # type: ignore[attr-defined]
return order
@@ -1238,10 +1304,6 @@ async def sync_pending_orders(db: AsyncSession) -> int:
updated_count += 1
elif trade_status in ("TRADE_CLOSED", "TRADE_CANCELLED"):
order.status = "cancelled"
if order.upgrade_period_ids_json:
await release_upgrade_reservation(
db, order=order, released_at=utc_now()
)
await db.commit()
updated_count += 1
@@ -1259,10 +1321,6 @@ async def sync_pending_orders(db: AsyncSession) -> int:
updated_count += 1
elif trade_state in ("CLOSED", "REVOKED"):
order.status = "cancelled"
if order.upgrade_period_ids_json:
await release_upgrade_reservation(
db, order=order, released_at=utc_now()
)
await db.commit()
updated_count += 1
except Exception as e:
@@ -1596,86 +1654,18 @@ async def process_refund(
db: AsyncSession,
order_no: str,
refund_amount: float | None = None,
refund_reason: str = "管理员退款"
refund_reason: str = "管理员退款",
) -> dict:
"""Process a refund for a paid order.
Args:
db: async database session
order_no: merchant order number
refund_amount: amount to refund (defaults to full order amount)
refund_reason: reason for refund
Returns:
dict with refund result
"""
result = await db.execute(
select(PaymentOrder).where(PaymentOrder.order_no == order_no).with_for_update().limit(1)
)
"""本版本保留退款 Service 入口，但主动订单退款统一关闭。"""
result = await db.execute(select(PaymentOrder).where(PaymentOrder.order_no == order_no).limit(1))
order = result.scalar_one_or_none()
if not order:
return {"success": False, "message": "订单不存在"}
if order.status != "paid":
return {"success": False, "message": f"订单状态为{order.status}，无法退款"}
if order.refunded_at is not None:
return {"success": False, "message": "订单已退款"}
refund_amount = to_credit_decimal(refund_amount if refund_amount is not None else order.amount)
# 金额校验
if refund_amount > to_credit_decimal(order.amount):
return {"success": False, "message": "退款金额超过订单金额"}
# 根据支付方式调用相应的退款API
db_configs = await _get_payment_configs(db)
if order.payment_method == "alipay":
refund_result = await _refund_alipay_order(
db, order, refund_amount, refund_reason, db_configs
)
if not refund_result.get("success"):
return refund_result
elif order.payment_method == "wechat":
refund_result = await _refund_wechat_order(
db, order, refund_amount, refund_reason, db_configs
)
if not refund_result.get("success"):
return refund_result
# 按原支付业务位置适配新积分账本；不改变支付渠道退款流程。
try:
if order.product_id:
await revoke_payment_order_credits(db, order=order, reason=refund_reason)
else:
await deduct_credits(
db,
order.user_id,
order.credits,
refund_reason,
related_id=order.id,
biz_key=_payment_biz_key(order, charge_kind=CreditRecordChargeKind.REFUND.value, action=CreditRecordAction.REFUND.value),
refund_for_biz_key=_payment_biz_key(order, charge_kind=CreditRecordChargeKind.RECHARGE.value, action=CreditRecordAction.CHARGE.value),
record_meta=_payment_refund_meta(order),
)
except Exception as e:
logger.exception(f"Failed to deduct credits for refund: {e}")
return {"success": False, "message": "积分扣除失败"}
# 更新订单状态
order.status = "refunded"
order.refund_amount = refund_amount
order.refunded_at = utc_now()
if order.payment_method == "alipay":
order.refund_trade_no = db_configs.get("refund_trade_no", "")
await db.commit()
logger.info(
f"REFUND_SUCCESS order_no={order_no} user={order.user_id} "
f"refund_amount={refund_amount}"
logger.warning(
"REFUND_BLOCKED order_no=%s user=%s requested_amount=%s reason=%s",
order_no, order.user_id, refund_amount, refund_reason,
)
return {"success": True, "message": "退款成功"}
return {"success": False, "message": "当前版本暂未开放订单退款"}
async def _refund_alipay_order(