thrillwiki_django_no_react/moderation/models.py

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
from django.utils.text import slugify
import pghistory
from core.models import HistoricalModel

UserType = Union[AbstractBaseUser, AnonymousUser]

@pghistory.track()  # Track all changes by default
class EditSubmission(HistoricalModel):
    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(id=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:
            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.id})"
                    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)
                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)
                obj.save()

            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(HistoricalModel):
    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 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 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()