salman/igny8
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of https://git.igny8.com/salman/igny8

Browse Source
This commit is contained in:
alorig
2025-12-24 01:58:22 +05:00
parent 646095da65 38bc015d96
commit 47777476b4
60 changed files with 12275 additions and 1272 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
backend/igny8_core/business/billing/management/commands/seed_ai_models.py Normal file
View File

@@ -0,0 +1,122 @@
"""
Management command to seed initial AI model configurations
"""
from django.core.management.base import BaseCommand
from django.db import transaction
from igny8_core.business.billing.models import AIModelConfig
class Command(BaseCommand):
help = 'Seeds initial AI model configurations with pricing data'
def handle(self, *args, **options):
self.stdout.write('Seeding AI model configurations...')
models_data = [
{
'model_name': 'gpt-4o-mini',
'provider': 'openai',
'model_type': 'text',
'cost_per_1k_input_tokens': 0.000150, # $0.15 per 1M tokens
'cost_per_1k_output_tokens': 0.000600, # $0.60 per 1M tokens
'tokens_per_credit': 50, # 50 tokens = 1 credit (more expensive)
'display_name': 'GPT-4o Mini',
'is_active': True,
'is_default': True, # Set as default text model
},
{
'model_name': 'gpt-4-turbo-2024-04-09',
'provider': 'openai',
'model_type': 'text',
'cost_per_1k_input_tokens': 0.010000, # $10 per 1M tokens
'cost_per_1k_output_tokens': 0.030000, # $30 per 1M tokens
'tokens_per_credit': 30, # 30 tokens = 1 credit (premium)
'display_name': 'GPT-4 Turbo',
'is_active': True,
'is_default': False,
},
{
'model_name': 'gpt-3.5-turbo',
'provider': 'openai',
'model_type': 'text',
'cost_per_1k_input_tokens': 0.000500, # $0.50 per 1M tokens
'cost_per_1k_output_tokens': 0.001500, # $1.50 per 1M tokens
'tokens_per_credit': 200, # 200 tokens = 1 credit (cheaper)
'display_name': 'GPT-3.5 Turbo',
'is_active': True,
'is_default': False,
},
{
'model_name': 'claude-3-5-sonnet-20241022',
'provider': 'anthropic',
'model_type': 'text',
'cost_per_1k_input_tokens': 0.003000, # $3 per 1M tokens
'cost_per_1k_output_tokens': 0.015000, # $15 per 1M tokens
'tokens_per_credit': 40, # 40 tokens = 1 credit
'display_name': 'Claude 3.5 Sonnet',
'is_active': True,
'is_default': False,
},
{
'model_name': 'claude-3-haiku-20240307',
'provider': 'anthropic',
'model_type': 'text',
'cost_per_1k_input_tokens': 0.000250, # $0.25 per 1M tokens
'cost_per_1k_output_tokens': 0.001250, # $1.25 per 1M tokens
'tokens_per_credit': 150, # 150 tokens = 1 credit (budget)
'display_name': 'Claude 3 Haiku',
'is_active': True,
'is_default': False,
},
{
'model_name': 'runware-flux-1.1-pro',
'provider': 'runware',
'model_type': 'image',
'cost_per_1k_input_tokens': 0.000000, # Image models don't use input tokens
'cost_per_1k_output_tokens': 0.040000, # $0.04 per image (treat as "tokens")
'tokens_per_credit': 1, # 1 "token" (image) = 1 credit
'display_name': 'Runware FLUX 1.1 Pro',
'is_active': True,
'is_default': True, # Set as default image model
},
{
'model_name': 'dall-e-3',
'provider': 'openai',
'model_type': 'image',
'cost_per_1k_input_tokens': 0.000000,
'cost_per_1k_output_tokens': 0.040000, # $0.040 per standard image
'tokens_per_credit': 1,
'display_name': 'DALL-E 3',
'is_active': True,
'is_default': False,
},
]
created_count = 0
updated_count = 0
with transaction.atomic():
for data in models_data:
model, created = AIModelConfig.objects.update_or_create(
model_name=data['model_name'],
defaults=data
)
if created:
created_count += 1
self.stdout.write(
self.style.SUCCESS(f'✓ Created: {model.display_name}')
)
else:
updated_count += 1
self.stdout.write(
self.style.WARNING(f'↻ Updated: {model.display_name}')
)
self.stdout.write('\n' + '='*60)
self.stdout.write(self.style.SUCCESS(
f'✓ Successfully processed {len(models_data)} AI models'
))
self.stdout.write(f' - Created: {created_count}')
self.stdout.write(f' - Updated: {updated_count}')
self.stdout.write('='*60)

156
backend/igny8_core/business/billing/migrations/0019_add_ai_model_config.py Normal file
View File

