300 lines
8.9 KiB
Python
300 lines
8.9 KiB
Python
"""
|
|
Promo Code API endpoints - user redemption and admin management
|
|
"""
|
|
import secrets
|
|
import string
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.api.deps import CurrentUser, DbSession, require_admin_with_2fa
|
|
from app.models import User, CoinTransaction, CoinTransactionType
|
|
from app.models.promo_code import PromoCode, PromoCodeRedemption
|
|
from app.schemas.promo_code import (
|
|
PromoCodeCreate,
|
|
PromoCodeUpdate,
|
|
PromoCodeResponse,
|
|
PromoCodeRedeemRequest,
|
|
PromoCodeRedeemResponse,
|
|
PromoCodeRedemptionResponse,
|
|
PromoCodeRedemptionUser,
|
|
)
|
|
from app.schemas.common import MessageResponse
|
|
|
|
router = APIRouter(prefix="/promo", tags=["promo"])
|
|
|
|
|
|
def generate_promo_code(length: int = 8) -> str:
|
|
"""Generate a random promo code"""
|
|
chars = string.ascii_uppercase + string.digits
|
|
return ''.join(secrets.choice(chars) for _ in range(length))
|
|
|
|
|
|
# === User endpoints ===
|
|
|
|
@router.post("/redeem", response_model=PromoCodeRedeemResponse)
|
|
async def redeem_promo_code(
|
|
data: PromoCodeRedeemRequest,
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
):
|
|
"""Redeem a promo code to receive coins"""
|
|
# Find promo code
|
|
result = await db.execute(
|
|
select(PromoCode).where(PromoCode.code == data.code.upper().strip())
|
|
)
|
|
promo = result.scalar_one_or_none()
|
|
|
|
if not promo:
|
|
raise HTTPException(status_code=404, detail="Промокод не найден")
|
|
|
|
# Check if valid
|
|
if not promo.is_active:
|
|
raise HTTPException(status_code=400, detail="Промокод деактивирован")
|
|
|
|
now = datetime.utcnow()
|
|
if promo.valid_from and now < promo.valid_from:
|
|
raise HTTPException(status_code=400, detail="Промокод ещё не активен")
|
|
|
|
if promo.valid_until and now > promo.valid_until:
|
|
raise HTTPException(status_code=400, detail="Промокод истёк")
|
|
|
|
if promo.max_uses is not None and promo.uses_count >= promo.max_uses:
|
|
raise HTTPException(status_code=400, detail="Лимит использований исчерпан")
|
|
|
|
# Check if user already redeemed
|
|
result = await db.execute(
|
|
select(PromoCodeRedemption).where(
|
|
PromoCodeRedemption.promo_code_id == promo.id,
|
|
PromoCodeRedemption.user_id == current_user.id,
|
|
)
|
|
)
|
|
existing = result.scalar_one_or_none()
|
|
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Вы уже использовали этот промокод")
|
|
|
|
# Create redemption record
|
|
redemption = PromoCodeRedemption(
|
|
promo_code_id=promo.id,
|
|
user_id=current_user.id,
|
|
coins_awarded=promo.coins_amount,
|
|
)
|
|
db.add(redemption)
|
|
|
|
# Update uses count
|
|
promo.uses_count += 1
|
|
|
|
# Award coins
|
|
transaction = CoinTransaction(
|
|
user_id=current_user.id,
|
|
amount=promo.coins_amount,
|
|
transaction_type=CoinTransactionType.PROMO_CODE.value,
|
|
reference_type="promo_code",
|
|
reference_id=promo.id,
|
|
description=f"Промокод: {promo.code}",
|
|
)
|
|
db.add(transaction)
|
|
|
|
current_user.coins_balance += promo.coins_amount
|
|
|
|
await db.commit()
|
|
await db.refresh(current_user)
|
|
|
|
return PromoCodeRedeemResponse(
|
|
success=True,
|
|
coins_awarded=promo.coins_amount,
|
|
new_balance=current_user.coins_balance,
|
|
message=f"Вы получили {promo.coins_amount} монет!",
|
|
)
|
|
|
|
|
|
# === Admin endpoints ===
|
|
|
|
@router.get("/admin/list", response_model=list[PromoCodeResponse])
|
|
async def admin_list_promo_codes(
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
include_inactive: bool = False,
|
|
):
|
|
"""Get all promo codes (admin only)"""
|
|
require_admin_with_2fa(current_user)
|
|
|
|
query = select(PromoCode).options(selectinload(PromoCode.created_by))
|
|
if not include_inactive:
|
|
query = query.where(PromoCode.is_active == True)
|
|
|
|
query = query.order_by(PromoCode.created_at.desc())
|
|
|
|
result = await db.execute(query)
|
|
promos = result.scalars().all()
|
|
|
|
return [
|
|
PromoCodeResponse(
|
|
id=p.id,
|
|
code=p.code,
|
|
coins_amount=p.coins_amount,
|
|
max_uses=p.max_uses,
|
|
uses_count=p.uses_count,
|
|
is_active=p.is_active,
|
|
valid_from=p.valid_from,
|
|
valid_until=p.valid_until,
|
|
created_at=p.created_at,
|
|
created_by_nickname=p.created_by.nickname if p.created_by else None,
|
|
)
|
|
for p in promos
|
|
]
|
|
|
|
|
|
@router.post("/admin/create", response_model=PromoCodeResponse)
|
|
async def admin_create_promo_code(
|
|
data: PromoCodeCreate,
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
):
|
|
"""Create a new promo code (admin only)"""
|
|
require_admin_with_2fa(current_user)
|
|
|
|
# Generate or use provided code
|
|
code = data.code.upper().strip() if data.code else generate_promo_code()
|
|
|
|
# Check uniqueness
|
|
result = await db.execute(
|
|
select(PromoCode).where(PromoCode.code == code)
|
|
)
|
|
if result.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail=f"Промокод '{code}' уже существует")
|
|
|
|
promo = PromoCode(
|
|
code=code,
|
|
coins_amount=data.coins_amount,
|
|
max_uses=data.max_uses,
|
|
valid_from=data.valid_from,
|
|
valid_until=data.valid_until,
|
|
created_by_id=current_user.id,
|
|
)
|
|
db.add(promo)
|
|
await db.commit()
|
|
await db.refresh(promo)
|
|
|
|
return PromoCodeResponse(
|
|
id=promo.id,
|
|
code=promo.code,
|
|
coins_amount=promo.coins_amount,
|
|
max_uses=promo.max_uses,
|
|
uses_count=promo.uses_count,
|
|
is_active=promo.is_active,
|
|
valid_from=promo.valid_from,
|
|
valid_until=promo.valid_until,
|
|
created_at=promo.created_at,
|
|
created_by_nickname=current_user.nickname,
|
|
)
|
|
|
|
|
|
@router.put("/admin/{promo_id}", response_model=PromoCodeResponse)
|
|
async def admin_update_promo_code(
|
|
promo_id: int,
|
|
data: PromoCodeUpdate,
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
):
|
|
"""Update a promo code (admin only)"""
|
|
require_admin_with_2fa(current_user)
|
|
|
|
result = await db.execute(
|
|
select(PromoCode)
|
|
.options(selectinload(PromoCode.created_by))
|
|
.where(PromoCode.id == promo_id)
|
|
)
|
|
promo = result.scalar_one_or_none()
|
|
|
|
if not promo:
|
|
raise HTTPException(status_code=404, detail="Промокод не найден")
|
|
|
|
if data.is_active is not None:
|
|
promo.is_active = data.is_active
|
|
if data.max_uses is not None:
|
|
promo.max_uses = data.max_uses
|
|
if data.valid_until is not None:
|
|
promo.valid_until = data.valid_until
|
|
|
|
await db.commit()
|
|
await db.refresh(promo)
|
|
|
|
return PromoCodeResponse(
|
|
id=promo.id,
|
|
code=promo.code,
|
|
coins_amount=promo.coins_amount,
|
|
max_uses=promo.max_uses,
|
|
uses_count=promo.uses_count,
|
|
is_active=promo.is_active,
|
|
valid_from=promo.valid_from,
|
|
valid_until=promo.valid_until,
|
|
created_at=promo.created_at,
|
|
created_by_nickname=promo.created_by.nickname if promo.created_by else None,
|
|
)
|
|
|
|
|
|
@router.delete("/admin/{promo_id}", response_model=MessageResponse)
|
|
async def admin_delete_promo_code(
|
|
promo_id: int,
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
):
|
|
"""Delete a promo code (admin only)"""
|
|
require_admin_with_2fa(current_user)
|
|
|
|
result = await db.execute(
|
|
select(PromoCode).where(PromoCode.id == promo_id)
|
|
)
|
|
promo = result.scalar_one_or_none()
|
|
|
|
if not promo:
|
|
raise HTTPException(status_code=404, detail="Промокод не найден")
|
|
|
|
await db.delete(promo)
|
|
await db.commit()
|
|
|
|
return MessageResponse(message=f"Промокод '{promo.code}' удалён")
|
|
|
|
|
|
@router.get("/admin/{promo_id}/redemptions", response_model=list[PromoCodeRedemptionResponse])
|
|
async def admin_get_promo_redemptions(
|
|
promo_id: int,
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
):
|
|
"""Get list of users who redeemed a promo code (admin only)"""
|
|
require_admin_with_2fa(current_user)
|
|
|
|
# Check promo exists
|
|
result = await db.execute(
|
|
select(PromoCode).where(PromoCode.id == promo_id)
|
|
)
|
|
if not result.scalar_one_or_none():
|
|
raise HTTPException(status_code=404, detail="Промокод не найден")
|
|
|
|
# Get redemptions
|
|
result = await db.execute(
|
|
select(PromoCodeRedemption)
|
|
.options(selectinload(PromoCodeRedemption.user))
|
|
.where(PromoCodeRedemption.promo_code_id == promo_id)
|
|
.order_by(PromoCodeRedemption.redeemed_at.desc())
|
|
)
|
|
redemptions = result.scalars().all()
|
|
|
|
return [
|
|
PromoCodeRedemptionResponse(
|
|
id=r.id,
|
|
user=PromoCodeRedemptionUser(
|
|
id=r.user.id,
|
|
nickname=r.user.nickname,
|
|
),
|
|
coins_awarded=r.coins_awarded,
|
|
redeemed_at=r.redeemed_at,
|
|
)
|
|
for r in redemptions
|
|
]
|