BillingConfiguration model doesn't exist in the codebase, causing ImportError and preventing backend from starting. Removed the import and admin class. This was causing the backend to be stuck in restarting state.
908 lines
35 KiB
Python
908 lines
35 KiB
Python
"""
|
|
Billing Module Admin
|
|
"""
|
|
from django.contrib import admin
|
|
from django.utils.html import format_html
|
|
from django.contrib import messages
|
|
from unfold.admin import ModelAdmin
|
|
from simple_history.admin import SimpleHistoryAdmin
|
|
from igny8_core.admin.base import AccountAdminMixin, Igny8ModelAdmin
|
|
from igny8_core.business.billing.models import (
|
|
CreditCostConfig,
|
|
Invoice,
|
|
Payment,
|
|
CreditPackage,
|
|
PaymentMethodConfig,
|
|
PlanLimitUsage,
|
|
AIModelConfig,
|
|
)
|
|
from .models import CreditTransaction, CreditUsageLog, AccountPaymentMethod
|
|
from import_export.admin import ExportMixin, ImportExportMixin
|
|
from import_export import resources
|
|
from rangefilter.filters import DateRangeFilter
|
|
|
|
|
|
class CreditTransactionResource(resources.ModelResource):
|
|
"""Resource class for exporting Credit Transactions"""
|
|
class Meta:
|
|
model = CreditTransaction
|
|
fields = ('id', 'account__name', 'transaction_type', 'amount', 'balance_after',
|
|
'description', 'reference_id', 'created_at')
|
|
export_order = fields
|
|
|
|
|
|
@admin.register(CreditTransaction)
|
|
class CreditTransactionAdmin(ExportMixin, AccountAdminMixin, Igny8ModelAdmin):
|
|
resource_class = CreditTransactionResource
|
|
list_display = ['id', 'account', 'transaction_type', 'amount', 'balance_after', 'description', 'created_at']
|
|
list_filter = ['transaction_type', ('created_at', DateRangeFilter), 'account']
|
|
search_fields = ['description', 'account__name']
|
|
readonly_fields = ['created_at']
|
|
date_hierarchy = 'created_at'
|
|
|
|
def get_account_display(self, obj):
|
|
"""Safely get account name"""
|
|
try:
|
|
account = getattr(obj, 'account', None)
|
|
return account.name if account else '-'
|
|
except:
|
|
return '-'
|
|
get_account_display.short_description = 'Account'
|
|
|
|
|
|
class CreditUsageLogResource(resources.ModelResource):
|
|
"""Resource class for exporting Credit Usage Logs"""
|
|
class Meta:
|
|
model = CreditUsageLog
|
|
fields = ('id', 'account__name', 'operation_type', 'credits_used', 'cost_usd',
|
|
'model_used', 'created_at')
|
|
export_order = fields
|
|
|
|
|
|
@admin.register(CreditUsageLog)
|
|
class CreditUsageLogAdmin(ExportMixin, AccountAdminMixin, Igny8ModelAdmin):
|
|
resource_class = CreditUsageLogResource
|
|
list_display = ['id', 'account', 'operation_type', 'credits_used', 'cost_usd', 'model_used', 'created_at']
|
|
list_filter = ['operation_type', 'created_at', 'account', 'model_used']
|
|
search_fields = ['account__name', 'model_used']
|
|
readonly_fields = ['created_at']
|
|
date_hierarchy = 'created_at'
|
|
|
|
def get_account_display(self, obj):
|
|
"""Safely get account name"""
|
|
try:
|
|
account = getattr(obj, 'account', None)
|
|
return account.name if account else '-'
|
|
except:
|
|
return '-'
|
|
get_account_display.short_description = 'Account'
|
|
|
|
|
|
class InvoiceResource(resources.ModelResource):
|
|
"""Resource class for exporting Invoices"""
|
|
class Meta:
|
|
model = Invoice
|
|
fields = ('id', 'invoice_number', 'account__name', 'status', 'total', 'currency',
|
|
'invoice_date', 'due_date', 'created_at', 'updated_at')
|
|
export_order = fields
|
|
|
|
|
|
@admin.register(Invoice)
|
|
class InvoiceAdmin(ExportMixin, AccountAdminMixin, Igny8ModelAdmin):
|
|
resource_class = InvoiceResource
|
|
list_display = [
|
|
'invoice_number',
|
|
'account',
|
|
'status',
|
|
'total',
|
|
'currency',
|
|
'invoice_date',
|
|
'due_date',
|
|
]
|
|
list_filter = ['status', 'currency', 'invoice_date', 'account']
|
|
search_fields = ['invoice_number', 'account__name']
|
|
readonly_fields = ['created_at', 'updated_at']
|
|
actions = [
|
|
'bulk_set_status_draft',
|
|
'bulk_set_status_sent',
|
|
'bulk_set_status_paid',
|
|
'bulk_set_status_overdue',
|
|
'bulk_set_status_cancelled',
|
|
'bulk_send_reminders',
|
|
]
|
|
|
|
def bulk_set_status_draft(self, request, queryset):
|
|
"""Set selected invoices to draft status"""
|
|
updated = queryset.update(status='draft')
|
|
self.message_user(request, f'{updated} invoice(s) set to draft.', messages.SUCCESS)
|
|
bulk_set_status_draft.short_description = 'Set status to Draft'
|
|
|
|
def bulk_set_status_sent(self, request, queryset):
|
|
"""Set selected invoices to sent status"""
|
|
updated = queryset.update(status='sent')
|
|
self.message_user(request, f'{updated} invoice(s) set to sent.', messages.SUCCESS)
|
|
bulk_set_status_sent.short_description = 'Set status to Sent'
|
|
|
|
def bulk_set_status_paid(self, request, queryset):
|
|
"""Set selected invoices to paid status"""
|
|
updated = queryset.update(status='paid')
|
|
self.message_user(request, f'{updated} invoice(s) set to paid.', messages.SUCCESS)
|
|
bulk_set_status_paid.short_description = 'Set status to Paid'
|
|
|
|
def bulk_set_status_overdue(self, request, queryset):
|
|
"""Set selected invoices to overdue status"""
|
|
updated = queryset.update(status='overdue')
|
|
self.message_user(request, f'{updated} invoice(s) set to overdue.', messages.SUCCESS)
|
|
bulk_set_status_overdue.short_description = 'Set status to Overdue'
|
|
|
|
def bulk_set_status_cancelled(self, request, queryset):
|
|
"""Set selected invoices to cancelled status"""
|
|
updated = queryset.update(status='cancelled')
|
|
self.message_user(request, f'{updated} invoice(s) set to cancelled.', messages.SUCCESS)
|
|
bulk_set_status_cancelled.short_description = 'Set status to Cancelled'
|
|
|
|
def bulk_send_reminders(self, request, queryset):
|
|
"""Send reminder emails for selected invoices"""
|
|
# TODO: Implement email sending logic when email service is configured
|
|
unpaid = queryset.filter(status__in=['sent', 'overdue'])
|
|
count = unpaid.count()
|
|
self.message_user(
|
|
request,
|
|
f'{count} invoice reminder(s) queued for sending. (Email integration required)',
|
|
messages.INFO
|
|
)
|
|
bulk_send_reminders.short_description = 'Send payment reminders'
|
|
|
|
|
|
class PaymentResource(resources.ModelResource):
|
|
"""Resource class for exporting Payments"""
|
|
class Meta:
|
|
model = Payment
|
|
fields = ('id', 'invoice__invoice_number', 'account__name', 'payment_method',
|
|
'status', 'amount', 'currency', 'manual_reference', 'approved_by__email',
|
|
'processed_at', 'created_at')
|
|
export_order = fields
|
|
|
|
|
|
@admin.register(Payment)
|
|
class PaymentAdmin(ExportMixin, AccountAdminMixin, SimpleHistoryAdmin, Igny8ModelAdmin):
|
|
"""
|
|
Main Payment Admin with approval workflow.
|
|
When you change status to 'succeeded', it automatically:
|
|
- Updates invoice to 'paid'
|
|
- Activates subscription
|
|
- Activates account
|
|
- Adds credits
|
|
"""
|
|
resource_class = PaymentResource
|
|
list_display = [
|
|
'id',
|
|
'invoice',
|
|
'account',
|
|
'payment_method',
|
|
'status',
|
|
'amount',
|
|
'currency',
|
|
'manual_reference',
|
|
'approved_by',
|
|
'processed_at',
|
|
]
|
|
list_filter = ['status', 'payment_method', 'currency', ('created_at', DateRangeFilter), ('processed_at', DateRangeFilter)]
|
|
search_fields = [
|
|
'invoice__invoice_number',
|
|
'account__name',
|
|
'stripe_payment_intent_id',
|
|
'paypal_order_id',
|
|
'manual_reference',
|
|
'admin_notes',
|
|
'manual_notes'
|
|
]
|
|
readonly_fields = ['created_at', 'updated_at', 'approved_at', 'processed_at', 'failed_at', 'refunded_at']
|
|
actions = ['approve_payments', 'reject_payments', 'bulk_refund']
|
|
|
|
fieldsets = (
|
|
('Payment Info', {
|
|
'fields': ('invoice', 'account', 'amount', 'currency', 'payment_method', 'status')
|
|
}),
|
|
('Manual Payment Details', {
|
|
'fields': ('manual_reference', 'manual_notes', 'admin_notes'),
|
|
'classes': ('collapse',),
|
|
}),
|
|
('Stripe/PayPal', {
|
|
'fields': ('stripe_payment_intent_id', 'stripe_charge_id', 'paypal_order_id', 'paypal_capture_id'),
|
|
'classes': ('collapse',),
|
|
}),
|
|
('Approval Info', {
|
|
'fields': ('approved_by', 'approved_at', 'processed_at', 'failed_at', 'refunded_at', 'failure_reason'),
|
|
'classes': ('collapse',),
|
|
}),
|
|
('Timestamps', {
|
|
'fields': ('created_at', 'updated_at'),
|
|
'classes': ('collapse',),
|
|
}),
|
|
)
|
|
|
|
def save_model(self, request, obj, form, change):
|
|
"""
|
|
Override save_model to trigger approval workflow when status changes to succeeded.
|
|
This ensures manual status changes in admin also activate accounts and add credits.
|
|
"""
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from igny8_core.business.billing.services.credit_service import CreditService
|
|
from igny8_core.auth.models import Subscription
|
|
|
|
# Check if status changed to 'succeeded'
|
|
status_changed_to_succeeded = False
|
|
if change and 'status' in form.changed_data:
|
|
if obj.status == 'succeeded' and form.initial.get('status') != 'succeeded':
|
|
status_changed_to_succeeded = True
|
|
elif not change and obj.status == 'succeeded':
|
|
status_changed_to_succeeded = True
|
|
|
|
# Save the payment first
|
|
if obj.status == 'succeeded' and not obj.approved_by:
|
|
obj.approved_by = request.user
|
|
if not obj.approved_at:
|
|
obj.approved_at = timezone.now()
|
|
if not obj.processed_at:
|
|
obj.processed_at = timezone.now()
|
|
|
|
super().save_model(request, obj, form, change)
|
|
|
|
# If status changed to succeeded, trigger the full approval workflow
|
|
if status_changed_to_succeeded:
|
|
try:
|
|
with transaction.atomic():
|
|
invoice = obj.invoice
|
|
account = obj.account
|
|
|
|
# Get subscription from invoice or account
|
|
subscription = None
|
|
if invoice and hasattr(invoice, 'subscription') and invoice.subscription:
|
|
subscription = invoice.subscription
|
|
elif account and hasattr(account, 'subscription'):
|
|
try:
|
|
subscription = account.subscription
|
|
except Subscription.DoesNotExist:
|
|
pass
|
|
|
|
# Update Invoice
|
|
if invoice and invoice.status != 'paid':
|
|
invoice.status = 'paid'
|
|
invoice.paid_at = timezone.now()
|
|
invoice.save()
|
|
|
|
# Update Subscription
|
|
if subscription and subscription.status != 'active':
|
|
subscription.status = 'active'
|
|
subscription.external_payment_id = obj.manual_reference
|
|
subscription.save()
|
|
|
|
# Update Account
|
|
if account.status != 'active':
|
|
account.status = 'active'
|
|
account.save()
|
|
|
|
# Add Credits (check if not already added)
|
|
from igny8_core.business.billing.models import CreditTransaction
|
|
existing_credit = CreditTransaction.objects.filter(
|
|
account=account,
|
|
metadata__payment_id=obj.id
|
|
).exists()
|
|
|
|
if not existing_credit:
|
|
credits_to_add = 0
|
|
plan_name = ''
|
|
|
|
if subscription and subscription.plan:
|
|
credits_to_add = subscription.plan.included_credits
|
|
plan_name = subscription.plan.name
|
|
elif account and account.plan:
|
|
credits_to_add = account.plan.included_credits
|
|
plan_name = account.plan.name
|
|
|
|
if credits_to_add > 0:
|
|
CreditService.add_credits(
|
|
account=account,
|
|
amount=credits_to_add,
|
|
transaction_type='subscription',
|
|
description=f'{plan_name} - Invoice {invoice.invoice_number}',
|
|
metadata={
|
|
'subscription_id': subscription.id if subscription else None,
|
|
'invoice_id': invoice.id,
|
|
'payment_id': obj.id,
|
|
'approved_by': request.user.email
|
|
}
|
|
)
|
|
self.message_user(
|
|
request,
|
|
f'✓ Payment approved: Account activated, {credits_to_add} credits added',
|
|
level='SUCCESS'
|
|
)
|
|
|
|
except Exception as e:
|
|
self.message_user(
|
|
request,
|
|
f'✗ Payment saved but workflow failed: {str(e)}',
|
|
level='ERROR'
|
|
)
|
|
|
|
def approve_payments(self, request, queryset):
|
|
"""Approve selected manual payments"""
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from igny8_core.business.billing.services.credit_service import CreditService
|
|
from igny8_core.auth.models import Subscription
|
|
|
|
successful = []
|
|
errors = []
|
|
|
|
for payment in queryset.filter(status='pending_approval'):
|
|
try:
|
|
with transaction.atomic():
|
|
invoice = payment.invoice
|
|
account = payment.account
|
|
|
|
# Get subscription from invoice or account
|
|
subscription = None
|
|
if invoice and hasattr(invoice, 'subscription') and invoice.subscription:
|
|
subscription = invoice.subscription
|
|
elif account and hasattr(account, 'subscription'):
|
|
try:
|
|
subscription = account.subscription
|
|
except Subscription.DoesNotExist:
|
|
pass
|
|
|
|
# Update Payment
|
|
payment.status = 'succeeded'
|
|
payment.approved_by = request.user
|
|
payment.approved_at = timezone.now()
|
|
payment.processed_at = timezone.now()
|
|
payment.admin_notes = f'Bulk approved by {request.user.email}'
|
|
payment.save()
|
|
|
|
# Update Invoice
|
|
invoice.status = 'paid'
|
|
invoice.paid_at = timezone.now()
|
|
invoice.save()
|
|
|
|
# Update Subscription
|
|
if subscription:
|
|
subscription.status = 'active'
|
|
subscription.external_payment_id = payment.manual_reference
|
|
subscription.save()
|
|
|
|
# Update Account
|
|
account.status = 'active'
|
|
account.save()
|
|
|
|
# Add Credits
|
|
credits_added = 0
|
|
if subscription and subscription.plan and subscription.plan.included_credits > 0:
|
|
credits_added = subscription.plan.included_credits
|
|
CreditService.add_credits(
|
|
account=account,
|
|
amount=credits_added,
|
|
transaction_type='subscription',
|
|
description=f'{subscription.plan.name} - Invoice {invoice.invoice_number}',
|
|
metadata={
|
|
'subscription_id': subscription.id,
|
|
'invoice_id': invoice.id,
|
|
'payment_id': payment.id,
|
|
'approved_by': request.user.email
|
|
}
|
|
)
|
|
elif account and account.plan and account.plan.included_credits > 0:
|
|
credits_added = account.plan.included_credits
|
|
CreditService.add_credits(
|
|
account=account,
|
|
amount=credits_added,
|
|
transaction_type='subscription',
|
|
description=f'{account.plan.name} - Invoice {invoice.invoice_number}',
|
|
metadata={
|
|
'invoice_id': invoice.id,
|
|
'payment_id': payment.id,
|
|
'approved_by': request.user.email
|
|
}
|
|
)
|
|
|
|
successful.append(f'Payment #{payment.id} - {account.name} - Invoice {invoice.invoice_number} - {credits_added} credits')
|
|
|
|
except Exception as e:
|
|
errors.append(f'Payment #{payment.id}: {str(e)}')
|
|
|
|
# Detailed success message
|
|
if successful:
|
|
self.message_user(request, f'✓ Successfully approved {len(successful)} payment(s):', level='SUCCESS')
|
|
for msg in successful[:10]: # Show first 10
|
|
self.message_user(request, f' • {msg}', level='SUCCESS')
|
|
if len(successful) > 10:
|
|
self.message_user(request, f' ... and {len(successful) - 10} more', level='SUCCESS')
|
|
|
|
# Detailed error messages
|
|
if errors:
|
|
self.message_user(request, f'✗ Failed to approve {len(errors)} payment(s):', level='ERROR')
|
|
for error in errors:
|
|
self.message_user(request, f' • {error}', level='ERROR')
|
|
|
|
approve_payments.short_description = 'Approve selected manual payments'
|
|
|
|
def reject_payments(self, request, queryset):
|
|
"""Reject selected manual payments"""
|
|
from django.utils import timezone
|
|
|
|
count = queryset.filter(status='pending_approval').update(
|
|
status='failed',
|
|
approved_by=request.user,
|
|
approved_at=timezone.now(),
|
|
failed_at=timezone.now(),
|
|
admin_notes=f'Bulk rejected by {request.user.email}',
|
|
failure_reason='Rejected by admin'
|
|
)
|
|
|
|
self.message_user(request, f'Rejected {count} payment(s)')
|
|
|
|
reject_payments.short_description = 'Reject selected manual payments'
|
|
|
|
def bulk_refund(self, request, queryset):
|
|
"""Refund selected payments"""
|
|
from django.utils import timezone
|
|
|
|
# Only refund succeeded payments
|
|
succeeded_payments = queryset.filter(status='succeeded')
|
|
count = 0
|
|
|
|
for payment in succeeded_payments:
|
|
# Mark as refunded
|
|
payment.status = 'refunded'
|
|
payment.refunded_at = timezone.now()
|
|
payment.admin_notes = f'{payment.admin_notes or ""}\nBulk refunded by {request.user.email} on {timezone.now()}'
|
|
payment.save()
|
|
|
|
# TODO: Process actual refund through payment gateway (Stripe/PayPal)
|
|
# For now, just marking as refunded in database
|
|
|
|
count += 1
|
|
|
|
self.message_user(
|
|
request,
|
|
f'{count} payment(s) marked as refunded. Note: Actual gateway refunds need to be processed separately.',
|
|
messages.WARNING
|
|
)
|
|
bulk_refund.short_description = 'Refund selected payments'
|
|
|
|
|
|
class CreditPackageResource(resources.ModelResource):
|
|
"""Resource class for importing/exporting Credit Packages"""
|
|
class Meta:
|
|
model = CreditPackage
|
|
fields = ('id', 'name', 'slug', 'credits', 'price', 'discount_percentage',
|
|
'is_active', 'is_featured', 'sort_order', 'created_at')
|
|
export_order = fields
|
|
import_id_fields = ('id',)
|
|
skip_unchanged = True
|
|
|
|
|
|
@admin.register(CreditPackage)
|
|
class CreditPackageAdmin(ImportExportMixin, Igny8ModelAdmin):
|
|
resource_class = CreditPackageResource
|
|
list_display = ['name', 'slug', 'credits', 'price', 'discount_percentage', 'is_active', 'is_featured', 'sort_order']
|
|
list_filter = ['is_active', 'is_featured']
|
|
search_fields = ['name', 'slug']
|
|
readonly_fields = ['created_at', 'updated_at']
|
|
actions = [
|
|
'bulk_activate',
|
|
'bulk_deactivate',
|
|
]
|
|
actions = [
|
|
'bulk_activate',
|
|
'bulk_deactivate',
|
|
]
|
|
|
|
def bulk_activate(self, request, queryset):
|
|
updated = queryset.update(is_active=True)
|
|
self.message_user(request, f'{updated} credit package(s) activated.', messages.SUCCESS)
|
|
bulk_activate.short_description = 'Activate selected packages'
|
|
|
|
def bulk_deactivate(self, request, queryset):
|
|
updated = queryset.update(is_active=False)
|
|
self.message_user(request, f'{updated} credit package(s) deactivated.', messages.SUCCESS)
|
|
bulk_deactivate.short_description = 'Deactivate selected packages'
|
|
|
|
|
|
@admin.register(PaymentMethodConfig)
|
|
class PaymentMethodConfigAdmin(Igny8ModelAdmin):
|
|
list_display = ['country_code', 'payment_method', 'display_name', 'is_enabled', 'sort_order', 'updated_at']
|
|
list_filter = ['payment_method', 'is_enabled', 'country_code']
|
|
search_fields = ['country_code', 'display_name', 'payment_method']
|
|
list_editable = ['is_enabled', 'sort_order']
|
|
readonly_fields = ['created_at', 'updated_at']
|
|
|
|
|
|
@admin.register(AccountPaymentMethod)
|
|
class AccountPaymentMethodAdmin(AccountAdminMixin, Igny8ModelAdmin):
|
|
list_display = [
|
|
'display_name',
|
|
'type',
|
|
'account',
|
|
'is_default',
|
|
'is_enabled',
|
|
'is_verified',
|
|
'country_code',
|
|
'updated_at',
|
|
]
|
|
list_filter = ['type', 'is_default', 'is_enabled', 'is_verified', 'country_code']
|
|
search_fields = ['display_name', 'account__name', 'account__id']
|
|
readonly_fields = ['created_at', 'updated_at']
|
|
fieldsets = (
|
|
('Payment Method', {
|
|
'fields': ('account', 'type', 'display_name', 'is_default', 'is_enabled', 'is_verified', 'country_code')
|
|
}),
|
|
('Instructions / Metadata', {
|
|
'fields': ('instructions', 'metadata')
|
|
}),
|
|
('Timestamps', {
|
|
'fields': ('created_at', 'updated_at'),
|
|
'classes': ('collapse',)
|
|
}),
|
|
)
|
|
|
|
|
|
@admin.register(CreditCostConfig)
|
|
class CreditCostConfigAdmin(SimpleHistoryAdmin, Igny8ModelAdmin):
|
|
list_display = [
|
|
'operation_type',
|
|
'display_name',
|
|
'tokens_per_credit_display',
|
|
'price_per_credit_usd',
|
|
'min_credits',
|
|
'is_active',
|
|
'cost_change_indicator',
|
|
'updated_at',
|
|
'updated_by'
|
|
]
|
|
|
|
list_filter = ['is_active', 'updated_at']
|
|
search_fields = ['operation_type', 'display_name', 'description']
|
|
|
|
fieldsets = (
|
|
('Operation', {
|
|
'fields': ('operation_type', 'display_name', 'description')
|
|
}),
|
|
('Token-to-Credit Configuration', {
|
|
'fields': ('tokens_per_credit', 'min_credits', 'price_per_credit_usd', 'is_active'),
|
|
'description': 'Configure how tokens are converted to credits for this operation'
|
|
}),
|
|
('Audit Trail', {
|
|
'fields': ('previous_tokens_per_credit', 'updated_by', 'created_at', 'updated_at'),
|
|
'classes': ('collapse',)
|
|
}),
|
|
)
|
|
|
|
readonly_fields = ['created_at', 'updated_at', 'previous_tokens_per_credit']
|
|
|
|
def tokens_per_credit_display(self, obj):
|
|
"""Show token ratio with color coding"""
|
|
if obj.tokens_per_credit <= 50:
|
|
color = 'red' # Expensive (low tokens per credit)
|
|
elif obj.tokens_per_credit <= 100:
|
|
color = 'orange'
|
|
else:
|
|
color = 'green' # Cheap (high tokens per credit)
|
|
return format_html(
|
|
'<span style="color: {}; font-weight: bold;">{} tokens/credit</span>',
|
|
color,
|
|
obj.tokens_per_credit
|
|
)
|
|
tokens_per_credit_display.short_description = 'Token Ratio'
|
|
|
|
def cost_change_indicator(self, obj):
|
|
"""Show if token ratio changed recently"""
|
|
if obj.previous_tokens_per_credit is not None:
|
|
if obj.tokens_per_credit < obj.previous_tokens_per_credit:
|
|
icon = '📈' # More expensive (fewer tokens per credit)
|
|
color = 'red'
|
|
elif obj.tokens_per_credit > obj.previous_tokens_per_credit:
|
|
icon = '📉' # Cheaper (more tokens per credit)
|
|
color = 'green'
|
|
else:
|
|
icon = '➡️' # Same
|
|
color = 'gray'
|
|
|
|
return format_html(
|
|
'{} <span style="color: {};">({} → {})</span>',
|
|
icon,
|
|
color,
|
|
obj.previous_tokens_per_credit,
|
|
obj.tokens_per_credit
|
|
)
|
|
return '—'
|
|
cost_change_indicator.short_description = 'Recent Change'
|
|
|
|
def save_model(self, request, obj, form, change):
|
|
"""Track who made the change"""
|
|
obj.updated_by = request.user
|
|
super().save_model(request, obj, form, change)
|
|
|
|
|
|
class PlanLimitUsageResource(resources.ModelResource):
|
|
"""Resource class for exporting Plan Limit Usage"""
|
|
class Meta:
|
|
model = PlanLimitUsage
|
|
fields = ('id', 'account__name', 'limit_type', 'amount_used',
|
|
'period_start', 'period_end', 'created_at')
|
|
export_order = fields
|
|
|
|
|
|
@admin.register(PlanLimitUsage)
|
|
class PlanLimitUsageAdmin(ExportMixin, AccountAdminMixin, Igny8ModelAdmin):
|
|
resource_class = PlanLimitUsageResource
|
|
"""Admin for tracking plan limit usage across billing periods"""
|
|
list_display = [
|
|
'account',
|
|
'limit_type',
|
|
'amount_used',
|
|
'period_display',
|
|
'created_at',
|
|
]
|
|
list_filter = [
|
|
'limit_type',
|
|
('period_start', DateRangeFilter),
|
|
('period_end', DateRangeFilter),
|
|
'account',
|
|
]
|
|
search_fields = ['account__name']
|
|
readonly_fields = ['created_at', 'updated_at']
|
|
date_hierarchy = 'period_start'
|
|
actions = [
|
|
'bulk_reset_usage',
|
|
'bulk_delete_old_records',
|
|
]
|
|
|
|
fieldsets = (
|
|
('Usage Info', {
|
|
'fields': ('account', 'limit_type', 'amount_used')
|
|
}),
|
|
('Billing Period', {
|
|
'fields': ('period_start', 'period_end')
|
|
}),
|
|
('Metadata', {
|
|
'fields': ('metadata',),
|
|
'classes': ('collapse',)
|
|
}),
|
|
('Timestamps', {
|
|
'fields': ('created_at', 'updated_at'),
|
|
'classes': ('collapse',)
|
|
}),
|
|
)
|
|
|
|
def period_display(self, obj):
|
|
"""Display billing period range"""
|
|
return f"{obj.period_start} to {obj.period_end}"
|
|
period_display.short_description = 'Billing Period'
|
|
|
|
def bulk_reset_usage(self, request, queryset):
|
|
"""Reset usage counters to zero"""
|
|
updated = queryset.update(amount_used=0)
|
|
self.message_user(request, f'{updated} usage counter(s) reset to zero.', messages.SUCCESS)
|
|
bulk_reset_usage.short_description = 'Reset usage counters'
|
|
|
|
def bulk_delete_old_records(self, request, queryset):
|
|
"""Delete usage records older than 1 year"""
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
|
|
cutoff_date = timezone.now() - timedelta(days=365)
|
|
old_records = queryset.filter(period_end__lt=cutoff_date)
|
|
count = old_records.count()
|
|
old_records.delete()
|
|
self.message_user(request, f'{count} old usage record(s) deleted (older than 1 year).', messages.SUCCESS)
|
|
bulk_delete_old_records.short_description = 'Delete old records (>1 year)'
|
|
|
|
|
|
@admin.register(AIModelConfig)
|
|
class AIModelConfigAdmin(SimpleHistoryAdmin, Igny8ModelAdmin):
|
|
"""
|
|
Admin for AI Model Configuration - Database-driven model pricing
|
|
Replaces hardcoded MODEL_RATES and IMAGE_MODEL_RATES
|
|
"""
|
|
list_display = [
|
|
'model_name',
|
|
'display_name_short',
|
|
'model_type_badge',
|
|
'provider_badge',
|
|
'pricing_display',
|
|
'is_active_icon',
|
|
'is_default_icon',
|
|
'sort_order',
|
|
'updated_at',
|
|
]
|
|
|
|
list_filter = [
|
|
'model_type',
|
|
'provider',
|
|
'is_active',
|
|
'is_default',
|
|
'supports_json_mode',
|
|
'supports_vision',
|
|
'supports_function_calling',
|
|
]
|
|
|
|
search_fields = ['model_name', 'display_name', 'description']
|
|
|
|
ordering = ['model_type', 'sort_order', 'model_name']
|
|
|
|
readonly_fields = ['created_at', 'updated_at', 'updated_by']
|
|
|
|
fieldsets = (
|
|
('Basic Information', {
|
|
'fields': ('model_name', 'display_name', 'model_type', 'provider', 'description'),
|
|
'description': 'Core model identification and classification'
|
|
}),
|
|
('Text Model Pricing', {
|
|
'fields': ('input_cost_per_1m', 'output_cost_per_1m', 'context_window', 'max_output_tokens'),
|
|
'description': 'Pricing and limits for TEXT models only (leave blank for image models)',
|
|
'classes': ('collapse',)
|
|
}),
|
|
('Image Model Pricing', {
|
|
'fields': ('cost_per_image', 'valid_sizes'),
|
|
'description': 'Pricing and configuration for IMAGE models only (leave blank for text models)',
|
|
'classes': ('collapse',)
|
|
}),
|
|
('Capabilities', {
|
|
'fields': ('supports_json_mode', 'supports_vision', 'supports_function_calling'),
|
|
'description': 'Model features and capabilities'
|
|
}),
|
|
('Status & Display', {
|
|
'fields': ('is_active', 'is_default', 'sort_order'),
|
|
'description': 'Control model availability and ordering in dropdowns'
|
|
}),
|
|
('Lifecycle', {
|
|
'fields': ('release_date', 'deprecation_date'),
|
|
'description': 'Model release and deprecation dates',
|
|
'classes': ('collapse',)
|
|
}),
|
|
('Audit Trail', {
|
|
'fields': ('created_at', 'updated_at', 'updated_by'),
|
|
'classes': ('collapse',)
|
|
}),
|
|
)
|
|
|
|
# Custom display methods
|
|
def display_name_short(self, obj):
|
|
"""Truncated display name for list view"""
|
|
if len(obj.display_name) > 50:
|
|
return obj.display_name[:47] + '...'
|
|
return obj.display_name
|
|
display_name_short.short_description = 'Display Name'
|
|
|
|
def model_type_badge(self, obj):
|
|
"""Colored badge for model type"""
|
|
colors = {
|
|
'text': '#3498db', # Blue
|
|
'image': '#e74c3c', # Red
|
|
'embedding': '#2ecc71', # Green
|
|
}
|
|
color = colors.get(obj.model_type, '#95a5a6')
|
|
return format_html(
|
|
'<span style="background-color: {}; color: white; padding: 3px 10px; '
|
|
'border-radius: 3px; font-weight: bold;">{}</span>',
|
|
color,
|
|
obj.get_model_type_display()
|
|
)
|
|
model_type_badge.short_description = 'Type'
|
|
|
|
def provider_badge(self, obj):
|
|
"""Colored badge for provider"""
|
|
colors = {
|
|
'openai': '#10a37f', # OpenAI green
|
|
'anthropic': '#d97757', # Anthropic orange
|
|
'runware': '#6366f1', # Purple
|
|
'google': '#4285f4', # Google blue
|
|
}
|
|
color = colors.get(obj.provider, '#95a5a6')
|
|
return format_html(
|
|
'<span style="background-color: {}; color: white; padding: 3px 10px; '
|
|
'border-radius: 3px; font-weight: bold;">{}</span>',
|
|
color,
|
|
obj.get_provider_display()
|
|
)
|
|
provider_badge.short_description = 'Provider'
|
|
|
|
def pricing_display(self, obj):
|
|
"""Format pricing based on model type"""
|
|
if obj.model_type == 'text':
|
|
return format_html(
|
|
'<span style="color: #2c3e50; font-family: monospace;">'
|
|
'${} / ${} per 1M</span>',
|
|
obj.input_cost_per_1m,
|
|
obj.output_cost_per_1m
|
|
)
|
|
elif obj.model_type == 'image':
|
|
return format_html(
|
|
'<span style="color: #2c3e50; font-family: monospace;">'
|
|
'${} per image</span>',
|
|
obj.cost_per_image
|
|
)
|
|
return '-'
|
|
pricing_display.short_description = 'Pricing'
|
|
|
|
def is_active_icon(self, obj):
|
|
"""Active status icon"""
|
|
if obj.is_active:
|
|
return format_html(
|
|
'<span style="color: green; font-size: 18px;" title="Active">●</span>'
|
|
)
|
|
return format_html(
|
|
'<span style="color: red; font-size: 18px;" title="Inactive">●</span>'
|
|
)
|
|
is_active_icon.short_description = 'Active'
|
|
|
|
def is_default_icon(self, obj):
|
|
"""Default status icon"""
|
|
if obj.is_default:
|
|
return format_html(
|
|
'<span style="color: gold; font-size: 18px;" title="Default">★</span>'
|
|
)
|
|
return format_html(
|
|
'<span style="color: #ddd; font-size: 18px;" title="Not Default">☆</span>'
|
|
)
|
|
is_default_icon.short_description = 'Default'
|
|
|
|
# Admin actions
|
|
actions = ['bulk_activate', 'bulk_deactivate', 'set_as_default']
|
|
|
|
def bulk_activate(self, request, queryset):
|
|
"""Enable selected models"""
|
|
count = queryset.update(is_active=True)
|
|
self.message_user(
|
|
request,
|
|
f'{count} model(s) activated successfully.',
|
|
messages.SUCCESS
|
|
)
|
|
bulk_activate.short_description = 'Activate selected models'
|
|
|
|
def bulk_deactivate(self, request, queryset):
|
|
"""Disable selected models"""
|
|
count = queryset.update(is_active=False)
|
|
self.message_user(
|
|
request,
|
|
f'{count} model(s) deactivated successfully.',
|
|
messages.WARNING
|
|
)
|
|
bulk_deactivate.short_description = 'Deactivate selected models'
|
|
|
|
def set_as_default(self, request, queryset):
|
|
"""Set one model as default for its type"""
|
|
if queryset.count() != 1:
|
|
self.message_user(
|
|
request,
|
|
'Please select exactly one model to set as default.',
|
|
messages.ERROR
|
|
)
|
|
return
|
|
|
|
model = queryset.first()
|
|
# Unset other defaults for same type
|
|
AIModelConfig.objects.filter(
|
|
model_type=model.model_type,
|
|
is_default=True
|
|
).exclude(pk=model.pk).update(is_default=False)
|
|
|
|
# Set this as default
|
|
model.is_default = True
|
|
model.save()
|
|
|
|
self.message_user(
|
|
request,
|
|
f'{model.model_name} is now the default {model.get_model_type_display()} model.',
|
|
messages.SUCCESS
|
|
)
|
|
set_as_default.short_description = 'Set as default model (for its type)'
|
|
|
|
def save_model(self, request, obj, form, change):
|
|
"""Track who made the change"""
|
|
obj.updated_by = request.user
|
|
super().save_model(request, obj, form, change)
|