@@ -0,0 +1,156 @@
# Generated manually for AI Model & Cost Configuration System
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('billing', '0018_update_operation_choices'),
]
operations = [
# Step 1: Create AIModelConfig table using raw SQL
migrations.RunSQL(
sql="""
CREATE TABLE "igny8_ai_model_config" (
"id" bigint NOT NULL PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY,
"model_name" varchar(100) NOT NULL UNIQUE,
"provider" varchar(50) NOT NULL,
"model_type" varchar(20) NOT NULL,
"cost_per_1k_input_tokens" numeric(10, 6) NOT NULL,
"cost_per_1k_output_tokens" numeric(10, 6) NOT NULL,
"tokens_per_credit" numeric(10, 2) NOT NULL,
"display_name" varchar(200) NOT NULL,
"is_active" boolean NOT NULL DEFAULT true,
"is_default" boolean NOT NULL DEFAULT false,
"created_at" timestamp with time zone NOT NULL DEFAULT NOW(),
"updated_at" timestamp with time zone NOT NULL DEFAULT NOW()
);
CREATE INDEX "igny8_ai_model_config_model_name_75645c19_like"
ON "igny8_ai_model_config" ("model_name" varchar_pattern_ops);
""",
reverse_sql='DROP TABLE "igny8_ai_model_config";',
),
# Step 2: Modify CreditUsageLog table
migrations.RunSQL(
sql="""
-- Add model_name column (copy of old model_used for backward compat)
ALTER TABLE igny8_credit_usage_logs ADD COLUMN model_name varchar(100);
UPDATE igny8_credit_usage_logs SET model_name = model_used;
-- Make old model_used nullable
ALTER TABLE igny8_credit_usage_logs ALTER COLUMN model_used DROP NOT NULL;
-- Add cost tracking fields
ALTER TABLE igny8_credit_usage_logs ADD COLUMN cost_usd_input numeric(10, 6) NULL;
ALTER TABLE igny8_credit_usage_logs ADD COLUMN cost_usd_output numeric(10, 6) NULL;
ALTER TABLE igny8_credit_usage_logs ADD COLUMN cost_usd_total numeric(10, 6) NULL;
-- Add model_config FK
ALTER TABLE igny8_credit_usage_logs ADD COLUMN model_config_id bigint NULL;
ALTER TABLE igny8_credit_usage_logs
ADD CONSTRAINT "igny8_credit_usage_l_model_config_id_fk_igny8_ai_"
FOREIGN KEY (model_config_id) REFERENCES igny8_ai_model_config(id)
DEFERRABLE INITIALLY DEFERRED;
CREATE INDEX "igny8_credit_usage_logs_model_config_id_idx"
ON "igny8_credit_usage_logs" ("model_config_id");
-- Rename old model_used to avoid conflicts
ALTER TABLE igny8_credit_usage_logs RENAME COLUMN model_used TO model_used_old;
""",
reverse_sql="""
ALTER TABLE igny8_credit_usage_logs RENAME COLUMN model_used_old TO model_used;
ALTER TABLE igny8_credit_usage_logs DROP CONSTRAINT "igny8_credit_usage_l_model_config_id_fk_igny8_ai_";
DROP INDEX "igny8_credit_usage_logs_model_config_id_idx";
ALTER TABLE igny8_credit_usage_logs DROP COLUMN model_config_id;
ALTER TABLE igny8_credit_usage_logs DROP COLUMN cost_usd_total;
ALTER TABLE igny8_credit_usage_logs DROP COLUMN cost_usd_output;
ALTER TABLE igny8_credit_usage_logs DROP COLUMN cost_usd_input;
ALTER TABLE igny8_credit_usage_logs DROP COLUMN model_name;
""",
),
# Step 3: Modify CreditCostConfig table
migrations.RunSQL(
sql="""
ALTER TABLE igny8_credit_cost_config ADD COLUMN default_model_id bigint NULL;
ALTER TABLE igny8_credit_cost_config
ADD CONSTRAINT "igny8_credit_cost_co_default_model_id_fk_igny8_ai_"
FOREIGN KEY (default_model_id) REFERENCES igny8_ai_model_config(id)
DEFERRABLE INITIALLY DEFERRED;
CREATE INDEX "igny8_credit_cost_config_default_model_id_idx"
ON "igny8_credit_cost_config" ("default_model_id");
""",
reverse_sql="""
ALTER TABLE igny8_credit_cost_config DROP CONSTRAINT "igny8_credit_cost_co_default_model_id_fk_igny8_ai_";
DROP INDEX "igny8_credit_cost_config_default_model_id_idx";
ALTER TABLE igny8_credit_cost_config DROP COLUMN default_model_id;
""",
),
# Step 4: Update model state (tell Django about the new structure)
migrations.SeparateDatabaseAndState(
state_operations=[
migrations.CreateModel(
name='AIModelConfig',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('model_name', models.CharField(help_text='Technical model identifier', max_length=100, unique=True)),
('provider', models.CharField(choices=[('openai', 'OpenAI'), ('anthropic', 'Anthropic'), ('runware', 'Runware'), ('other', 'Other')], max_length=50)),
('model_type', models.CharField(choices=[('text', 'Text Generation'), ('image', 'Image Generation'), ('embedding', 'Embeddings')], max_length=20)),
('cost_per_1k_input_tokens', models.DecimalField(decimal_places=6, help_text='Cost in USD per 1,000 input tokens', max_digits=10)),
('cost_per_1k_output_tokens', models.DecimalField(decimal_places=6, help_text='Cost in USD per 1,000 output tokens', max_digits=10)),
('tokens_per_credit', models.DecimalField(decimal_places=2, help_text='How many tokens equal 1 credit', max_digits=10)),
('display_name', models.CharField(help_text='Human-readable model name', max_length=200)),
('is_active', models.BooleanField(default=True)),
('is_default', models.BooleanField(default=False)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
],
options={
'db_table': 'igny8_ai_model_config',
'ordering': ['provider', 'model_name'],
},
),
migrations.AddField(
model_name='creditusagelog',
name='model_name',
field=models.CharField(blank=True, max_length=100),
),
migrations.AddField(
model_name='creditusagelog',
name='cost_usd_input',
field=models.DecimalField(blank=True, decimal_places=6, max_digits=10, null=True),
),
migrations.AddField(
model_name='creditusagelog',
name='cost_usd_output',
field=models.DecimalField(blank=True, decimal_places=6, max_digits=10, null=True),
),
migrations.AddField(
model_name='creditusagelog',
name='cost_usd_total',
field=models.DecimalField(blank=True, decimal_places=6, max_digits=10, null=True),
),
migrations.AddField(
model_name='creditusagelog',
name='model_config',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='usage_logs', to='billing.aimodelconfig', db_column='model_config_id'),
),
migrations.AlterField(
model_name='creditcostconfig',
name='unit',
field=models.CharField(choices=[('per_operation', 'Per Operation'), ('per_item', 'Per Item'), ('per_100_tokens', 'Per 100 Tokens'), ('per_1000_tokens', 'Per 1000 Tokens')], default='per_operation', max_length=50),
),
migrations.AddField(
model_name='creditcostconfig',
name='default_model',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='cost_configs', to='billing.aimodelconfig'),
),
],
database_operations=[], # Already done with RunSQL above
),
]

