Files
aivideo/backend/app/modules/wallets/service.py

431 lines
16 KiB
Python

from __future__ import annotations
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.common.config.settings import get_settings
from app.common.errors.app_error import BusinessAppError, NotFoundAppError
from app.common.utils.id_gen import new_order_no
from app.models.entities import (
GrowthRewardRule,
InviteRelation,
PaymentChannel,
RechargeOrder,
RechargePlan,
RedeemCode,
Wallet,
WalletTransaction,
)
from app.modules.wallets.repository import WalletRepository
settings = get_settings()
class WalletService:
def __init__(self, db: Session) -> None:
self.db = db
self.repository = WalletRepository(db)
def get_wallet_summary(self, user_id: int) -> dict:
wallet = self.repository.lock_wallet(user_id)
return {
"balancePoints": wallet.balance_points,
"frozenPoints": wallet.frozen_points,
"availablePoints": wallet.balance_points - wallet.frozen_points,
"pointExchangeRatio": settings.point_exchange_ratio,
}
def list_transactions(self, user_id: int) -> list[dict]:
records = (
self.repository.wallet_transactions(user_id)
.order_by(WalletTransaction.id.desc())
.limit(100)
.all()
)
return [
{
"transactionNo": item.transaction_no,
"bizType": item.biz_type,
"direction": item.direction,
"amountPoints": item.amount_points,
"remark": item.remark,
"createdAt": item.created_at.isoformat(),
}
for item in records
]
def list_recharge_orders(self, user_id: int) -> list[dict]:
records = (
self.repository.recharge_orders(user_id)
.order_by(RechargeOrder.id.desc())
.limit(100)
.all()
)
return [
{
"orderNo": item.order_no,
"payAmount": f"{item.pay_amount:.2f}",
"arrivalPoints": item.arrival_points,
"status": item.status,
"paymentChannelCode": item.payment_channel_code,
"paidAt": item.paid_at.isoformat() if item.paid_at else None,
"createdAt": item.created_at.isoformat(),
}
for item in records
]
def recharge_options(self) -> dict:
plans = (
self.db.query(RechargePlan)
.filter(RechargePlan.status == 1)
.order_by(RechargePlan.sort_order.asc(), RechargePlan.id.asc())
.all()
)
channels = (
self.db.query(PaymentChannel)
.filter(PaymentChannel.status == 1)
.order_by(PaymentChannel.sort_order.asc(), PaymentChannel.id.asc())
.all()
)
return {
"plans": [
{
"id": item.id,
"name": item.name,
"payAmount": f"{item.pay_amount:.2f}",
"arrivalPoints": item.give_points + item.bonus_points,
"bonusPoints": item.bonus_points,
}
for item in plans
],
"channels": [
{
"id": item.id,
"channelCode": item.channel_code,
"channelName": item.channel_name,
}
for item in channels
],
}
def list_redeem_records(self, user_id: int) -> list[dict]:
records = (
self.db.query(RedeemCode)
.filter(RedeemCode.used_by_user_id == user_id)
.order_by(RedeemCode.id.desc())
.all()
)
return [
{
"redeemCode": item.redeem_code,
"points": item.points,
"usedAt": item.used_at.isoformat() if item.used_at else None,
}
for item in records
]
def create_recharge_order(self, user_id: int, payload) -> dict:
plan = self.repository.get_recharge_plan(payload.rechargePlanId)
if not plan or plan.status != 1:
raise NotFoundAppError("recharge plan not found", code=30001)
channel = self.db.scalar(
select(PaymentChannel).where(
PaymentChannel.channel_code == payload.paymentChannelCode
)
)
if not channel or channel.status != 1:
raise NotFoundAppError("payment channel not found", code=30002)
arrival_points = plan.give_points + plan.bonus_points
order = RechargeOrder(
order_no=new_order_no("rc"),
user_id=user_id,
recharge_plan_id=plan.id,
payment_channel_id=channel.id,
payment_channel_code=channel.channel_code,
pay_amount=plan.pay_amount,
point_ratio_snapshot=plan.point_ratio,
give_points=plan.give_points,
bonus_points=plan.bonus_points,
arrival_points=arrival_points,
status="pending",
)
self.db.add(order)
self.db.commit()
return {
"orderNo": order.order_no,
"payAmount": f"{order.pay_amount:.2f}",
"arrivalPoints": order.arrival_points,
"payUrl": f"/api/v1/payments/mock-pay?orderNo={order.order_no}",
}
def handle_mock_payment(self, order_no: str) -> dict:
order = self.repository.get_order_by_no(order_no)
if not order:
raise NotFoundAppError("order not found", code=30001)
if order.status == "paid":
return {"orderNo": order.order_no, "status": order.status, "idempotent": True}
wallet = self.repository.lock_wallet(order.user_id)
before_balance = wallet.balance_points
wallet.balance_points += order.arrival_points
wallet.total_recharged_points += order.arrival_points
order.status = "paid"
order.paid_at = datetime.utcnow()
self.db.add(
WalletTransaction(
transaction_no=new_order_no("wt"),
user_id=order.user_id,
wallet_id=wallet.id,
biz_type="recharge",
direction="in",
amount_points=order.arrival_points,
balance_before_points=before_balance,
balance_after_points=wallet.balance_points,
frozen_before_points=wallet.frozen_points,
frozen_after_points=wallet.frozen_points,
related_type="recharge_order",
related_id=order.id,
remark=f"recharge order {order.order_no}",
operator_type="system",
created_at=datetime.utcnow(),
)
)
self.db.commit()
return {
"orderNo": order.order_no,
"status": order.status,
"arrivalPoints": order.arrival_points,
"idempotent": False,
}
def exchange_redeem_code(self, user_id: int, payload, request) -> dict:
redeem_code = self.repository.lock_redeem_code(payload.redeemCode)
if not redeem_code:
raise BusinessAppError("redeem code not found", code=20004)
if redeem_code.status == "used":
raise BusinessAppError("redeem code already used", code=20005)
if redeem_code.status in {"expired", "disabled"}:
raise BusinessAppError("redeem code unavailable", code=20006)
if redeem_code.expired_at and redeem_code.expired_at < datetime.utcnow():
redeem_code.status = "expired"
self.db.commit()
raise BusinessAppError("redeem code expired", code=20006)
wallet = self.repository.lock_wallet(user_id)
before_balance = wallet.balance_points
wallet.balance_points += redeem_code.points
tx = WalletTransaction(
transaction_no=new_order_no("wt"),
user_id=user_id,
wallet_id=wallet.id,
biz_type="redeem_code",
direction="in",
amount_points=redeem_code.points,
balance_before_points=before_balance,
balance_after_points=wallet.balance_points,
frozen_before_points=wallet.frozen_points,
frozen_after_points=wallet.frozen_points,
related_type="redeem_code",
related_id=redeem_code.id,
remark=f"redeem {redeem_code.redeem_code}",
operator_type="user",
operator_id=user_id,
created_at=datetime.utcnow(),
)
self.db.add(tx)
self.db.flush()
redeem_code.status = "used"
redeem_code.used_by_user_id = user_id
redeem_code.wallet_transaction_id = tx.id
redeem_code.used_at = datetime.utcnow()
redeem_code.used_ip = request.client.host if request.client else ""
redeem_code.used_user_agent = request.headers.get("user-agent", "")
self.db.commit()
return {
"redeemCode": redeem_code.redeem_code,
"points": redeem_code.points,
"walletBalance": wallet.balance_points,
}
def add_points(
self,
user_id: int,
amount_points: int,
*,
biz_type: str,
related_type: str,
related_id: int | None,
remark: str,
operator_type: str = "system",
operator_id: int | None = None,
) -> WalletTransaction:
wallet = self.repository.lock_wallet(user_id)
before_balance = wallet.balance_points
wallet.balance_points += amount_points
if biz_type == "recharge":
wallet.total_recharged_points += amount_points
if biz_type in {"refund", "unfreeze"}:
wallet.total_refunded_points += amount_points
tx = WalletTransaction(
transaction_no=new_order_no("wt"),
user_id=user_id,
wallet_id=wallet.id,
biz_type=biz_type,
direction="in",
amount_points=amount_points,
balance_before_points=before_balance,
balance_after_points=wallet.balance_points,
frozen_before_points=wallet.frozen_points,
frozen_after_points=wallet.frozen_points,
related_type=related_type,
related_id=related_id,
remark=remark,
operator_type=operator_type,
operator_id=operator_id,
created_at=datetime.utcnow(),
)
self.db.add(tx)
self.db.flush()
return tx
def freeze_points(self, user_id: int, amount_points: int, *, related_type: str, related_id: int | None, remark: str) -> None:
wallet = self.repository.lock_wallet(user_id)
available_points = wallet.balance_points - wallet.frozen_points
if available_points < amount_points:
raise BusinessAppError("insufficient balance", code=20001)
balance_before = wallet.balance_points
frozen_before = wallet.frozen_points
wallet.frozen_points += amount_points
self.db.add(
WalletTransaction(
transaction_no=new_order_no("wt"),
user_id=user_id,
wallet_id=wallet.id,
biz_type="freeze",
direction="freeze",
amount_points=amount_points,
balance_before_points=balance_before,
balance_after_points=wallet.balance_points,
frozen_before_points=frozen_before,
frozen_after_points=wallet.frozen_points,
related_type=related_type,
related_id=related_id,
remark=remark,
operator_type="system",
created_at=datetime.utcnow(),
)
)
def consume_frozen_points(self, user_id: int, amount_points: int, *, related_type: str, related_id: int | None, remark: str) -> None:
wallet = self.repository.lock_wallet(user_id)
if wallet.frozen_points < amount_points:
raise BusinessAppError("frozen points not enough", code=20003)
balance_before = wallet.balance_points
frozen_before = wallet.frozen_points
wallet.balance_points -= amount_points
wallet.frozen_points -= amount_points
wallet.total_consumed_points += amount_points
self.db.add(
WalletTransaction(
transaction_no=new_order_no("wt"),
user_id=user_id,
wallet_id=wallet.id,
biz_type="consume",
direction="out",
amount_points=amount_points,
balance_before_points=balance_before,
balance_after_points=wallet.balance_points,
frozen_before_points=frozen_before,
frozen_after_points=wallet.frozen_points,
related_type=related_type,
related_id=related_id,
remark=remark,
operator_type="system",
created_at=datetime.utcnow(),
)
)
def release_frozen_points(self, user_id: int, amount_points: int, *, related_type: str, related_id: int | None, remark: str) -> None:
wallet = self.repository.lock_wallet(user_id)
if wallet.frozen_points < amount_points:
amount_points = wallet.frozen_points
balance_before = wallet.balance_points
frozen_before = wallet.frozen_points
wallet.frozen_points -= amount_points
self.db.add(
WalletTransaction(
transaction_no=new_order_no("wt"),
user_id=user_id,
wallet_id=wallet.id,
biz_type="unfreeze",
direction="unfreeze",
amount_points=amount_points,
balance_before_points=balance_before,
balance_after_points=wallet.balance_points,
frozen_before_points=frozen_before,
frozen_after_points=wallet.frozen_points,
related_type=related_type,
related_id=related_id,
remark=remark,
operator_type="system",
created_at=datetime.utcnow(),
)
)
def try_issue_signup_reward(self, user_id: int) -> None:
rule = self.db.scalar(
select(GrowthRewardRule).where(GrowthRewardRule.rule_type == "signup_reward")
)
if not rule or not rule.enabled or rule.reward_points <= 0:
return
exists = self.db.scalar(
select(WalletTransaction).where(
WalletTransaction.user_id == user_id,
WalletTransaction.biz_type == "signup_reward",
)
)
if exists:
return
self.add_points(
user_id,
rule.reward_points,
biz_type="signup_reward",
related_type="growth_rule",
related_id=rule.id,
remark="signup reward",
)
def try_issue_invite_reward(self, user_id: int, task_id: int, final_points: int) -> None:
relation = self.db.scalar(
select(InviteRelation).where(InviteRelation.invitee_user_id == user_id)
)
if not relation or relation.reward_status == "rewarded":
return
rule = self.db.scalar(
select(GrowthRewardRule).where(GrowthRewardRule.rule_type == "invite_reward")
)
if (
not rule
or not rule.enabled
or final_points <= 0
or final_points < rule.min_consume_points
):
return
tx = self.add_points(
relation.inviter_user_id,
rule.reward_points,
biz_type="invite_reward",
related_type="invite_relation",
related_id=relation.id,
remark="invite reward",
)
relation.reward_status = "rewarded"
relation.reward_points = rule.reward_points
relation.first_consumed_task_id = task_id
relation.first_consumed_at = datetime.utcnow()
relation.rewarded_at = datetime.utcnow()
relation.reward_wallet_transaction_id = tx.id