"""
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,
)
from .models import CreditTransaction, CreditUsageLog, AccountPaymentMethod
from import_export.admin import ExportMixin
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'
@admin.register(CreditUsageLog)
class CreditUsageLogAdmin(AccountAdminMixin, Igny8ModelAdmin):
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'
@admin.register(Invoice)
class InvoiceAdmin(AccountAdminMixin, Igny8ModelAdmin):
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']
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']
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'
@admin.register(CreditPackage)
class CreditPackageAdmin(Igny8ModelAdmin):
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']
@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(
'{} tokens/credit',
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(
'{} ({} → {})',
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)
@admin.register(PlanLimitUsage)
class PlanLimitUsageAdmin(AccountAdminMixin, Igny8ModelAdmin):
"""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'
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'
@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)