mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 09:31:09 -05:00
9e0259f739
Enable the avatar field in the UserProfile model and associated event tracking, alongside new migrations for core, parks, and rides modules. Replit-Commit-Author: Agent Replit-Commit-Session-Id: eff39de1-3afa-446d-a965-acaf61837fc7 Replit-Commit-Checkpoint-Type: intermediate_checkpoint Replit-Commit-Screenshot-Url: https://storage.googleapis.com/screenshot-production-us-central1/d6d61dac-164d-45dd-929f-7dcdfd771b64/eff39de1-3afa-446d-a965-acaf61837fc7/RPze4Xv
639 lines
22 KiB
Python
639 lines
22 KiB
Python
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
|
|
from apps.core.choices import RichChoiceField
|
|
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):
|
|
# 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 = RichChoiceField(
|
|
choice_group="user_roles",
|
|
domain="accounts",
|
|
max_length=10,
|
|
default="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 = RichChoiceField(
|
|
choice_group="theme_preferences",
|
|
domain="accounts",
|
|
max_length=5,
|
|
default="light",
|
|
)
|
|
|
|
# Notification preferences
|
|
email_notifications = models.BooleanField(default=True)
|
|
push_notifications = models.BooleanField(default=False)
|
|
|
|
# Privacy settings
|
|
privacy_level = RichChoiceField(
|
|
choice_group="privacy_levels",
|
|
domain="accounts",
|
|
max_length=10,
|
|
default="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 = RichChoiceField(
|
|
choice_group="privacy_levels",
|
|
domain="accounts",
|
|
max_length=10,
|
|
default="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):
|
|
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 = RichChoiceField(
|
|
choice_group="top_list_categories",
|
|
domain="accounts",
|
|
max_length=2,
|
|
)
|
|
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.
|
|
"""
|
|
|
|
# Core fields
|
|
user = models.ForeignKey(
|
|
User, on_delete=models.CASCADE, related_name="notifications"
|
|
)
|
|
|
|
notification_type = RichChoiceField(
|
|
choice_group="notification_types",
|
|
domain="accounts",
|
|
max_length=30,
|
|
)
|
|
|
|
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 = RichChoiceField(
|
|
choice_group="notification_priorities",
|
|
domain="accounts",
|
|
max_length=10,
|
|
default="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)
|