314
backend/igny8_core/business/billing/models.py
View File

@@ -19,6 +19,102 @@ PAYMENT_METHOD_CHOICES = [
]
class AIModelConfig(models.Model):
"""
AI Model Configuration - Centralized pricing and token ratios
Single source of truth for all AI model costs
"""
# Model identification
model_name = models.CharField(
max_length=100,
unique=True,
help_text="Technical model name (e.g., gpt-4-turbo, gpt-3.5-turbo)"
)
provider = models.CharField(
max_length=50,
choices=[
('openai', 'OpenAI'),
('anthropic', 'Anthropic'),
('runware', 'Runware'),
('other', 'Other'),
],
help_text="AI provider"
)
model_type = models.CharField(
max_length=20,
choices=[
('text', 'Text Generation'),
('image', 'Image Generation'),
('embedding', 'Embeddings'),
],
default='text',
help_text="Type of AI model"
)
# Pricing (per 1K tokens for text models)
cost_per_1k_input_tokens = models.DecimalField(
max_digits=10,
decimal_places=6,
default=Decimal('0.001'),
validators=[MinValueValidator(Decimal('0'))],
help_text="Cost in USD per 1,000 input tokens"
)
cost_per_1k_output_tokens = models.DecimalField(
max_digits=10,
decimal_places=6,
default=Decimal('0.002'),
validators=[MinValueValidator(Decimal('0'))],
help_text="Cost in USD per 1,000 output tokens"
)
# Token-to-credit ratio
tokens_per_credit = models.IntegerField(
default=100,
validators=[MinValueValidator(1)],
help_text="How many tokens equal 1 credit (e.g., 100 tokens = 1 credit)"
)
# Display
display_name = models.CharField(
max_length=150,
help_text="Human-readable name (e.g., 'GPT-4 Turbo (Premium)')"
)
description = models.TextField(
blank=True,
help_text="Model description and use cases"
)
# Status
is_active = models.BooleanField(
default=True,
help_text="Enable/disable this model"
)
is_default = models.BooleanField(
default=False,
help_text="Use as system-wide default model"
)
# Metadata
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'billing'
db_table = 'igny8_ai_model_config'
verbose_name = 'AI Model Configuration'
verbose_name_plural = 'AI Model Configurations'
ordering = ['provider', 'model_name']
def __str__(self):
return f"{self.display_name} ({self.model_name})"
def calculate_cost(self, tokens_input, tokens_output):
"""Calculate USD cost for given token usage"""
cost_input = (tokens_input / 1000) * self.cost_per_1k_input_tokens
cost_output = (tokens_output / 1000) * self.cost_per_1k_output_tokens
return float(cost_input + cost_output)
class CreditTransaction(AccountBaseModel):
"""Track all credit transactions (additions, deductions)"""
TRANSACTION_TYPE_CHOICES = [
@@ -75,18 +171,73 @@ class CreditUsageLog(AccountBaseModel):
('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'),
('ideas', 'Content Ideas Generation'), # Legacy
('content', 'Content Generation'), # Legacy
('images', 'Image Generation'), # Legacy
('site_structure_generation', 'Site Structure Generation'),
('site_page_generation', 'Site Page Generation'),
# Legacy aliases for backward compatibility (don't show in new dropdowns)
('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)
# Model tracking
model_config = models.ForeignKey(
'billing.AIModelConfig',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='usage_logs',
help_text="AI model configuration used",
db_column='model_config_id'
)
model_name = models.CharField(
max_length=100,
blank=True,
help_text="Model name (deprecated, use model_config FK)"
)
# Token tracking
tokens_input = models.IntegerField(null=True, blank=True, validators=[MinValueValidator(0)])
tokens_output = models.IntegerField(null=True, blank=True, validators=[MinValueValidator(0)])
# Cost tracking (USD)
cost_usd_input = models.DecimalField(
max_digits=10,
decimal_places=6,
null=True,
blank=True,
help_text="USD cost for input tokens"
)
cost_usd_output = models.DecimalField(
max_digits=10,
decimal_places=6,
null=True,
blank=True,
help_text="USD cost for output tokens"
)
cost_usd_total = models.DecimalField(
max_digits=10,
decimal_places=6,
null=True,
blank=True,
help_text="Total USD cost (input + output)"
)
# Legacy cost field (deprecated)
cost_usd = models.DecimalField(
max_digits=10,
decimal_places=4,
null=True,
blank=True,
help_text="Deprecated, use cost_usd_total"
)
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)
@@ -109,38 +260,64 @@ class CreditUsageLog(AccountBaseModel):
class CreditCostConfig(models.Model):
"""
Token-based credit pricing configuration.
ALL operations use token-to-credit conversion.
Configurable credit costs per AI function
Admin-editable alternative to hardcoded constants
"""
# Active operation types (excludes legacy aliases)
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_structure_generation', 'Site Structure Generation'),
('site_page_generation', 'Site Page Generation'),
]
# Operation identification
operation_type = models.CharField(
max_length=50,
unique=True,
choices=CreditUsageLog.OPERATION_TYPE_CHOICES,
choices=OPERATION_TYPE_CHOICES,
help_text="AI operation type"
)
# Token-to-credit ratio (tokens per 1 credit)
tokens_per_credit = models.IntegerField(
default=100,
validators=[MinValueValidator(1)],
help_text="Number of tokens that equal 1 credit (e.g., 100 tokens = 1 credit)"
)
# Minimum credits (for very small token usage)
min_credits = models.IntegerField(
default=1,
# Cost configuration
credits_cost = models.IntegerField(
validators=[MinValueValidator(0)],
help_text="Minimum credits to charge regardless of token usage"
help_text="Credits required for this operation"
)
# Price per credit (for revenue reporting)
price_per_credit_usd = models.DecimalField(
max_digits=10,
decimal_places=4,
default=Decimal('0.01'),
validators=[MinValueValidator(Decimal('0.0001'))],
help_text="USD price per credit (for revenue reporting)"
# Unit of measurement
UNIT_CHOICES = [
('per_request', 'Per Request'),
('per_100_words', 'Per 100 Words'),
('per_200_words', 'Per 200 Words'),
('per_item', 'Per Item'),
('per_image', 'Per Image'),
('per_100_tokens', 'Per 100 Tokens'), # NEW: Token-based
('per_1000_tokens', 'Per 1000 Tokens'), # NEW: Token-based
]
unit = models.CharField(
max_length=50,
default='per_request',
choices=UNIT_CHOICES,
help_text="What the cost applies to"
)
# Model configuration
default_model = models.ForeignKey(
'billing.AIModelConfig',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='operation_configs',
help_text="Default AI model for this operation (optional)"
)
# Metadata
@@ -150,7 +327,6 @@ class CreditCostConfig(models.Model):
# Status
is_active = models.BooleanField(default=True, help_text="Enable/disable this operation")
# Audit fields
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
@@ -164,10 +340,10 @@ class CreditCostConfig(models.Model):
)
# Change tracking
previous_tokens_per_credit = models.IntegerField(
previous_cost = models.IntegerField(
null=True,
blank=True,
help_text="Tokens per credit before last update (for audit trail)"
help_text="Cost before last update (for audit trail)"
)
# History tracking
@@ -181,92 +357,20 @@ class CreditCostConfig(models.Model):
ordering = ['operation_type']
def __str__(self):
return f"{self.display_name} - {self.tokens_per_credit} tokens/credit"
return f"{self.display_name} - {self.credits_cost} credits {self.unit}"
def save(self, *args, **kwargs):
# Track token ratio changes
# Track cost changes
if self.pk:
try:
old = CreditCostConfig.objects.get(pk=self.pk)
if old.tokens_per_credit != self.tokens_per_credit:
self.previous_tokens_per_credit = old.tokens_per_credit
if old.credits_cost != self.credits_cost:
self.previous_cost = old.credits_cost
except CreditCostConfig.DoesNotExist:
pass
super().save(*args, **kwargs)
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)

455
backend/igny8_core/business/billing/services/credit_service.py
View File

@@ -3,111 +3,126 @@ Credit Service for managing credit transactions and deductions
"""
from django.db import transaction
from django.utils import timezone
from igny8_core.business.billing.models import CreditTransaction, CreditUsageLog
from decimal import Decimal
import math
from igny8_core.business.billing.models import CreditTransaction, CreditUsageLog, AIModelConfig
from igny8_core.business.billing.constants import CREDIT_COSTS
from igny8_core.business.billing.exceptions import InsufficientCreditsError, CreditCalculationError
from igny8_core.auth.models import Account
class CreditService:
"""Service for managing credits - Token-based only"""
"""Service for managing credits"""
@staticmethod
def calculate_credits_from_tokens(operation_type, tokens_input, tokens_output):
def get_credit_cost(operation_type, amount=None):
"""
Calculate credits from actual token usage using configured ratio.
This is the ONLY way credits are calculated in the system.
Get credit cost for operation.
Now checks database config first, falls back to constants.
Args:
operation_type: Type of operation
tokens_input: Input tokens used
tokens_output: Output tokens used
operation_type: Type of operation (from CREDIT_COSTS)
amount: Optional amount (word count, image count, etc.)
Returns:
int: Credits to deduct
int: Number of credits required
Raises:
CreditCalculationError: If configuration error
CreditCalculationError: If operation type is unknown
"""
import logging
import math
from igny8_core.business.billing.models import CreditCostConfig, BillingConfiguration
logger = logging.getLogger(__name__)
# Get operation config (use global default if not found)
config = CreditCostConfig.objects.filter(
operation_type=operation_type,
is_active=True
).first()
# Try to get from database config first
try:
from igny8_core.business.billing.models import CreditCostConfig
config = CreditCostConfig.objects.filter(
operation_type=operation_type,
is_active=True
).first()
if config:
base_cost = config.credits_cost
# Apply unit-based calculation
if config.unit == 'per_100_words' and amount:
return max(1, int(base_cost * (amount / 100)))
elif config.unit == 'per_200_words' and amount:
return max(1, int(base_cost * (amount / 200)))
elif config.unit in ['per_item', 'per_image'] and amount:
return base_cost * amount
else:
return base_cost
if not config:
# Use global billing config as fallback
billing_config = BillingConfiguration.get_config()
tokens_per_credit = billing_config.default_tokens_per_credit
min_credits = 1
logger.info(f"No config for {operation_type}, using default: {tokens_per_credit} tokens/credit")
else:
tokens_per_credit = config.tokens_per_credit
min_credits = config.min_credits
except Exception as e:
logger.warning(f"Failed to get cost from database, using constants: {e}")
# Calculate total tokens
total_tokens = (tokens_input or 0) + (tokens_output or 0)
# Fallback to hardcoded constants
base_cost = CREDIT_COSTS.get(operation_type, 0)
if base_cost == 0:
raise CreditCalculationError(f"Unknown operation type: {operation_type}")
# Calculate credits (fractional)
if tokens_per_credit <= 0:
raise CreditCalculationError(f"Invalid tokens_per_credit: {tokens_per_credit}")
# Variable cost operations (legacy logic)
if operation_type == 'content_generation' and amount:
# Per 100 words
return max(1, int(base_cost * (amount / 100)))
elif operation_type == 'optimization' and amount:
# Per 200 words
return max(1, int(base_cost * (amount / 200)))
elif operation_type == 'image_generation' and amount:
# Per image
return base_cost * amount
elif operation_type == 'idea_generation' and amount:
# Per idea
return base_cost * amount
credits_float = total_tokens / tokens_per_credit
# Get rounding mode from global config
billing_config = BillingConfiguration.get_config()
rounding_mode = billing_config.credit_rounding_mode
if rounding_mode == 'up':
credits = math.ceil(credits_float)
elif rounding_mode == 'down':
credits = math.floor(credits_float)
else: # nearest
credits = round(credits_float)
# Apply minimum
credits = max(credits, min_credits)
logger.info(
f"Calculated credits for {operation_type}: "
f"{total_tokens} tokens ({tokens_input} in, {tokens_output} out) "
f"÷ {tokens_per_credit} = {credits} credits"
)
return credits
# Fixed cost operations
return base_cost
@staticmethod
def check_credits_for_tokens(account, operation_type, estimated_tokens_input, estimated_tokens_output):
def check_credits(account, operation_type, amount=None):
"""
Check if account has sufficient credits based on estimated token usage.
Check if account has sufficient credits for an operation.
Args:
account: Account instance
operation_type: Type of operation
estimated_tokens_input: Estimated input tokens
estimated_tokens_output: Estimated output tokens
amount: Optional amount (word count, image count, etc.)
Raises:
InsufficientCreditsError: If account doesn't have enough credits
"""
required = CreditService.calculate_credits_from_tokens(
operation_type, estimated_tokens_input, estimated_tokens_output
)
required = CreditService.get_credit_cost(operation_type, amount)
if account.credits < required:
raise InsufficientCreditsError(
f"Insufficient credits. Required: {required}, Available: {account.credits}"
)
return True
@staticmethod
def check_credits_legacy(account, required_credits):
"""
Legacy method: Check if account has enough credits (for backward compatibility).
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):
def deduct_credits(account, amount, operation_type, description, metadata=None,
cost_usd_input=None, cost_usd_output=None, cost_usd_total=None,
model_config=None, tokens_input=None, tokens_output=None,
related_object_type=None, related_object_id=None):
"""
Deduct credits and log transaction.
@@ -117,8 +132,10 @@ class CreditService:
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
cost_usd_input: Optional input cost in USD
cost_usd_output: Optional output cost in USD
cost_usd_total: Optional total cost in USD
model_config: Optional AIModelConfig instance
tokens_input: Optional input tokens
tokens_output: Optional output tokens
related_object_type: Optional related object type
@@ -144,83 +161,93 @@ class CreditService:
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 {}
)
# Create CreditUsageLog with new model_config FK
log_data = {
'account': account,
'operation_type': operation_type,
'credits_used': amount,
'tokens_input': tokens_input,
'tokens_output': tokens_output,
'related_object_type': related_object_type or '',
'related_object_id': related_object_id,
'metadata': metadata or {},
}
# Add model tracking (new FK)
if model_config:
log_data['model_config'] = model_config
log_data['model_name'] = model_config.model_name
# Add cost tracking (new fields)
if cost_usd_input is not None:
log_data['cost_usd_input'] = cost_usd_input
if cost_usd_output is not None:
log_data['cost_usd_output'] = cost_usd_output
if cost_usd_total is not None:
log_data['cost_usd_total'] = cost_usd_total
# Legacy cost_usd field (backward compatibility)
if cost_usd_total is not None:
log_data['cost_usd'] = cost_usd_total
CreditUsageLog.objects.create(**log_data)
return account.credits
@staticmethod
@transaction.atomic
def deduct_credits_for_operation(
account,
operation_type,
tokens_input,
tokens_output,
description=None,
metadata=None,
cost_usd=None,
model_used=None,
related_object_type=None,
related_object_id=None
):
def deduct_credits_for_operation(account, operation_type, amount=None, description=None,
metadata=None, cost_usd_input=None, cost_usd_output=None,
cost_usd_total=None, model_config=None, tokens_input=None,
tokens_output=None, related_object_type=None, related_object_id=None):
"""
Deduct credits for an operation based on actual token usage.
This is the ONLY way to deduct credits in the token-based system.
Deduct credits for an operation (convenience method that calculates cost automatically).
Args:
account: Account instance
operation_type: Type of operation
tokens_input: REQUIRED - Actual input tokens used
tokens_output: REQUIRED - Actual output tokens used
amount: Optional amount (word count, image count, etc.)
description: Optional description (auto-generated if not provided)
metadata: Optional metadata dict
cost_usd: Optional cost in USD
model_used: Optional AI model used
cost_usd_input: Optional input cost in USD
cost_usd_output: Optional output cost in USD
cost_usd_total: Optional total cost in USD
model_config: Optional AIModelConfig instance
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
Raises:
ValueError: If tokens_input or tokens_output not provided
"""
# Validate token inputs
if tokens_input is None or tokens_output is None:
raise ValueError(
f"tokens_input and tokens_output are REQUIRED for credit deduction. "
f"Got: tokens_input={tokens_input}, tokens_output={tokens_output}"
# Calculate credit cost - use token-based if tokens provided
if tokens_input is not None and tokens_output is not None and model_config:
credits_required = CreditService.calculate_credits_from_tokens(
operation_type, tokens_input, tokens_output, model_config
)
# Calculate credits from actual token usage
credits_required = CreditService.calculate_credits_from_tokens(
operation_type, tokens_input, tokens_output
)
else:
credits_required = CreditService.get_credit_cost(operation_type, amount)
# Check sufficient credits
if account.credits < credits_required:
raise InsufficientCreditsError(
f"Insufficient credits. Required: {credits_required}, Available: {account.credits}"
)
CreditService.check_credits_legacy(account, credits_required)
# Auto-generate description if not provided
if not description:
total_tokens = tokens_input + tokens_output
description = (
f"{operation_type}: {total_tokens} tokens "
f"({tokens_input} in, {tokens_output} out) = {credits_required} credits"
)
model_name = model_config.display_name if model_config else "AI"
if operation_type == 'clustering':
description = f"Clustering operation ({model_name})"
elif operation_type == 'idea_generation':
description = f"Generated {amount or 1} idea(s) ({model_name})"
elif operation_type == 'content_generation':
if tokens_input and tokens_output:
description = f"Generated content ({tokens_input + tokens_output} tokens, {model_name})"
else:
description = f"Generated content ({amount or 0} words, {model_name})"
elif operation_type == 'image_generation':
description = f"Generated {amount or 1} image(s) ({model_name})"
else:
description = f"{operation_type} operation ({model_name})"
return CreditService.deduct_credits(
account=account,
@@ -228,8 +255,10 @@ class CreditService:
operation_type=operation_type,
description=description,
metadata=metadata,
cost_usd=cost_usd,
model_used=model_used,
cost_usd_input=cost_usd_input,
cost_usd_output=cost_usd_output,
cost_usd_total=cost_usd_total,
model_config=model_config,
tokens_input=tokens_input,
tokens_output=tokens_output,
related_object_type=related_object_type,
@@ -267,4 +296,188 @@ class CreditService:
)
return account.credits
@staticmethod
def calculate_credits_for_operation(operation_type, **kwargs):
"""
Calculate credits needed for an operation.
Legacy method - use get_credit_cost() instead.
Args:
operation_type: Type of operation
**kwargs: Operation-specific parameters
Returns:
int: Number of credits required
Raises:
CreditCalculationError: If calculation fails
"""
# Map legacy operation types
if operation_type == 'ideas':
operation_type = 'idea_generation'
elif operation_type == 'content':
operation_type = 'content_generation'
elif operation_type == 'images':
operation_type = 'image_generation'
# Extract amount from kwargs
amount = None
if 'word_count' in kwargs:
amount = kwargs.get('word_count')
elif 'image_count' in kwargs:
amount = kwargs.get('image_count')
elif 'idea_count' in kwargs:
amount = kwargs.get('idea_count')
return CreditService.get_credit_cost(operation_type, amount)
@staticmethod
def calculate_credits_from_tokens(operation_type, tokens_input, tokens_output, model_config):
"""
Calculate credits based on actual token usage and AI model configuration.
This is the new token-aware calculation method.
Args:
operation_type: Type of operation (e.g., 'content_generation')
tokens_input: Number of input tokens used
tokens_output: Number of output tokens used
model_config: AIModelConfig instance
Returns:
int: Number of credits to deduct
Raises:
CreditCalculationError: If calculation fails
"""
import logging
logger = logging.getLogger(__name__)
try:
from igny8_core.business.billing.models import CreditCostConfig
# Get operation config
config = CreditCostConfig.objects.filter(
operation_type=operation_type,
is_active=True
).first()
if not config:
raise CreditCalculationError(f"No active config found for operation: {operation_type}")
# Check if operation uses token-based billing
if config.unit in ['per_100_tokens', 'per_1000_tokens']:
total_tokens = tokens_input + tokens_output
# Get model's tokens-per-credit ratio
tokens_per_credit = model_config.tokens_per_credit
if tokens_per_credit <= 0:
raise CreditCalculationError(f"Invalid tokens_per_credit: {tokens_per_credit}")
# Calculate credits (float)
credits_float = Decimal(total_tokens) / Decimal(tokens_per_credit)
# Apply rounding (always round up to avoid undercharging)
credits = math.ceil(credits_float)
# Apply minimum cost from config (if set)
credits = max(credits, config.credits_cost)
logger.info(
f"Token-based calculation: {total_tokens} tokens / {tokens_per_credit} = {credits} credits "
f"(model: {model_config.model_name}, operation: {operation_type})"
)
return credits
else:
# Fall back to legacy calculation for non-token operations
logger.warning(
f"Operation {operation_type} uses unit {config.unit}, falling back to legacy calculation"
)
return config.credits_cost
except Exception as e:
logger.error(f"Failed to calculate credits from tokens: {e}")
raise CreditCalculationError(f"Credit calculation failed: {e}")
@staticmethod
def get_model_for_operation(account, operation_type, task_model_override=None):
"""
Determine which AI model to use for an operation.
Priority: Task Override > Account Default > Operation Default > System Default
Args:
account: Account instance
operation_type: Type of operation
task_model_override: Optional AIModelConfig instance from task
Returns:
AIModelConfig: The model to use
"""
import logging
logger = logging.getLogger(__name__)
# 1. Task-level override (highest priority)
if task_model_override:
logger.info(f"Using task-level model override: {task_model_override.model_name}")
return task_model_override
# 2. Account default model (from IntegrationSettings)
try:
from igny8_core.modules.system.models import IntegrationSettings
from igny8_core.business.billing.models import CreditCostConfig
integration = IntegrationSettings.objects.filter(account=account).first()
if integration:
# Determine if this is text or image operation
config = CreditCostConfig.objects.filter(
operation_type=operation_type,
is_active=True
).first()
if config and config.default_model:
model_type = config.default_model.model_type
if model_type == 'text' and integration.default_text_model:
logger.info(f"Using account default text model: {integration.default_text_model.model_name}")
return integration.default_text_model
elif model_type == 'image' and integration.default_image_model:
logger.info(f"Using account default image model: {integration.default_image_model.model_name}")
return integration.default_image_model
except Exception as e:
logger.warning(f"Failed to get account default model: {e}")
# 3. Operation default model
try:
from igny8_core.business.billing.models import CreditCostConfig
config = CreditCostConfig.objects.filter(
operation_type=operation_type,
is_active=True
).first()
if config and config.default_model:
logger.info(f"Using operation default model: {config.default_model.model_name}")
return config.default_model
except Exception as e:
logger.warning(f"Failed to get operation default model: {e}")
# 4. System-wide default (fallback)
try:
default_model = AIModelConfig.objects.filter(
is_default=True,
is_active=True
).first()
if default_model:
logger.info(f"Using system default model: {default_model.model_name}")
return default_model
except Exception as e:
logger.warning(f"Failed to get system default model: {e}")
# 5. Hard-coded fallback
logger.warning("All model selection failed, using hard-coded fallback: gpt-4o-mini")
return AIModelConfig.objects.filter(model_name='gpt-4o-mini').first()

3
backend/igny8_core/business/system/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
System app package
"""

