pacnpal/thrillwiki_django_no_react
1
1
Fork 0
mirror of https://github.com/pacnpal/thrillwiki_django_no_react.git synced 2025-12-20 04:31:09 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
9bed7827847f106146b38ac1382725df091df907
thrillwiki_django_no_react/backend/apps/accounts/models.py
pacnpal 9bed782784 feat: Implement avatar upload system with Cloudflare integration 
- Added migration to transition avatar data from CloudflareImageField to ForeignKey structure in UserProfile.
- Fixed UserProfileEvent avatar field to align with new avatar structure.
- Created serializers for social authentication, including connected and available providers.
- Developed request logging middleware for comprehensive request/response logging.
- Updated moderation and parks migrations to remove outdated triggers and adjust foreign key relationships.
- Enhanced rides migrations to ensure proper handling of image uploads and triggers.
- Introduced a test script for the 3-step avatar upload process, ensuring functionality with Cloudflare.
- Documented the fix for avatar upload issues, detailing root cause, implementation, and verification steps.
- Implemented automatic deletion of Cloudflare images upon avatar, park, and ride photo changes or removals.
2025-08-30 21:20:25 -04:00

679 lines
24 KiB
Python
Raw Blame History

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
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 = models.ForeignKey(
'django_cloudflareimages_toolkit.CloudflareImage',
on_delete=models.SET_NULL,
null=True,
blank=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 and self.avatar.is_uploaded:
# Try to get avatar variant first, fallback to public
avatar_url = self.avatar.get_url('avatar')
if avatar_url:
return avatar_url
# Fallback to public variant
public_url = self.avatar.get_url('public')
if public_url:
return public_url
# Last fallback - try any available variant
if self.avatar.variants:
if isinstance(self.avatar.variants, list) and self.avatar.variants:
return self.avatar.variants[0]
elif isinstance(self.avatar.variants, dict):
# Return first available variant
for variant_url in self.avatar.variants.values():
if variant_url:
return variant_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 and self.avatar.is_uploaded:
variants = {}
# Try to get specific variants
thumbnail_url = self.avatar.get_url('thumbnail')
avatar_url = self.avatar.get_url('avatar')
large_url = self.avatar.get_url('large')
public_url = self.avatar.get_url('public')
# Use specific variants if available, otherwise fallback to public or first available
fallback_url = public_url
if not fallback_url and self.avatar.variants:
if isinstance(self.avatar.variants, list) and self.avatar.variants:
fallback_url = self.avatar.variants[0]
elif isinstance(self.avatar.variants, dict):
fallback_url = next(iter(self.avatar.variants.values()), None)
variants = {
"thumbnail": thumbnail_url or fallback_url,
"avatar": avatar_url or fallback_url,
"large": large_url or fallback_url,
}
# Only return variants if we have at least one valid URL
if any(variants.values()):
return variants
# 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)
Reference in New Issue View Git Blame Copy Permalink