from django.dispatch import receiver from django.db.models.signals import post_save from django.contrib.auth.models import AbstractUser from django.contrib.contenttypes.fields import GenericForeignKey from django.db import models from django.urls import reverse from django.utils.translation import gettext_lazy as _ import secrets from datetime import timedelta from django.utils import timezone from apps.core.history import TrackedModel import pghistory from cloudflare_images.field import CloudflareImagesField def generate_random_id(model_class, id_field): """Generate a random ID starting at 4 digits, expanding to 5 if needed""" while True: # Try to get a 4-digit number first new_id = str(secrets.SystemRandom().randint(1000, 9999)) if not model_class.objects.filter(**{id_field: new_id}).exists(): return new_id # If all 4-digit numbers are taken, try 5 digits new_id = str(secrets.SystemRandom().randint(10000, 99999)) if not model_class.objects.filter(**{id_field: new_id}).exists(): return new_id @pghistory.track() class User(AbstractUser): class Roles(models.TextChoices): USER = "USER", _("User") MODERATOR = "MODERATOR", _("Moderator") ADMIN = "ADMIN", _("Admin") SUPERUSER = "SUPERUSER", _("Superuser") class ThemePreference(models.TextChoices): LIGHT = "light", _("Light") DARK = "dark", _("Dark") class PrivacyLevel(models.TextChoices): PUBLIC = "public", _("Public") FRIENDS = "friends", _("Friends Only") PRIVATE = "private", _("Private") # Override inherited fields to remove them first_name = None last_name = None # Read-only ID user_id = models.CharField( max_length=10, unique=True, editable=False, help_text=( "Unique identifier for this user that remains constant even if the " "username changes" ), ) role = models.CharField( max_length=10, choices=Roles.choices, default=Roles.USER, ) is_banned = models.BooleanField(default=False) ban_reason = models.TextField(blank=True) ban_date = models.DateTimeField(null=True, blank=True) pending_email = models.EmailField(blank=True, null=True) theme_preference = models.CharField( max_length=5, choices=ThemePreference.choices, default=ThemePreference.LIGHT, ) # Notification preferences email_notifications = models.BooleanField(default=True) push_notifications = models.BooleanField(default=False) # Privacy settings privacy_level = models.CharField( max_length=10, choices=PrivacyLevel.choices, default=PrivacyLevel.PUBLIC, ) show_email = models.BooleanField(default=False) show_real_name = models.BooleanField(default=True) show_join_date = models.BooleanField(default=True) show_statistics = models.BooleanField(default=True) show_reviews = models.BooleanField(default=True) show_photos = models.BooleanField(default=True) show_top_lists = models.BooleanField(default=True) allow_friend_requests = models.BooleanField(default=True) allow_messages = models.BooleanField(default=True) allow_profile_comments = models.BooleanField(default=False) search_visibility = models.BooleanField(default=True) activity_visibility = models.CharField( max_length=10, choices=PrivacyLevel.choices, default=PrivacyLevel.FRIENDS, ) # Security settings two_factor_enabled = models.BooleanField(default=False) login_notifications = models.BooleanField(default=True) session_timeout = models.IntegerField(default=30) # days login_history_retention = models.IntegerField(default=90) # days last_password_change = models.DateTimeField(auto_now_add=True) # Display name - core user data for better performance display_name = models.CharField( max_length=50, blank=True, help_text="Display name shown throughout the site. Falls back to username if not set.", ) # Detailed notification preferences (JSON field for flexibility) notification_preferences = models.JSONField( default=dict, blank=True, help_text="Detailed notification preferences stored as JSON", ) def __str__(self): return self.get_display_name() def get_absolute_url(self): return reverse("profile", kwargs={"username": self.username}) def get_display_name(self): """Get the user's display name, falling back to username if not set""" if self.display_name: return self.display_name # Fallback to profile display_name for backward compatibility profile = getattr(self, "profile", None) if profile and profile.display_name: return profile.display_name return self.username def save(self, *args, **kwargs): if not self.user_id: self.user_id = generate_random_id(User, "user_id") super().save(*args, **kwargs) @pghistory.track() class UserProfile(models.Model): # Read-only ID profile_id = models.CharField( max_length=10, unique=True, editable=False, help_text="Unique identifier for this profile that remains constant", ) user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="profile") display_name = models.CharField( max_length=50, blank=True, help_text="Legacy display name field - use User.display_name instead", ) avatar = CloudflareImagesField(blank=True, null=True) pronouns = models.CharField(max_length=50, blank=True) bio = models.TextField(max_length=500, blank=True) # Social media links twitter = models.URLField(blank=True) instagram = models.URLField(blank=True) youtube = models.URLField(blank=True) discord = models.CharField(max_length=100, blank=True) # Ride statistics coaster_credits = models.IntegerField(default=0) dark_ride_credits = models.IntegerField(default=0) flat_ride_credits = models.IntegerField(default=0) water_ride_credits = models.IntegerField(default=0) def get_avatar_url(self): """ Return the avatar URL or generate a default letter-based avatar URL """ if self.avatar: # Return Cloudflare Images URL with avatar variant base_url = self.avatar.url if '/public' in base_url: return base_url.replace('/public', '/avatar') return base_url # Generate default letter-based avatar using first letter of username first_letter = self.user.username[0].upper() if self.user.username else "U" # Use a service like UI Avatars or generate a simple colored avatar return f"https://ui-avatars.com/api/?name={first_letter}&size=200&background=random&color=fff&bold=true" def get_avatar_variants(self): """ Return avatar variants for different use cases """ if self.avatar: base_url = self.avatar.url if '/public' in base_url: return { "thumbnail": base_url.replace('/public', '/thumbnail'), "avatar": base_url.replace('/public', '/avatar'), "large": base_url.replace('/public', '/large'), } else: # If no variant in URL, return the same URL for all variants return { "thumbnail": base_url, "avatar": base_url, "large": base_url, } # For default avatars, return the same URL for all variants default_url = self.get_avatar_url() return { "thumbnail": default_url, "avatar": default_url, "large": default_url, } def save(self, *args, **kwargs): # If no display name is set, use the username if not self.display_name: self.display_name = self.user.username if not self.profile_id: self.profile_id = generate_random_id(UserProfile, "profile_id") super().save(*args, **kwargs) def __str__(self): return self.display_name @pghistory.track() class EmailVerification(models.Model): user = models.OneToOneField(User, on_delete=models.CASCADE) token = models.CharField(max_length=64, unique=True) created_at = models.DateTimeField(auto_now_add=True) last_sent = models.DateTimeField(auto_now_add=True) def __str__(self): return f"Email verification for {self.user.username}" class Meta: verbose_name = "Email Verification" verbose_name_plural = "Email Verifications" @pghistory.track() class PasswordReset(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) token = models.CharField(max_length=64) created_at = models.DateTimeField(auto_now_add=True) expires_at = models.DateTimeField() used = models.BooleanField(default=False) def __str__(self): return f"Password reset for {self.user.username}" class Meta: verbose_name = "Password Reset" verbose_name_plural = "Password Resets" # @pghistory.track() class TopList(TrackedModel): class Categories(models.TextChoices): ROLLER_COASTER = "RC", _("Roller Coaster") DARK_RIDE = "DR", _("Dark Ride") FLAT_RIDE = "FR", _("Flat Ride") WATER_RIDE = "WR", _("Water Ride") PARK = "PK", _("Park") user = models.ForeignKey( User, on_delete=models.CASCADE, related_name="top_lists", # Added related_name for User model access ) title = models.CharField(max_length=100) category = models.CharField(max_length=2, choices=Categories.choices) description = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta(TrackedModel.Meta): ordering = ["-updated_at"] def __str__(self): return ( f"{self.user.get_display_name()}'s {self.category} Top List: {self.title}" ) # @pghistory.track() class TopListItem(TrackedModel): top_list = models.ForeignKey( TopList, on_delete=models.CASCADE, related_name="items" ) content_type = models.ForeignKey( "contenttypes.ContentType", on_delete=models.CASCADE ) object_id = models.PositiveIntegerField() rank = models.PositiveIntegerField() notes = models.TextField(blank=True) class Meta(TrackedModel.Meta): ordering = ["rank"] unique_together = [["top_list", "rank"]] def __str__(self): return f"#{self.rank} in {self.top_list.title}" @pghistory.track() class UserDeletionRequest(models.Model): """ Model to track user deletion requests with email verification. When a user requests to delete their account, a verification code is sent to their email. The deletion is only processed when they provide the correct code. """ user = models.OneToOneField( User, on_delete=models.CASCADE, related_name="deletion_request" ) verification_code = models.CharField( max_length=32, unique=True, help_text="Unique verification code sent to user's email", ) created_at = models.DateTimeField(auto_now_add=True) expires_at = models.DateTimeField(help_text="When this deletion request expires") email_sent_at = models.DateTimeField( null=True, blank=True, help_text="When the verification email was sent" ) attempts = models.PositiveIntegerField( default=0, help_text="Number of verification attempts made" ) max_attempts = models.PositiveIntegerField( default=5, help_text="Maximum number of verification attempts allowed" ) is_used = models.BooleanField( default=False, help_text="Whether this deletion request has been used" ) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["verification_code"]), models.Index(fields=["expires_at"]), models.Index(fields=["user", "is_used"]), ] def __str__(self): return f"Deletion request for {self.user.username} - {self.verification_code}" def save(self, *args, **kwargs): if not self.verification_code: self.verification_code = self.generate_verification_code() if not self.expires_at: # Deletion requests expire after 24 hours self.expires_at = timezone.now() + timedelta(hours=24) super().save(*args, **kwargs) @staticmethod def generate_verification_code(): """Generate a unique 8-character verification code.""" while True: # Generate a random 8-character alphanumeric code code = "".join( secrets.choice("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") for _ in range(8) ) # Ensure it's unique if not UserDeletionRequest.objects.filter(verification_code=code).exists(): return code def is_expired(self): """Check if this deletion request has expired.""" return timezone.now() > self.expires_at def is_valid(self): """Check if this deletion request is still valid.""" return ( not self.is_used and not self.is_expired() and self.attempts < self.max_attempts ) def increment_attempts(self): """Increment the number of verification attempts.""" self.attempts += 1 self.save(update_fields=["attempts"]) def mark_as_used(self): """Mark this deletion request as used.""" self.is_used = True self.save(update_fields=["is_used"]) @classmethod def cleanup_expired(cls): """Remove expired deletion requests.""" expired_requests = cls.objects.filter( expires_at__lt=timezone.now(), is_used=False ) count = expired_requests.count() expired_requests.delete() return count @pghistory.track() class UserNotification(TrackedModel): """ Model to store user notifications for various events. This includes submission approvals, rejections, system announcements, and other user-relevant notifications. """ class NotificationType(models.TextChoices): # Submission related SUBMISSION_APPROVED = "submission_approved", _("Submission Approved") SUBMISSION_REJECTED = "submission_rejected", _("Submission Rejected") SUBMISSION_PENDING = "submission_pending", _("Submission Pending Review") # Review related REVIEW_REPLY = "review_reply", _("Review Reply") REVIEW_HELPFUL = "review_helpful", _("Review Marked Helpful") # Social related FRIEND_REQUEST = "friend_request", _("Friend Request") FRIEND_ACCEPTED = "friend_accepted", _("Friend Request Accepted") MESSAGE_RECEIVED = "message_received", _("Message Received") PROFILE_COMMENT = "profile_comment", _("Profile Comment") # System related SYSTEM_ANNOUNCEMENT = "system_announcement", _("System Announcement") ACCOUNT_SECURITY = "account_security", _("Account Security") FEATURE_UPDATE = "feature_update", _("Feature Update") MAINTENANCE = "maintenance", _("Maintenance Notice") # Achievement related ACHIEVEMENT_UNLOCKED = "achievement_unlocked", _("Achievement Unlocked") MILESTONE_REACHED = "milestone_reached", _("Milestone Reached") class Priority(models.TextChoices): LOW = "low", _("Low") NORMAL = "normal", _("Normal") HIGH = "high", _("High") URGENT = "urgent", _("Urgent") # Core fields user = models.ForeignKey( User, on_delete=models.CASCADE, related_name="notifications" ) notification_type = models.CharField( max_length=30, choices=NotificationType.choices ) title = models.CharField(max_length=200) message = models.TextField() # Optional related object (submission, review, etc.) content_type = models.ForeignKey( "contenttypes.ContentType", on_delete=models.CASCADE, null=True, blank=True ) object_id = models.PositiveIntegerField(null=True, blank=True) related_object = GenericForeignKey("content_type", "object_id") # Metadata priority = models.CharField( max_length=10, choices=Priority.choices, default=Priority.NORMAL ) # Status tracking is_read = models.BooleanField(default=False) read_at = models.DateTimeField(null=True, blank=True) # Delivery tracking email_sent = models.BooleanField(default=False) email_sent_at = models.DateTimeField(null=True, blank=True) push_sent = models.BooleanField(default=False) push_sent_at = models.DateTimeField(null=True, blank=True) # Additional data (JSON field for flexibility) extra_data = models.JSONField(default=dict, blank=True) # Timestamps created_at = models.DateTimeField(auto_now_add=True) expires_at = models.DateTimeField(null=True, blank=True) class Meta(TrackedModel.Meta): ordering = ["-created_at"] indexes = [ models.Index(fields=["user", "is_read"]), models.Index(fields=["user", "notification_type"]), models.Index(fields=["created_at"]), models.Index(fields=["expires_at"]), ] def __str__(self): return f"{self.user.username}: {self.title}" def mark_as_read(self): """Mark notification as read.""" if not self.is_read: self.is_read = True self.read_at = timezone.now() self.save(update_fields=["is_read", "read_at"]) def is_expired(self): """Check if notification has expired.""" if not self.expires_at: return False return timezone.now() > self.expires_at @classmethod def cleanup_expired(cls): """Remove expired notifications.""" expired_notifications = cls.objects.filter(expires_at__lt=timezone.now()) count = expired_notifications.count() expired_notifications.delete() return count @classmethod def mark_all_read_for_user(cls, user): """Mark all notifications as read for a specific user.""" return cls.objects.filter(user=user, is_read=False).update( is_read=True, read_at=timezone.now() ) @pghistory.track() class NotificationPreference(TrackedModel): """ User preferences for different types of notifications. This allows users to control which notifications they receive and through which channels (email, push, in-app). """ user = models.OneToOneField( User, on_delete=models.CASCADE, related_name="notification_preference" ) # Submission notifications submission_approved_email = models.BooleanField(default=True) submission_approved_push = models.BooleanField(default=True) submission_approved_inapp = models.BooleanField(default=True) submission_rejected_email = models.BooleanField(default=True) submission_rejected_push = models.BooleanField(default=True) submission_rejected_inapp = models.BooleanField(default=True) submission_pending_email = models.BooleanField(default=False) submission_pending_push = models.BooleanField(default=False) submission_pending_inapp = models.BooleanField(default=True) # Review notifications review_reply_email = models.BooleanField(default=True) review_reply_push = models.BooleanField(default=True) review_reply_inapp = models.BooleanField(default=True) review_helpful_email = models.BooleanField(default=False) review_helpful_push = models.BooleanField(default=True) review_helpful_inapp = models.BooleanField(default=True) # Social notifications friend_request_email = models.BooleanField(default=True) friend_request_push = models.BooleanField(default=True) friend_request_inapp = models.BooleanField(default=True) friend_accepted_email = models.BooleanField(default=False) friend_accepted_push = models.BooleanField(default=True) friend_accepted_inapp = models.BooleanField(default=True) message_received_email = models.BooleanField(default=True) message_received_push = models.BooleanField(default=True) message_received_inapp = models.BooleanField(default=True) # System notifications system_announcement_email = models.BooleanField(default=True) system_announcement_push = models.BooleanField(default=False) system_announcement_inapp = models.BooleanField(default=True) account_security_email = models.BooleanField(default=True) account_security_push = models.BooleanField(default=True) account_security_inapp = models.BooleanField(default=True) feature_update_email = models.BooleanField(default=True) feature_update_push = models.BooleanField(default=False) feature_update_inapp = models.BooleanField(default=True) # Achievement notifications achievement_unlocked_email = models.BooleanField(default=False) achievement_unlocked_push = models.BooleanField(default=True) achievement_unlocked_inapp = models.BooleanField(default=True) milestone_reached_email = models.BooleanField(default=False) milestone_reached_push = models.BooleanField(default=True) milestone_reached_inapp = models.BooleanField(default=True) class Meta(TrackedModel.Meta): verbose_name = "Notification Preference" verbose_name_plural = "Notification Preferences" def __str__(self): return f"Notification preferences for {self.user.username}" def should_send_notification(self, notification_type, channel): """ Check if a notification should be sent for a specific type and channel. Args: notification_type: The type of notification (from UserNotification.NotificationType) channel: The delivery channel ('email', 'push', 'inapp') Returns: bool: True if notification should be sent, False otherwise """ field_name = f"{notification_type}_{channel}" return getattr(self, field_name, False) # Signal handlers for automatic notification preference creation @receiver(post_save, sender=User) def create_notification_preference(sender, instance, created, **kwargs): """Create notification preferences when a new user is created.""" if created: NotificationPreference.objects.create(user=instance)