65
backend/igny8_core/business/system/admin.py Normal file
View File

@@ -0,0 +1,65 @@
"""
System admin configuration
"""
from django.contrib import admin
from django.urls import reverse
from django.utils.html import format_html
from igny8_core.business.system.models import DebugConfiguration
@admin.register(DebugConfiguration)
class DebugConfigurationAdmin(admin.ModelAdmin):
"""Admin for debug configuration (singleton)"""
def has_add_permission(self, request):
# Only allow one instance
return not DebugConfiguration.objects.exists()
def has_delete_permission(self, request, obj=None):
# Don't allow deletion
return False
def changelist_view(self, request, extra_context=None):
# Redirect to edit view for singleton
if DebugConfiguration.objects.exists():
obj = DebugConfiguration.objects.first()
return self.changeform_view(request, str(obj.pk), '', extra_context)
return super().changelist_view(request, extra_context)
fieldsets = (
('Debug Logging Control', {
'fields': ('enable_debug_logging',),
'description': '⚠️ <strong>Master Switch:</strong> When DISABLED, all logging below is completely skipped (zero overhead). When ENABLED, logs appear in console output.'
}),
('Logging Categories', {
'fields': (
'log_ai_steps',
'log_api_requests',
'log_database_queries',
'log_celery_tasks',
),
'description': 'Fine-tune what gets logged when debug logging is enabled'
}),
('Audit', {
'fields': ('updated_at', 'updated_by'),
'classes': ('collapse',)
}),
)
readonly_fields = ('updated_at', 'updated_by')
def save_model(self, request, obj, form, change):
obj.updated_by = request.user
super().save_model(request, obj, form, change)
# Show message about cache clearing
if change:
self.message_user(request,
"Debug configuration updated. Cache cleared. Changes take effect immediately.",
level='success'
)
class Media:
css = {
'all': ('admin/css/forms.css',)
}

