""" Module Core - Modèles de base et utilitaires Fournit les modèles de base avec audit trail pour tous les autres modules """ import uuid import os import logging import string import secrets from django.db import models from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.fields import GenericForeignKey from django.utils import timezone from django.core.mail import send_mail from django.template.loader import render_to_string from django.conf import settings def generate_nano_id(): """ Génère un Nano ID segmenté pour les modèles Format: xxxx-xxxx-xxxx (12 caractères + 2 tirets = 14 total) Utilise un alphabet sans caractères ambigus (pas de 0, o, 1, l) pour éviter la confusion lors de la saisie manuelle """ # Alphabet sans caractères ambigus alphabet = string.ascii_lowercase + string.digits alphabet = ( alphabet.replace("0", "").replace("o", "").replace("1", "").replace("l", "") ) # Générer 3 segments de 4 caractères chacun segments = [] for _ in range(3): segment = "".join(secrets.choice(alphabet) for _ in range(4)) segments.append(segment) return "-".join(segments) # ✅ IMPORT DU NOUVEAU SYSTÈME DE STOCKAGE ADAPTATIF from .storage_backend import ( create_file_field, create_image_field, create_video_field, get_document_upload_path, get_image_upload_path, get_video_upload_path, get_audio_upload_path, get_archive_upload_path, get_media_upload_path, get_storage_info, StorageBackendMixin, ) class ActiveManager(models.Manager): """Manager pour filtrer les objets actifs""" def get_queryset(self): return super().get_queryset().filter(is_active=True) class BaseModel(models.Model): """ Modèle de base avec audit trail complet Tous les modèles de l'application doivent hériter de ce modèle pour garantir la traçabilité et la cohérence """ # ✅ NOUVEAU : ID COURT NANO ID (12 caractères + tirets) # Format: v1st-gxr8-z5jh (14 caractères total) id = models.CharField( max_length=14, # 12 chars + 2 tirets primary_key=True, default=generate_nano_id, editable=False, help_text="Identifiant unique court (Nano ID segmenté)", ) # Audit temporel created_at = models.DateTimeField( auto_now_add=True, help_text="Date et heure de création" ) updated_at = models.DateTimeField( auto_now=True, help_text="Date et heure de dernière modification" ) # Audit utilisateur created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="%(class)s_created", null=True, blank=True, help_text="Utilisateur ayant créé cet enregistrement", ) updated_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="%(class)s_updated", null=True, blank=True, help_text="Utilisateur ayant modifié cet enregistrement", ) # Soft delete is_active = models.BooleanField( default=True, help_text="Indique si l'enregistrement est actif" ) # Managers objects = models.Manager() # Manager par défaut active_objects = ActiveManager() # Manager pour les objets actifs class Meta: abstract = True ordering = ["-created_at"] def soft_delete(self, user=None): """Suppression logique de l'enregistrement""" self.is_active = False self.updated_by = user self.save(update_fields=["is_active", "updated_by", "updated_at"]) def restore(self, user=None): """Restauration d'un enregistrement supprimé logiquement""" self.is_active = True self.updated_by = user self.save(update_fields=["is_active", "updated_by", "updated_at"]) class AuditLog(BaseModel): """ Log d'audit des opérations critiques Enregistre toutes les actions importantes dans le système pour la conformité et la traçabilité """ class ActionType(models.TextChoices): # Actions de base CREATE = "create", "Création" UPDATE = "update", "Modification" DELETE = "delete", "Suppression" LOGIN = "login", "Connexion" LOGOUT = "logout", "Déconnexion" TRANSACTION = "transaction", "Transaction" APPROVAL = "approval", "Approbation" REJECTION = "rejection", "Rejet" SUSPENSION = "suspension", "Suspension" ACTIVATION = "activation", "Activation" PASSWORD_CHANGE = "password_change", "Changement Mot de Passe" EXPORT = "export", "Export de Données" IMPORT = "import", "Import de Données" # Actions spécialisées rental RENTAL_ACTION = "rental_action", "Action de Location" # Actions spécialisées opérations ASSIGNED = "assigned", "Assignation" STARTED = "started", "Démarrage" COMPLETED = "completed", "Terminaison" CANCELLED = "cancelled", "Annulation" VALIDATION_REQUESTED = "validation_requested", "Demande de validation" APPROVED = "approved", "Approbation" REJECTED = "rejected", "Rejet" # Actions sur les pièces jointes ATTACHMENT_ADDED = "attachment_added", "Pièce jointe ajoutée" ATTACHMENT_REMOVED = "attachment_removed", "Pièce jointe supprimée" # Actions de communication COMMENT_ADDED = "comment_added", "Commentaire ajouté" NOTIFICATION_SENT = "notification_sent", "Notification envoyée" # Actions de planification SCHEDULED = "scheduled", "Planification" RESCHEDULED = "rescheduled", "Reprogrammation" # Actions budgétaires COST_UPDATED = "cost_updated", "Coût mis à jour" user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="audit_logs", null=True, # ✅ CORRECTION : Permettre NULL pour les actions système blank=True, help_text="Utilisateur ayant effectué l'action (NULL pour les actions système)", ) action_type = models.CharField( max_length=50, choices=ActionType.choices, help_text="Type d'action effectuée" ) # Objet concerné (Generic Foreign Key) object_type = models.ForeignKey( ContentType, on_delete=models.CASCADE, null=True, blank=True ) object_id = models.CharField(max_length=50, blank=True) content_object = GenericForeignKey("object_type", "object_id") description = models.TextField(help_text="Description détaillée de l'action") # Métadonnées ip_address = models.GenericIPAddressField( null=True, blank=True, help_text="Adresse IP de l'utilisateur" ) user_agent = models.TextField(blank=True, help_text="User Agent du navigateur") metadata = models.JSONField( default=dict, blank=True, help_text="Données supplémentaires (JSON)" ) class Meta: db_table = "audit_log" verbose_name = "5.2 - Log d'Audit" verbose_name_plural = "5.2 - Logs d'Audit" indexes = [ models.Index(fields=["user", "action_type"]), models.Index(fields=["object_type", "object_id"]), models.Index(fields=["created_at"]), models.Index(fields=["action_type", "created_at"]), ] def __str__(self): user_display = self.user.email if self.user else "SYSTEME" return f"{user_display} - {self.get_action_type_display()} - {self.created_at}" class SystemConfiguration(BaseModel): """ Configuration système globale Stocke les paramètres de configuration de l'application """ key = models.CharField( max_length=100, unique=True, help_text="Clé de configuration" ) value = models.TextField(help_text="Valeur de configuration") description = models.TextField(blank=True, help_text="Description de ce paramètre") is_sensitive = models.BooleanField( default=False, help_text="Valeur sensible (masquée dans l'admin)" ) is_system = models.BooleanField( default=False, help_text="Paramètre système (non modifiable via l'interface)" ) class Meta: db_table = "system_configuration" verbose_name = "5.1 - System Config." verbose_name_plural = "5.1 - System Config." ordering = ["key"] def __str__(self): return f"{self.key}" @classmethod def get_value(cls, key, default=None): """Récupère une valeur de configuration""" try: config = cls.objects.get(key=key, is_active=True) return config.value except cls.DoesNotExist: return default @classmethod def set_value(cls, key, value, description="", user=None): """Définit une valeur de configuration""" config, created = cls.objects.get_or_create( key=key, defaults={ "value": value, "description": description, "created_by": user, }, ) if not created: config.value = value config.updated_by = user config.save() return config class Notification(BaseModel): """ Système de notifications Gère les notifications internes du système """ class NotificationType(models.TextChoices): INFO = "info", "Information" WARNING = "warning", "Avertissement" ERROR = "error", "Erreur" SUCCESS = "success", "Succès" CRITICAL = "critical", "Critique" class Priority(models.TextChoices): LOW = "low", "Basse" NORMAL = "normal", "Normale" HIGH = "high", "Haute" URGENT = "urgent", "Urgente" recipient = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="notifications", help_text="Destinataire de la notification", ) title = models.CharField(max_length=200, help_text="Titre de la notification") message = models.TextField(help_text="Message de la notification") notification_type = models.CharField( max_length=10, choices=NotificationType.choices, default=NotificationType.INFO ) priority = models.CharField( max_length=10, choices=Priority.choices, default=Priority.NORMAL ) # État is_read = models.BooleanField(default=False, help_text="Notification lue") read_at = models.DateTimeField(null=True, blank=True, help_text="Date de lecture") # Actions action_url = models.URLField(blank=True, help_text="URL d'action associée") action_text = models.CharField( max_length=100, blank=True, help_text="Texte du bouton d'action" ) # Expiration expires_at = models.DateTimeField( null=True, blank=True, help_text="Date d'expiration" ) class Meta: db_table = "notifications" verbose_name = "5.3 - Notification" verbose_name_plural = "5.3 - Notifications" ordering = ["-created_at"] indexes = [ models.Index(fields=["recipient", "is_read"]), models.Index(fields=["created_at", "priority"]), models.Index(fields=["expires_at"]), ] def __str__(self): return f"{self.title} - {self.recipient.email}" def mark_as_read(self): """Marquer comme lue""" if not self.is_read: self.is_read = True self.read_at = timezone.now() self.save(update_fields=["is_read", "read_at"]) @property def is_expired(self): """Vérifie si la notification a expiré""" if self.expires_at: return timezone.now() > self.expires_at return False def send_email_notification(self): """Envoie un email de notification au destinataire""" try: if not self.recipient.email: logging.warning( f"Aucun email configuré pour l'utilisateur {self.recipient.email}" ) return False # Préparer le contexte pour le template context = { "notification": self, "recipient": self.recipient, "site_name": "COOP-CA AMIFOND", "current_year": timezone.now().year, "notification_url": self._get_notification_url(), } # Sujet de l'email selon le type de notification subject_prefix = self._get_email_subject_prefix() email_subject = f"[AMIFOND] {subject_prefix} {self.title}" # Contenu HTML de l'email html_content = self._generate_email_html_content(context) # Contenu texte de l'email (fallback) text_content = self._generate_email_text_content(context) # Envoi de l'email success = send_mail( subject=email_subject, message=text_content, from_email=getattr( settings, "DEFAULT_FROM_EMAIL", "noreply@amifond.com" ), recipient_list=[self.recipient.email], fail_silently=False, html_message=html_content, ) if success: logging.info( f"Email de notification envoyé à {self.recipient.email} pour: {self.title}" ) return True else: logging.error( f"Échec de l'envoi d'email à {self.recipient.email} pour: {self.title}" ) return False except Exception as e: logging.error(f"Erreur lors de l'envoi d'email de notification: {str(e)}") return False def _get_email_subject_prefix(self): """Retourne le préfixe du sujet selon le type de notification""" prefixes = { "INFO": "📋 Information", "WARNING": "⚠️ Avertissement", "ERROR": "❌ Erreur", "SUCCESS": "✅ Succès", "CRITICAL": "🚨 Critique", } return prefixes.get(self.notification_type, "📬 Notification") def _get_notification_url(self): """Génère l'URL complète de la notification""" base_url = getattr(settings, "BASE_URL", "http://localhost:8000") if self.action_url: return f"{base_url}{self.action_url}" return f"{base_url}/notifications/" def _generate_email_html_content(self, context): """Génère le contenu HTML de l'email""" return f""" Notification AMIFOND

