mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2026-02-05 09:25:18 -05:00
d631f3183c
feat: add passkey authentication and enhance user preferences - Add passkey login security event type with fingerprint icon - Include request and site context in email confirmation for backend - Add user_id exact match filter to prevent incorrect user lookups - Enable PATCH method for updating user preferences via API - Add moderation_preferences support to user settings - Optimize ticket queries with select_related and prefetch_related This commit introduces passkey authentication tracking, improves user profile filtering accuracy, and extends the preferences API to support updates. Query optimizations reduce database hits for ticket listings.
417 lines
16 KiB
Python
417 lines
16 KiB
Python
from typing import TYPE_CHECKING, Any, Optional
|
||
|
||
import pghistory
|
||
from django.core.exceptions import ValidationError
|
||
from django.db import models
|
||
from django.urls import reverse
|
||
from django.utils.text import slugify
|
||
|
||
from apps.core.choices import RichChoiceField
|
||
from apps.core.history import TrackedModel
|
||
from apps.core.state_machine import RichFSMField, StateMachineMixin
|
||
from config.django import base as settings
|
||
|
||
if TYPE_CHECKING:
|
||
from django.contrib.auth.models import AbstractBaseUser
|
||
|
||
from apps.rides.models import Ride
|
||
|
||
from . import ParkArea
|
||
|
||
|
||
@pghistory.track()
|
||
class Park(StateMachineMixin, TrackedModel):
|
||
# Import managers
|
||
from ..managers import ParkManager
|
||
|
||
objects = ParkManager()
|
||
id: int # Type hint for Django's automatic id field
|
||
|
||
name = models.CharField(max_length=255, help_text="Park name")
|
||
slug = models.SlugField(max_length=255, unique=True, help_text="URL-friendly identifier")
|
||
description = models.TextField(blank=True, help_text="Park description")
|
||
state_field_name = "status"
|
||
|
||
status = RichFSMField(
|
||
choice_group="statuses",
|
||
domain="parks",
|
||
max_length=20,
|
||
default="OPERATING",
|
||
)
|
||
|
||
park_type = RichChoiceField(
|
||
choice_group="types",
|
||
domain="parks",
|
||
max_length=30,
|
||
default="THEME_PARK",
|
||
db_index=True,
|
||
help_text="Type/category of the park",
|
||
)
|
||
|
||
# Location relationship - reverse relation from ParkLocation
|
||
# location will be available via the 'location' related_name on
|
||
# ParkLocation
|
||
|
||
# Details
|
||
opening_date = models.DateField(null=True, blank=True, help_text="Opening date")
|
||
opening_date_precision = RichChoiceField(
|
||
choice_group="date_precision",
|
||
domain="parks",
|
||
max_length=20,
|
||
default="exact",
|
||
blank=True,
|
||
help_text="Precision of the opening date",
|
||
)
|
||
closing_date = models.DateField(null=True, blank=True, help_text="Closing date")
|
||
closing_date_precision = RichChoiceField(
|
||
choice_group="date_precision",
|
||
domain="parks",
|
||
max_length=20,
|
||
default="exact",
|
||
blank=True,
|
||
help_text="Precision of the closing date",
|
||
)
|
||
operating_season = models.CharField(max_length=255, blank=True, help_text="Operating season")
|
||
size_acres = models.DecimalField(
|
||
max_digits=10, decimal_places=2, null=True, blank=True, help_text="Park size in acres"
|
||
)
|
||
website = models.URLField(blank=True, help_text="Official website URL")
|
||
phone = models.CharField(max_length=30, blank=True, help_text="Contact phone number")
|
||
email = models.EmailField(blank=True, help_text="Contact email address")
|
||
|
||
# Statistics
|
||
average_rating = models.DecimalField(
|
||
max_digits=3, decimal_places=2, null=True, blank=True, help_text="Average user rating (1–10)"
|
||
)
|
||
ride_count = models.IntegerField(null=True, blank=True, help_text="Total ride count")
|
||
coaster_count = models.IntegerField(null=True, blank=True, help_text="Total coaster count")
|
||
|
||
# Image settings - references to existing photos
|
||
banner_image = models.ForeignKey(
|
||
"ParkPhoto",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="parks_using_as_banner",
|
||
help_text="Photo to use as banner image for this park",
|
||
)
|
||
card_image = models.ForeignKey(
|
||
"ParkPhoto",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="parks_using_as_card",
|
||
help_text="Photo to use as card image for this park",
|
||
)
|
||
|
||
# Relationships
|
||
operator = models.ForeignKey(
|
||
"Company",
|
||
on_delete=models.PROTECT,
|
||
related_name="operated_parks",
|
||
help_text="Company that operates this park",
|
||
limit_choices_to={"roles__contains": ["OPERATOR"]},
|
||
)
|
||
property_owner = models.ForeignKey(
|
||
"Company",
|
||
on_delete=models.PROTECT,
|
||
related_name="owned_parks",
|
||
null=True,
|
||
blank=True,
|
||
help_text="Company that owns the property (if different from operator)",
|
||
limit_choices_to={"roles__contains": ["PROPERTY_OWNER"]},
|
||
)
|
||
areas: models.Manager["ParkArea"] # Type hint for reverse relation
|
||
# Type hint for reverse relation from rides app
|
||
rides: models.Manager["Ride"]
|
||
|
||
# Metadata
|
||
created_at = models.DateTimeField(auto_now_add=True, null=True)
|
||
updated_at = models.DateTimeField(auto_now=True)
|
||
|
||
# Frontend URL
|
||
url = models.URLField(blank=True, help_text="Frontend URL for this park")
|
||
|
||
# Computed fields for hybrid filtering
|
||
opening_year = models.IntegerField(
|
||
null=True, blank=True, db_index=True, help_text="Year the park opened (computed from opening_date)"
|
||
)
|
||
search_text = models.TextField(
|
||
blank=True, db_index=True, help_text="Searchable text combining name, description, location, and operator"
|
||
)
|
||
|
||
# Timezone for park operations
|
||
timezone = models.CharField(
|
||
max_length=50,
|
||
default="UTC",
|
||
blank=True,
|
||
help_text="Timezone identifier for park operations (e.g., 'America/New_York')",
|
||
)
|
||
|
||
# Submission metadata fields (from frontend schema)
|
||
source_url = models.URLField(
|
||
blank=True,
|
||
help_text="Source URL for the data (e.g., official website, Wikipedia)",
|
||
)
|
||
is_test_data = models.BooleanField(
|
||
default=False,
|
||
help_text="Whether this is test/development data",
|
||
)
|
||
|
||
class Meta:
|
||
verbose_name = "Park"
|
||
verbose_name_plural = "Parks"
|
||
ordering = ["name"]
|
||
constraints = [
|
||
# Business rule: Closing date must be after opening date
|
||
models.CheckConstraint(
|
||
name="park_closing_after_opening",
|
||
check=models.Q(closing_date__isnull=True)
|
||
| models.Q(opening_date__isnull=True)
|
||
| models.Q(closing_date__gte=models.F("opening_date")),
|
||
violation_error_message="Closing date must be after opening date",
|
||
),
|
||
# Business rule: Size must be positive
|
||
models.CheckConstraint(
|
||
name="park_size_positive",
|
||
check=models.Q(size_acres__isnull=True) | models.Q(size_acres__gt=0),
|
||
violation_error_message="Park size must be positive",
|
||
),
|
||
# Business rule: Rating must be between 1 and 10
|
||
models.CheckConstraint(
|
||
name="park_rating_range",
|
||
check=models.Q(average_rating__isnull=True)
|
||
| (models.Q(average_rating__gte=1) & models.Q(average_rating__lte=10)),
|
||
violation_error_message="Average rating must be between 1 and 10",
|
||
),
|
||
# Business rule: Counts must be non-negative
|
||
models.CheckConstraint(
|
||
name="park_ride_count_non_negative",
|
||
check=models.Q(ride_count__isnull=True) | models.Q(ride_count__gte=0),
|
||
violation_error_message="Ride count must be non-negative",
|
||
),
|
||
models.CheckConstraint(
|
||
name="park_coaster_count_non_negative",
|
||
check=models.Q(coaster_count__isnull=True) | models.Q(coaster_count__gte=0),
|
||
violation_error_message="Coaster count must be non-negative",
|
||
),
|
||
# Business rule: Coaster count cannot exceed ride count
|
||
models.CheckConstraint(
|
||
name="park_coaster_count_lte_ride_count",
|
||
check=models.Q(coaster_count__isnull=True)
|
||
| models.Q(ride_count__isnull=True)
|
||
| models.Q(coaster_count__lte=models.F("ride_count")),
|
||
violation_error_message="Coaster count cannot exceed total ride count",
|
||
),
|
||
]
|
||
|
||
def __str__(self) -> str:
|
||
return self.name
|
||
|
||
# FSM Transition Wrapper Methods
|
||
def reopen(self, *, user: Optional["AbstractBaseUser"] = None) -> None:
|
||
"""Transition park to OPERATING status."""
|
||
self.transition_to_operating(user=user)
|
||
self.save()
|
||
|
||
def close_temporarily(self, *, user: Optional["AbstractBaseUser"] = None) -> None:
|
||
"""Transition park to CLOSED_TEMP status."""
|
||
self.transition_to_closed_temp(user=user)
|
||
self.save()
|
||
|
||
def start_construction(self, *, user: Optional["AbstractBaseUser"] = None) -> None:
|
||
"""Transition park to UNDER_CONSTRUCTION status."""
|
||
self.transition_to_under_construction(user=user)
|
||
self.save()
|
||
|
||
def close_permanently(self, *, closing_date=None, user: Optional["AbstractBaseUser"] = None) -> None:
|
||
"""Transition park to CLOSED_PERM status."""
|
||
self.transition_to_closed_perm(user=user)
|
||
if closing_date:
|
||
self.closing_date = closing_date
|
||
self.save()
|
||
|
||
def demolish(self, *, user: Optional["AbstractBaseUser"] = None) -> None:
|
||
"""Transition park to DEMOLISHED status."""
|
||
self.transition_to_demolished(user=user)
|
||
self.save()
|
||
|
||
def relocate(self, *, user: Optional["AbstractBaseUser"] = None) -> None:
|
||
"""Transition park to RELOCATED status."""
|
||
self.transition_to_relocated(user=user)
|
||
self.save()
|
||
|
||
def save(self, *args: Any, **kwargs: Any) -> None:
|
||
from django.contrib.contenttypes.models import ContentType
|
||
|
||
from apps.core.history import HistoricalSlug
|
||
|
||
# Get old instance if it exists
|
||
if self.pk:
|
||
try:
|
||
old_instance = type(self).objects.get(pk=self.pk)
|
||
old_name = old_instance.name
|
||
old_slug = old_instance.slug
|
||
except type(self).DoesNotExist:
|
||
old_name = None
|
||
old_slug = None
|
||
else:
|
||
old_name = None
|
||
old_slug = None
|
||
|
||
# Generate new slug if name has changed or slug is missing
|
||
if not self.slug or (old_name and old_name != self.name):
|
||
self.slug = slugify(self.name)
|
||
|
||
# Generate frontend URL
|
||
frontend_domain = getattr(settings, "FRONTEND_DOMAIN", "https://thrillwiki.com")
|
||
self.url = f"{frontend_domain}/parks/{self.slug}/"
|
||
|
||
# Populate computed fields for hybrid filtering
|
||
self._populate_computed_fields()
|
||
|
||
# Save the model
|
||
super().save(*args, **kwargs)
|
||
|
||
# If slug has changed, save historical record
|
||
if old_slug and old_slug != self.slug:
|
||
HistoricalSlug.objects.create(
|
||
content_type=ContentType.objects.get_for_model(self),
|
||
object_id=self.pk,
|
||
slug=old_slug,
|
||
)
|
||
|
||
def _populate_computed_fields(self) -> None:
|
||
"""Populate computed fields for hybrid filtering"""
|
||
# Populate opening_year from opening_date
|
||
if self.opening_date:
|
||
self.opening_year = self.opening_date.year
|
||
else:
|
||
self.opening_year = None
|
||
|
||
# Populate search_text for client-side filtering
|
||
search_parts = [self.name]
|
||
|
||
if self.description:
|
||
search_parts.append(self.description)
|
||
|
||
# Add location information if available
|
||
try:
|
||
if hasattr(self, "location") and self.location:
|
||
if self.location.city:
|
||
search_parts.append(self.location.city)
|
||
if self.location.state:
|
||
search_parts.append(self.location.state)
|
||
if self.location.country:
|
||
search_parts.append(self.location.country)
|
||
except Exception:
|
||
# Handle case where location relationship doesn't exist yet
|
||
pass
|
||
|
||
# Add operator information
|
||
if self.operator:
|
||
search_parts.append(self.operator.name)
|
||
|
||
# Add property owner information if different
|
||
if self.property_owner and self.property_owner != self.operator:
|
||
search_parts.append(self.property_owner.name)
|
||
|
||
# Combine all parts into searchable text
|
||
self.search_text = " ".join(filter(None, search_parts)).lower()
|
||
|
||
def clean(self):
|
||
super().clean()
|
||
if self.operator and "OPERATOR" not in self.operator.roles:
|
||
raise ValidationError({"operator": "Company must have the OPERATOR role."})
|
||
if self.property_owner and "PROPERTY_OWNER" not in self.property_owner.roles:
|
||
raise ValidationError({"property_owner": "Company must have the PROPERTY_OWNER role."})
|
||
|
||
def get_absolute_url(self) -> str:
|
||
return reverse("parks:park_detail", kwargs={"slug": self.slug})
|
||
|
||
@property
|
||
def formatted_location(self) -> str:
|
||
"""Get formatted address from ParkLocation if it exists"""
|
||
if hasattr(self, "location") and self.location:
|
||
return self.location.formatted_address
|
||
return ""
|
||
|
||
@property
|
||
def is_closing(self) -> bool:
|
||
"""Returns True if this park has a closing date in the future (announced closure)."""
|
||
from django.utils import timezone
|
||
if self.closing_date:
|
||
return self.closing_date > timezone.now().date()
|
||
return False
|
||
|
||
@property
|
||
def coordinates(self) -> list[float] | None:
|
||
"""Returns coordinates as a list [latitude, longitude]"""
|
||
if hasattr(self, "location") and self.location:
|
||
coords = self.location.coordinates
|
||
if coords and isinstance(coords, tuple | list):
|
||
return list(coords)
|
||
return None
|
||
|
||
@classmethod
|
||
def get_by_slug(cls, slug: str) -> tuple["Park", bool]:
|
||
"""Get park by current or historical slug"""
|
||
from django.contrib.contenttypes.models import ContentType
|
||
|
||
from apps.core.history import HistoricalSlug
|
||
|
||
print(f"\nLooking up slug: {slug}")
|
||
|
||
try:
|
||
park = cls.objects.get(slug=slug)
|
||
print(f"Found current park with slug: {slug}")
|
||
return park, False
|
||
except cls.DoesNotExist:
|
||
print(f"No current park found with slug: {slug}")
|
||
|
||
# Try historical slugs in HistoricalSlug model
|
||
content_type = ContentType.objects.get_for_model(cls)
|
||
print(f"Searching HistoricalSlug with content_type: {content_type}")
|
||
historical = (
|
||
HistoricalSlug.objects.filter(content_type=content_type, slug=slug).order_by("-created_at").first()
|
||
)
|
||
|
||
if historical:
|
||
print(
|
||
f"Found historical slug record for object_id: {
|
||
historical.object_id
|
||
}"
|
||
)
|
||
try:
|
||
park = cls.objects.get(pk=historical.object_id)
|
||
print(f"Found park from historical slug: {park.name}")
|
||
return park, True
|
||
except cls.DoesNotExist:
|
||
print("Park not found for historical slug record")
|
||
else:
|
||
print("No historical slug record found")
|
||
|
||
# Try pghistory events
|
||
print("Searching pghistory events")
|
||
event_model = getattr(cls, "event_model", None)
|
||
historical_event = None
|
||
if event_model:
|
||
historical_event = event_model.objects.filter(slug=slug).order_by("-pgh_created_at").first()
|
||
|
||
if historical_event:
|
||
print(
|
||
f"Found pghistory event for pgh_obj_id: {
|
||
historical_event.pgh_obj_id
|
||
}"
|
||
)
|
||
try:
|
||
park = cls.objects.get(pk=historical_event.pgh_obj_id)
|
||
print(f"Found park from pghistory: {park.name}")
|
||
return park, True
|
||
except cls.DoesNotExist:
|
||
print("Park not found for pghistory event")
|
||
else:
|
||
print("No pghistory event found")
|
||
|
||
raise cls.DoesNotExist("No park found with this slug") from None
|