Promocode system
This commit is contained in:
299
backend/app/api/v1/promo.py
Normal file
299
backend/app/api/v1/promo.py
Normal file
@@ -0,0 +1,299 @@
|
||||
"""
|
||||
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
|
||||
]
|
||||
Reference in New Issue
Block a user