"""
Admin interface for auth models
"""
from django import forms
from django.contrib import admin
from django.contrib.auth.admin import UserAdmin as BaseUserAdmin
from django.db import models
from unfold.admin import ModelAdmin, TabularInline
from simple_history.admin import SimpleHistoryAdmin
from igny8_core.admin.base import AccountAdminMixin, Igny8ModelAdmin
from .models import User, Account, Plan, Subscription, Site, Sector, SiteUserAccess, Industry, IndustrySector, SeedKeyword, PasswordResetToken
from import_export.admin import ExportMixin, ImportExportMixin
from import_export import resources
class AccountAdminForm(forms.ModelForm):
"""Custom form for Account admin with dynamic payment method choices from PaymentMethodConfig"""
class Meta:
model = Account
fields = '__all__'
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
from igny8_core.business.billing.models import PaymentMethodConfig, AccountPaymentMethod
if self.instance and self.instance.pk:
# Get country from billing_country, fallback to wildcard '*' for global
country = self.instance.billing_country or '*'
# Get enabled payment methods for this country OR global (*)
available_methods = PaymentMethodConfig.objects.filter(
country_code__in=[country, '*'],
is_enabled=True
).order_by('country_code', 'sort_order').values_list('payment_method', 'display_name')
if available_methods:
# Build choices from PaymentMethodConfig
choices = []
seen = set()
for method_type, display_name in available_methods:
if method_type not in seen:
choices.append((method_type, display_name or method_type.replace('_', ' ').title()))
seen.add(method_type)
else:
# Fallback to model choices if no configs
choices = Account.PAYMENT_METHOD_CHOICES
self.fields['payment_method'].widget = forms.Select(choices=choices)
# Get current default from AccountPaymentMethod
default_method = AccountPaymentMethod.objects.filter(
account=self.instance,
is_default=True,
is_enabled=True
).first()
if default_method:
self.fields['payment_method'].initial = default_method.type
self.fields['payment_method'].help_text = f'✓ Current: {default_method.display_name} ({default_method.get_type_display()})'
else:
self.fields['payment_method'].help_text = 'Select from available payment methods based on country'
def save(self, commit=True):
"""When payment_method changes, update/create AccountPaymentMethod"""
from igny8_core.business.billing.models import AccountPaymentMethod, PaymentMethodConfig
instance = super().save(commit=False)
if commit:
instance.save()
# Get selected payment method
selected_type = self.cleaned_data.get('payment_method')
if selected_type:
# Get config for display name and instructions
country = instance.billing_country or '*'
config = PaymentMethodConfig.objects.filter(
country_code__in=[country, '*'],
payment_method=selected_type,
is_enabled=True
).first()
# Create or update AccountPaymentMethod
account_method, created = AccountPaymentMethod.objects.get_or_create(
account=instance,
type=selected_type,
defaults={
'display_name': config.display_name if config else selected_type.replace('_', ' ').title(),
'is_default': True,
'is_enabled': True,
'instructions': config.instructions if config else '',
'country_code': instance.billing_country or '',
}
)
if not created:
# Update existing and set as default
account_method.is_default = True
account_method.is_enabled = True
if config:
account_method.display_name = config.display_name
account_method.instructions = config.instructions
account_method.save()
# Unset other methods as default
AccountPaymentMethod.objects.filter(
account=instance
).exclude(id=account_method.id).update(is_default=False)
return instance
class PlanResource(resources.ModelResource):
"""Resource class for importing/exporting Plans"""
class Meta:
model = Plan
fields = ('id', 'name', 'slug', 'price', 'billing_cycle', 'max_sites', 'max_users',
'max_keywords', 'max_ahrefs_queries', 'included_credits', 'is_active', 'is_featured')
export_order = fields
import_id_fields = ('id',)
skip_unchanged = True
@admin.register(Plan)
class PlanAdmin(ImportExportMixin, Igny8ModelAdmin):
resource_class = PlanResource
"""Plan admin - Global, no account filtering needed"""
list_display = ['name', 'slug', 'price', 'billing_cycle', 'max_sites', 'max_users', 'max_keywords', 'max_ahrefs_queries', 'included_credits', 'is_active', 'is_featured']
list_filter = ['is_active', 'billing_cycle', 'is_internal', 'is_featured']
search_fields = ['name', 'slug']
readonly_fields = ['created_at']
actions = [
'bulk_set_active',
'bulk_set_inactive',
'bulk_clone_plans',
]
fieldsets = (
('Plan Info', {
'fields': ('name', 'slug', 'price', 'original_price', 'annual_discount_percent', 'billing_cycle', 'features', 'is_active', 'is_featured', 'is_internal'),
'description': 'Price: Current price | Original Price: Crossed-out price (optional) | Annual Discount %: For annual billing | Is Featured: Show as popular/recommended plan'
}),
('Account Management Limits', {
'fields': ('max_users', 'max_sites', 'max_industries', 'max_author_profiles'),
'description': 'Persistent limits for account-level resources'
}),
('Hard Limits (Persistent)', {
'fields': ('max_keywords',),
'description': 'Total allowed - never reset'
}),
('Monthly Limits (Reset on Billing Cycle)', {
'fields': ('max_ahrefs_queries',),
'description': 'Monthly Ahrefs keyword research queries (0 = disabled)'
}),
('Billing & Credits', {
'fields': ('included_credits', 'extra_credit_price', 'allow_credit_topup', 'auto_credit_topup_threshold', 'auto_credit_topup_amount', 'credits_per_month')
}),
('Stripe Integration', {
'fields': ('stripe_product_id', 'stripe_price_id')
}),
('PayPal Integration', {
'fields': ('paypal_plan_id',),
'description': 'PayPal subscription plan ID (required for PayPal subscriptions)'
}),
)
def bulk_set_active(self, request, queryset):
"""Set selected plans to active"""
updated = queryset.update(is_active=True)
self.message_user(request, f'{updated} plan(s) set to active.', messages.SUCCESS)
bulk_set_active.short_description = 'Set plans to Active'
def bulk_set_inactive(self, request, queryset):
"""Set selected plans to inactive"""
updated = queryset.update(is_active=False)
self.message_user(request, f'{updated} plan(s) set to inactive.', messages.SUCCESS)
bulk_set_inactive.short_description = 'Set plans to Inactive'
def bulk_clone_plans(self, request, queryset):
"""Clone selected plans"""
count = 0
for plan in queryset:
plan_copy = Plan.objects.get(pk=plan.pk)
plan_copy.pk = None
plan_copy.name = f"{plan.name} (Copy)"
plan_copy.slug = f"{plan.slug}-copy"
plan_copy.is_active = False
plan_copy.save()
count += 1
self.message_user(request, f'{count} plan(s) cloned.', messages.SUCCESS)
bulk_clone_plans.short_description = 'Clone selected plans'
class AccountResource(resources.ModelResource):
"""Resource class for exporting Accounts"""
class Meta:
model = Account
fields = ('id', 'name', 'slug', 'owner__email', 'plan__name', 'status',
'credits', 'billing_country', 'created_at', 'updated_at')
export_order = fields
@admin.register(Account)
class AccountAdmin(ExportMixin, AccountAdminMixin, SimpleHistoryAdmin, Igny8ModelAdmin):
resource_class = AccountResource
form = AccountAdminForm
list_display = ['name', 'slug', 'owner', 'plan', 'status', 'health_indicator', 'credits', 'created_at']
list_filter = ['status', 'plan']
search_fields = ['name', 'slug']
readonly_fields = ['created_at', 'updated_at', 'health_indicator', 'health_details']
actions = [
'bulk_set_status_active',
'bulk_set_status_suspended',
'bulk_set_status_trial',
'bulk_set_status_cancelled',
'bulk_add_credits',
'bulk_subtract_credits',
]
def get_queryset(self, request):
"""Override to filter by account for non-superusers"""
qs = super().get_queryset(request)
if request.user.is_superuser or (hasattr(request.user, 'is_developer') and request.user.is_developer()):
return qs
# Owners can see their own accounts
if hasattr(request.user, 'role') and request.user.role == 'owner':
return qs.filter(owner=request.user)
# Admins can see their account
try:
user_account = getattr(request.user, 'account', None)
if user_account:
return qs.filter(id=user_account.id)
except (AttributeError, Exception):
# If account access fails (e.g., column mismatch), return empty
pass
return qs.none()
def health_indicator(self, obj):
"""Display health status with visual indicator"""
from django.utils.html import format_html
from django.utils import timezone
from datetime import timedelta
# Check credits
if obj.credits < 10:
status = 'critical'
message = 'Critical: Very low credits'
elif obj.credits < 100:
status = 'warning'
message = 'Warning: Low credits'
else:
status = 'good'
message = 'Good'
# Check for recent failed automations
try:
from igny8_core.business.automation.models import AutomationRun
week_ago = timezone.now() - timedelta(days=7)
failed_runs = AutomationRun.objects.filter(
account=obj,
status='failed',
created_at__gte=week_ago
).count()
if failed_runs > 5:
status = 'critical'
message = f'Critical: {failed_runs} automation failures'
elif failed_runs > 0:
if status == 'good':
status = 'warning'
message = f'Warning: {failed_runs} automation failures'
except:
pass
# Check account status
if obj.status != 'active':
status = 'critical'
message = f'Critical: Account {obj.status}'
colors = {
'good': '#0bbf87',
'warning': '#ff7a00',
'critical': '#ef4444'
}
return format_html(
'{}',
colors[status], message
)
health_indicator.short_description = 'Health'
def health_details(self, obj):
"""Detailed health information"""
from django.utils.html import format_html
from django.utils import timezone
from datetime import timedelta
details = []
# Credits status
colors = {
'critical': '#ef4444',
'warning': '#ff7a00',
'good': '#0bbf87'
}
if obj.credits < 10:
details.append(f'Critical: Only {obj.credits} credits remaining')
elif obj.credits < 100:
details.append(f'Warning: Only {obj.credits} credits remaining')
else:
details.append(f'Credits: {obj.credits} available')
# Recent activity
try:
from igny8_core.modules.writer.models import Content
week_ago = timezone.now() - timedelta(days=7)
recent_content = Content.objects.filter(
site__account=obj,
created_at__gte=week_ago
).count()
details.append(f'📚 Activity: {recent_content} content pieces created this week')
except:
pass
# Failed automations
try:
from igny8_core.business.automation.models import AutomationRun
week_ago = timezone.now() - timedelta(days=7)
failed_runs = AutomationRun.objects.filter(
account=obj,
status='failed',
created_at__gte=week_ago
).count()
if failed_runs > 0:
details.append(f'🔴 Automations: {failed_runs} failures this week')
else:
details.append(f'✅ Automations: No failures this week')
except:
pass
# Failed syncs
try:
from igny8_core.business.integration.models import SyncEvent
today = timezone.now().date()
failed_syncs = SyncEvent.objects.filter(
site__account=obj,
success=False,
created_at__date=today
).count()
if failed_syncs > 0:
details.append(f'⚠️ Syncs: {failed_syncs} failures today')
else:
details.append(f'✅ Syncs: No failures today')
except:
pass
# Account status
if obj.status == 'active':
details.append(f'✅ Status: Active')
else:
details.append(f'🔴 Status: {obj.status.title()}')
return format_html('
'.join(details))
health_details.short_description = 'Health Details'
def has_delete_permission(self, request, obj=None):
if obj and getattr(obj, 'slug', '') == 'aws-admin':
return False
return super().has_delete_permission(request, obj)
# Bulk Actions
def bulk_set_status_active(self, request, queryset):
"""Set selected accounts to active status"""
updated = queryset.update(status='active')
self.message_user(request, f'{updated} account(s) set to active.', messages.SUCCESS)
bulk_set_status_active.short_description = 'Set status to Active'
def bulk_set_status_suspended(self, request, queryset):
"""Set selected accounts to suspended status"""
updated = queryset.update(status='suspended')
self.message_user(request, f'{updated} account(s) set to suspended.', messages.SUCCESS)
bulk_set_status_suspended.short_description = 'Set status to Suspended'
def bulk_set_status_trial(self, request, queryset):
"""Set selected accounts to trial status"""
updated = queryset.update(status='trial')
self.message_user(request, f'{updated} account(s) set to trial.', messages.SUCCESS)
bulk_set_status_trial.short_description = 'Set status to Trial'
def bulk_set_status_cancelled(self, request, queryset):
"""Set selected accounts to cancelled status"""
updated = queryset.update(status='cancelled')
self.message_user(request, f'{updated} account(s) set to cancelled.', messages.SUCCESS)
bulk_set_status_cancelled.short_description = 'Set status to Cancelled'
def bulk_add_credits(self, request, queryset):
"""Add credits to selected accounts"""
from django import forms
if 'apply' in request.POST:
amount = int(request.POST.get('credits', 0))
if amount > 0:
for account in queryset:
account.credits += amount
account.save()
self.message_user(request, f'Added {amount} credits to {queryset.count()} account(s).', messages.SUCCESS)
return
class CreditForm(forms.Form):
credits = forms.IntegerField(
min_value=1,
label="Credits to Add",
help_text=f"Add credits to {queryset.count()} selected account(s)"
)
from django.shortcuts import render
return render(request, 'admin/bulk_action_form.html', {
'title': 'Add Credits to Accounts',
'queryset': queryset,
'form': CreditForm(),
'action': 'bulk_add_credits',
})
bulk_add_credits.short_description = 'Add credits to accounts'
def bulk_subtract_credits(self, request, queryset):
"""Subtract credits from selected accounts"""
from django import forms
if 'apply' in request.POST:
amount = int(request.POST.get('credits', 0))
if amount > 0:
for account in queryset:
account.credits = max(0, account.credits - amount)
account.save()
self.message_user(request, f'Subtracted {amount} credits from {queryset.count()} account(s).', messages.SUCCESS)
return
class CreditForm(forms.Form):
credits = forms.IntegerField(
min_value=1,
label="Credits to Subtract",
help_text=f"Subtract credits from {queryset.count()} selected account(s)"
)
from django.shortcuts import render
return render(request, 'admin/bulk_action_form.html', {
'title': 'Subtract Credits from Accounts',
'queryset': queryset,
'form': CreditForm(),
'action': 'bulk_subtract_credits',
})
bulk_subtract_credits.short_description = 'Subtract credits from accounts'
class SubscriptionResource(resources.ModelResource):
"""Resource class for exporting Subscriptions"""
class Meta:
model = Subscription
fields = ('id', 'account__name', 'status', 'current_period_start', 'current_period_end',
'stripe_subscription_id', 'created_at')
export_order = fields
@admin.register(Subscription)
class SubscriptionAdmin(ExportMixin, AccountAdminMixin, Igny8ModelAdmin):
resource_class = SubscriptionResource
list_display = ['account', 'status', 'current_period_start', 'current_period_end']
list_filter = ['status']
search_fields = ['account__name', 'stripe_subscription_id']
readonly_fields = ['created_at', 'updated_at']
actions = [
'bulk_set_status_active',
'bulk_set_status_cancelled',
'bulk_set_status_suspended',
'bulk_set_status_trialing',
'bulk_renew',
]
actions = [
'bulk_set_status_active',
'bulk_set_status_cancelled',
'bulk_set_status_suspended',
'bulk_set_status_trialing',
'bulk_renew',
]
def bulk_set_status_active(self, request, queryset):
"""Set subscriptions to active"""
updated = queryset.update(status='active')
self.message_user(request, f'{updated} subscription(s) set to active.', messages.SUCCESS)
bulk_set_status_active.short_description = 'Set status to Active'
def bulk_set_status_cancelled(self, request, queryset):
"""Set subscriptions to cancelled"""
updated = queryset.update(status='cancelled')
self.message_user(request, f'{updated} subscription(s) set to cancelled.', messages.SUCCESS)
bulk_set_status_cancelled.short_description = 'Set status to Cancelled'
def bulk_set_status_suspended(self, request, queryset):
"""Set subscriptions to suspended"""
updated = queryset.update(status='suspended')
self.message_user(request, f'{updated} subscription(s) set to suspended.', messages.SUCCESS)
bulk_set_status_suspended.short_description = 'Set status to Suspended'
def bulk_set_status_trialing(self, request, queryset):
"""Set subscriptions to trialing"""
updated = queryset.update(status='trialing')
self.message_user(request, f'{updated} subscription(s) set to trialing.', messages.SUCCESS)
bulk_set_status_trialing.short_description = 'Set status to Trialing'
def bulk_renew(self, request, queryset):
"""Renew selected subscriptions"""
from django.utils import timezone
from datetime import timedelta
count = 0
for subscription in queryset:
# Extend subscription by one billing period
if subscription.current_period_end:
subscription.current_period_end = subscription.current_period_end + timedelta(days=30)
subscription.status = 'active'
subscription.save()
count += 1
self.message_user(request, f'{count} subscription(s) renewed for 30 days.', messages.SUCCESS)
bulk_renew.short_description = 'Renew subscriptions'
@admin.register(PasswordResetToken)
class PasswordResetTokenAdmin(Igny8ModelAdmin):
list_display = ['user', 'token', 'used', 'expires_at', 'created_at']
list_filter = ['used', 'expires_at', 'created_at']
search_fields = ['user__email', 'token']
readonly_fields = ['created_at', 'token']
def get_queryset(self, request):
"""Filter by account for non-superusers"""
qs = super().get_queryset(request)
if request.user.is_superuser or (hasattr(request.user, 'is_developer') and request.user.is_developer()):
return qs
user_account = getattr(request.user, 'account', None)
if user_account:
return qs.filter(user__account=user_account)
return qs.none()
class SectorInline(TabularInline):
"""Inline admin for sectors within Site admin."""
model = Sector
extra = 0
fields = ['industry_sector', 'name', 'slug', 'status', 'is_active', 'get_keywords_count', 'get_clusters_count']
readonly_fields = ['get_keywords_count', 'get_clusters_count']
def get_keywords_count(self, obj):
if obj.pk:
try:
return obj.keywords_set.count()
except (AttributeError, Exception):
return 0
return 0
get_keywords_count.short_description = 'Keywords'
def get_clusters_count(self, obj):
if obj.pk:
try:
return obj.clusters_set.count()
except (AttributeError, Exception):
return 0
return 0
get_clusters_count.short_description = 'Clusters'
class SiteResource(resources.ModelResource):
"""Resource class for importing/exporting Sites"""
class Meta:
model = Site
fields = ('id', 'name', 'slug', 'account__name', 'industry__name', 'domain',
'status', 'is_active', 'site_type', 'hosting_type', 'description', 'created_at')
export_order = fields
import_id_fields = ('id',)
skip_unchanged = True
@admin.register(Site)
class SiteAdmin(ImportExportMixin, AccountAdminMixin, Igny8ModelAdmin):
resource_class = SiteResource
list_display = ['name', 'slug', 'account', 'industry', 'domain', 'status', 'is_active', 'get_api_key_status', 'get_sectors_count']
list_filter = ['status', 'is_active', 'account', 'industry', 'hosting_type']
search_fields = ['name', 'slug', 'domain', 'industry__name']
readonly_fields = ['created_at', 'updated_at', 'get_api_key_display']
inlines = [SectorInline]
actions = [
'generate_api_keys',
'bulk_set_status_active',
'bulk_set_status_inactive',
'bulk_set_status_maintenance',
]
fieldsets = (
('Site Info', {
'fields': ('name', 'slug', 'account', 'domain', 'description', 'industry', 'site_type', 'hosting_type', 'status', 'is_active')
}),
('WordPress Integration', {
'fields': ('get_api_key_display',),
'description': 'WordPress integration API key. Use SiteIntegration model for full integration settings.'
}),
('SEO Metadata', {
'fields': ('seo_metadata',),
'classes': ('collapse',)
}),
('Timestamps', {
'fields': ('created_at', 'updated_at'),
'classes': ('collapse',)
}),
)
def get_api_key_display(self, obj):
"""Display API key with copy button"""
from django.utils.html import format_html
if obj.wp_api_key:
return format_html(
'
{}'
''
'