🏦 COOP-CA AMIFOND

{self.get_notification_type_display()}

{self.title}

{self.message}
{f'{self.action_text}' if self.action_text else ''}

Destinataire: {self.recipient.get_full_name() or self.recipient.email}

Priorité: {self.get_priority_display()}

Date: {self.created_at.strftime('%d/%m/%Y à %H:%M')}

{f'

Expire le: {self.expires_at.strftime("%d/%m/%Y à %H:%M")}

' if self.expires_at else ''}

© {context['current_year']} COOP-CA AMIFOND - Système de gestion coopérative

Cette notification a été générée automatiquement. Ne pas répondre à cet email.

""" def _generate_email_text_content(self, context): """Génère le contenu texte de l'email (fallback)""" content = f""" 🏦 COOP-CA AMIFOND - Notification {self.get_notification_type_display().upper()}: {self.title} {self.message} Détails: - Destinataire: {self.recipient.get_full_name() or self.recipient.email} - Priorité: {self.get_priority_display()} - Date: {self.created_at.strftime('%d/%m/%Y à %H:%M')} """ if self.expires_at: content += f"- Expire le: {self.expires_at.strftime('%d/%m/%Y à %H:%M')}\n" if self.action_url and self.action_text: content += f"\nAction disponible: {self.action_text}\nLien: {context['notification_url']}\n" content += f""" --- © {context['current_year']} COOP-CA AMIFOND - Système de gestion coopérative Cette notification a été générée automatiquement. """ return content def _get_notification_color(self): """Retourne la couleur selon le type de notification""" colors = { "INFO": "#17a2b8", "WARNING": "#ffc107", "ERROR": "#dc3545", "SUCCESS": "#28a745", "CRITICAL": "#dc3545", } return colors.get(self.notification_type, "#17a2b8") def _get_priority_color(self): """Retourne la couleur selon la priorité""" colors = { "LOW": "#6c757d", "NORMAL": "#007bff", "HIGH": "#fd7e14", "URGENT": "#dc3545", } return colors.get(self.priority, "#007bff") def save(self, *args, **kwargs): """Override save pour envoyer automatiquement l'email""" from django.conf import settings is_new = self.pk is None super().save(*args, **kwargs) # Envoyer l'email seulement pour les nouvelles notifications ET si les emails ne sont pas désactivés if is_new and not getattr(settings, "DISABLE_EMAIL_NOTIFICATIONS", False): # Utiliser transaction.on_commit pour s'assurer que l'email # n'est envoyé qu'après le commit de la transaction from django.db import transaction transaction.on_commit(lambda: self.send_email_notification()) class Document(BaseModel, StorageBackendMixin): """Modèle pour les documents uploadés avec stockage adaptatif""" title = models.CharField( max_length=200, verbose_name="Titre", blank=True, help_text="Titre du document (généré automatiquement à partir du fichier si vide)", ) description = models.TextField(blank=True, verbose_name="Description") # ✅ NOUVEAU : Champ adaptatif selon la configuration file = create_file_field( upload_path=get_document_upload_path, resource_type="auto", folder_prefix="amifond/documents", ) class Meta: verbose_name = "5.4 - Document" verbose_name_plural = "5.4 - Documents" ordering = ["-created_at"] db_table = "core_document" def __str__(self): return self.title or "Document sans titre" def save(self, *args, **kwargs): """Génère automatiquement le titre si non fourni""" if not self.title and self.file: # Extraire le nom du fichier sans l'extension filename = os.path.basename(str(self.file)) name, _ = os.path.splitext(filename) # Nettoyer le nom du fichier pour un titre plus lisible self.title = name.replace("_", " ").replace("-", " ").title() super().save(*args, **kwargs) @property def file_url(self): """URL du fichier, compatible avec tous les environnements""" return self.get_file_url("file") class MediaFile(BaseModel, StorageBackendMixin): """Modèle générique pour tous types de fichiers media avec stockage adaptatif""" MEDIA_TYPES = [ ("document", "Document"), ("image", "Image"), ("video", "Vidéo"), ("audio", "Audio"), ("archive", "Archive"), ("other", "Autre"), ] title = models.CharField(max_length=200, verbose_name="Titre") description = models.TextField(blank=True, verbose_name="Description") media_type = models.CharField( max_length=20, choices=MEDIA_TYPES, default="document", verbose_name="Type de média", ) # ✅ NOUVEAU : Champ adaptatif selon la configuration file = create_file_field( upload_path=get_media_upload_path, resource_type="auto", folder_prefix="amifond/media", ) class Meta: verbose_name = "5.5 - Fichier média" verbose_name_plural = "5.5 - Fichiers média" ordering = ["-created_at"] db_table = "core_media_file" def __str__(self): return f"{self.title} ({self.get_media_type_display()})" @property def file_url(self): """URL du fichier, compatible avec tous les environnements""" return self.get_file_url("file") def get_file_extension(self): """Récupère l'extension du fichier""" try: return self.file.name.split(".")[-1].lower() except Exception: return "" def is_image(self): """Vérifie si le fichier est une image""" image_extensions = ["jpg", "jpeg", "png", "gif", "webp", "svg"] return self.get_file_extension() in image_extensions def is_video(self): """Vérifie si le fichier est une vidéo""" video_extensions = ["mp4", "avi", "mov", "wmv", "flv", "webm"] return self.get_file_extension() in video_extensions @property def thumbnail_url(self): """URL de la miniature selon l'environnement""" return self.get_thumbnail_url("file", 300, 200) @property def large_url(self): """URL de l'image en grande taille selon l'environnement""" return self.get_large_image_url("file", 1920, 1280) def get_file_size_formatted(self): """Taille du fichier formatée""" try: if hasattr(self.file, "size") and self.file.size: size = self.file.size for unit in ["B", "KB", "MB", "GB"]: if size < 1024.0: return f"{size:.1f} {unit}" size /= 1024.0 return f"{size:.1f} TB" except Exception: pass return "Taille inconnue"