""" Billing Models for Credit System """ from decimal import Decimal from django.db import models from django.core.validators import MinValueValidator from django.conf import settings from igny8_core.auth.models import AccountBaseModel from simple_history.models import HistoricalRecords # Centralized payment method choices - single source of truth PAYMENT_METHOD_CHOICES = [ ('stripe', 'Stripe (Credit/Debit Card)'), ('paypal', 'PayPal'), ('bank_transfer', 'Bank Transfer (Manual)'), ('local_wallet', 'Local Wallet (Manual)'), ('manual', 'Manual Payment'), ] class CreditTransaction(AccountBaseModel): """Track all credit transactions (additions, deductions)""" TRANSACTION_TYPE_CHOICES = [ ('purchase', 'Purchase'), ('subscription', 'Subscription Renewal'), ('refund', 'Refund'), ('deduction', 'Usage Deduction'), ('adjustment', 'Manual Adjustment'), ] transaction_type = models.CharField(max_length=20, choices=TRANSACTION_TYPE_CHOICES, db_index=True) amount = models.IntegerField(help_text="Positive for additions, negative for deductions") balance_after = models.IntegerField(help_text="Credit balance after this transaction") description = models.CharField(max_length=255) metadata = models.JSONField(default=dict, help_text="Additional context (AI call details, etc.)") # Payment FK - preferred over reference_id string payment = models.ForeignKey( 'billing.Payment', on_delete=models.SET_NULL, null=True, blank=True, related_name='credit_transactions', help_text='Payment that triggered this credit transaction' ) # Deprecated: Use payment FK instead reference_id = models.CharField( max_length=255, blank=True, help_text="DEPRECATED: Use payment FK. Legacy reference (e.g., payment id, invoice id)" ) created_at = models.DateTimeField(auto_now_add=True) class Meta: app_label = 'billing' db_table = 'igny8_credit_transactions' ordering = ['-created_at'] indexes = [ models.Index(fields=['account', 'transaction_type']), models.Index(fields=['account', 'created_at']), ] def __str__(self): account = getattr(self, 'account', None) return f"{self.get_transaction_type_display()} - {self.amount} credits - {account.name if account else 'No Account'}" class CreditUsageLog(AccountBaseModel): """Detailed log of credit usage per AI operation""" OPERATION_TYPE_CHOICES = [ ('clustering', 'Keyword Clustering'), ('idea_generation', 'Content Ideas Generation'), ('content_generation', 'Content Generation'), ('image_generation', 'Image Generation'), ('image_prompt_extraction', 'Image Prompt Extraction'), ('linking', 'Internal Linking'), ('optimization', 'Content Optimization'), ('reparse', 'Content Reparse'), ('site_page_generation', 'Site Page Generation'), ('site_structure_generation', 'Site Structure Generation'), ('ideas', 'Content Ideas Generation'), # Legacy ('content', 'Content Generation'), # Legacy ('images', 'Image Generation'), # Legacy ] operation_type = models.CharField(max_length=50, choices=OPERATION_TYPE_CHOICES, db_index=True) credits_used = models.IntegerField(validators=[MinValueValidator(0)]) cost_usd = models.DecimalField(max_digits=10, decimal_places=4, null=True, blank=True) model_used = models.CharField(max_length=100, blank=True) tokens_input = models.IntegerField(null=True, blank=True, validators=[MinValueValidator(0)]) tokens_output = models.IntegerField(null=True, blank=True, validators=[MinValueValidator(0)]) related_object_type = models.CharField(max_length=50, blank=True) # 'keyword', 'cluster', 'task' related_object_id = models.IntegerField(null=True, blank=True) metadata = models.JSONField(default=dict) created_at = models.DateTimeField(auto_now_add=True) class Meta: app_label = 'billing' db_table = 'igny8_credit_usage_logs' ordering = ['-created_at'] indexes = [ models.Index(fields=['account', 'operation_type']), models.Index(fields=['account', 'created_at']), models.Index(fields=['account', 'operation_type', 'created_at']), ] def __str__(self): account = getattr(self, 'account', None) return f"{self.get_operation_type_display()} - {self.credits_used} credits - {account.name if account else 'No Account'}" class CreditCostConfig(models.Model): """ Fixed credit costs per operation type. Per final-model-schemas.md: | Field | Type | Required | Notes | |-------|------|----------|-------| | operation_type | CharField(50) PK | Yes | Unique operation ID | | display_name | CharField(100) | Yes | Human-readable | | base_credits | IntegerField | Yes | Fixed credits per operation | | is_active | BooleanField | Yes | Enable/disable | | description | TextField | No | Admin notes | """ # Operation identification (Primary Key) operation_type = models.CharField( max_length=50, unique=True, primary_key=True, help_text="Unique operation ID (e.g., 'article_generation', 'image_generation')" ) # Human-readable name display_name = models.CharField( max_length=100, help_text="Human-readable name" ) # Fixed credits per operation base_credits = models.IntegerField( default=1, validators=[MinValueValidator(0)], help_text="Fixed credits per operation" ) # Status is_active = models.BooleanField( default=True, help_text="Enable/disable this operation" ) # Admin notes description = models.TextField( blank=True, help_text="Admin notes about this operation" ) # History tracking history = HistoricalRecords() class Meta: app_label = 'billing' db_table = 'igny8_credit_cost_config' verbose_name = 'Credit Cost Configuration' verbose_name_plural = 'Credit Cost Configurations' ordering = ['operation_type'] def __str__(self): return f"{self.display_name} - {self.base_credits} credits" class BillingConfiguration(models.Model): """ System-wide billing configuration (Singleton). Global settings for token-credit pricing. """ # Default token-to-credit ratio default_tokens_per_credit = models.IntegerField( default=100, validators=[MinValueValidator(1)], help_text="Default: How many tokens equal 1 credit (e.g., 100)" ) # Credit pricing default_credit_price_usd = models.DecimalField( max_digits=10, decimal_places=4, default=Decimal('0.01'), validators=[MinValueValidator(Decimal('0.0001'))], help_text="Default price per credit in USD" ) # Reporting settings enable_token_based_reporting = models.BooleanField( default=True, help_text="Show token metrics in all reports" ) # Rounding settings ROUNDING_CHOICES = [ ('up', 'Round Up'), ('down', 'Round Down'), ('nearest', 'Round to Nearest'), ] credit_rounding_mode = models.CharField( max_length=10, default='up', choices=ROUNDING_CHOICES, help_text="How to round fractional credits" ) # Audit fields updated_at = models.DateTimeField(auto_now=True) updated_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, help_text="Admin who last updated" ) class Meta: app_label = 'billing' db_table = 'igny8_billing_configuration' verbose_name = 'Billing Configuration' verbose_name_plural = 'Billing Configuration' def save(self, *args, **kwargs): """Enforce singleton pattern""" self.pk = 1 super().save(*args, **kwargs) @classmethod def get_config(cls): """Get or create the singleton config""" config, created = cls.objects.get_or_create(pk=1) return config def __str__(self): return f"Billing Configuration (1 credit = {self.default_tokens_per_credit} tokens)" class PlanLimitUsage(AccountBaseModel): """ Track monthly usage of plan limits (ideas, words, images, prompts) Resets at start of each billing period """ LIMIT_TYPE_CHOICES = [ ('content_ideas', 'Content Ideas'), ('content_words', 'Content Words'), ('images_basic', 'Basic Images'), ('images_premium', 'Premium Images'), ('image_prompts', 'Image Prompts'), ] limit_type = models.CharField( max_length=50, choices=LIMIT_TYPE_CHOICES, db_index=True, help_text="Type of limit being tracked" ) amount_used = models.IntegerField( default=0, validators=[MinValueValidator(0)], help_text="Amount used in current period" ) # Billing period tracking period_start = models.DateField( help_text="Start date of billing period" ) period_end = models.DateField( help_text="End date of billing period" ) # Metadata metadata = models.JSONField( default=dict, blank=True, help_text="Additional tracking data (e.g., breakdown by site)" ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: app_label = 'billing' db_table = 'igny8_plan_limit_usage' verbose_name = 'Plan Limit Usage' verbose_name_plural = 'Plan Limit Usage Records' unique_together = [['account', 'limit_type', 'period_start']] ordering = ['-period_start', 'limit_type'] indexes = [ models.Index(fields=['account', 'limit_type']), models.Index(fields=['account', 'period_start', 'period_end']), models.Index(fields=['limit_type', 'period_start']), ] def __str__(self): account = getattr(self, 'account', None) return f"{account.name if account else 'No Account'} - {self.get_limit_type_display()} - {self.amount_used} used" def is_current_period(self): """Check if this record is for the current billing period""" from django.utils import timezone today = timezone.now().date() return self.period_start <= today <= self.period_end def remaining_allowance(self, plan_limit): """Calculate remaining allowance""" return max(0, plan_limit - self.amount_used) def percentage_used(self, plan_limit): """Calculate percentage of limit used""" if plan_limit == 0: return 0 return min(100, int((self.amount_used / plan_limit) * 100)) class Invoice(AccountBaseModel): """ Invoice for subscription or credit purchases Tracks billing invoices with line items and payment status """ STATUS_CHOICES = [ ('draft', 'Draft'), ('pending', 'Pending'), ('paid', 'Paid'), ('void', 'Void'), ('uncollectible', 'Uncollectible'), ] invoice_number = models.CharField(max_length=50, unique=True, db_index=True) # Subscription relationship subscription = models.ForeignKey( 'igny8_core_auth.Subscription', on_delete=models.SET_NULL, null=True, blank=True, related_name='invoices', help_text='Subscription this invoice is for (if subscription-based)' ) # Amounts subtotal = models.DecimalField(max_digits=10, decimal_places=2, default=0) tax = models.DecimalField(max_digits=10, decimal_places=2, default=0) total = models.DecimalField(max_digits=10, decimal_places=2, default=0) currency = models.CharField(max_length=3, default='USD') # Status status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending', db_index=True) # Dates invoice_date = models.DateField(db_index=True) due_date = models.DateField() paid_at = models.DateTimeField(null=True, blank=True) # Line items line_items = models.JSONField(default=list, help_text="Invoice line items: [{description, amount, quantity}]") # Payment integration stripe_invoice_id = models.CharField(max_length=255, null=True, blank=True) payment_method = models.CharField(max_length=50, null=True, blank=True) # Metadata notes = models.TextField(blank=True) metadata = models.JSONField(default=dict) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: app_label = 'billing' db_table = 'igny8_invoices' ordering = ['-invoice_date', '-created_at'] indexes = [ models.Index(fields=['account', 'status']), models.Index(fields=['account', 'invoice_date']), models.Index(fields=['invoice_number']), ] def __str__(self): return f"Invoice {self.invoice_number} - {self.account.name if self.account else 'No Account'}" # ------------------------------------------------------------------ # Helpers to keep service code working with legacy field names # ------------------------------------------------------------------ @property def subtotal_amount(self): return self.subtotal @property def tax_amount(self): return self.tax @property def tax_rate(self): """Get tax rate from metadata if stored""" if self.metadata and 'tax_rate' in self.metadata: return self.metadata['tax_rate'] return 0 @property def discount_amount(self): """Get discount amount from metadata if stored""" if self.metadata and 'discount_amount' in self.metadata: return self.metadata['discount_amount'] return 0 @property def total_amount(self): return self.total @property def billing_period_start(self): """Get from subscription - single source of truth""" if self.account and hasattr(self.account, 'subscription'): return self.account.subscription.current_period_start return None @property def billing_period_end(self): """Get from subscription - single source of truth""" if self.account and hasattr(self.account, 'subscription'): return self.account.subscription.current_period_end return None @property def billing_email(self): """Get from metadata snapshot or account""" if self.metadata and 'billing_snapshot' in self.metadata: return self.metadata['billing_snapshot'].get('email') return self.account.billing_email if self.account else None def add_line_item(self, description: str, quantity: int, unit_price: Decimal, amount: Decimal = None): """Append a line item and keep JSON shape consistent.""" items = list(self.line_items or []) qty = quantity or 1 amt = Decimal(amount) if amount is not None else Decimal(unit_price) * qty items.append({ 'description': description, 'quantity': qty, 'unit_price': str(unit_price), 'amount': str(amt), }) self.line_items = items def calculate_totals(self): """Recompute subtotal, tax, and total from line_items.""" subtotal = Decimal('0') for item in self.line_items or []: try: subtotal += Decimal(str(item.get('amount') or 0)) except Exception: pass self.subtotal = subtotal self.total = subtotal + (self.tax or Decimal('0')) class Payment(AccountBaseModel): """ Payment record for invoices Supports: Stripe, PayPal, Manual (Bank Transfer, Local Wallet) """ STATUS_CHOICES = [ ('pending_approval', 'Pending Approval'), # Manual payment submitted by user ('succeeded', 'Succeeded'), # Payment approved and processed ('failed', 'Failed'), # Payment rejected or failed ('refunded', 'Refunded'), # Payment refunded (rare) ] # Use centralized payment method choices PAYMENT_METHOD_CHOICES = PAYMENT_METHOD_CHOICES invoice = models.ForeignKey(Invoice, on_delete=models.CASCADE, related_name='payments') # Amount amount = models.DecimalField(max_digits=10, decimal_places=2) currency = models.CharField(max_length=3, default='USD') # Status status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending_approval', db_index=True) # Payment method payment_method = models.CharField(max_length=50, choices=PAYMENT_METHOD_CHOICES, db_index=True) # Stripe integration stripe_payment_intent_id = models.CharField(max_length=255, null=True, blank=True) stripe_charge_id = models.CharField(max_length=255, null=True, blank=True) # PayPal integration paypal_order_id = models.CharField(max_length=255, null=True, blank=True) paypal_capture_id = models.CharField(max_length=255, null=True, blank=True) # Manual payment details manual_reference = models.CharField( max_length=255, blank=True, null=True, help_text="Bank transfer reference, wallet transaction ID, etc." ) manual_notes = models.TextField(blank=True, help_text="Admin notes for manual payments") admin_notes = models.TextField(blank=True, help_text="Internal notes on approval/rejection") approved_by = models.ForeignKey( settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=models.SET_NULL, related_name='approved_payments' ) approved_at = models.DateTimeField(null=True, blank=True) # Timestamps processed_at = models.DateTimeField(null=True, blank=True) failed_at = models.DateTimeField(null=True, blank=True) refunded_at = models.DateTimeField(null=True, blank=True) # Error tracking failure_reason = models.TextField(blank=True) # Metadata metadata = models.JSONField(default=dict) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) # History tracking history = HistoricalRecords() class Meta: app_label = 'billing' db_table = 'igny8_payments' ordering = ['-created_at'] indexes = [ models.Index(fields=['account', 'status']), models.Index(fields=['account', 'payment_method']), models.Index(fields=['invoice', 'status']), ] constraints = [ # Ensure manual_reference is unique when not null/empty # This prevents duplicate bank transfer references models.UniqueConstraint( fields=['manual_reference'], name='unique_manual_reference_when_not_null', condition=models.Q(manual_reference__isnull=False) & ~models.Q(manual_reference='') ), ] def __str__(self): return f"Payment {self.id} - {self.get_payment_method_display()} - {self.amount} {self.currency}" def save(self, *args, **kwargs): """Normalize empty manual_reference to NULL for proper uniqueness handling""" if self.manual_reference == '': self.manual_reference = None super().save(*args, **kwargs) class CreditPackage(models.Model): """ One-time credit purchase packages Defines available credit bundles for purchase """ name = models.CharField(max_length=100) slug = models.SlugField(unique=True, db_index=True) # Credits credits = models.IntegerField(validators=[MinValueValidator(1)]) # Pricing price = models.DecimalField(max_digits=10, decimal_places=2) discount_percentage = models.IntegerField(default=0, help_text="Discount percentage (0-100)") # Stripe stripe_product_id = models.CharField(max_length=255, null=True, blank=True) stripe_price_id = models.CharField(max_length=255, null=True, blank=True) # PayPal paypal_plan_id = models.CharField(max_length=255, null=True, blank=True) # Status is_active = models.BooleanField(default=True, db_index=True) is_featured = models.BooleanField(default=False, help_text="Show as featured package") # Display description = models.TextField(blank=True) features = models.JSONField(default=list, help_text="Bonus features or highlights") # Sort order sort_order = models.IntegerField(default=0, help_text="Display order (lower = first)") created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: app_label = 'billing' db_table = 'igny8_credit_packages' ordering = ['sort_order', 'price'] def __str__(self): return f"{self.name} - {self.credits} credits - ${self.price}" class PaymentMethodConfig(models.Model): """ Configure payment methods availability per country. For online payments (stripe, paypal): Credentials stored in IntegrationProvider. For manual payments (bank_transfer, local_wallet): Bank/wallet details stored here. """ # Use centralized choices PAYMENT_METHOD_CHOICES = PAYMENT_METHOD_CHOICES country_code = models.CharField( max_length=2, db_index=True, help_text="ISO 2-letter country code (e.g., US, GB, PK) or '*' for global" ) payment_method = models.CharField(max_length=50, choices=PAYMENT_METHOD_CHOICES) is_enabled = models.BooleanField(default=True) # Display info display_name = models.CharField(max_length=100, blank=True) instructions = models.TextField(blank=True, help_text="Payment instructions for users") # Manual payment details (for bank_transfer only) bank_name = models.CharField(max_length=255, blank=True) account_number = models.CharField(max_length=255, blank=True) account_title = models.CharField(max_length=255, blank=True, help_text="Account holder name") routing_number = models.CharField(max_length=255, blank=True, help_text="Routing/Sort code") swift_code = models.CharField(max_length=255, blank=True, help_text="SWIFT/BIC code for international") iban = models.CharField(max_length=255, blank=True, help_text="IBAN for international transfers") # Additional fields for local wallets wallet_type = models.CharField(max_length=100, blank=True, help_text="E.g., JazzCash, EasyPaisa, etc.") wallet_id = models.CharField(max_length=255, blank=True, help_text="Mobile number or wallet ID") # Order/priority sort_order = models.IntegerField(default=0) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: app_label = 'billing' db_table = 'igny8_payment_method_config' unique_together = [['country_code', 'payment_method']] ordering = ['country_code', 'sort_order'] verbose_name = 'Payment Method Configuration' verbose_name_plural = 'Payment Method Configurations' def __str__(self): return f"{self.country_code} - {self.get_payment_method_display()}" class AccountPaymentMethod(AccountBaseModel): """ Account-scoped payment methods (Stripe/PayPal/manual bank/wallet). Only metadata/refs are stored here; no secrets. """ # Use centralized choices PAYMENT_METHOD_CHOICES = PAYMENT_METHOD_CHOICES type = models.CharField(max_length=50, choices=PAYMENT_METHOD_CHOICES, db_index=True) display_name = models.CharField(max_length=100, help_text="User-visible label", default='') is_default = models.BooleanField(default=False, db_index=True) is_enabled = models.BooleanField(default=True, db_index=True) is_verified = models.BooleanField(default=False, db_index=True) country_code = models.CharField(max_length=2, blank=True, default='', help_text="ISO-2 country code (optional)") # Manual/bank/local wallet details (non-sensitive metadata) instructions = models.TextField(blank=True, default='') metadata = models.JSONField(default=dict, blank=True, help_text="Provider references or display metadata") created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: app_label = 'billing' db_table = 'igny8_account_payment_methods' ordering = ['-is_default', 'display_name', 'id'] indexes = [ models.Index(fields=['account', 'is_default']), models.Index(fields=['account', 'type']), ] unique_together = [['account', 'display_name']] def __str__(self): return f"{self.account_id} - {self.display_name} ({self.type})" class AIModelConfig(models.Model): """ All AI models (text + image) with pricing and credit configuration. Single Source of Truth for Models. Per final-model-schemas.md: | Field | Type | Required | Notes | |-------|------|----------|-------| | id | AutoField PK | Auto | | | model_name | CharField(100) | Yes | gpt-5.1, dall-e-3, runware:97@1 | | model_type | CharField(20) | Yes | text / image | | provider | CharField(50) | Yes | Links to IntegrationProvider | | display_name | CharField(200) | Yes | Human-readable | | is_default | BooleanField | Yes | One default per type | | is_active | BooleanField | Yes | Enable/disable | | cost_per_1k_input | DecimalField | No | Provider cost (USD) - text models | | cost_per_1k_output | DecimalField | No | Provider cost (USD) - text models | | tokens_per_credit | IntegerField | No | Text: tokens per 1 credit (e.g., 1000) | | credits_per_image | IntegerField | No | Image: credits per image (e.g., 1, 5, 15) | | quality_tier | CharField(20) | No | basic / quality / premium | | max_tokens | IntegerField | No | Model token limit | | context_window | IntegerField | No | Model context size | | capabilities | JSONField | No | vision, function_calling, etc. | | created_at | DateTime | Auto | | | updated_at | DateTime | Auto | | """ MODEL_TYPE_CHOICES = [ ('text', 'Text Generation'), ('image', 'Image Generation'), ] PROVIDER_CHOICES = [ ('openai', 'OpenAI'), ('anthropic', 'Anthropic'), ('runware', 'Runware'), ('google', 'Google'), ] QUALITY_TIER_CHOICES = [ ('basic', 'Basic'), ('quality', 'Quality'), ('premium', 'Premium'), ] # Basic Information model_name = models.CharField( max_length=100, unique=True, db_index=True, help_text="Model identifier (e.g., 'gpt-5.1', 'dall-e-3', 'runware:97@1')" ) model_type = models.CharField( max_length=20, choices=MODEL_TYPE_CHOICES, db_index=True, help_text="text / image" ) provider = models.CharField( max_length=50, choices=PROVIDER_CHOICES, db_index=True, help_text="Links to IntegrationProvider" ) display_name = models.CharField( max_length=200, help_text="Human-readable name" ) is_default = models.BooleanField( default=False, db_index=True, help_text="One default per type" ) is_active = models.BooleanField( default=True, db_index=True, help_text="Enable/disable" ) # Text Model Pricing (cost per 1K tokens) cost_per_1k_input = models.DecimalField( max_digits=10, decimal_places=6, null=True, blank=True, help_text="Provider cost per 1K input tokens (USD) - text models" ) cost_per_1k_output = models.DecimalField( max_digits=10, decimal_places=6, null=True, blank=True, help_text="Provider cost per 1K output tokens (USD) - text models" ) # Credit Configuration tokens_per_credit = models.IntegerField( null=True, blank=True, help_text="Text: tokens per 1 credit (e.g., 1000, 10000)" ) credits_per_image = models.IntegerField( null=True, blank=True, help_text="Image: credits per image (e.g., 1, 5, 15)" ) quality_tier = models.CharField( max_length=20, choices=QUALITY_TIER_CHOICES, null=True, blank=True, help_text="basic / quality / premium - for image models" ) # Image Size Configuration (for image models) landscape_size = models.CharField( max_length=20, null=True, blank=True, help_text="Landscape image size for this model (e.g., '1792x1024', '1280x768')" ) square_size = models.CharField( max_length=20, default='1024x1024', blank=True, help_text="Square image size for this model (e.g., '1024x1024')" ) valid_sizes = models.JSONField( default=list, blank=True, help_text="List of valid sizes for this model (e.g., ['1024x1024', '1792x1024'])" ) # Model Limits max_tokens = models.IntegerField( null=True, blank=True, help_text="Model token limit" ) context_window = models.IntegerField( null=True, blank=True, help_text="Model context size" ) # Capabilities capabilities = models.JSONField( default=dict, blank=True, help_text="Capabilities: vision, function_calling, json_mode, etc." ) # Timestamps created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) # History tracking history = HistoricalRecords() class Meta: app_label = 'billing' db_table = 'igny8_ai_model_config' verbose_name = 'AI Model Configuration' verbose_name_plural = 'AI Model Configurations' ordering = ['model_type', 'model_name'] indexes = [ models.Index(fields=['model_type', 'is_active']), models.Index(fields=['provider', 'is_active']), models.Index(fields=['is_default', 'model_type']), ] def __str__(self): return self.display_name def save(self, *args, **kwargs): """Ensure only one is_default per model_type""" if self.is_default: AIModelConfig.objects.filter( model_type=self.model_type, is_default=True ).exclude(pk=self.pk).update(is_default=False) super().save(*args, **kwargs) @classmethod def get_default_text_model(cls): """Get the default text generation model""" return cls.objects.filter(model_type='text', is_default=True, is_active=True).first() @classmethod def get_default_image_model(cls): """Get the default image generation model""" return cls.objects.filter(model_type='image', is_default=True, is_active=True).first() @classmethod def get_image_models_by_tier(cls): """Get all active image models grouped by quality tier""" return cls.objects.filter( model_type='image', is_active=True ).order_by('quality_tier', 'model_name') def validate_size(self, size: str) -> bool: """Validate that the given size is valid for this image model""" if not self.valid_sizes: # If no valid_sizes defined, accept common sizes return True return size in self.valid_sizes def get_landscape_size(self) -> str: """Get the landscape size for this model""" return self.landscape_size or '1792x1024' def get_square_size(self) -> str: """Get the square size for this model""" return self.square_size or '1024x1024' class WebhookEvent(models.Model): """ Store all incoming webhook events for audit and replay capability. This model provides: - Audit trail of all webhook events - Idempotency verification (via event_id) - Ability to replay failed events - Debugging and monitoring """ PROVIDER_CHOICES = [ ('stripe', 'Stripe'), ('paypal', 'PayPal'), ] # Unique identifier from the payment provider event_id = models.CharField( max_length=255, unique=True, db_index=True, help_text="Unique event ID from the payment provider" ) # Payment provider provider = models.CharField( max_length=20, choices=PROVIDER_CHOICES, db_index=True, help_text="Payment provider (stripe or paypal)" ) # Event type (e.g., 'checkout.session.completed', 'PAYMENT.CAPTURE.COMPLETED') event_type = models.CharField( max_length=100, db_index=True, help_text="Event type from the provider" ) # Full payload for debugging and replay payload = models.JSONField( help_text="Full webhook payload" ) # Processing status processed = models.BooleanField( default=False, db_index=True, help_text="Whether this event has been successfully processed" ) processed_at = models.DateTimeField( null=True, blank=True, help_text="When the event was processed" ) # Error tracking error_message = models.TextField( blank=True, help_text="Error message if processing failed" ) retry_count = models.IntegerField( default=0, help_text="Number of processing attempts" ) # Timestamps created_at = models.DateTimeField(auto_now_add=True) class Meta: app_label = 'billing' db_table = 'igny8_webhook_events' verbose_name = 'Webhook Event' verbose_name_plural = 'Webhook Events' ordering = ['-created_at'] indexes = [ models.Index(fields=['provider', 'event_type']), models.Index(fields=['processed', 'created_at']), models.Index(fields=['provider', 'processed']), ] def __str__(self): return f"{self.provider}:{self.event_type} - {self.event_id[:20]}..." @classmethod def record_event(cls, event_id: str, provider: str, event_type: str, payload: dict): """ Record a webhook event. Returns (event, created) tuple. If the event already exists, returns the existing event. """ return cls.objects.get_or_create( event_id=event_id, defaults={ 'provider': provider, 'event_type': event_type, 'payload': payload, } ) def mark_processed(self): """Mark the event as successfully processed""" from django.utils import timezone self.processed = True self.processed_at = timezone.now() self.save(update_fields=['processed', 'processed_at']) def mark_failed(self, error_message: str): """Mark the event as failed with error message""" self.error_message = error_message self.retry_count += 1 self.save(update_fields=['error_message', 'retry_count'])