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

feat: add Usage Limits Panel component with usage tracking and visual indicators for limits

Browse Source 
style: implement custom color schemes and gradients for account section, enhancing visual hierarchy
This commit is contained in:
IGNY8 VPS (Salman)
2025-12-12 13:15:15 +00:00
parent 12956ec64a
commit 6e2101d019
29 changed files with 3622 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -189,6 +189,83 @@ class CreditCostConfig(models.Model):
super().save(*args, **kwargs)
class PlanLimitUsage(AccountBaseModel):
"""
Track monthly usage of plan limits (ideas, words, images, prompts)
Resets at start of each billing period
"""
LIMIT_TYPE_CHOICES = [
('content_ideas', 'Content Ideas'),
('content_words', 'Content Words'),
('images_basic', 'Basic Images'),
('images_premium', 'Premium Images'),
('image_prompts', 'Image Prompts'),
]
limit_type = models.CharField(
max_length=50,
choices=LIMIT_TYPE_CHOICES,
db_index=True,
help_text="Type of limit being tracked"
)
amount_used = models.IntegerField(
default=0,
validators=[MinValueValidator(0)],
help_text="Amount used in current period"
)
# Billing period tracking
period_start = models.DateField(
help_text="Start date of billing period"
)
period_end = models.DateField(
help_text="End date of billing period"
)
# Metadata
metadata = models.JSONField(
default=dict,
blank=True,
help_text="Additional tracking data (e.g., breakdown by site)"
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'billing'
db_table = 'igny8_plan_limit_usage'
verbose_name = 'Plan Limit Usage'
verbose_name_plural = 'Plan Limit Usage Records'
unique_together = [['account', 'limit_type', 'period_start']]
ordering = ['-period_start', 'limit_type']
indexes = [
models.Index(fields=['account', 'limit_type']),
models.Index(fields=['account', 'period_start', 'period_end']),
models.Index(fields=['limit_type', 'period_start']),
]
def __str__(self):
account = getattr(self, 'account', None)
return f"{account.name if account else 'No Account'} - {self.get_limit_type_display()} - {self.amount_used} used"
def is_current_period(self):
"""Check if this record is for the current billing period"""
from django.utils import timezone
today = timezone.now().date()
return self.period_start <= today <= self.period_end
def remaining_allowance(self, plan_limit):
"""Calculate remaining allowance"""
return max(0, plan_limit - self.amount_used)
def percentage_used(self, plan_limit):
"""Calculate percentage of limit used"""
if plan_limit == 0:
return 0
return min(100, int((self.amount_used / plan_limit) * 100))
class Invoice(AccountBaseModel):
"""
Invoice for subscription or credit purchases

357
backend/igny8_core/business/billing/services/limit_service.py Normal file
View File

@@ -0,0 +1,357 @@
"""
Limit Service for Plan Limit Enforcement
Manages hard limits (sites, users, keywords, clusters) and monthly limits (ideas, words, images, prompts)
"""
from django.db import transaction
from django.utils import timezone
from datetime import timedelta
import logging
from igny8_core.auth.models import Account
logger = logging.getLogger(__name__)
class LimitExceededError(Exception):
"""Base exception for limit exceeded errors"""
pass
class HardLimitExceededError(LimitExceededError):
"""Raised when a hard limit (sites, users, keywords, clusters) is exceeded"""
pass
class MonthlyLimitExceededError(LimitExceededError):
"""Raised when a monthly limit (ideas, words, images, prompts) is exceeded"""
pass
class LimitService:
"""Service for managing and enforcing plan limits"""
# Map limit types to model/field names
HARD_LIMIT_MAPPINGS = {
'sites': {
'model': 'igny8_core_auth.Site',
'plan_field': 'max_sites',
'display_name': 'Sites',
'filter_field': 'account',
},
'users': {
'model': 'igny8_core_auth.SiteUserAccess',
'plan_field': 'max_users',
'display_name': 'Team Users',
'filter_field': 'site__account',
},
'keywords': {
'model': 'planner.Keywords',
'plan_field': 'max_keywords',
'display_name': 'Keywords',
'filter_field': 'account',
},
'clusters': {
'model': 'planner.Clusters',
'plan_field': 'max_clusters',
'display_name': 'Clusters',
'filter_field': 'account',
},
}
MONTHLY_LIMIT_MAPPINGS = {
'content_ideas': {
'plan_field': 'max_content_ideas',
'usage_field': 'usage_content_ideas',
'display_name': 'Content Ideas',
},
'content_words': {
'plan_field': 'max_content_words',
'usage_field': 'usage_content_words',
'display_name': 'Content Words',
},
'images_basic': {
'plan_field': 'max_images_basic',
'usage_field': 'usage_images_basic',
'display_name': 'Basic Images',
},
'images_premium': {
'plan_field': 'max_images_premium',
'usage_field': 'usage_images_premium',
'display_name': 'Premium Images',
},
'image_prompts': {
'plan_field': 'max_image_prompts',
'usage_field': 'usage_image_prompts',
'display_name': 'Image Prompts',
},
}
@staticmethod
def check_hard_limit(account: Account, limit_type: str, additional_count: int = 1) -> bool:
"""
Check if adding items would exceed hard limit.
Args:
account: Account instance
limit_type: Type of limit
additional_count: Number of items to add
Returns:
bool: True if within limit
Raises:
HardLimitExceededError: If limit would be exceeded
"""
from django.apps import apps
if limit_type not in LimitService.HARD_LIMIT_MAPPINGS:
raise ValueError(f"Invalid hard limit type: {limit_type}")
config = LimitService.HARD_LIMIT_MAPPINGS[limit_type]
plan = account.plan
if not plan:
raise ValueError("Account has no plan")
plan_limit = getattr(plan, config['plan_field'])
model_path = config['model']
app_label, model_name = model_path.split('.')
Model = apps.get_model(app_label, model_name)
filter_field = config.get('filter_field', 'account')
filter_kwargs = {filter_field: account}
current_count = Model.objects.filter(**filter_kwargs).count()
new_count = current_count + additional_count
logger.info(f"Hard limit check: {limit_type} - Current: {current_count}, Requested: {additional_count}, Limit: {plan_limit}")
if new_count > plan_limit:
raise HardLimitExceededError(
f"{config['display_name']} limit exceeded. "
f"Current: {current_count}, Limit: {plan_limit}. "
f"Upgrade your plan to increase this limit."
)
return True
@staticmethod
def check_monthly_limit(account: Account, limit_type: str, amount: int = 1) -> bool:
"""
Check if operation would exceed monthly limit.
Args:
account: Account instance
limit_type: Type of limit
amount: Amount to use
Returns:
bool: True if within limit
Raises:
MonthlyLimitExceededError: If limit would be exceeded
"""
if limit_type not in LimitService.MONTHLY_LIMIT_MAPPINGS:
raise ValueError(f"Invalid monthly limit type: {limit_type}")
config = LimitService.MONTHLY_LIMIT_MAPPINGS[limit_type]
plan = account.plan
if not plan:
raise ValueError("Account has no plan")
plan_limit = getattr(plan, config['plan_field'])
current_usage = getattr(account, config['usage_field'], 0)
new_usage = current_usage + amount
logger.info(f"Monthly limit check: {limit_type} - Current: {current_usage}, Requested: {amount}, Limit: {plan_limit}")
if new_usage > plan_limit:
period_end = account.usage_period_end or timezone.now().date()
raise MonthlyLimitExceededError(
f"{config['display_name']} limit exceeded. "
f"Used: {current_usage}, Requested: {amount}, Limit: {plan_limit}. "
f"Resets on {period_end.strftime('%B %d, %Y')}. "
f"Upgrade your plan or wait for reset."
)
return True
@staticmethod
@transaction.atomic
def increment_usage(account: Account, limit_type: str, amount: int = 1, metadata: dict = None) -> int:
"""
Increment monthly usage after successful operation.
Args:
account: Account instance
limit_type: Type of limit
amount: Amount to increment
metadata: Optional metadata
Returns:
int: New usage amount
"""
if limit_type not in LimitService.MONTHLY_LIMIT_MAPPINGS:
raise ValueError(f"Invalid monthly limit type: {limit_type}")
config = LimitService.MONTHLY_LIMIT_MAPPINGS[limit_type]
usage_field = config['usage_field']
current_usage = getattr(account, usage_field, 0)
new_usage = current_usage + amount
setattr(account, usage_field, new_usage)
account.save(update_fields=[usage_field, 'updated_at'])
logger.info(f"Incremented {limit_type} usage by {amount}. New total: {new_usage}")
return new_usage
@staticmethod
def get_current_period(account: Account) -> tuple:
"""
Get current billing period start and end dates from account.
Args:
account: Account instance
Returns:
tuple: (period_start, period_end) as datetime objects
"""
if account.usage_period_start and account.usage_period_end:
return account.usage_period_start, account.usage_period_end
subscription = getattr(account, 'subscription', None)
if subscription and hasattr(subscription, 'current_period_start'):
period_start = subscription.current_period_start
period_end = subscription.current_period_end
else:
now = timezone.now()
period_start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
if now.month == 12:
next_month = now.replace(year=now.year + 1, month=1, day=1)
else:
next_month = now.replace(month=now.month + 1, day=1)
period_end = next_month - timedelta(days=1)
period_end = period_end.replace(hour=23, minute=59, second=59)
account.usage_period_start = period_start
account.usage_period_end = period_end
account.save(update_fields=['usage_period_start', 'usage_period_end', 'updated_at'])
return period_start, period_end
@staticmethod
def get_usage_summary(account: Account) -> dict:
"""
Get comprehensive usage summary for all limits.
Args:
account: Account instance
Returns:
dict: Usage summary with hard and monthly limits
"""
from django.apps import apps
plan = account.plan
if not plan:
return {'error': 'No plan assigned to account'}
period_start, period_end = LimitService.get_current_period(account)
days_until_reset = (period_end.date() - timezone.now().date()).days if period_end else 0
summary = {
'account_id': account.id,
'account_name': account.name,
'plan_name': plan.name,
'period_start': period_start.isoformat() if period_start else None,
'period_end': period_end.isoformat() if period_end else None,
'days_until_reset': days_until_reset,
'hard_limits': {},
'monthly_limits': {},
}
for limit_type, config in LimitService.HARD_LIMIT_MAPPINGS.items():
model_path = config['model']
app_label, model_name = model_path.split('.')
Model = apps.get_model(app_label, model_name)
filter_field = config.get('filter_field', 'account')
filter_kwargs = {filter_field: account}
current_count = Model.objects.filter(**filter_kwargs).count()
plan_limit = getattr(plan, config['plan_field'])
summary['hard_limits'][limit_type] = {
'display_name': config['display_name'],
'current': current_count,
'limit': plan_limit,
'remaining': max(0, plan_limit - current_count),
'percentage_used': int((current_count / plan_limit) * 100) if plan_limit > 0 else 0,
}
for limit_type, config in LimitService.MONTHLY_LIMIT_MAPPINGS.items():
plan_limit = getattr(plan, config['plan_field'])
current_usage = getattr(account, config['usage_field'], 0)
summary['monthly_limits'][limit_type] = {
'display_name': config['display_name'],
'current': current_usage,
'limit': plan_limit,
'remaining': max(0, plan_limit - current_usage),
'percentage_used': int((current_usage / plan_limit) * 100) if plan_limit > 0 else 0,
}
return summary
@staticmethod
@transaction.atomic
def reset_monthly_limits(account: Account) -> dict:
"""
Reset all monthly limits for an account.
Args:
account: Account instance
Returns:
dict: Summary of reset operation
"""
account.usage_content_ideas = 0
account.usage_content_words = 0
account.usage_images_basic = 0
account.usage_images_premium = 0
account.usage_image_prompts = 0
old_period_end = account.usage_period_end
now = timezone.now()
new_period_start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
if now.month == 12:
next_month = now.replace(year=now.year + 1, month=1, day=1)
else:
next_month = now.replace(month=now.month + 1, day=1)
new_period_end = next_month - timedelta(days=1)
new_period_end = new_period_end.replace(hour=23, minute=59, second=59)
account.usage_period_start = new_period_start
account.usage_period_end = new_period_end
account.save(update_fields=[
'usage_content_ideas', 'usage_content_words',
'usage_images_basic', 'usage_images_premium', 'usage_image_prompts',
'usage_period_start', 'usage_period_end', 'updated_at'
])
logger.info(f"Reset monthly limits for account {account.id}")
return {
'account_id': account.id,
'old_period_end': old_period_end.isoformat() if old_period_end else None,
'new_period_start': new_period_start.isoformat(),
'new_period_end': new_period_end.isoformat(),
'limits_reset': 5,
}

3
backend/igny8_core/business/billing/urls.py
View File

@@ -7,6 +7,7 @@ from .views import (
PaymentViewSet,
CreditPackageViewSet,
AccountPaymentMethodViewSet,
get_usage_summary,
)
from igny8_core.modules.billing.views import (
CreditBalanceViewSet,
@@ -29,4 +30,6 @@ router.register(r'payment-configs', BillingViewSet, basename='payment-configs')
urlpatterns = [
path('', include(router.urls)),
# User-facing usage summary endpoint for plan limits
path('usage-summary/', get_usage_summary, name='usage-summary'),
]

41
backend/igny8_core/business/billing/views.py
View File

@@ -866,3 +866,44 @@ class AccountPaymentMethodViewSet(AccountModelViewSet):
{'count': paginator.page.paginator.count, 'next': paginator.get_next_link(), 'previous': paginator.get_previous_link(), 'results': results},
request=request
)
# ============================================================================
# USAGE SUMMARY (Plan Limits) - User-facing endpoint
# ============================================================================
from rest_framework.decorators import api_view, permission_classes
@api_view(['GET'])
@permission_classes([IsAuthenticatedAndActive, HasTenantAccess])
def get_usage_summary(request):
"""
Get comprehensive usage summary for current account.
Includes hard limits (sites, users, keywords, clusters) and monthly limits (ideas, words, images).
GET /api/v1/billing/usage-summary/
"""
try:
account = getattr(request, 'account', None)
if not account:
return error_response(
error='Account not found.',
status_code=status.HTTP_400_BAD_REQUEST,
request=request
)
from igny8_core.business.billing.services.limit_service import LimitService
summary = LimitService.get_usage_summary(account)
return success_response(
data=summary,
message='Usage summary retrieved successfully.',
request=request
)
except Exception as e:
logger.error(f'Error getting usage summary: {str(e)}', exc_info=True)
return error_response(
error=f'Failed to retrieve usage summary: {str(e)}',
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
request=request
)

49
backend/igny8_core/business/content/models.py
View File

@@ -276,6 +276,55 @@ class Content(SoftDeletableModel, SiteSectorBaseModel):
def __str__(self):
return self.title or f"Content {self.id}"
def save(self, *args, **kwargs):
"""Override save to auto-calculate word_count from content_html"""
is_new = self.pk is None
old_word_count = 0
# Get old word count if updating
if not is_new and self.content_html:
try:
old_instance = Content.objects.get(pk=self.pk)
old_word_count = old_instance.word_count or 0
except Content.DoesNotExist:
pass
# Auto-calculate word count if content_html has changed
if self.content_html:
from igny8_core.utils.word_counter import calculate_word_count
calculated_count = calculate_word_count(self.content_html)
# Only update if different to avoid unnecessary saves
if self.word_count != calculated_count:
self.word_count = calculated_count
super().save(*args, **kwargs)
# Increment usage for new content or if word count increased
if self.content_html and self.word_count:
# Only count newly generated words
new_words = self.word_count - old_word_count if not is_new else self.word_count
if new_words > 0:
from igny8_core.business.billing.services.limit_service import LimitService
try:
# Get account from site
account = self.site.account if self.site else None
if account:
LimitService.increment_usage(
account=account,
limit_type='content_words',
amount=new_words,
metadata={
'content_id': self.id,
'content_title': self.title,
'site_id': self.site.id if self.site else None,
}
)
except Exception as e:
import logging
logger = logging.getLogger(__name__)
logger.error(f"Error incrementing word usage for content {self.id}: {str(e)}")
class ContentTaxonomy(SiteSectorBaseModel):

8
backend/igny8_core/business/content/services/content_generation_service.py
View File

@@ -30,12 +30,20 @@ class ContentGenerationService:
Raises:
InsufficientCreditsError: If account doesn't have enough credits
"""
from igny8_core.business.billing.services.limit_service import LimitService, MonthlyLimitExceededError
# Get tasks
tasks = Tasks.objects.filter(id__in=task_ids, account=account)
# Calculate estimated credits needed based on word count
total_word_count = sum(task.word_count or 1000 for task in tasks)
# Check monthly word count limit
try:
LimitService.check_monthly_limit(account, 'content_words', amount=total_word_count)
except MonthlyLimitExceededError as e:
raise InsufficientCreditsError(str(e))
# Check credits
try:
self.credit_service.check_credits(account, 'content_generation', total_word_count)