307 lines
13 KiB
Python
307 lines
13 KiB
Python
"""
|
|
Notification Service
|
|
|
|
Provides methods to create notifications for various events in the system.
|
|
"""
|
|
|
|
from .models import Notification, NotificationType, NotificationSeverity
|
|
|
|
|
|
class NotificationService:
|
|
"""Service for creating notifications"""
|
|
|
|
@staticmethod
|
|
def notify_clustering_complete(account, site=None, cluster_count=0, keyword_count=0, user=None):
|
|
"""Create notification when keyword clustering completes"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_CLUSTER_COMPLETE,
|
|
title='Clustering Complete',
|
|
message=f'Created {cluster_count} clusters from {keyword_count} keywords',
|
|
severity=NotificationSeverity.SUCCESS,
|
|
user=user,
|
|
site=site,
|
|
action_url='/planner/clusters',
|
|
action_label='View Clusters',
|
|
metadata={'cluster_count': cluster_count, 'keyword_count': keyword_count}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_clustering_failed(account, site=None, error=None, user=None):
|
|
"""Create notification when keyword clustering fails"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_CLUSTER_FAILED,
|
|
title='Clustering Failed',
|
|
message=f'Failed to cluster keywords: {error}' if error else 'Failed to cluster keywords',
|
|
severity=NotificationSeverity.ERROR,
|
|
user=user,
|
|
site=site,
|
|
action_url='/planner/keywords',
|
|
action_label='View Keywords',
|
|
metadata={'error': str(error) if error else None}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_ideas_complete(account, site=None, idea_count=0, cluster_count=0, user=None):
|
|
"""Create notification when idea generation completes"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_IDEAS_COMPLETE,
|
|
title='Ideas Generated',
|
|
message=f'Generated {idea_count} content ideas from {cluster_count} clusters',
|
|
severity=NotificationSeverity.SUCCESS,
|
|
user=user,
|
|
site=site,
|
|
action_url='/planner/ideas',
|
|
action_label='View Ideas',
|
|
metadata={'idea_count': idea_count, 'cluster_count': cluster_count}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_ideas_failed(account, site=None, error=None, user=None):
|
|
"""Create notification when idea generation fails"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_IDEAS_FAILED,
|
|
title='Idea Generation Failed',
|
|
message=f'Failed to generate ideas: {error}' if error else 'Failed to generate ideas',
|
|
severity=NotificationSeverity.ERROR,
|
|
user=user,
|
|
site=site,
|
|
action_url='/planner/clusters',
|
|
action_label='View Clusters',
|
|
metadata={'error': str(error) if error else None}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_content_complete(account, site=None, article_count=0, word_count=0, user=None):
|
|
"""Create notification when content generation completes"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_CONTENT_COMPLETE,
|
|
title='Content Generated',
|
|
message=f'Generated {article_count} article{"s" if article_count != 1 else ""} ({word_count:,} words)',
|
|
severity=NotificationSeverity.SUCCESS,
|
|
user=user,
|
|
site=site,
|
|
action_url='/writer/content',
|
|
action_label='View Content',
|
|
metadata={'article_count': article_count, 'word_count': word_count}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_content_failed(account, site=None, error=None, user=None):
|
|
"""Create notification when content generation fails"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_CONTENT_FAILED,
|
|
title='Content Generation Failed',
|
|
message=f'Failed to generate content: {error}' if error else 'Failed to generate content',
|
|
severity=NotificationSeverity.ERROR,
|
|
user=user,
|
|
site=site,
|
|
action_url='/writer/tasks',
|
|
action_label='View Tasks',
|
|
metadata={'error': str(error) if error else None}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_images_complete(account, site=None, image_count=0, user=None):
|
|
"""Create notification when image generation completes"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_IMAGES_COMPLETE,
|
|
title='Images Generated',
|
|
message=f'Generated {image_count} image{"s" if image_count != 1 else ""}',
|
|
severity=NotificationSeverity.SUCCESS,
|
|
user=user,
|
|
site=site,
|
|
action_url='/writer/images',
|
|
action_label='View Images',
|
|
metadata={'image_count': image_count}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_images_failed(account, site=None, error=None, image_count=0, user=None):
|
|
"""Create notification when image generation fails"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_IMAGES_FAILED,
|
|
title='Image Generation Failed',
|
|
message=f'Failed to generate {image_count} image{"s" if image_count != 1 else ""}: {error}' if error else f'Failed to generate images',
|
|
severity=NotificationSeverity.ERROR,
|
|
user=user,
|
|
site=site,
|
|
action_url='/writer/images',
|
|
action_label='View Images',
|
|
metadata={'error': str(error) if error else None, 'image_count': image_count}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_prompts_complete(account, site=None, prompt_count=0, user=None):
|
|
"""Create notification when image prompt generation completes"""
|
|
in_article_count = prompt_count - 1 if prompt_count > 1 else 0
|
|
message = f'{prompt_count} image prompts ready (1 featured + {in_article_count} in-article)' if in_article_count > 0 else '1 image prompt ready'
|
|
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_PROMPTS_COMPLETE,
|
|
title='Image Prompts Created',
|
|
message=message,
|
|
severity=NotificationSeverity.SUCCESS,
|
|
user=user,
|
|
site=site,
|
|
action_url='/writer/images',
|
|
action_label='Generate Images',
|
|
metadata={'prompt_count': prompt_count, 'in_article_count': in_article_count}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_prompts_failed(account, site=None, error=None, user=None):
|
|
"""Create notification when image prompt generation fails"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.AI_PROMPTS_FAILED,
|
|
title='Image Prompts Failed',
|
|
message=f'Failed to create image prompts: {error}' if error else 'Failed to create image prompts',
|
|
severity=NotificationSeverity.ERROR,
|
|
user=user,
|
|
site=site,
|
|
action_url='/writer/content',
|
|
action_label='View Content',
|
|
metadata={'error': str(error) if error else None}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_content_published(account, site=None, title='', content_object=None, user=None):
|
|
"""Create notification when content is published"""
|
|
site_name = site.name if site else 'site'
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.CONTENT_PUBLISHED,
|
|
title='Content Published',
|
|
message=f'"{title}" published to {site_name}',
|
|
severity=NotificationSeverity.SUCCESS,
|
|
user=user,
|
|
site=site,
|
|
content_object=content_object,
|
|
action_url='/writer/published',
|
|
action_label='View Published',
|
|
metadata={'content_title': title}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_publish_failed(account, site=None, title='', error=None, user=None):
|
|
"""Create notification when publishing fails"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.CONTENT_PUBLISH_FAILED,
|
|
title='Publishing Failed',
|
|
message=f'Failed to publish "{title}": {error}' if error else f'Failed to publish "{title}"',
|
|
severity=NotificationSeverity.ERROR,
|
|
user=user,
|
|
site=site,
|
|
action_url='/writer/review',
|
|
action_label='View Review',
|
|
metadata={'content_title': title, 'error': str(error) if error else None}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_wordpress_sync_success(account, site=None, count=0, user=None):
|
|
"""Create notification when WordPress sync succeeds"""
|
|
site_name = site.name if site else 'site'
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.WORDPRESS_SYNC_SUCCESS,
|
|
title='WordPress Synced',
|
|
message=f'Synced {count} item{"s" if count != 1 else ""} with {site_name}',
|
|
severity=NotificationSeverity.SUCCESS,
|
|
user=user,
|
|
site=site,
|
|
action_url='/writer/published',
|
|
action_label='View Published',
|
|
metadata={'sync_count': count}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_wordpress_sync_failed(account, site=None, error=None, user=None):
|
|
"""Create notification when WordPress sync fails"""
|
|
site_name = site.name if site else 'site'
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.WORDPRESS_SYNC_FAILED,
|
|
title='Sync Failed',
|
|
message=f'WordPress sync failed for {site_name}: {error}' if error else f'WordPress sync failed for {site_name}',
|
|
severity=NotificationSeverity.ERROR,
|
|
user=user,
|
|
site=site,
|
|
action_url=f'/sites/{site.id}/integrations' if site else '/sites',
|
|
action_label='Check Integration',
|
|
metadata={'error': str(error) if error else None}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_credits_low(account, percentage_used=80, credits_remaining=0, user=None):
|
|
"""Create notification when credits are running low"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.CREDITS_LOW,
|
|
title='Credits Running Low',
|
|
message=f"You've used {percentage_used}% of your credits. {credits_remaining} credits remaining.",
|
|
severity=NotificationSeverity.WARNING,
|
|
user=user,
|
|
action_url='/account/billing',
|
|
action_label='Upgrade Plan',
|
|
metadata={'percentage_used': percentage_used, 'credits_remaining': credits_remaining}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_credits_depleted(account, user=None):
|
|
"""Create notification when credits are depleted"""
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.CREDITS_DEPLETED,
|
|
title='Credits Depleted',
|
|
message='Your credits are exhausted. Upgrade to continue using AI features.',
|
|
severity=NotificationSeverity.ERROR,
|
|
user=user,
|
|
action_url='/account/billing',
|
|
action_label='Upgrade Now',
|
|
metadata={}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_site_setup_complete(account, site=None, user=None):
|
|
"""Create notification when site setup is complete"""
|
|
site_name = site.name if site else 'Site'
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.SITE_SETUP_COMPLETE,
|
|
title='Site Ready',
|
|
message=f'{site_name} is fully configured and ready!',
|
|
severity=NotificationSeverity.SUCCESS,
|
|
user=user,
|
|
site=site,
|
|
action_url=f'/sites/{site.id}' if site else '/sites',
|
|
action_label='View Site',
|
|
metadata={}
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_keywords_imported(account, site=None, count=0, user=None):
|
|
"""Create notification when keywords are imported"""
|
|
site_name = site.name if site else 'site'
|
|
return Notification.create_notification(
|
|
account=account,
|
|
notification_type=NotificationType.KEYWORDS_IMPORTED,
|
|
title='Keywords Imported',
|
|
message=f'Added {count} keyword{"s" if count != 1 else ""} to {site_name}',
|
|
severity=NotificationSeverity.INFO,
|
|
user=user,
|
|
site=site,
|
|
action_url='/planner/keywords',
|
|
action_label='View Keywords',
|
|
metadata={'keyword_count': count}
|
|
)
|