11
backend/igny8_core/business/system/apps.py Normal file
View File

@@ -0,0 +1,11 @@
"""
System app configuration
"""
from django.apps import AppConfig
class SystemConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'igny8_core.business.system'
label = 'debug_system'
verbose_name = 'Debug & System Settings'

35
backend/igny8_core/business/system/migrations/0001_initial.py Normal file
View File

@@ -0,0 +1,35 @@
# Generated by Django 5.2.9 on 2025-12-23 02:32
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='DebugConfiguration',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('enable_debug_logging', models.BooleanField(default=False, help_text='Enable verbose debug logging to console (AI steps, detailed execution)')),
('log_ai_steps', models.BooleanField(default=True, help_text='Log AI function execution steps (only when debug logging enabled)')),
('log_api_requests', models.BooleanField(default=False, help_text='Log all API requests and responses (only when debug logging enabled)')),
('log_database_queries', models.BooleanField(default=False, help_text='Log database queries (only when debug logging enabled)')),
('log_celery_tasks', models.BooleanField(default=True, help_text='Log Celery task execution (only when debug logging enabled)')),
('updated_at', models.DateTimeField(auto_now=True)),
('updated_by', models.ForeignKey(blank=True, help_text='Admin who last updated', null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL)),
],
options={
'verbose_name': 'Debug Configuration',
'verbose_name_plural': 'Debug Configuration',
'db_table': 'igny8_debug_configuration',
},
),
]

