""" Moderation Models This module contains models for the ThrillWiki moderation system, including: - EditSubmission: Original content submission and approval workflow - ModerationReport: User reports for content moderation - ModerationQueue: Workflow management for moderation tasks - ModerationAction: Actions taken against users/content - BulkOperation: Administrative bulk operations All models use pghistory for change tracking and TrackedModel base class. """ from typing import Any, Dict, Optional, 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 from datetime import timedelta import pghistory from apps.core.history import TrackedModel UserType = Union[AbstractBaseUser, AnonymousUser] # ============================================================================ # Original EditSubmission Model (Preserved) # ============================================================================ @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(TrackedModel.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: field = model_class._meta.get_field(field_name) if isinstance(field, models.ForeignKey) and value is not None: try: related_obj = field.related_model.objects.get(pk=value) resolved_data[field_name] = related_obj except ObjectDoesNotExist: raise ValueError( f"Related object {field.related_model.__name__} with pk={value} does not exist" ) except FieldDoesNotExist: # Field doesn't exist on model, skip it continue return resolved_data def _get_final_changes(self) -> Dict[str, Any]: """Get the final changes to apply (moderator changes if available, otherwise original changes)""" return self.moderator_changes or self.changes def approve(self, moderator: UserType) -> Optional[models.Model]: """ Approve this submission and apply the changes. Args: moderator: The user approving the submission Returns: The created or updated model instance Raises: ValueError: If submission cannot be approved ValidationError: If the data is invalid """ if self.status != "PENDING": raise ValueError(f"Cannot approve submission with status {self.status}") model_class = self.content_type.model_class() if not model_class: raise ValueError("Could not resolve model class") final_changes = self._get_final_changes() resolved_changes = self._resolve_foreign_keys(final_changes) try: if self.submission_type == "CREATE": # Create new object obj = model_class(**resolved_changes) obj.full_clean() obj.save() else: # Update existing object if not self.content_object: raise ValueError("Cannot update: content object not found") obj = self.content_object for field_name, value in resolved_changes.items(): if hasattr(obj, field_name): setattr(obj, field_name, value) obj.full_clean() obj.save() # Mark submission as approved self.status = "APPROVED" self.handled_by = moderator self.handled_at = timezone.now() self.save() return obj except Exception as e: # Mark as rejected on any error self.status = "REJECTED" self.handled_by = moderator self.handled_at = timezone.now() self.notes = f"Approval failed: {str(e)}" self.save() raise def reject(self, moderator: UserType, reason: str) -> None: """ Reject this submission. Args: moderator: The user rejecting the submission reason: Reason for rejection """ if self.status != "PENDING": raise ValueError(f"Cannot reject submission with status {self.status}") self.status = "REJECTED" self.handled_by = moderator self.handled_at = timezone.now() self.notes = f"Rejected: {reason}" self.save() def escalate(self, moderator: UserType, reason: str) -> None: """ Escalate this submission for higher-level review. Args: moderator: The user escalating the submission reason: Reason for escalation """ if self.status != "PENDING": raise ValueError(f"Cannot escalate submission with status {self.status}") self.status = "ESCALATED" self.handled_by = moderator self.handled_at = timezone.now() self.notes = f"Escalated: {reason}" self.save() @property def submitted_by(self): """Alias for user field to maintain compatibility""" return self.user @property def submitted_at(self): """Alias for created_at field to maintain compatibility""" return self.created_at # ============================================================================ # New Moderation System Models # ============================================================================ @pghistory.track() class ModerationReport(TrackedModel): """ Model for tracking user reports about content, users, or behavior. This handles the initial reporting phase where users flag content or behavior that needs moderator attention. """ STATUS_CHOICES = [ ('PENDING', 'Pending Review'), ('UNDER_REVIEW', 'Under Review'), ('RESOLVED', 'Resolved'), ('DISMISSED', 'Dismissed'), ] PRIORITY_CHOICES = [ ('LOW', 'Low'), ('MEDIUM', 'Medium'), ('HIGH', 'High'), ('URGENT', 'Urgent'), ] REPORT_TYPE_CHOICES = [ ('SPAM', 'Spam'), ('HARASSMENT', 'Harassment'), ('INAPPROPRIATE_CONTENT', 'Inappropriate Content'), ('MISINFORMATION', 'Misinformation'), ('COPYRIGHT', 'Copyright Violation'), ('PRIVACY', 'Privacy Violation'), ('HATE_SPEECH', 'Hate Speech'), ('VIOLENCE', 'Violence or Threats'), ('OTHER', 'Other'), ] # Report details report_type = models.CharField(max_length=50, choices=REPORT_TYPE_CHOICES) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING') priority = models.CharField( max_length=10, choices=PRIORITY_CHOICES, default='MEDIUM') # What is being reported reported_entity_type = models.CharField( max_length=50, help_text="Type of entity being reported (park, ride, user, etc.)") reported_entity_id = models.PositiveIntegerField( help_text="ID of the entity being reported") content_type = models.ForeignKey( ContentType, on_delete=models.CASCADE, null=True, blank=True) # Report content reason = models.CharField(max_length=200, help_text="Brief reason for the report") description = models.TextField(help_text="Detailed description of the issue") evidence_urls = models.JSONField( default=list, blank=True, help_text="URLs to evidence (screenshots, etc.)") # Users involved reported_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='moderation_reports_made' ) assigned_moderator = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name='assigned_moderation_reports' ) # Resolution resolution_action = models.CharField( max_length=100, blank=True, help_text="Action taken to resolve") resolution_notes = models.TextField( blank=True, help_text="Notes about the resolution") resolved_at = models.DateTimeField(null=True, blank=True) # Timestamps created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta(TrackedModel.Meta): ordering = ['-created_at'] indexes = [ models.Index(fields=['status', 'priority']), models.Index(fields=['reported_by']), models.Index(fields=['assigned_moderator']), models.Index(fields=['created_at']), ] def __str__(self): return f"{self.get_report_type_display()} report by {self.reported_by.username}" @pghistory.track() class ModerationQueue(TrackedModel): """ Model for managing moderation workflow and task assignment. This represents items in the moderation queue that need attention, separate from the initial reports. """ STATUS_CHOICES = [ ('PENDING', 'Pending'), ('IN_PROGRESS', 'In Progress'), ('COMPLETED', 'Completed'), ('CANCELLED', 'Cancelled'), ] PRIORITY_CHOICES = [ ('LOW', 'Low'), ('MEDIUM', 'Medium'), ('HIGH', 'High'), ('URGENT', 'Urgent'), ] ITEM_TYPE_CHOICES = [ ('CONTENT_REVIEW', 'Content Review'), ('USER_REVIEW', 'User Review'), ('BULK_ACTION', 'Bulk Action'), ('POLICY_VIOLATION', 'Policy Violation'), ('APPEAL', 'Appeal'), ('OTHER', 'Other'), ] # Queue item details item_type = models.CharField(max_length=50, choices=ITEM_TYPE_CHOICES) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING') priority = models.CharField( max_length=10, choices=PRIORITY_CHOICES, default='MEDIUM') title = models.CharField(max_length=200, help_text="Brief title for the queue item") description = models.TextField( help_text="Detailed description of what needs to be done") # What entity this relates to entity_type = models.CharField( max_length=50, blank=True, help_text="Type of entity (park, ride, user, etc.)") entity_id = models.PositiveIntegerField( null=True, blank=True, help_text="ID of the related entity") entity_preview = models.JSONField( default=dict, blank=True, help_text="Preview data for the entity") content_type = models.ForeignKey( ContentType, on_delete=models.CASCADE, null=True, blank=True) # Assignment and timing assigned_to = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name='assigned_queue_items' ) assigned_at = models.DateTimeField(null=True, blank=True) estimated_review_time = models.PositiveIntegerField( default=30, help_text="Estimated time in minutes") # Metadata flagged_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name='flagged_queue_items' ) tags = models.JSONField(default=list, blank=True, help_text="Tags for categorization") # Related objects related_report = models.ForeignKey( ModerationReport, on_delete=models.CASCADE, null=True, blank=True, related_name='queue_items' ) # Timestamps created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta(TrackedModel.Meta): ordering = ['priority', 'created_at'] indexes = [ models.Index(fields=['status', 'priority']), models.Index(fields=['assigned_to']), models.Index(fields=['created_at']), ] def __str__(self): return f"{self.get_item_type_display()}: {self.title}" @pghistory.track() class ModerationAction(TrackedModel): """ Model for tracking actions taken against users or content. This records what actions moderators have taken, including warnings, suspensions, content removal, etc. """ ACTION_TYPE_CHOICES = [ ('WARNING', 'Warning'), ('USER_SUSPENSION', 'User Suspension'), ('USER_BAN', 'User Ban'), ('CONTENT_REMOVAL', 'Content Removal'), ('CONTENT_EDIT', 'Content Edit'), ('CONTENT_RESTRICTION', 'Content Restriction'), ('ACCOUNT_RESTRICTION', 'Account Restriction'), ('OTHER', 'Other'), ] # Action details action_type = models.CharField(max_length=50, choices=ACTION_TYPE_CHOICES) reason = models.CharField(max_length=200, help_text="Brief reason for the action") details = models.TextField(help_text="Detailed explanation of the action") # Duration (for temporary actions) duration_hours = models.PositiveIntegerField( null=True, blank=True, help_text="Duration in hours for temporary actions" ) expires_at = models.DateTimeField( null=True, blank=True, help_text="When this action expires") is_active = models.BooleanField( default=True, help_text="Whether this action is currently active") # Users involved moderator = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='moderation_actions_taken' ) target_user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='moderation_actions_received' ) # Related objects related_report = models.ForeignKey( ModerationReport, on_delete=models.SET_NULL, null=True, blank=True, related_name='actions_taken' ) # Timestamps created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta(TrackedModel.Meta): ordering = ['-created_at'] indexes = [ models.Index(fields=['target_user', 'is_active']), models.Index(fields=['moderator']), models.Index(fields=['expires_at']), models.Index(fields=['created_at']), ] def __str__(self): return f"{self.get_action_type_display()} against {self.target_user.username} by {self.moderator.username}" def save(self, *args, **kwargs): # Set expiration time if duration is provided if self.duration_hours and not self.expires_at: self.expires_at = timezone.now() + timedelta(hours=self.duration_hours) super().save(*args, **kwargs) @pghistory.track() class BulkOperation(TrackedModel): """ Model for tracking bulk administrative operations. This handles large-scale operations like bulk updates, imports, exports, or mass moderation actions. """ STATUS_CHOICES = [ ('PENDING', 'Pending'), ('RUNNING', 'Running'), ('COMPLETED', 'Completed'), ('FAILED', 'Failed'), ('CANCELLED', 'Cancelled'), ] PRIORITY_CHOICES = [ ('LOW', 'Low'), ('MEDIUM', 'Medium'), ('HIGH', 'High'), ('URGENT', 'Urgent'), ] OPERATION_TYPE_CHOICES = [ ('UPDATE_PARKS', 'Update Parks'), ('UPDATE_RIDES', 'Update Rides'), ('IMPORT_DATA', 'Import Data'), ('EXPORT_DATA', 'Export Data'), ('MODERATE_CONTENT', 'Moderate Content'), ('USER_ACTIONS', 'User Actions'), ('CLEANUP', 'Cleanup'), ('OTHER', 'Other'), ] # Operation details operation_type = models.CharField(max_length=50, choices=OPERATION_TYPE_CHOICES) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING') priority = models.CharField( max_length=10, choices=PRIORITY_CHOICES, default='MEDIUM') description = models.TextField(help_text="Description of what this operation does") # Operation parameters and results parameters = models.JSONField( default=dict, help_text="Parameters for the operation") results = models.JSONField(default=dict, blank=True, help_text="Results and output from the operation") # Progress tracking total_items = models.PositiveIntegerField( default=0, help_text="Total number of items to process") processed_items = models.PositiveIntegerField( default=0, help_text="Number of items processed") failed_items = models.PositiveIntegerField( default=0, help_text="Number of items that failed") # Timing estimated_duration_minutes = models.PositiveIntegerField( null=True, blank=True, help_text="Estimated duration in minutes" ) schedule_for = models.DateTimeField( null=True, blank=True, help_text="When to run this operation") # Control can_cancel = models.BooleanField( default=True, help_text="Whether this operation can be cancelled") # User who created the operation created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='bulk_operations_created' ) # Timestamps created_at = models.DateTimeField(auto_now_add=True) started_at = models.DateTimeField(null=True, blank=True) completed_at = models.DateTimeField(null=True, blank=True) updated_at = models.DateTimeField(auto_now=True) class Meta(TrackedModel.Meta): ordering = ['-created_at'] indexes = [ models.Index(fields=['status', 'priority']), models.Index(fields=['created_by']), models.Index(fields=['schedule_for']), models.Index(fields=['created_at']), ] def __str__(self): return f"{self.get_operation_type_display()}: {self.description[:50]}" @property def progress_percentage(self): """Calculate progress percentage.""" if self.total_items == 0: return 0.0 return round((self.processed_items / self.total_items) * 100, 2) @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.ForeignKey( 'django_cloudflareimages_toolkit.CloudflareImage', on_delete=models.CASCADE, help_text="Photo submission stored on Cloudflare Images" ) 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(TrackedModel.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()