Add scheduled automation task and update URL routing

- Introduced a new scheduled task for executing automation rules every 5 minutes in the Celery beat schedule.
- Updated URL routing to include a new endpoint for automation-related functionalities.
- Refactored imports in various modules to align with the new business layer structure, ensuring backward compatibility for billing models, exceptions, and services.
This commit is contained in:
IGNY8 VPS (Salman)
2025-11-16 22:11:05 +00:00
parent 455358ecfc
commit 7f8982a0ab
40 changed files with 1822 additions and 778 deletions

View File

@@ -0,0 +1,75 @@
"""
Billing Models for Credit System
"""
from django.db import models
from django.core.validators import MinValueValidator
from igny8_core.auth.models import AccountBaseModel
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.)")
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
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'),
('reparse', 'Content Reparse'),
('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:
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'}"