""" Credit Service for managing credit transactions and deductions """ from django.db import transaction from django.utils import timezone from .models import CreditTransaction, CreditUsageLog from .constants import CREDIT_COSTS from .exceptions import InsufficientCreditsError, CreditCalculationError from igny8_core.auth.models import Account class CreditService: """Service for managing credits""" @staticmethod def check_credits(account, required_credits): """ Check if account has enough credits. Args: account: Account instance required_credits: Number of credits required Raises: InsufficientCreditsError: If account doesn't have enough credits """ if account.credits < required_credits: raise InsufficientCreditsError( f"Insufficient credits. Required: {required_credits}, Available: {account.credits}" ) @staticmethod @transaction.atomic def deduct_credits(account, amount, operation_type, description, metadata=None, cost_usd=None, model_used=None, tokens_input=None, tokens_output=None, related_object_type=None, related_object_id=None): """ Deduct credits and log transaction. Args: account: Account instance amount: Number of credits to deduct operation_type: Type of operation (from CreditUsageLog.OPERATION_TYPE_CHOICES) description: Description of the transaction metadata: Optional metadata dict cost_usd: Optional cost in USD model_used: Optional AI model used tokens_input: Optional input tokens tokens_output: Optional output tokens related_object_type: Optional related object type related_object_id: Optional related object ID Returns: int: New credit balance """ # Check sufficient credits CreditService.check_credits(account, amount) # Deduct from account.credits account.credits -= amount account.save(update_fields=['credits']) # Create CreditTransaction CreditTransaction.objects.create( account=account, transaction_type='deduction', amount=-amount, # Negative for deduction balance_after=account.credits, description=description, metadata=metadata or {} ) # Create CreditUsageLog CreditUsageLog.objects.create( account=account, operation_type=operation_type, credits_used=amount, cost_usd=cost_usd, model_used=model_used or '', tokens_input=tokens_input, tokens_output=tokens_output, related_object_type=related_object_type or '', related_object_id=related_object_id, metadata=metadata or {} ) return account.credits @staticmethod @transaction.atomic def add_credits(account, amount, transaction_type, description, metadata=None): """ Add credits (purchase, subscription, etc.). Args: account: Account instance amount: Number of credits to add transaction_type: Type of transaction (from CreditTransaction.TRANSACTION_TYPE_CHOICES) description: Description of the transaction metadata: Optional metadata dict Returns: int: New credit balance """ # Add to account.credits account.credits += amount account.save(update_fields=['credits']) # Create CreditTransaction CreditTransaction.objects.create( account=account, transaction_type=transaction_type, amount=amount, # Positive for addition balance_after=account.credits, description=description, metadata=metadata or {} ) return account.credits @staticmethod def calculate_credits_for_operation(operation_type, **kwargs): """ Calculate credits needed for an operation. Args: operation_type: Type of operation **kwargs: Operation-specific parameters Returns: int: Number of credits required Raises: CreditCalculationError: If calculation fails """ if operation_type not in CREDIT_COSTS: raise CreditCalculationError(f"Unknown operation type: {operation_type}") cost_config = CREDIT_COSTS[operation_type] if operation_type == 'clustering': # 1 credit per 30 keywords keyword_count = kwargs.get('keyword_count', 0) credits = max(1, int(keyword_count * cost_config['per_keyword'])) return credits elif operation_type == 'ideas': # 1 credit per idea idea_count = kwargs.get('idea_count', 1) return cost_config['base'] * idea_count elif operation_type == 'content': # 3 credits per content piece content_count = kwargs.get('content_count', 1) return cost_config['base'] * content_count elif operation_type == 'images': # 1 credit per image image_count = kwargs.get('image_count', 1) return cost_config['base'] * image_count elif operation_type == 'reparse': # 1 credit per reparse return cost_config['base'] return cost_config['base']