import os import secrets import uuid from pdf2image import convert_from_path from PIL import Image, ExifTags from django.apps import apps from django.conf import settings from django.contrib.auth import get_user_model from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.contenttypes.models import ContentType from django.core.files.base import ContentFile from django.db import models, transaction from django.utils import timezone from django.utils.text import slugify from django.utils.translation import gettext_lazy as _ from common.models import Document as LegacyDocument from common.storages import PrivateMediaStorage private_storage = PrivateMediaStorage() User = get_user_model() class DocumentType(models.Model): code = models.SlugField(max_length=64, unique=True, verbose_name=_("Code")) label = models.CharField(max_length=255, verbose_name=_("Label")) class Meta: ordering = ["label"] verbose_name = _("Document type") verbose_name_plural = _("Document types") def __str__(self): return self.label class DocumentTag(models.Model): name_fr = models.CharField(max_length=100, unique=True, verbose_name=_("Name")) name_nl = models.CharField(max_length=100, unique=True, verbose_name=_("Name (NL)"), null=True, blank=True) slug = models.SlugField(max_length=120, unique=True, verbose_name=_("Slug")) order = models.PositiveSmallIntegerField( default=0, verbose_name=_("Display order"), help_text=_("Lower values appear first in the document viewer. Tags with the same value are sorted alphabetically.") ) is_unique = models.BooleanField( default=False, verbose_name=_("Unique tag"), help_text=_("If checked, uploading a document with this tag will create a new version of the existing document instead of a new document.") ) show_in_asset_detail = models.BooleanField( default=False, verbose_name=_("Show in asset detail"), help_text=_("If checked, documents with this tag will be displayed in the asset detail page.") ) thematics = models.ManyToManyField( "common.Thematic", blank=True, related_name="document_tags", verbose_name=_("Thematics"), help_text=_("Thematics where this tag is available.") ) related_content_types = models.ManyToManyField( "contenttypes.ContentType", blank=True, related_name="document_tags", verbose_name=_("Related models"), help_text=_( "When set, this tag will only appear for uploads on the selected model types. " "Leave empty to make it available for all models of the selected thematics." ), ) extra_params_schema = models.JSONField( default=list, blank=True, verbose_name=_("Extra parameters schema"), help_text=_( 'JSON list of extra parameter definitions shown in the upload modal when this tag is selected. ' 'Supported types: "text", "number", "textarea", "select" (requires "options" list). ' 'Example: [{"key": "cables", "label_fr": "Câbles concernés", "label_nl": "Betrokken kabels", ' '"type": "text", "required": false, "placeholder_fr": "Ex: C1, C2"}]' ), ) @property def name(self): lang = timezone.get_current_timezone_name() if lang.startswith("nl") and self.name_nl: return self.name_nl return self.name_fr class Meta: ordering = ["order", "name_fr"] verbose_name = _("Tag") verbose_name_plural = _("Tags") def __str__(self): return self.name def save(self, *args, **kwargs): if not self.slug: base_slug = slugify(self.name) or str(uuid.uuid4()) slug = base_slug counter = 1 while DocumentTag.objects.exclude(pk=self.pk).filter(slug=slug).exists(): slug = f"{base_slug}-{counter}" counter += 1 self.slug = slug super().save(*args, **kwargs) class DocumentFolder(models.Model): uuid = models.UUIDField(default=uuid.uuid4, unique=True, editable=False) name = models.CharField(max_length=255, verbose_name=_("Name")) slug = models.SlugField(max_length=150, unique=True, verbose_name=_("Slug")) description = models.TextField(blank=True, verbose_name=_("Description")) is_predefined = models.BooleanField(default=False, verbose_name=_("Predefined")) application_label = models.CharField( max_length=100, blank=True, verbose_name=_("Application label") ) thematics = models.ManyToManyField( "common.Thematic", blank=True, related_name="document_folders", verbose_name=_("Thematics"), ) created_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name="created_folders", verbose_name=_("Created by"), ) created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Created at")) updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated at")) parent_folders = models.ManyToManyField( "self", symmetrical=False, related_name="child_folders", blank=True, verbose_name=_("Parent folders"), ) class Meta: ordering = ["name"] verbose_name = _("Document folder") verbose_name_plural = _("Document folders") def __str__(self): return self.name def save(self, *args, **kwargs): if not self.slug: self.slug = slugify(self.name) super().save(*args, **kwargs) @property def storage_segment(self): return self.slug or slugify(self.name) class ManagedDocument(models.Model): PERMISSION_VIEW = "view" PERMISSION_COMMENT = "comment" PERMISSION_EDIT = "edit" PERMISSION_APPROVE = "approve" PERMISSION_CHOICES = ( (PERMISSION_VIEW, _("View")), (PERMISSION_COMMENT, _("Comment")), (PERMISSION_EDIT, _("Edit")), (PERMISSION_APPROVE, _("Approve")), ) uuid = models.UUIDField(default=uuid.uuid4, unique=True, editable=False) title = models.CharField(max_length=255, verbose_name=_("Title")) description = models.TextField(blank=True, verbose_name=_("Description")) created_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name="created_documents", verbose_name=_("Created by"), ) created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Created at")) updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated at")) current_version = models.ForeignKey( "documents.DocumentVersion", on_delete=models.SET_NULL, null=True, blank=True, related_name="current_for_documents", verbose_name=_("Current version"), ) is_archived = models.BooleanField(default=False, verbose_name=_("Archived")) unique_params = models.JSONField( default=dict, blank=True, verbose_name=_("Unique parameters"), help_text=_("Additional parameters to determine uniqueness for documents with unique tags (e.g., {'format': 'pdf', 'cables': 'C1,C2'})"), ) georeference_coords = models.JSONField( null=True, blank=True, verbose_name=_("Georeference Coordinates"), help_text=_("Coordinates of the 4 corners: [[lng, lat], [lng, lat], [lng, lat], [lng, lat]]"), ) georeference_page = models.IntegerField( default=1, verbose_name=_("Georeference Page"), help_text=_("Specific page of the PDF that has been georeferenced"), ) types = models.ManyToManyField(DocumentType, blank=True, related_name="documents") tags = models.ManyToManyField(DocumentTag, blank=True, related_name="documents") folders = models.ManyToManyField( DocumentFolder, related_name="documents", blank=True, verbose_name=_("Folders"), ) class Meta: ordering = ["-updated_at", "title"] verbose_name = _("Managed document") verbose_name_plural = _("Managed documents") def __str__(self): return self.title @property def latest_version(self): return self.current_version or self.versions.order_by("-version_number").first() def get_file_extension(self): """Retourne l'extension du fichier de la version actuelle (ex: .pdf, .docx)""" version = self.latest_version if version and version.filename: _, ext = os.path.splitext(version.filename) return ext return "" def touch(self, version): self.current_version = version self.updated_at = timezone.now() self.save(update_fields=["current_version", "updated_at"]) def get_absolute_url(self): from django.urls import reverse return reverse("documents:detail", args=[self.pk]) def build_public_url(self, token): base_url = getattr(settings, "ROOT_URL", "") return f"{base_url.rstrip('/')}{token}" if token.startswith("http") else token def get_primary_folder(self): default_folder = ( self.folders.filter(attachments__is_default=True) .order_by("name") .first() ) if default_folder: return default_folder return self.folders.order_by("name").first() def ensure_folder(self, folder): if folder and not self.folders.filter(pk=folder.pk).exists(): self.folders.add(folder) self.updated_at = timezone.now() self.save(update_fields=["updated_at"]) def document_version_upload_to(instance, filename): filename = os.path.basename(filename) document = getattr(instance, "document", None) folder_segment = None if document: folder = document.get_primary_folder() if folder: folder_segment = folder.storage_segment if not folder_segment: folder_segment = "unclassified" return f"document_versions/{folder_segment}/{timezone.now():%Y/%m}/{filename}" def document_thumbnail_upload_to(instance, filename): filename = os.path.basename(filename) document = getattr(instance, "document", None) folder_segment = None if document: folder = document.get_primary_folder() if folder: folder_segment = folder.storage_segment if not folder_segment: folder_segment = "unclassified" return f"document_versions/thumbnails/{folder_segment}/{timezone.now():%Y/%m}/{filename}" class DocumentVersion(LegacyDocument): STATUS_DRAFT = "draft" STATUS_PENDING = "pending" STATUS_APPROVED = "approved" STATUS_REJECTED = "rejected" STATUS_CHOICES = ( (STATUS_DRAFT, _("Draft")), (STATUS_PENDING, _("Pending validation")), (STATUS_APPROVED, _("Approved")), (STATUS_REJECTED, _("Rejected")), ) document = models.ForeignKey( "documents.ManagedDocument", related_name="versions", on_delete=models.CASCADE, verbose_name=_("Document"), ) version_number = models.PositiveIntegerField(verbose_name=_("Version")) file = models.FileField(upload_to=document_version_upload_to, max_length=255, storage=private_storage) thumbnail = models.ImageField(upload_to=document_thumbnail_upload_to, blank=True, null=True, max_length=255, storage=private_storage) uploaded_by = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=models.SET_NULL) description = models.TextField( null=True, blank=True, verbose_name=_("Description")) uploaded_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Uploaded at")) comment = models.TextField(blank=True, verbose_name=_("Comment")) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=STATUS_PENDING) validated_at = models.DateTimeField(null=True, blank=True, verbose_name=_("Validated at")) class Meta: ordering = ["-version_number"] unique_together = ("document", "version_number") verbose_name = _("Document version") verbose_name_plural = _("Document versions") def get_title(self): return getattr(self, "description", None) or self.filename def __str__(self): return f"{self.get_title()} v{self.version_number}" def generate_thumbnail(self): """Générer une vignette pour les fichiers PDF, DXF, DWG ou images.""" ext = os.path.splitext(self.file.name)[1].lower() if ext == ".pdf": try: pdf_path = self.file.path images = convert_from_path(pdf_path, first_page=1, last_page=1, size=(200, 200)) thumbnail_path = f"{self.file.path}_thumbnail.jpg" images[0].save(thumbnail_path, "JPEG") self.thumbnail.name = f"{self.file.name}_thumbnail.jpg" self.save() except Exception as e: print(f"Error generating thumbnail for {self.file.name}: {e}") elif ext in [".dxf", ".dwg"]: # Skip CAD thumbnail rendering to avoid slow CPU processing during save. # Explorer will fall back to default CAD/generic file icon. pass else: try: with Image.open(self.file.path) as img: # Conversion en mode RGB si nécessaire (évite les erreurs avec le format JPEG) if img.mode in ("RGBA", "P"): # "P" correspond aux images en palette (GIF) img = img.convert("RGB") img.thumbnail((200, 200)) thumbnail_path = f"{self.file.path}_thumbnail.jpg" img.save(thumbnail_path, "JPEG") self.thumbnail.name = f"{self.file.name}_thumbnail.jpg" self.save() except Exception as e: print(f"Error generating thumbnail for {self.file.name}: {e}") def save(self, *args, **kwargs): if not self.version_number: with transaction.atomic(): latest = ( self.__class__.objects.select_for_update() .filter(document=self.document) .order_by("-version_number") .first() ) self.version_number = 1 if latest is None else latest.version_number + 1 # Mettre à jour le titre du document parent avec le nom du fichier (sans extension) if self.file and self.document: filename = os.path.basename(self.file.name) title_without_ext = os.path.splitext(filename)[0] if title_without_ext and self.document.title != title_without_ext: self.document.title = title_without_ext self.document.save(update_fields=["title"]) super().save(*args, **kwargs) if self.document.current_version_id != self.pk: self.document.touch(self) if self.file: # Vérifier qu'un fichier est bien présent ext = os.path.splitext(self.file.name)[1].lower() if ext in ['.png', '.jpg', '.jpeg', '.gif', '.bmp', '.svg', '.webp']: try: with Image.open(self.file.path) as img: # Vérifier et corriger l'orientation EXIF si nécessaire try: exif = img._getexif() if exif: for tag, value in exif.items(): if tag in ExifTags.TAGS and ExifTags.TAGS[tag] == "Orientation": if value == 3: img = img.rotate(180, expand=True) elif value == 6: img = img.rotate(270, expand=True) elif value == 8: img = img.rotate(90, expand=True) except AttributeError: pass # Pas d'EXIF, on ignore l'orientation # Sauvegarde de l'image corrigée en écrasant l'originale img.save(self.file.path, format=img.format) except Exception as e: print(f"Erreur saving picture {self.file.name}: {e}") if not self.thumbnail: self.generate_thumbnail() class DocumentValidation(models.Model): STATUS_PENDING = "pending" STATUS_APPROVED = "approved" STATUS_REJECTED = "rejected" STATUS_CHOICES = ( (STATUS_PENDING, _("Pending")), (STATUS_APPROVED, _("Approved")), (STATUS_REJECTED, _("Rejected")), ) version = models.ForeignKey(DocumentVersion, related_name="validations", on_delete=models.CASCADE) role = models.ForeignKey("common.Role", related_name="document_validations", on_delete=models.CASCADE) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=STATUS_PENDING) validated_by = models.ForeignKey(User, null=True, blank=True, on_delete=models.SET_NULL, related_name="document_validations") validated_at = models.DateTimeField(null=True, blank=True) comment = models.TextField(blank=True) class Meta: unique_together = ("version", "role") verbose_name = _("Validation") verbose_name_plural = _("Validations") def __str__(self): return f"{self.role} - {self.get_status_display()}" def approve(self, user, comment=""): self.status = self.STATUS_APPROVED self.validated_by = user self.validated_at = timezone.now() self.comment = comment self.save(update_fields=["status", "validated_by", "validated_at", "comment"]) def reject(self, user, comment=""): self.status = self.STATUS_REJECTED self.validated_by = user self.validated_at = timezone.now() self.comment = comment self.save(update_fields=["status", "validated_by", "validated_at", "comment"]) class DocumentShare(models.Model): document = models.ForeignKey(ManagedDocument, related_name="shares", on_delete=models.CASCADE) user = models.ForeignKey(User, related_name="document_shares", on_delete=models.CASCADE) permission = models.CharField(max_length=20, choices=ManagedDocument.PERMISSION_CHOICES, default=ManagedDocument.PERMISSION_VIEW) created_at = models.DateTimeField(auto_now_add=True) created_by = models.ForeignKey(User, null=True, blank=True, on_delete=models.SET_NULL, related_name="created_document_shares") class Meta: unique_together = ("document", "user") verbose_name = _("Share") verbose_name_plural = _("Shares") def __str__(self): return f"{self.user} - {self.document} ({self.get_permission_display()})" class DocumentFolderShare(models.Model): folder = models.ForeignKey(DocumentFolder, related_name="shares", on_delete=models.CASCADE) user = models.ForeignKey(User, related_name="document_folder_shares", on_delete=models.CASCADE) permission = models.CharField( max_length=20, choices=ManagedDocument.PERMISSION_CHOICES, default=ManagedDocument.PERMISSION_VIEW, ) created_at = models.DateTimeField(auto_now_add=True) created_by = models.ForeignKey( User, null=True, blank=True, on_delete=models.SET_NULL, related_name="created_document_folder_shares", ) class Meta: unique_together = ("folder", "user") verbose_name = _("Folder share") verbose_name_plural = _("Folder shares") def __str__(self): return f"{self.user} - {self.folder} ({self.get_permission_display()})" class DocumentFolderAccessLink(models.Model): folder = models.ForeignKey( DocumentFolder, related_name="access_links", on_delete=models.CASCADE, verbose_name=_("Folder"), ) token = models.CharField(max_length=64, unique=True, editable=False) permission = models.CharField( max_length=20, choices=ManagedDocument.PERMISSION_CHOICES, default=ManagedDocument.PERMISSION_VIEW, ) created_at = models.DateTimeField(auto_now_add=True) created_by = models.ForeignKey( User, null=True, blank=True, on_delete=models.SET_NULL, related_name="created_document_folder_links", ) expires_at = models.DateTimeField(null=True, blank=True) class Meta: verbose_name = _("Folder access link") verbose_name_plural = _("Folder access links") def __str__(self): return f"{self.folder} ({self.get_permission_display()})" def save(self, *args, **kwargs): if not self.token: self.token = secrets.token_urlsafe(32) super().save(*args, **kwargs) @property def is_expired(self): return self.expires_at and timezone.now() > self.expires_at class DocumentAccessLink(models.Model): document = models.ForeignKey(ManagedDocument, related_name="access_links", on_delete=models.CASCADE) token = models.CharField(max_length=64, unique=True, editable=False) permission = models.CharField(max_length=20, choices=ManagedDocument.PERMISSION_CHOICES, default=ManagedDocument.PERMISSION_VIEW) created_at = models.DateTimeField(auto_now_add=True) created_by = models.ForeignKey(User, null=True, blank=True, on_delete=models.SET_NULL, related_name="created_document_links") expires_at = models.DateTimeField(null=True, blank=True) class Meta: verbose_name = _("Access link") verbose_name_plural = _("Access links") def __str__(self): return f"{self.document} ({self.get_permission_display()})" def save(self, *args, **kwargs): if not self.token: self.token = secrets.token_urlsafe(32) super().save(*args, **kwargs) @property def is_expired(self): return self.expires_at and timezone.now() > self.expires_at class DocumentAttachment(models.Model): document = models.ForeignKey(ManagedDocument, related_name="attachments", on_delete=models.CASCADE) content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.PositiveIntegerField() content_object = GenericForeignKey("content_type", "object_id") attached_at = models.DateTimeField(auto_now_add=True) attached_by = models.ForeignKey(User, null=True, blank=True, on_delete=models.SET_NULL, related_name="document_attachments") note = models.CharField(max_length=255, blank=True) class Meta: unique_together = ("document", "content_type", "object_id") verbose_name = _("Attachment") verbose_name_plural = _("Attachments") def __str__(self): return f"{self.document} -> {self.content_object}" def save(self, *args, **kwargs): super().save(*args, **kwargs) self.apply_default_folder() def apply_default_folder(self): DocumentFolderAttachment = apps.get_model("documents", "DocumentFolderAttachment") default_folder_link = ( DocumentFolderAttachment.objects.select_related("folder") .filter( content_type=self.content_type, object_id=self.object_id, is_default=True, ) .first() ) if default_folder_link: self.document.ensure_folder(default_folder_link.folder) class DocumentFolderAttachment(models.Model): folder = models.ForeignKey(DocumentFolder, related_name="attachments", on_delete=models.CASCADE) content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.PositiveIntegerField() content_object = GenericForeignKey("content_type", "object_id") is_default = models.BooleanField(default=False, verbose_name=_("Default folder")) attached_at = models.DateTimeField(auto_now_add=True) attached_by = models.ForeignKey( User, null=True, blank=True, on_delete=models.SET_NULL, related_name="document_folder_attachments", ) class Meta: verbose_name = _("Folder attachment") verbose_name_plural = _("Folder attachments") constraints = [ models.UniqueConstraint( fields=["content_type", "object_id"], condition=models.Q(is_default=True), name="documents_unique_default_folder_per_object", ) ] unique_together = ("folder", "content_type", "object_id") def __str__(self): return f"{self.folder} -> {self.content_object}" def save(self, *args, **kwargs): super().save(*args, **kwargs) if self.is_default: # ensure default folder is applied to all attached documents DocumentAttachment = apps.get_model("documents", "DocumentAttachment") attachments = DocumentAttachment.objects.filter( content_type=self.content_type, object_id=self.object_id ).select_related("document") for attachment in attachments: attachment.document.ensure_folder(self.folder) def duplicate_file(field): if not field: return None file_name = os.path.basename(field.name) with field.open("rb") as f: return ContentFile(f.read(), name=file_name) def ensure_document_for_legacy(instance, old_file_name=None): """Synchronise a legacy :class:`common.models.Document` with the new document management models. The function is intentionally idempotent so that calling it on every save only creates new versions when a file has been replaced. """ if not instance.file: return ManagedDocument = apps.get_model("documents", "ManagedDocument") DocumentVersion = apps.get_model("documents", "DocumentVersion") DocumentAttachment = apps.get_model("documents", "DocumentAttachment") DocumentType = apps.get_model("documents", "DocumentType") managed = instance.managed_document context_object = instance.get_document_context() title = instance.get_document_title() if managed is None: managed = ManagedDocument.objects.create( title=title, description=getattr(instance, "description", "") or "", created_by=getattr(instance, "uploaded_by", None), ) instance.__class__.objects.filter(pk=instance.pk).update(managed_document=managed) instance.managed_document = managed if context_object is not None: DocumentAttachment.objects.get_or_create( document=managed, content_type=ContentType.objects.get_for_model(context_object), object_id=context_object.pk, defaults={"attached_by": getattr(instance, "uploaded_by", None)}, ) if hasattr(instance, "document_type") and instance.document_type: label = instance.document_type if hasattr(instance, "get_document_type_display"): label = instance.get_document_type_display() doc_type, _created = DocumentType.objects.get_or_create( code=slugify(instance.document_type), defaults={"label": label}, ) managed.types.add(doc_type) file_changed = bool(old_file_name and old_file_name != instance.file.name) should_create_version = not managed.versions.exists() or file_changed if should_create_version: new_file = duplicate_file(instance.file) thumbnail_file = duplicate_file(getattr(instance, "thumbnail", None)) version = DocumentVersion( document=managed, file=new_file, thumbnail=thumbnail_file, uploaded_by=getattr(instance, "uploaded_by", None), description=getattr(instance, "description", "") or "", comment=_("Version synchronised from legacy module"), status=DocumentVersion.STATUS_APPROVED, validated_at=timezone.now(), ) version.save() elif managed.current_version is None: managed.touch(managed.versions.order_by("-version_number").first()) LegacyDocument.ensure_document_for_legacy = ensure_document_for_legacy