0
backend/igny8_core/business/system/migrations/__init__.py Normal file
View File

86
backend/igny8_core/business/system/models.py Normal file
View File

@@ -0,0 +1,86 @@
"""
System-wide settings and configuration models
"""
from django.db import models
from django.conf import settings
from django.core.cache import cache
class DebugConfiguration(models.Model):
"""
System-wide debug configuration (Singleton).
Controls verbose logging and debugging features.
"""
# Debug settings
enable_debug_logging = models.BooleanField(
default=False,
help_text="Enable verbose debug logging to console (AI steps, detailed execution)"
)
log_ai_steps = models.BooleanField(
default=True,
help_text="Log AI function execution steps (only when debug logging enabled)"
)
log_api_requests = models.BooleanField(
default=False,
help_text="Log all API requests and responses (only when debug logging enabled)"
)
log_database_queries = models.BooleanField(
default=False,
help_text="Log database queries (only when debug logging enabled)"
)
log_celery_tasks = models.BooleanField(
default=True,
help_text="Log Celery task execution (only when debug logging enabled)"
)
# 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 = 'debug_system'
db_table = 'igny8_debug_configuration'
verbose_name = 'Debug Configuration'
verbose_name_plural = 'Debug Configuration'
def save(self, *args, **kwargs):
"""Enforce singleton pattern and clear cache on save"""
self.pk = 1
super().save(*args, **kwargs)
# Clear ALL debug-related caches when settings change
cache.delete('debug_config')
cache.delete('debug_enabled')
cache.delete('debug_first_worker_pid') # Reset worker selection
@classmethod
def get_config(cls):
"""Get or create the singleton config (cached)"""
config = cache.get('debug_config')
if config is None:
config, created = cls.objects.get_or_create(pk=1)
cache.set('debug_config', config, 300) # Cache for 5 minutes
return config
@classmethod
def is_debug_enabled(cls):
"""Fast check if debug logging is enabled (cached for performance)"""
enabled = cache.get('debug_enabled')
if enabled is None:
config = cls.get_config()
enabled = config.enable_debug_logging
cache.set('debug_enabled', enabled, 60) # Cache for 1 minute
return enabled
def __str__(self):
status = "ENABLED" if self.enable_debug_logging else "DISABLED"
return f"Debug Configuration ({status})"