from typing import Any, Dict, Optional, Type, Union, cast 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.apps import apps from django.core.exceptions import ObjectDoesNotExist, FieldDoesNotExist from django.contrib.auth.base_user import AbstractBaseUser from django.contrib.auth.models import AnonymousUser UserType = Union[AbstractBaseUser, AnonymousUser] class EditSubmission(models.Model): STATUS_CHOICES = [ ("NEW", "New"), ("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" ) # 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="NEW") 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" model_class = self.content_type.model_class() target = self.content_object or ( model_class.__name__ if model_class else "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""" model_class = self.content_type.model_class() if not model_class: raise ValueError("Could not resolve model class") resolved_data = data.copy() for field_name, value in data.items(): try: field = model_class._meta.get_field(field_name) if isinstance(field, models.ForeignKey) and value is not None: related_model = field.related_model if related_model: resolved_data[field_name] = related_model.objects.get(id=value) except (FieldDoesNotExist, ObjectDoesNotExist): continue return resolved_data def approve(self, user: UserType) -> Optional[models.Model]: """Approve the submission and apply the changes""" self.status = "APPROVED" self.handled_by = user # type: ignore self.handled_at = timezone.now() model_class = self.content_type.model_class() if not model_class: raise ValueError("Could not resolve model class") try: resolved_data = self._resolve_foreign_keys(self.changes) if self.submission_type == "CREATE": # Create new object obj = model_class(**resolved_data) obj.save() # Update object_id after creation self.object_id = getattr(obj, "id", None) else: # Apply changes to existing object obj = self.content_object if not obj: raise ValueError("Content object not found") for field, value in resolved_data.items(): setattr(obj, field, value) obj.save() self.save() return obj except Exception as e: 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() class PhotoSubmission(models.Model): STATUS_CHOICES = [ ("NEW", "New"), ("APPROVED", "Approved"), ("REJECTED", "Rejected"), ("AUTO_APPROVED", "Auto Approved"), ] # 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="NEW") 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 media.models import Photo self.status = "APPROVED" self.handled_by = moderator # type: ignore self.handled_at = timezone.now() self.notes = notes # Create the approved photo Photo.objects.create( uploaded_by=self.user, content_type=self.content_type, object_id=self.object_id, 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 the photo submission (for moderators/admins)""" from media.models import Photo self.status = "AUTO_APPROVED" self.handled_by = self.user self.handled_at = timezone.now() # Create the approved photo Photo.objects.create( uploaded_by=self.user, content_type=self.content_type, object_id=self.object_id, image=self.photo, caption=self.caption, is_approved=True, ) self.save()