from typing import Any, Dict, Optional, Type, Union from django.db import models from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.contenttypes.models import ContentType from django.conf import settings from django.utils import timezone from django.core.exceptions import ObjectDoesNotExist, FieldDoesNotExist from django.contrib.auth.base_user import AbstractBaseUser from django.contrib.auth.models import AnonymousUser import pghistory from apps.core.history import TrackedModel UserType = Union[AbstractBaseUser, AnonymousUser] @pghistory.track() # Track all changes by default class EditSubmission(TrackedModel): STATUS_CHOICES = [ ("PENDING", "Pending"), ("APPROVED", "Approved"), ("REJECTED", "Rejected"), ("ESCALATED", "Escalated"), ] SUBMISSION_TYPE_CHOICES = [ ("EDIT", "Edit Existing"), ("CREATE", "Create New"), ] # Who submitted the edit user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="edit_submissions", ) # What is being edited (Park or Ride) content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.PositiveIntegerField( null=True, blank=True ) # Null for new objects content_object = GenericForeignKey("content_type", "object_id") # Type of submission submission_type = models.CharField( max_length=10, choices=SUBMISSION_TYPE_CHOICES, default="EDIT" ) # The actual changes/data changes = models.JSONField( help_text="JSON representation of the changes or new object data" ) # Moderator's edited version of changes before approval moderator_changes = models.JSONField( null=True, blank=True, help_text="Moderator's edited version of the changes before approval", ) # Metadata reason = models.TextField(help_text="Why this edit/addition is needed") source = models.TextField( blank=True, help_text="Source of information (if applicable)" ) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="PENDING") created_at = models.DateTimeField(auto_now_add=True) # Review details handled_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="handled_submissions", ) handled_at = models.DateTimeField(null=True, blank=True) notes = models.TextField( blank=True, help_text="Notes from the moderator about this submission" ) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["content_type", "object_id"]), models.Index(fields=["status"]), ] def __str__(self) -> str: action = "creation" if self.submission_type == "CREATE" else "edit" if model_class := self.content_type.model_class(): target = self.content_object or model_class.__name__ else: target = "Unknown" return f"{action} by {self.user.username} on {target}" def _resolve_foreign_keys(self, data: Dict[str, Any]) -> Dict[str, Any]: """Convert foreign key IDs to model instances""" if not (model_class := self.content_type.model_class()): raise ValueError("Could not resolve model class") resolved_data = data.copy() for field_name, value in data.items(): try: if ( (field := model_class._meta.get_field(field_name)) and isinstance(field, models.ForeignKey) and value is not None ): if related_model := field.related_model: resolved_data[field_name] = related_model.objects.get(pk=value) except (FieldDoesNotExist, ObjectDoesNotExist): continue return resolved_data def _prepare_model_data( self, data: Dict[str, Any], model_class: Type[models.Model] ) -> Dict[str, Any]: """Prepare data for model creation/update by filtering out auto-generated fields""" prepared_data = data.copy() # Remove fields that are auto-generated or handled by the model's save # method auto_fields = {"created_at", "updated_at", "slug"} for field in auto_fields: prepared_data.pop(field, None) # Set default values for required fields if not provided for field in model_class._meta.fields: if not field.auto_created and not field.blank and not field.null: if field.name not in prepared_data and field.has_default(): prepared_data[field.name] = field.get_default() return prepared_data def _check_duplicate_name( self, model_class: Type[models.Model], name: str ) -> Optional[models.Model]: """Check if an object with the same name already exists""" try: return model_class.objects.filter(name=name).first() except BaseException as e: print(f"Error checking for duplicate name '{name}': {e}") raise e return None def approve(self, user: UserType) -> Optional[models.Model]: """Approve the submission and apply the changes""" if not (model_class := self.content_type.model_class()): raise ValueError("Could not resolve model class") try: # Use moderator_changes if available, otherwise use original # changes changes_to_apply = ( self.moderator_changes if self.moderator_changes is not None else self.changes ) resolved_data = self._resolve_foreign_keys(changes_to_apply) prepared_data = self._prepare_model_data(resolved_data, model_class) # For CREATE submissions, check for duplicates by name if self.submission_type == "CREATE" and "name" in prepared_data: if existing_obj := self._check_duplicate_name( model_class, prepared_data["name"] ): self.status = "REJECTED" self.handled_by = user # type: ignore self.handled_at = timezone.now() self.notes = f"A { model_class.__name__} with the name '{ prepared_data['name']}' already exists (ID: { existing_obj.pk})" self.save() raise ValueError(self.notes) self.status = "APPROVED" self.handled_by = user # type: ignore self.handled_at = timezone.now() if self.submission_type == "CREATE": # Create new object obj = model_class(**prepared_data) # CRITICAL STYLEGUIDE FIX: Call full_clean before save obj.full_clean() obj.save() # Update object_id after creation self.object_id = getattr(obj, "id", None) else: # Apply changes to existing object if not (obj := self.content_object): raise ValueError("Content object not found") for field, value in prepared_data.items(): setattr(obj, field, value) # CRITICAL STYLEGUIDE FIX: Call full_clean before save obj.full_clean() obj.save() # CRITICAL STYLEGUIDE FIX: Call full_clean before save self.full_clean() self.save() return obj except Exception as e: if ( self.status != "REJECTED" ): # Don't override if already rejected due to duplicate self.status = "PENDING" # Reset status if approval failed self.save() raise ValueError(f"Error approving submission: {str(e)}") from e def reject(self, user: UserType) -> None: """Reject the submission""" self.status = "REJECTED" self.handled_by = user # type: ignore self.handled_at = timezone.now() self.save() def escalate(self, user: UserType) -> None: """Escalate the submission to admin""" self.status = "ESCALATED" self.handled_by = user # type: ignore self.handled_at = timezone.now() self.save() @pghistory.track() # Track all changes by default class PhotoSubmission(TrackedModel): STATUS_CHOICES = [ ("PENDING", "Pending"), ("APPROVED", "Approved"), ("REJECTED", "Rejected"), ("ESCALATED", "Escalated"), ] # Who submitted the photo user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="photo_submissions", ) # What the photo is for (Park or Ride) content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.PositiveIntegerField() content_object = GenericForeignKey("content_type", "object_id") # The photo itself photo = models.ImageField(upload_to="submissions/photos/") caption = models.CharField(max_length=255, blank=True) date_taken = models.DateField(null=True, blank=True) # Metadata status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="PENDING") created_at = models.DateTimeField(auto_now_add=True) # Review details handled_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="handled_photos", ) handled_at = models.DateTimeField(null=True, blank=True) notes = models.TextField( blank=True, help_text="Notes from the moderator about this photo submission", ) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["content_type", "object_id"]), models.Index(fields=["status"]), ] def __str__(self) -> str: return f"Photo submission by { self.user.username} for { self.content_object}" def approve(self, moderator: UserType, notes: str = "") -> None: """Approve the photo submission""" from apps.parks.models.media import ParkPhoto from apps.rides.models.media import RidePhoto self.status = "APPROVED" self.handled_by = moderator # type: ignore self.handled_at = timezone.now() self.notes = notes # Determine the correct photo model based on the content type model_class = self.content_type.model_class() if model_class.__name__ == "Park": PhotoModel = ParkPhoto elif model_class.__name__ == "Ride": PhotoModel = RidePhoto else: raise ValueError(f"Unsupported content type: {model_class.__name__}") # Create the approved photo PhotoModel.objects.create( uploaded_by=self.user, content_object=self.content_object, image=self.photo, caption=self.caption, is_approved=True, ) self.save() def reject(self, moderator: UserType, notes: str) -> None: """Reject the photo submission""" self.status = "REJECTED" self.handled_by = moderator # type: ignore self.handled_at = timezone.now() self.notes = notes self.save() def auto_approve(self) -> None: """Auto-approve submissions from moderators""" # Get user role safely user_role = getattr(self.user, "role", None) # If user is moderator or above, auto-approve if user_role in ["MODERATOR", "ADMIN", "SUPERUSER"]: self.approve(self.user) def escalate(self, moderator: UserType, notes: str = "") -> None: """Escalate the photo submission to admin""" self.status = "ESCALATED" self.handled_by = moderator # type: ignore self.handled_at = timezone.now() self.notes = notes self.save()