109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
"""
|
|
Coupon Service - Coupon Operations
|
|
This service handles ONLY coupon validation and redemption
|
|
NO payment provider interactions, NO subscription logic
|
|
"""
|
|
|
|
import structlog
|
|
from typing import Dict, Any, Optional, Tuple
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.repositories.coupon_repository import CouponRepository
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class CouponService:
|
|
"""Service for handling coupon validation and redemption"""
|
|
|
|
def __init__(self, db_session: AsyncSession):
|
|
self.db_session = db_session
|
|
self.coupon_repo = CouponRepository(db_session)
|
|
|
|
async def validate_coupon_code(
|
|
self,
|
|
coupon_code: str,
|
|
tenant_id: str
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validate a coupon code for a tenant
|
|
|
|
Args:
|
|
coupon_code: Coupon code to validate
|
|
tenant_id: Tenant ID
|
|
|
|
Returns:
|
|
Dictionary with validation results
|
|
"""
|
|
try:
|
|
validation = await self.coupon_repo.validate_coupon(coupon_code, tenant_id)
|
|
|
|
return {
|
|
"valid": validation.valid,
|
|
"error_message": validation.error_message,
|
|
"discount_preview": validation.discount_preview,
|
|
"coupon": {
|
|
"code": validation.coupon.code,
|
|
"discount_type": validation.coupon.discount_type.value,
|
|
"discount_value": validation.coupon.discount_value
|
|
} if validation.coupon else None
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to validate coupon", error=str(e), coupon_code=coupon_code)
|
|
return {
|
|
"valid": False,
|
|
"error_message": "Error al validar el cupón",
|
|
"discount_preview": None,
|
|
"coupon": None
|
|
}
|
|
|
|
async def redeem_coupon(
|
|
self,
|
|
coupon_code: str,
|
|
tenant_id: str,
|
|
base_trial_days: int = 0
|
|
) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
|
|
"""
|
|
Redeem a coupon for a tenant
|
|
|
|
Args:
|
|
coupon_code: Coupon code to redeem
|
|
tenant_id: Tenant ID
|
|
base_trial_days: Base trial days without coupon
|
|
|
|
Returns:
|
|
Tuple of (success, discount_applied, error_message)
|
|
"""
|
|
try:
|
|
success, redemption, error = await self.coupon_repo.redeem_coupon(
|
|
coupon_code,
|
|
tenant_id,
|
|
base_trial_days
|
|
)
|
|
|
|
if success and redemption:
|
|
return True, redemption.discount_applied, None
|
|
else:
|
|
return False, None, error
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to redeem coupon", error=str(e), coupon_code=coupon_code)
|
|
return False, None, f"Error al aplicar el cupón: {str(e)}"
|
|
|
|
async def get_coupon_by_code(self, coupon_code: str) -> Optional[Any]:
|
|
"""
|
|
Get coupon details by code
|
|
|
|
Args:
|
|
coupon_code: Coupon code to retrieve
|
|
|
|
Returns:
|
|
Coupon object or None
|
|
"""
|
|
try:
|
|
return await self.coupon_repo.get_coupon_by_code(coupon_code)
|
|
except Exception as e:
|
|
logger.error("Failed to get coupon by code", error=str(e), coupon_code=coupon_code)
|
|
return None
|