929 lines
36 KiB
Python
929 lines
36 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,
|
|
BillingConfiguration,
|
|
Invoice,
|
|
Payment,
|
|
CreditPackage,
|
|
PaymentMethodConfig,
|
|
PlanLimitUsage,
|
|
AIModelConfig,
|
|
)
|
|
from .models import CreditTransaction, CreditUsageLog, AccountPaymentMethod
|
|
from import_export.admin import ExportMixin, ImportExportMixin
|
|
from import_export import resources
|
|
|
|
|
|
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', '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', 'processed_at']
|
|
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']
|
|
|
|
fieldsets = (
|
|
('Payment Method', {
|
|
'fields': ('country_code', 'payment_method', 'display_name', 'is_enabled', 'sort_order')
|
|
}),
|
|
('Instructions', {
|
|
'fields': ('instructions',),
|
|
'description': 'Instructions shown to users for this payment method'
|
|
}),
|
|
('Bank Transfer Details', {
|
|
'fields': ('bank_name', 'account_title', 'account_number', 'routing_number', 'swift_code', 'iban'),
|
|
'classes': ('collapse',),
|
|
'description': 'Only for bank_transfer payment method'
|
|
}),
|
|
('Local Wallet Details', {
|
|
'fields': ('wallet_type', 'wallet_id'),
|
|
'classes': ('collapse',),
|
|
'description': 'Only for local_wallet payment method (JazzCash, EasyPaisa, etc.)'
|
|
}),
|
|
('Timestamps', {
|
|
'fields': ('created_at', 'updated_at'),
|
|
'classes': ('collapse',)
|
|
}),
|
|
)
|
|
|
|
|
|
@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):
|
|
"""
|
|
Admin for Credit Cost Configuration.
|
|
Per final-model-schemas.md - Fixed credits per operation type.
|
|
"""
|
|
list_display = [
|
|
'operation_type',
|
|
'display_name',
|
|
'base_credits_display',
|
|
'is_active_icon',
|
|
]
|
|
|
|
list_filter = ['is_active']
|
|
search_fields = ['operation_type', 'display_name', 'description']
|
|
actions = ['bulk_activate', 'bulk_deactivate']
|
|
|
|
fieldsets = (
|
|
('Operation', {
|
|
'fields': ('operation_type', 'display_name', 'description')
|
|
}),
|
|
('Credits', {
|
|
'fields': ('base_credits', 'is_active'),
|
|
'description': 'Fixed credits charged per operation'
|
|
}),
|
|
)
|
|
|
|
def base_credits_display(self, obj):
|
|
"""Show base credits with formatting"""
|
|
return format_html(
|
|
'<span style="font-weight: bold;">{} credits</span>',
|
|
obj.base_credits
|
|
)
|
|
base_credits_display.short_description = 'Credits'
|
|
|
|
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'
|
|
|
|
@admin.action(description='Activate selected configurations')
|
|
def bulk_activate(self, request, queryset):
|
|
"""Bulk activate credit cost configurations"""
|
|
updated = queryset.update(is_active=True)
|
|
self.message_user(request, f'{updated} configuration(s) activated.', messages.SUCCESS)
|
|
|
|
@admin.action(description='Deactivate selected configurations')
|
|
def bulk_deactivate(self, request, queryset):
|
|
"""Bulk deactivate credit cost configurations"""
|
|
updated = queryset.update(is_active=False)
|
|
self.message_user(request, f'{updated} configuration(s) deactivated.', messages.WARNING)
|
|
|
|
|
|
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', 'period_end', '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(BillingConfiguration)
|
|
class BillingConfigurationAdmin(Igny8ModelAdmin):
|
|
"""Admin for global billing configuration (Singleton)"""
|
|
list_display = [
|
|
'id',
|
|
'default_tokens_per_credit',
|
|
'default_credit_price_usd',
|
|
'credit_rounding_mode',
|
|
'enable_token_based_reporting',
|
|
'updated_at',
|
|
'updated_by'
|
|
]
|
|
|
|
fieldsets = (
|
|
('Global Token-to-Credit Settings', {
|
|
'fields': ('default_tokens_per_credit', 'default_credit_price_usd', 'credit_rounding_mode'),
|
|
'description': 'These settings apply when no operation-specific config exists'
|
|
}),
|
|
('Reporting Settings', {
|
|
'fields': ('enable_token_based_reporting',),
|
|
'description': 'Control token-based reporting features'
|
|
}),
|
|
('Audit Trail', {
|
|
'fields': ('updated_by', 'updated_at'),
|
|
'classes': ('collapse',)
|
|
}),
|
|
)
|
|
|
|
readonly_fields = ['updated_at']
|
|
|
|
def has_add_permission(self, request):
|
|
"""Only allow one instance (singleton)"""
|
|
from igny8_core.business.billing.models import BillingConfiguration
|
|
return not BillingConfiguration.objects.exists()
|
|
|
|
def has_delete_permission(self, request, obj=None):
|
|
"""Prevent deletion of the singleton"""
|
|
return False
|
|
|
|
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)
|
|
|
|
|
|
@admin.register(AIModelConfig)
|
|
class AIModelConfigAdmin(SimpleHistoryAdmin, Igny8ModelAdmin):
|
|
"""
|
|
Admin for AI Model Configuration - Single Source of Truth for Models.
|
|
Per final-model-schemas.md
|
|
"""
|
|
list_display = [
|
|
'model_name',
|
|
'display_name_short',
|
|
'model_type_badge',
|
|
'provider_badge',
|
|
'credit_display',
|
|
'quality_tier',
|
|
'is_active_icon',
|
|
'is_default_icon',
|
|
'updated_at',
|
|
]
|
|
|
|
list_filter = [
|
|
'model_type',
|
|
'provider',
|
|
'quality_tier',
|
|
'is_active',
|
|
'is_default',
|
|
]
|
|
|
|
search_fields = ['model_name', 'display_name']
|
|
|
|
ordering = ['model_type', 'model_name']
|
|
|
|
readonly_fields = ['created_at', 'updated_at']
|
|
|
|
fieldsets = (
|
|
('Basic Information', {
|
|
'fields': ('model_name', 'model_type', 'provider', 'display_name'),
|
|
'description': 'Core model identification'
|
|
}),
|
|
('Text Model Pricing', {
|
|
'fields': ('cost_per_1k_input', 'cost_per_1k_output', 'tokens_per_credit', 'max_tokens', 'context_window'),
|
|
'description': 'For TEXT models only',
|
|
'classes': ('collapse',)
|
|
}),
|
|
('Image Model Pricing', {
|
|
'fields': ('credits_per_image', 'quality_tier'),
|
|
'description': 'For IMAGE models only',
|
|
'classes': ('collapse',)
|
|
}),
|
|
('Image Model Sizes', {
|
|
'fields': ('landscape_size', 'square_size', 'valid_sizes'),
|
|
'description': 'For IMAGE models: specify supported image dimensions',
|
|
'classes': ('collapse',)
|
|
}),
|
|
('Capabilities', {
|
|
'fields': ('capabilities',),
|
|
'description': 'JSON: vision, function_calling, json_mode, etc.',
|
|
'classes': ('collapse',)
|
|
}),
|
|
('Status', {
|
|
'fields': ('is_active', 'is_default'),
|
|
}),
|
|
('Timestamps', {
|
|
'fields': ('created_at', 'updated_at'),
|
|
'classes': ('collapse',)
|
|
}),
|
|
)
|
|
|
|
# Custom display methods
|
|
def display_name_short(self, obj):
|
|
"""Truncated display name for list view"""
|
|
if len(obj.display_name) > 40:
|
|
return obj.display_name[:37] + '...'
|
|
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
|
|
}
|
|
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',
|
|
'anthropic': '#d97757',
|
|
'runware': '#6366f1',
|
|
'google': '#4285f4',
|
|
}
|
|
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 credit_display(self, obj):
|
|
"""Format credit info based on model type"""
|
|
if obj.model_type == 'text' and obj.tokens_per_credit:
|
|
return format_html(
|
|
'<span style="font-family: monospace;">{} tokens/credit</span>',
|
|
obj.tokens_per_credit
|
|
)
|
|
elif obj.model_type == 'image' and obj.credits_per_image:
|
|
return format_html(
|
|
'<span style="font-family: monospace;">{} credits/image</span>',
|
|
obj.credits_per_image
|
|
)
|
|
return '-'
|
|
credit_display.short_description = 'Credits'
|
|
|
|
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.', 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.', 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, 'Select exactly one model.', messages.ERROR)
|
|
return
|
|
|
|
model = queryset.first()
|
|
AIModelConfig.objects.filter(
|
|
model_type=model.model_type,
|
|
is_default=True
|
|
).exclude(pk=model.pk).update(is_default=False)
|
|
|
|
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'
|