""" Core building blocks shared by every FablePool app: * ``UUIDModel`` — UUIDv4 primary keys everywhere (safe to expose publicly, merge-friendly for federated/imported content). * ``TimeStampedModel`` — created/updated audit timestamps. * ``SoftDeleteModel`` — tombstone pattern. ``Model.objects`` hides deleted rows; ``Model.all_objects`` sees everything; ``hard_delete()`` is explicit. * ``MediaAsset`` — S3-key-backed uploaded files with licensing/attribution. * ``AuditLog`` — append-only record of privileged actions (immutable rows). """ import uuid from django.conf import settings from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.contenttypes.models import ContentType from django.db import models from django.utils import timezone # --------------------------------------------------------------------------- # Abstract mixins # --------------------------------------------------------------------------- class UUIDModel(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) class Meta: abstract = True class TimeStampedModel(models.Model): created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: abstract = True class SoftDeleteQuerySet(models.QuerySet): def delete(self): """Bulk soft delete — stamps ``deleted_at`` instead of removing rows.""" return self.update(deleted_at=timezone.now()) def hard_delete(self): return super().delete() def alive(self): return self.filter(deleted_at__isnull=True) def dead(self): return self.filter(deleted_at__isnull=False) class SoftDeleteManager(models.Manager): """Default manager that hides soft-deleted rows.""" def get_queryset(self): return SoftDeleteQuerySet(self.model, using=self._db).filter( deleted_at__isnull=True ) class AllObjectsManager(models.Manager): def get_queryset(self): return SoftDeleteQuerySet(self.model, using=self._db) class SoftDeleteModel(models.Model): deleted_at = models.DateTimeField(null=True, blank=True, db_index=True) objects = SoftDeleteManager() all_objects = AllObjectsManager() class Meta: abstract = True @property def is_deleted(self) -> bool: return self.deleted_at is not None def delete(self, using=None, keep_parents=False): """Soft delete by default. Use ``hard_delete()`` to truly remove.""" self.deleted_at = timezone.now() self.save(update_fields=["deleted_at"]) def hard_delete(self, using=None, keep_parents=False): return super().delete(using=using, keep_parents=keep_parents) def restore(self): self.deleted_at = None self.save(update_fields=["deleted_at"]) # --------------------------------------------------------------------------- # MediaAsset # --------------------------------------------------------------------------- class MediaAsset(UUIDModel, TimeStampedModel, SoftDeleteModel): """An uploaded file stored in S3-compatible object storage. Only the storage key is persisted — never the bytes. Every asset carries license and attribution metadata because all published content must be redistributable under CC BY-SA. """ class Kind(models.TextChoices): IMAGE = "image", "Image" VIDEO = "video", "Video" AUDIO = "audio", "Audio" DOCUMENT = "document", "Document" DATASET = "dataset", "Dataset" WIDGET_BUNDLE = "widget_bundle", "Widget bundle" OTHER = "other", "Other" class ScanStatus(models.TextChoices): PENDING = "pending", "Pending scan" CLEAN = "clean", "Clean" FLAGGED = "flagged", "Flagged" uploaded_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, related_name="media_assets", ) kind = models.CharField(max_length=20, choices=Kind.choices, default=Kind.IMAGE) storage_key = models.CharField(max_length=512, unique=True) original_filename = models.CharField(max_length=255) mime_type = models.CharField(max_length=127) size_bytes = models.PositiveBigIntegerField() sha256 = models.CharField(max_length=64) alt_text = models.CharField( max_length=500, blank=True, help_text="Accessibility description; required by the API for images.", ) width = models.PositiveIntegerField(null=True, blank=True) height = models.PositiveIntegerField(null=True, blank=True) duration_seconds = models.FloatField(null=True, blank=True) license = models.CharField(max_length=64, default="CC-BY-SA-4.0") attribution = models.TextField( blank=True, help_text="Required when the uploader is not the author." ) scan_status = models.CharField( max_length=10, choices=ScanStatus.choices, default=ScanStatus.PENDING ) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["sha256"], name="media_sha256_idx"), models.Index(fields=["kind", "created_at"], name="media_kind_created_idx"), ] def __str__(self) -> str: return f"{self.get_kind_display()}: {self.original_filename}" # --------------------------------------------------------------------------- # AuditLog # --------------------------------------------------------------------------- class AuditLog(UUIDModel): """Append-only record of privileged actions. Rows are immutable: ``save()`` refuses updates and ``delete()`` raises. Use ``AuditLog.record(...)`` as the single write path. """ actor = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, related_name="audit_logs", ) action = models.CharField( max_length=100, db_index=True, help_text="Dotted action name, e.g. 'problem.publish' or 'user.role_grant'.", ) target_content_type = models.ForeignKey( ContentType, on_delete=models.SET_NULL, null=True, blank=True ) target_object_id = models.CharField(max_length=64, blank=True) target = GenericForeignKey("target_content_type", "target_object_id") summary = models.CharField(max_length=500, blank=True) metadata = models.JSONField(default=dict, blank=True) ip_address = models.GenericIPAddressField(null=True, blank=True) user_agent = models.CharField(max_length=500, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) class Meta: ordering = ["-created_at"] indexes = [ models.Index( fields=["target_content_type", "target_object_id"], name="audit_target_idx", ), ] def __str__(self) -> str: return f"{self.action} by {self.actor_id} at {self.created_at}" def save(self, *args, **kwargs): if not self._state.adding: raise ValueError("AuditLog entries are immutable and cannot be updated.") super().save(*args, **kwargs) def delete(self, *args, **kwargs): raise ValueError("AuditLog entries cannot be deleted.") @classmethod def record( cls, *, actor=None, action: str, target=None, summary: str = "", metadata: dict | None = None, ip_address: str | None = None, user_agent: str = "", ) -> "AuditLog": entry = cls( actor=actor, action=action, summary=summary, metadata=metadata or {}, ip_address=ip_address, user_agent=user_agent[:500], ) if target is not None: entry.target_content_type = ContentType.objects.get_for_model( type(target) ) entry.target_object_id = str(target.pk) entry.save() return entry