325 lines
12 KiB
Python
325 lines
12 KiB
Python
"""激励服务：资产钱包 / 推荐绑定 / 提现（对应《03-接口文档》第 9 章、《02-技术文档》6.6）"""
|
||
from datetime import datetime, timedelta
|
||
from decimal import Decimal
|
||
|
||
import redis.asyncio as aioredis
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import biz_config
|
||
from app.core import errors
|
||
from app.core.response import BizException
|
||
from app.models.incentive import AssetTransaction, AssetWallet, Referral, Withdrawal
|
||
from app.models.user import User
|
||
from app.services import config_service, membership_service
|
||
|
||
# 资产流水业务类型（与 asset_transactions.biz_type 注释一致）
|
||
ASSET_BIZ_REBATE = 1 # 充值返利
|
||
ASSET_BIZ_COMPLETE = 2 # 完整观看收益
|
||
ASSET_BIZ_LIKE = 3 # 点赞收益
|
||
ASSET_BIZ_WITHDRAW = 4 # 提现
|
||
ASSET_BIZ_WITHDRAW_BACK = 5 # 提现退回
|
||
ASSET_BIZ_CLAWBACK = 6 # 返利扣回
|
||
ASSET_BIZ_ADJUST = 7 # 后台调整
|
||
|
||
# 提现单在途锁兜底过期（正常由后台审核后释放）
|
||
WD_LOCK_TTL = 7 * 24 * 3600
|
||
# 补绑邀请码时限：注册后 24h 内
|
||
BIND_WINDOW_HOURS = 24
|
||
|
||
REFERRAL_STATUS_TEXT = {0: "待激活", 1: "有效", 2: "无效", 3: "风控冻结"}
|
||
WITHDRAW_STATUS_TEXT = {0: "待审核", 1: "已打款", 2: "已拒绝"}
|
||
|
||
|
||
def _fmt(dt: datetime | None) -> str | None:
|
||
return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None
|
||
|
||
|
||
def _asset_cfg() -> dict:
|
||
return biz_config.get("asset") or {}
|
||
|
||
|
||
def _withdraw_cfg() -> dict:
|
||
return biz_config.get("withdraw") or {}
|
||
|
||
|
||
# ---------- 资产钱包变更（与积分钱包同一套"行锁 + 流水 + 乐观锁"规范） ----------
|
||
|
||
async def get_or_create_asset_wallet(db: AsyncSession, user_id: int, for_update: bool = False) -> AssetWallet:
|
||
stmt = select(AssetWallet).where(AssetWallet.user_id == user_id)
|
||
if for_update:
|
||
stmt = stmt.with_for_update()
|
||
wallet = (await db.execute(stmt)).scalar_one_or_none()
|
||
if wallet is None:
|
||
wallet = AssetWallet(user_id=user_id)
|
||
db.add(wallet)
|
||
await db.flush()
|
||
if for_update:
|
||
wallet = (await db.execute(stmt)).scalar_one()
|
||
return wallet
|
||
|
||
|
||
async def change_assets(
|
||
db: AsyncSession,
|
||
user_id: int,
|
||
amount: Decimal,
|
||
biz_type: int,
|
||
biz_id: str = "",
|
||
remark: str = "",
|
||
allow_negative: bool = False,
|
||
) -> Decimal:
|
||
"""变更资产余额，返回变动后余额。
|
||
|
||
- amount 正增负减；扣减默认不允许透支（退款/返利扣回场景传 allow_negative=True）
|
||
- 同事务内：行锁 → 校验/更新余额与累计 → 写流水
|
||
"""
|
||
wallet = await get_or_create_asset_wallet(db, user_id, for_update=True)
|
||
new_balance = wallet.assets + amount
|
||
if amount < 0 and new_balance < 0 and not allow_negative:
|
||
raise BizException(errors.ASSET_NOT_ENOUGH)
|
||
|
||
wallet.assets = new_balance
|
||
wallet.version += 1
|
||
if amount > 0:
|
||
wallet.total_earned += amount
|
||
elif biz_type == ASSET_BIZ_WITHDRAW:
|
||
wallet.total_withdrawn += -amount
|
||
|
||
db.add(
|
||
AssetTransaction(
|
||
user_id=user_id,
|
||
change_amount=amount,
|
||
balance_after=new_balance,
|
||
biz_type=biz_type,
|
||
biz_id=biz_id,
|
||
remark=remark,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return new_balance
|
||
|
||
|
||
# ---------- 推荐 ----------
|
||
|
||
async def get_my_referral(db: AsyncSession, redis: aioredis.Redis, user_id: int) -> dict:
|
||
"""我的邀请（9.1）：邀请码 + 统计 + 当前会员身份（eligible）"""
|
||
user = await db.get(User, user_id)
|
||
if user is None:
|
||
raise BizException(errors.UNAUTHORIZED, http_status=401)
|
||
|
||
base_url = await config_service.get_config(db, redis, "app.invite_base_url", "https://app.example.com/r/")
|
||
total_invitees = (await db.execute(
|
||
select(func.count(Referral.id)).where(Referral.inviter_id == user_id)
|
||
)).scalar() or 0
|
||
activated_count = (await db.execute(
|
||
select(func.count(Referral.id)).where(
|
||
Referral.inviter_id == user_id, Referral.status == 1
|
||
)
|
||
)).scalar() or 0
|
||
# 邀请积分奖励累计（point_transactions biz_type=8）
|
||
from app.models.billing import PointTransaction
|
||
|
||
reward_points_total = (await db.execute(
|
||
select(func.coalesce(func.sum(PointTransaction.change_amount), 0)).where(
|
||
PointTransaction.user_id == user_id, PointTransaction.biz_type == 8
|
||
)
|
||
)).scalar() or 0
|
||
# 充值返利资产累计（asset_transactions biz_type=1）
|
||
rebate_assets_total = (await db.execute(
|
||
select(func.coalesce(func.sum(AssetTransaction.change_amount), 0)).where(
|
||
AssetTransaction.user_id == user_id, AssetTransaction.biz_type == ASSET_BIZ_REBATE
|
||
)
|
||
)).scalar() or 0
|
||
|
||
return {
|
||
"invite_code": user.invite_code,
|
||
"invite_url": f"{base_url}{user.invite_code}",
|
||
"eligible": await membership_service.is_member(db, user_id),
|
||
"total_invitees": total_invitees,
|
||
"activated_count": activated_count,
|
||
"reward_points_total": int(reward_points_total),
|
||
"rebate_assets_total": float(rebate_assets_total),
|
||
}
|
||
|
||
|
||
async def bind_invite_code(
|
||
db: AsyncSession, redis: aioredis.Redis, user_id: int, invite_code: str, device_id: str | None, ip: str
|
||
) -> dict:
|
||
"""补绑邀请码（9.2）：注册 24h 内仅一次；无效码/已绑定/绑自己/同设备均 2101"""
|
||
user = await db.get(User, user_id)
|
||
if user is None:
|
||
raise BizException(errors.UNAUTHORIZED, http_status=401)
|
||
|
||
bound = (await db.execute(
|
||
select(Referral.id).where(Referral.invitee_id == user_id)
|
||
)).first()
|
||
if bound:
|
||
raise BizException(errors.INVITE_CODE_INVALID, "已绑定过邀请码")
|
||
if user.created_at < datetime.now() - timedelta(hours=BIND_WINDOW_HOURS):
|
||
raise BizException(errors.INVITE_CODE_INVALID, "超过绑定时限（注册后24小时内可绑定）")
|
||
|
||
inviter = (await db.execute(
|
||
select(User).where(User.invite_code == invite_code)
|
||
)).scalar_one_or_none()
|
||
if inviter is None:
|
||
raise BizException(errors.INVITE_CODE_INVALID, "邀请码无效")
|
||
if inviter.id == user_id:
|
||
raise BizException(errors.INVITE_CODE_INVALID, "不能绑定自己的邀请码")
|
||
if device_id and await redis.exists(f"ref:device:{device_id}"):
|
||
old = int(await redis.get(f"ref:device:{device_id}") or 0)
|
||
if old and old != user_id:
|
||
raise BizException(errors.INVITE_CODE_INVALID, "当前设备已注册其他账号")
|
||
|
||
# 风控规则与注册时一致（邀请人单日超限冻结 / 同 IP 批量注册标记）
|
||
status, risk_flag = 0, ""
|
||
risk_key = f"ref:risk:{inviter.id}:{datetime.now():%Y%m%d}"
|
||
bind_count = await redis.incr(risk_key)
|
||
await redis.expire(risk_key, 48 * 3600)
|
||
max_bind = (biz_config.get("referral") or {}).get("risk_max_bind_per_day", 10)
|
||
if bind_count > max_bind:
|
||
status, risk_flag = 3, "inviter_burst"
|
||
else:
|
||
max_reg = (biz_config.get("referral") or {}).get("risk_max_register_per_ip", 5)
|
||
cnt = (await db.execute(
|
||
select(func.count(User.id)).where(
|
||
User.register_ip == ip, User.created_at > datetime.now() - timedelta(hours=24)
|
||
)
|
||
)).scalar() or 0
|
||
if cnt > max_reg:
|
||
status, risk_flag = 3, "batch_register"
|
||
|
||
db.add(Referral(
|
||
inviter_id=inviter.id,
|
||
invitee_id=user_id,
|
||
invite_code=invite_code,
|
||
status=status,
|
||
risk_flag=risk_flag,
|
||
register_ip=ip,
|
||
))
|
||
await db.commit()
|
||
return {"bound": True, "status": status, "status_text": REFERRAL_STATUS_TEXT[status]}
|
||
|
||
|
||
# ---------- 资产 / 提现 ----------
|
||
|
||
async def get_my_assets(db: AsyncSession, redis: aioredis.Redis, user_id: int) -> dict:
|
||
"""资产钱包（9.3）"""
|
||
wallet = await get_or_create_asset_wallet(db, user_id)
|
||
per_yuan = Decimal(str(_asset_cfg().get("assets_per_yuan", 10)))
|
||
has_pending = bool(await redis.exists(f"wd:pending:{user_id}")) or bool(
|
||
(await db.execute(
|
||
select(Withdrawal.id).where(Withdrawal.user_id == user_id, Withdrawal.status == 0)
|
||
)).first()
|
||
)
|
||
return {
|
||
"assets": float(wallet.assets),
|
||
"assets_value_yuan": float(wallet.assets / per_yuan),
|
||
"total_earned": float(wallet.total_earned),
|
||
"total_withdrawn": float(wallet.total_withdrawn),
|
||
"min_withdraw_assets": int(_withdraw_cfg().get("min_assets", 100)),
|
||
"has_pending_withdraw": has_pending,
|
||
}
|
||
|
||
|
||
async def list_asset_records(
|
||
db: AsyncSession, user_id: int, page: int, page_size: int, biz_type: int | None
|
||
) -> dict:
|
||
"""资产流水（9.4）"""
|
||
stmt = select(AssetTransaction).where(AssetTransaction.user_id == user_id)
|
||
if biz_type is not None:
|
||
stmt = stmt.where(AssetTransaction.biz_type == biz_type)
|
||
total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
|
||
rows = (
|
||
await db.execute(
|
||
stmt.order_by(AssetTransaction.id.desc()).offset((page - 1) * page_size).limit(page_size)
|
||
)
|
||
).scalars().all()
|
||
return {
|
||
"total": total,
|
||
"list": [
|
||
{
|
||
"id": r.id,
|
||
"change_amount": float(r.change_amount),
|
||
"balance_after": float(r.balance_after),
|
||
"biz_type": r.biz_type,
|
||
"remark": r.remark,
|
||
"created_at": _fmt(r.created_at),
|
||
}
|
||
for r in rows
|
||
],
|
||
}
|
||
|
||
|
||
async def apply_withdraw(
|
||
db: AsyncSession, redis: aioredis.Redis, user_id: int, assets: Decimal, channel: int, account: str
|
||
) -> dict:
|
||
"""申请提现（9.5）：在途锁防并发 → 事务内冻结扣减 + 建单；审核拒绝由后台退回并释放锁"""
|
||
min_assets = int(_withdraw_cfg().get("min_assets", 100))
|
||
if assets < min_assets:
|
||
raise BizException(errors.WITHDRAW_TOO_LOW)
|
||
|
||
lock_key = f"wd:pending:{user_id}"
|
||
if not await redis.set(lock_key, 1, nx=True, ex=WD_LOCK_TTL):
|
||
raise BizException(errors.WITHDRAW_PENDING)
|
||
try:
|
||
# 双保险：DB 层存在待审核单也拦截（锁丢失场景）
|
||
if (await db.execute(
|
||
select(Withdrawal.id).where(Withdrawal.user_id == user_id, Withdrawal.status == 0)
|
||
)).first():
|
||
raise BizException(errors.WITHDRAW_PENDING)
|
||
|
||
per_yuan = Decimal(str(_asset_cfg().get("assets_per_yuan", 10)))
|
||
amount_fen = int((assets / per_yuan) * 100) # 10资产=1元 → 1资产=10分
|
||
# 冻结扣减（余额不足抛 2102，事务回滚）
|
||
await change_assets(
|
||
db, user_id, -assets, ASSET_BIZ_WITHDRAW,
|
||
remark=f"申请提现冻结（{account}）",
|
||
)
|
||
wd = Withdrawal(
|
||
withdraw_no=membership_service.gen_order_no("W"),
|
||
user_id=user_id,
|
||
assets=assets,
|
||
amount_fen=amount_fen,
|
||
channel=channel,
|
||
account=account,
|
||
)
|
||
db.add(wd)
|
||
await db.commit()
|
||
except Exception:
|
||
await redis.delete(lock_key) # 建单失败释放锁
|
||
raise
|
||
|
||
return {
|
||
"withdraw_no": wd.withdraw_no,
|
||
"assets": float(wd.assets),
|
||
"amount_yuan": float(Decimal(wd.amount_fen) / 100),
|
||
"status": 0,
|
||
"status_text": WITHDRAW_STATUS_TEXT[0],
|
||
}
|
||
|
||
|
||
async def list_withdraw_records(db: AsyncSession, user_id: int, page: int, page_size: int) -> dict:
|
||
"""提现记录（9.6）"""
|
||
stmt = select(Withdrawal).where(Withdrawal.user_id == user_id)
|
||
total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
|
||
rows = (
|
||
await db.execute(
|
||
stmt.order_by(Withdrawal.id.desc()).offset((page - 1) * page_size).limit(page_size)
|
||
)
|
||
).scalars().all()
|
||
return {
|
||
"total": total,
|
||
"list": [
|
||
{
|
||
"withdraw_no": w.withdraw_no,
|
||
"assets": float(w.assets),
|
||
"amount_yuan": float(Decimal(w.amount_fen) / 100),
|
||
"channel": w.channel,
|
||
"status": w.status,
|
||
"status_text": WITHDRAW_STATUS_TEXT.get(w.status, ""),
|
||
"created_at": _fmt(w.created_at),
|
||
}
|
||
for w in rows
|
||
],
|
||
}
|