1291 lines
55 KiB
Python
1291 lines
55 KiB
Python
from django.contrib.gis.db import models
|
||
from django.contrib.auth.models import User
|
||
from django.utils.translation import get_language, gettext_lazy as _
|
||
from django.contrib.postgres.indexes import GinIndex, OpClass
|
||
from django.contrib.contenttypes.fields import GenericForeignKey
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from django.contrib.postgres.fields import ArrayField
|
||
from django.db.models.functions import Lower
|
||
from django.conf import settings
|
||
|
||
import os
|
||
from PIL import Image, ExifTags
|
||
from pdf2image import convert_from_path
|
||
|
||
|
||
class LanguageManager(models.Manager):
|
||
def get_queryset(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return super().get_queryset().order_by('name_nl')
|
||
return super().get_queryset().order_by('name_fr')
|
||
|
||
class Thematic(models.Model):
|
||
code = models.CharField(max_length=100, unique=True, verbose_name="Code")
|
||
name_fr = models.CharField(max_length=255, verbose_name="Nom")
|
||
name_nl = models.CharField(max_length=255, verbose_name="Naam")
|
||
description = models.TextField(blank=True, null=True, verbose_name="Description")
|
||
icon = models.CharField(max_length=100, blank=True, null=True, verbose_name="Icône Bootstrap")
|
||
color = models.CharField(max_length=7, default="#000000", verbose_name="Couleur associée")
|
||
uses_occupations = models.BooleanField(default=False, verbose_name="Utilise les occupations")
|
||
include_in_daily_report = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Inclure dans rapport journalier",
|
||
help_text="Si coché, cette thématique apparaît toujours dans le rapport journalier, même si le nombre d'interventions est 0.",
|
||
)
|
||
allow_planning_without_contract = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Planification sans contrat autorisée",
|
||
help_text="Si coché, les interventions de cette thématique peuvent passer au statut 'À planifier' sans qu'un prestataire ou un contrat soit renseigné. Le prestataire et le contrat restent modifiables dans ce statut.",
|
||
)
|
||
|
||
@property
|
||
def name(self):
|
||
"""Renvoie le nom de la thématique dans la langue active."""
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl
|
||
return self.name_fr
|
||
|
||
objects = LanguageManager()
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
class Meta:
|
||
verbose_name = "Thématique"
|
||
verbose_name_plural = "Thématiques"
|
||
|
||
|
||
class Role(models.Model):
|
||
name = models.CharField(max_length=50, unique=True) # ex: 'manager', 'technician', 'admin'
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
|
||
class AppView(models.Model):
|
||
"""
|
||
Defines available application views/modules.
|
||
This allows for a flexible permission system for different views.
|
||
"""
|
||
code = models.CharField(max_length=50, unique=True, verbose_name="Code")
|
||
name_fr = models.CharField(max_length=100, verbose_name="Nom")
|
||
name_nl = models.CharField(max_length=100, verbose_name="Naam")
|
||
icon = models.CharField(max_length=50, blank=True, null=True, verbose_name="Icône Bootstrap")
|
||
url_name = models.CharField(max_length=100, verbose_name="Nom de l'URL", help_text="Ex: 'home:index' ou 'sign:index'")
|
||
is_default = models.BooleanField(default=False, verbose_name="Vue par défaut", help_text="Vue affichée par défaut pour les nouveaux utilisateurs")
|
||
order = models.PositiveIntegerField(default=0, verbose_name="Ordre d'affichage")
|
||
|
||
@property
|
||
def name(self):
|
||
"""Renvoie le nom dans la langue active."""
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl
|
||
return self.name_fr
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
class Meta:
|
||
verbose_name = "Vue applicative"
|
||
verbose_name_plural = "Vues applicatives"
|
||
ordering = ['order', 'name_fr']
|
||
|
||
class UserConfig(models.Model):
|
||
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="config")
|
||
is_intern = models.BooleanField(default=False)
|
||
roles = models.ManyToManyField(Role, blank=True)
|
||
default_thematic = models.ForeignKey(Thematic, on_delete=models.SET_NULL, null=True, blank=True, related_name="default_user_accesses")
|
||
default_contract = models.ForeignKey("contracts.Contract", on_delete=models.SET_NULL, null=True, blank=True, related_name="default_user_accesses", verbose_name="Contrat par défaut", help_text="Contrat assigné par défaut lors de la création d'une intervention si aucun contrat n'est résolu automatiquement.")
|
||
thematics = models.ManyToManyField("common.Thematic", through="UserThematics", related_name="user_accesses")
|
||
contracts = models.ManyToManyField("contracts.Contract", through="UserContractAccess", related_name="user_accesses")
|
||
warehouses = models.ManyToManyField("stock.Warehouse", through="UserWarehouseAccess", related_name="user_accesses", blank=True)
|
||
warehouse_locations = models.ManyToManyField("stock.WarehouseLocation", through="UserWarehouseLocationAccess", related_name="user_configs", blank=True)
|
||
limit_assets_to_contracts = models.BooleanField(default=True)
|
||
limit_interventions_to_contracts = models.BooleanField(default=True)
|
||
limit_shortcuts_to_contracts = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Limiter la page d'accueil aux contrats",
|
||
help_text="Limite les raccourcis et les compteurs de la page d'accueil aux contrats pour lesquels l'utilisateur dispose de permissions explicites.",
|
||
)
|
||
generic_email = models.EmailField(blank=True, null=True) # Email générique de l'utilisateur
|
||
REGIONAL_STANDARD_CHOICES = [
|
||
('BE', 'Belgique (PEB, Maison Communale / Hôtel de Ville, Sécurité PMR)'),
|
||
('FR', 'France (DPE, GES, Mairie, Normes ERP)'),
|
||
('INT', 'International / Générique'),
|
||
]
|
||
regional_standard = models.CharField(
|
||
max_length=10,
|
||
choices=REGIONAL_STANDARD_CHOICES,
|
||
default='BE',
|
||
verbose_name="Norme régionale & vocabulaire",
|
||
help_text="Définit la terminologie et les normes réglementaires adaptées à la région.",
|
||
)
|
||
hide_intervention_map_by_default = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Masquer la carte des interventions par défaut",
|
||
help_text="Affiche un bouton permettant d'afficher la carte au besoin.",
|
||
)
|
||
INTERVENTION_LIST_LAYOUT_CHOICES = [
|
||
('both', 'Tableau et carte'),
|
||
('table_only', 'Tableau seul'),
|
||
('map_only', 'Carte seule'),
|
||
]
|
||
intervention_list_layout = models.CharField(
|
||
max_length=20,
|
||
choices=INTERVENTION_LIST_LAYOUT_CHOICES,
|
||
default='both',
|
||
verbose_name="Disposition de la liste des interventions",
|
||
help_text="Préférence de disposition pour la vue liste des interventions.",
|
||
)
|
||
intervention_list_filters = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Filtres de la liste des interventions",
|
||
help_text="Préférences de filtrage pour la vue liste des interventions.",
|
||
)
|
||
mobile_intervention_list_filters = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Filtres de la liste des interventions mobile",
|
||
help_text="Préférences de filtrage pour la vue mobile de la liste des interventions.",
|
||
)
|
||
intervention_list_sort = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Tri de la liste des interventions",
|
||
help_text="Préférences de tri pour la vue liste des interventions.",
|
||
)
|
||
PREFERRED_LANGUAGE_CHOICES = [
|
||
('fr', _('Français')),
|
||
('nl', _('Nederlands')),
|
||
('en', _('English')),
|
||
]
|
||
preferred_language = models.CharField(
|
||
max_length=5,
|
||
choices=PREFERRED_LANGUAGE_CHOICES,
|
||
default='fr',
|
||
verbose_name=_("Langue préférée"),
|
||
help_text=_("Langue utilisée pour l'interface, les e-mails et les notifications."),
|
||
)
|
||
notification_reminder_enabled = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Activer les rappels par e-mail"),
|
||
help_text=_("Recevoir un e-mail de rappel pour les notifications non lues après un certain délai."),
|
||
)
|
||
notification_reminder_emails = models.CharField(
|
||
max_length=500,
|
||
blank=True,
|
||
verbose_name=_("E-mail(s) de réception des rappels"),
|
||
help_text=_("Adresses e-mail séparées par des virgules. Si vide, l'adresse de votre compte sera utilisée."),
|
||
)
|
||
notification_reminder_delay_hours = models.PositiveIntegerField(
|
||
default=24,
|
||
verbose_name=_("Délai avant rappel (en heures)"),
|
||
help_text=_("Nombre d'heures écoulées après la réception d'une notification avant l'envoi d'un rappel par e-mail."),
|
||
)
|
||
intervention_list_columns = models.JSONField(
|
||
default=None,
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Colonnes masquées de la liste des interventions",
|
||
help_text="Liste des champs masqués dans le tableau de la vue liste des interventions.",
|
||
)
|
||
observation_list_filters = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Filtres de la liste des observations",
|
||
help_text="Préférences de filtrage pour la vue liste des observations.",
|
||
)
|
||
observation_list_sort = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Tri de la liste des observations",
|
||
help_text="Préférences de tri pour la vue liste des observations.",
|
||
)
|
||
project_list_filters = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Filtres de la liste des projets",
|
||
help_text="Préférences de filtrage pour la vue liste des projets.",
|
||
)
|
||
project_list_sort = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Tri de la liste des projets",
|
||
help_text="Préférences de tri pour la vue liste des projets.",
|
||
)
|
||
control_list_filters = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Filtres de la liste des contrôles",
|
||
help_text="Préférences de filtrage pour la vue liste des contrôles.",
|
||
)
|
||
control_list_sort = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Tri de la liste des contrôles",
|
||
help_text="Préférences de tri pour la vue liste des contrôles.",
|
||
)
|
||
ASSETS_LIST_LAYOUT_CHOICES = [
|
||
('both', 'Tableau et carte'),
|
||
('table_only', 'Tableau seul'),
|
||
('map_only', 'Carte seule'),
|
||
]
|
||
assets_list_layout = models.CharField(
|
||
max_length=20,
|
||
choices=ASSETS_LIST_LAYOUT_CHOICES,
|
||
default='both',
|
||
verbose_name="Disposition de la liste des assets",
|
||
help_text="Préférence de disposition pour la vue liste des assets.",
|
||
)
|
||
linked_configs = models.ManyToManyField(
|
||
"self",
|
||
blank=True,
|
||
symmetrical=False,
|
||
verbose_name="Comptes liés",
|
||
)
|
||
auto_logout_hours = models.PositiveIntegerField(
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Déconnexion automatique (heures)",
|
||
help_text="Nombre d'heures après la connexion avant déconnexion automatique. Laisser vide pour désactiver.",
|
||
)
|
||
# Generic view access permissions
|
||
accessible_views = models.ManyToManyField(
|
||
AppView,
|
||
blank=True,
|
||
verbose_name="Vues accessibles",
|
||
help_text="Les vues/applications auxquelles l'utilisateur a accès.",
|
||
related_name="user_configs",
|
||
)
|
||
panoramax_can_upload = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut uploader des photos 360",
|
||
help_text="Permet à l'utilisateur d'uploader de nouvelles photos ou vidéos 360 pour Panoramax."
|
||
)
|
||
# Société à laquelle appartient cet utilisateur (ex: prestataire)
|
||
company = models.ForeignKey(
|
||
'contracts.Company',
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='user_configs',
|
||
verbose_name="Société",
|
||
help_text="Société/organisme auquel appartient cet utilisateur (utile pour filtrer les membres visibles).",
|
||
)
|
||
# Document management permissions for external users
|
||
can_add_unique_tag_documents = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut ajouter des documents avec tags uniques",
|
||
help_text="Permet aux utilisateurs externes d'ajouter/modifier des documents avec des tags uniques (CME, PIS, As-Built, etc.).",
|
||
)
|
||
esign_can_edit = models.BooleanField(
|
||
default=False,
|
||
verbose_name="eSign : Peut éditer",
|
||
help_text="Permission pour éditer les panneaux et les poteaux.",
|
||
)
|
||
esign_can_manage_rcp = models.BooleanField(
|
||
default=False,
|
||
verbose_name="eSign : Peut gérer les RCP",
|
||
help_text="Permission pour créer, modifier et supprimer les RCP.",
|
||
)
|
||
esign_can_validate_rcp = models.BooleanField(
|
||
default=False,
|
||
verbose_name="eSign : Peut valider les RCP",
|
||
help_text="Permission pour valider les RCP.",
|
||
)
|
||
esign_organizations = models.ManyToManyField(
|
||
"sign.Organization",
|
||
blank=True,
|
||
related_name="user_configs",
|
||
verbose_name="eSign : Gestionnaires",
|
||
help_text="Les gestionnaires (communes, Bruxelles Mobilité) pour lesquels l'utilisateur a des permissions."
|
||
)
|
||
# Catégories d'assets concernées pour l'approbation des interventions
|
||
approval_categories = models.ManyToManyField(
|
||
'assets.AssetCategory',
|
||
blank=True,
|
||
verbose_name="Catégories d'approbation",
|
||
help_text="Catégories d'assets pour lesquelles l'utilisateur approuve des interventions. Si vide, aucun filtre de catégorie n'est appliqué.",
|
||
related_name='approver_user_configs',
|
||
)
|
||
# Préférences de filtrage du reporting (structurées par page/widget)
|
||
# Ex : {"dashboard": {"granularity": "week", "contract_ids": [], "thematic_ids": [], ...}}
|
||
reporting_preferences = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Préférences du reporting",
|
||
help_text="Préférences de filtrage pour les différentes pages du reporting.",
|
||
)
|
||
assets_index_preferences = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name="Préférences de la page index unifiée",
|
||
help_text="Préférences de la page assets (thématiques actives, catégories, filtres, position carte).",
|
||
)
|
||
calendar_preferences = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
null=True,
|
||
verbose_name="Préférences du calendrier",
|
||
help_text="Préférences de la page calendrier (mode d'affichage, zoom, filtres actifs, état d'expansion du backlog).",
|
||
)
|
||
|
||
def __str__(self):
|
||
return f"Configuration de {self.user.username}"
|
||
|
||
def has_role(self, role_str):
|
||
return self.roles.filter(name=role_str).exists()
|
||
|
||
@property
|
||
def can_approve_any_intervention(self):
|
||
"""Détermine si l'utilisateur dispose de permissions d'approbation sur au moins un contrat ou un rôle supérieur."""
|
||
if self.has_role('admin') or self.has_role('top_manager') or self.user.is_superuser:
|
||
return True
|
||
return self.usercontractaccess_set.filter(can_approve=True).exists()
|
||
|
||
def can_access_view(self, view_code):
|
||
"""Check if the user has access to a specific view by its code."""
|
||
return self.accessible_views.filter(code=view_code).exists()
|
||
|
||
def get_accessible_views(self):
|
||
"""Returns all accessible views ordered by their display order."""
|
||
return self.accessible_views.all().order_by('order')
|
||
|
||
def can_view_warehouse(self, warehouse):
|
||
"""Check if the user can view a specific warehouse."""
|
||
if self.user.is_superuser or self.user.is_staff or self.has_role('admin'):
|
||
return True
|
||
if not self.userwarehouses.exists():
|
||
if self.userlocations.exists():
|
||
return self.userlocations.filter(warehouse_location__warehouse=warehouse).exists()
|
||
return True
|
||
return self.userwarehouses.filter(warehouse=warehouse, can_view=True).exists()
|
||
|
||
def can_manage_warehouse(self, warehouse):
|
||
"""Check if the user can manage a specific warehouse."""
|
||
if self.user.is_superuser or self.user.is_staff or self.has_role('admin'):
|
||
return True
|
||
if not self.userwarehouses.exists():
|
||
if self.userlocations.exists():
|
||
return False
|
||
return True
|
||
return self.userwarehouses.filter(warehouse=warehouse, can_manage=True).exists()
|
||
|
||
def can_manage_preparation_orders_any(self):
|
||
"""Vérifie si l'utilisateur peut gérer des bons de préparation sur au moins un entrepôt."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
if not self.userwarehouses.exists():
|
||
if self.userlocations.exists():
|
||
return False
|
||
return True
|
||
return self.userwarehouses.filter(can_manage_preparation_orders=True).exists()
|
||
|
||
def can_manage_preparation_orders_for(self, warehouse):
|
||
"""Vérifie si l'utilisateur peut gérer des bons de préparation pour un entrepôt spécifique."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
if not self.userwarehouses.exists():
|
||
if self.userlocations.exists():
|
||
return False
|
||
return True
|
||
return self.userwarehouses.filter(warehouse=warehouse, can_manage_preparation_orders=True).exists()
|
||
|
||
def can_manage_purchase_orders_any(self):
|
||
"""Vérifie si l'utilisateur peut gérer des bons de commande sur au moins un entrepôt."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
if not self.userwarehouses.exists():
|
||
if self.userlocations.exists():
|
||
return False
|
||
return True
|
||
return self.userwarehouses.filter(can_manage_purchase_orders=True).exists()
|
||
|
||
def can_manage_purchase_orders_for(self, warehouse):
|
||
"""Vérifie si l'utilisateur peut gérer des bons de commande pour un entrepôt spécifique."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
if not self.userwarehouses.exists():
|
||
if self.userlocations.exists():
|
||
return False
|
||
return True
|
||
return self.userwarehouses.filter(warehouse=warehouse, can_manage_purchase_orders=True).exists()
|
||
|
||
def can_validate_inventory_for(self, warehouse):
|
||
"""Vérifie si l'utilisateur peut valider les inventaires d'un entrepôt."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
if not self.userwarehouses.exists():
|
||
if self.userlocations.exists():
|
||
return False
|
||
return True
|
||
return self.userwarehouses.filter(warehouse=warehouse, can_validate_inventory=True).exists()
|
||
|
||
def can_perform_inventory_on(self, location):
|
||
"""Vérifie si l'utilisateur peut effectuer l'inventaire d'un emplacement spécifique."""
|
||
if self.user.is_superuser or self.user.is_staff or self.has_role('admin'):
|
||
return True
|
||
if self.can_manage_warehouse(location.warehouse) or self.can_validate_inventory_for(location.warehouse):
|
||
return True
|
||
return self.userlocations.filter(warehouse_location=location, can_perform_inventory=True).exists()
|
||
|
||
def can_validate_inventory_on(self, location):
|
||
"""Vérifie si l'utilisateur peut valider les inventaires d'un emplacement spécifique."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
if self.can_validate_inventory_for(location.warehouse):
|
||
return True
|
||
return self.userlocations.filter(warehouse_location=location, can_validate_inventory=True).exists()
|
||
|
||
def get_accessible_warehouses(self, can_manage=False):
|
||
"""Returns all warehouses the user can access."""
|
||
from stock.models import Warehouse
|
||
if self.user.is_superuser or self.user.is_staff or self.has_role('admin'):
|
||
return Warehouse.objects.all()
|
||
|
||
if not self.userwarehouses.exists():
|
||
if self.userlocations.exists():
|
||
if can_manage:
|
||
from django.db.models import Q
|
||
return Warehouse.objects.filter(
|
||
Q(locations__user_accesses__user_config=self, locations__user_accesses__can_perform_inventory=True) |
|
||
Q(locations__user_accesses__user_config=self, locations__user_accesses__can_validate_inventory=True)
|
||
).distinct()
|
||
return Warehouse.objects.filter(
|
||
locations__user_accesses__user_config=self
|
||
).distinct()
|
||
return Warehouse.objects.all()
|
||
|
||
if can_manage:
|
||
return Warehouse.objects.filter(
|
||
userwarehouseaccess__user_config=self,
|
||
userwarehouseaccess__can_manage=True
|
||
).distinct()
|
||
|
||
return Warehouse.objects.filter(
|
||
userwarehouseaccess__user_config=self,
|
||
userwarehouseaccess__can_view=True
|
||
).distinct()
|
||
|
||
def get_default_warehouse(self):
|
||
"""Returns the user's default warehouse if set."""
|
||
default_access = self.userwarehouses.filter(is_default=True).first()
|
||
return default_access.warehouse if default_access else None
|
||
|
||
def can_administer_warehouse(self, warehouse):
|
||
"""Check if the user can administer a specific warehouse."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
return self.userwarehouses.filter(warehouse=warehouse, can_administer=True).exists()
|
||
|
||
def can_administer_any_warehouse(self):
|
||
"""Check if the user can administer at least one warehouse."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
return self.userwarehouses.filter(can_administer=True).exists()
|
||
|
||
def get_viewable_assets_thematics(self):
|
||
"""Retourne le QuerySet des thématiques pour lesquelles l'utilisateur est autorisé à voir les assets."""
|
||
from common.models import Thematic
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return Thematic.objects.all()
|
||
|
||
if not self.userthematics.exists():
|
||
return None
|
||
|
||
return Thematic.objects.filter(
|
||
userthematics__user_config=self,
|
||
userthematics__can_view_assets=True
|
||
).distinct()
|
||
|
||
def get_inspectable_thematics(self):
|
||
"""Retourne le QuerySet des thématiques pour lesquelles l'utilisateur est autorisé à réaliser des inspections."""
|
||
from common.models import Thematic
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return Thematic.objects.all()
|
||
|
||
return Thematic.objects.filter(
|
||
userthematics__user_config=self,
|
||
userthematics__can_inspect=True
|
||
).distinct()
|
||
|
||
def can_inspect_thematic(self, thematic):
|
||
"""Vérifie si l'utilisateur peut réaliser des inspections pour une thématique donnée."""
|
||
if self.user.is_superuser or self.has_role('admin'):
|
||
return True
|
||
|
||
return self.userthematics.filter(thematic=thematic, can_inspect=True).exists()
|
||
|
||
def get_default_contract(self):
|
||
"""
|
||
Retourne le contrat par défaut de l'utilisateur.
|
||
Si aucun contrat par défaut n'est spécifié mais que l'utilisateur a limit_interventions_to_contracts=True
|
||
et n'a accès qu'à un seul contrat (via UserContractAccess), retourne ce contrat unique.
|
||
"""
|
||
if self.default_contract_id:
|
||
return self.default_contract
|
||
if self.limit_interventions_to_contracts:
|
||
allowed_contracts = list(self.contracts.all()[:2])
|
||
if len(allowed_contracts) == 1:
|
||
return allowed_contracts[0]
|
||
return None
|
||
|
||
|
||
|
||
class UserThematics(models.Model):
|
||
"""Table intermédiaire pour gérer les thématiques d'un utilisateur"""
|
||
user_config = models.ForeignKey(UserConfig, on_delete=models.CASCADE, related_name="userthematics")
|
||
thematic = models.ForeignKey(Thematic, on_delete=models.CASCADE)
|
||
can_view_assets = models.BooleanField(default=True) # Peut voir les assets de la thématique
|
||
can_edit_assets = models.BooleanField(default=False) # Peut éditer les assets de la thématique
|
||
can_view_interventions = models.BooleanField(default=True) # Peut voir les interventions de la thématique
|
||
can_edit_interventions = models.BooleanField(default=False) # Peut éditer les interventions de la thématique
|
||
can_edit_locations = models.BooleanField(default=False) # Peut créer/modifier/supprimer les localisations de la thématique
|
||
can_process_observations = models.BooleanField(default=False, verbose_name="Peut traiter les observations") # Peut changer le statut des observations (traiter, rediriger, archiver, etc.)
|
||
can_view_projects = models.BooleanField(default=False, verbose_name="Peut voir les projets") # Peut voir les projets de la thématique
|
||
can_edit_projects = models.BooleanField(default=False, verbose_name="Peut éditer les projets") # Peut créer/modifier les projets de la thématique
|
||
can_validate_assets = models.BooleanField(default=False, verbose_name="Peut valider les assets")
|
||
can_inspect = models.BooleanField(default=False, verbose_name="Peut réaliser des inspections")
|
||
|
||
class Meta:
|
||
unique_together = ("user_config", "thematic") # Empêche les doublons
|
||
verbose_name = "Thématique utilisateur"
|
||
verbose_name_plural = "Thématiques utilisateur"
|
||
|
||
def __str__(self):
|
||
return f"{self.user_config.user.username} - {self.thematic.name}"
|
||
|
||
|
||
# Statuts d'intervention (copie locale pour éviter import circulaire avec interventions)
|
||
INTERVENTION_STATUS_CHOICES = [
|
||
('in_preparation', 'En préparation'),
|
||
('to_be_approved', 'À approuver'),
|
||
('to_be_planned', 'À planifier'),
|
||
('to_be_processed', 'À traiter'),
|
||
('assigned', 'Pris en charge'),
|
||
('in_progress', 'Démarré'),
|
||
('on_pause', 'En pause'),
|
||
('finished', 'Terminé'),
|
||
('processed', 'Traité'),
|
||
('to_be_corrected', 'À corriger'),
|
||
('corrected', 'Corrigé'),
|
||
('validated', 'Validé'),
|
||
('invoiced', 'Comptabilisé'),
|
||
('closed', 'Finalisé'),
|
||
('canceled', 'Annulé'),
|
||
]
|
||
|
||
# Ordre d'affichage des statuts pour les colonnes de la table admin
|
||
INTERVENTION_STATUS_ORDER = [
|
||
'in_preparation', 'to_be_approved', 'to_be_planned', 'to_be_processed',
|
||
'assigned', 'in_progress', 'on_pause', 'finished', 'processed',
|
||
'to_be_corrected', 'corrected', 'validated', 'invoiced', 'closed', 'canceled',
|
||
]
|
||
|
||
|
||
class UserThematicStatusPermission(models.Model):
|
||
"""Permissions par statut d'intervention pour un accès thématique utilisateur."""
|
||
user_thematic = models.ForeignKey(
|
||
UserThematics,
|
||
on_delete=models.CASCADE,
|
||
related_name='status_permissions',
|
||
)
|
||
status = models.CharField(
|
||
max_length=30,
|
||
choices=INTERVENTION_STATUS_CHOICES,
|
||
verbose_name="Statut",
|
||
)
|
||
can_view = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Voir",
|
||
help_text="Peut voir les interventions à ce statut.",
|
||
)
|
||
can_edit = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Modifier",
|
||
help_text="Peut modifier les interventions à ce statut.",
|
||
)
|
||
can_change_status_to = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Changer vers",
|
||
help_text="Peut déclencher une transition vers ce statut (selon le graphe global des transitions).",
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = ('user_thematic', 'status')
|
||
verbose_name = "Permission par statut (thématique)"
|
||
verbose_name_plural = "Permissions par statut (thématique)"
|
||
indexes = [
|
||
# Index couvrant can_view : permet un index-only scan dans les sous-requêtes EXISTS
|
||
models.Index(fields=['user_thematic', 'status', 'can_view'], name='utsp_status_canview_idx'),
|
||
]
|
||
|
||
def __str__(self):
|
||
return f"{self.user_thematic} – {self.status}"
|
||
|
||
|
||
class UserContractAccess(models.Model):
|
||
"""Table intermédiaire pour gérer les droits d'accès d'un utilisateur à un contrat"""
|
||
user_config = models.ForeignKey(UserConfig, on_delete=models.CASCADE)
|
||
contract = models.ForeignKey("contracts.Contract", on_delete=models.CASCADE)
|
||
can_view_posts = models.BooleanField(default=False) # Peut voir les postes du contrat
|
||
can_view_prices = models.BooleanField(default=False) # Peut voir les prix des postes
|
||
can_view_interventions = models.BooleanField(default=True) # Peut voir les interventions du contrat
|
||
can_view_assets = models.BooleanField(default=True) # Peut voir les assets du contrat
|
||
can_create_interventions = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut créer des interventions",
|
||
help_text="Permet au prestataire de créer des interventions pour ce contrat."
|
||
) # Permet aux prestataires de créer des interventions
|
||
can_check_interventions = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut vérifier les interventions",
|
||
help_text="Permet de vérifier les interventions pour ce contrat."
|
||
) # Permet de vérifier les interventions
|
||
can_approve = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut approuver les interventions",
|
||
help_text="Permet de faire passer les interventions du statut 'à approuver' vers un autre statut pour ce contrat."
|
||
) # Permet d'approuver les interventions (passage depuis 'to_be_approved')
|
||
|
||
class Meta:
|
||
unique_together = ("user_config", "contract") # Empêche les doublons
|
||
|
||
def __str__(self):
|
||
return f"{self.user_config.user.username} - {self.contract.contract_number}"
|
||
|
||
|
||
class UserContractStatusPermission(models.Model):
|
||
"""Permissions par statut d'intervention pour un accès contrat utilisateur."""
|
||
user_contract = models.ForeignKey(
|
||
UserContractAccess,
|
||
on_delete=models.CASCADE,
|
||
related_name='status_permissions',
|
||
)
|
||
status = models.CharField(
|
||
max_length=30,
|
||
choices=INTERVENTION_STATUS_CHOICES,
|
||
verbose_name="Statut",
|
||
)
|
||
can_view = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Voir",
|
||
help_text="Peut voir les interventions à ce statut.",
|
||
)
|
||
can_edit = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Modifier",
|
||
help_text="Peut modifier les interventions à ce statut.",
|
||
)
|
||
can_change_status_to = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Changer vers",
|
||
help_text="Peut déclencher une transition vers ce statut (selon le graphe global des transitions).",
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = ('user_contract', 'status')
|
||
verbose_name = "Permission par statut (contrat)"
|
||
verbose_name_plural = "Permissions par statut (contrat)"
|
||
indexes = [
|
||
# Index couvrant can_view : permet un index-only scan dans les sous-requêtes EXISTS
|
||
models.Index(fields=['user_contract', 'status', 'can_view'], name='ucsp_status_canview_idx'),
|
||
]
|
||
|
||
def __str__(self):
|
||
return f"{self.user_contract} – {self.status}"
|
||
|
||
|
||
class UserWarehouseAccess(models.Model):
|
||
"""Table intermédiaire pour gérer les droits d'accès d'un utilisateur à un entrepôt"""
|
||
user_config = models.ForeignKey(UserConfig, on_delete=models.CASCADE, related_name="userwarehouses")
|
||
warehouse = models.ForeignKey("stock.Warehouse", on_delete=models.CASCADE)
|
||
can_view = models.BooleanField(default=True, verbose_name="Peut visualiser") # Peut voir l'entrepôt
|
||
can_manage = models.BooleanField(default=False, verbose_name="Peut gérer") # Peut gérer l'entrepôt (ajouter/modifier stock, commandes, etc.)
|
||
can_administer = models.BooleanField(default=False, verbose_name="Peut administrer") # Peut administrer l'entrepôt (gérer les accès, config, etc.)
|
||
can_manage_preparation_orders = models.BooleanField(default=False, verbose_name="Peut gérer les bons de préparation")
|
||
can_manage_purchase_orders = models.BooleanField(default=False, verbose_name="Peut gérer les bons de commande")
|
||
can_validate_inventory = models.BooleanField(default=False, verbose_name="Peut valider les inventaires")
|
||
is_default = models.BooleanField(default=False, verbose_name="Entrepôt par défaut") # Entrepôt par défaut pour cet utilisateur
|
||
|
||
class Meta:
|
||
unique_together = ("user_config", "warehouse") # Empêche les doublons
|
||
verbose_name = "Accès Entrepôt"
|
||
verbose_name_plural = "Accès Entrepôts"
|
||
|
||
def __str__(self):
|
||
return f"{self.user_config.user.username} - {self.warehouse.name}"
|
||
|
||
def save(self, *args, **kwargs):
|
||
# Si cet entrepôt est défini comme par défaut, retirer le flag des autres
|
||
if self.is_default:
|
||
UserWarehouseAccess.objects.filter(
|
||
user_config=self.user_config,
|
||
is_default=True
|
||
).exclude(pk=self.pk).update(is_default=False)
|
||
super().save(*args, **kwargs)
|
||
|
||
|
||
class UserWarehouseLocationAccess(models.Model):
|
||
"""Table intermédiaire pour gérer les droits d'accès d'un utilisateur à un emplacement d'entrepôt (camionnette, etc.)"""
|
||
user_config = models.ForeignKey(UserConfig, on_delete=models.CASCADE, related_name="userlocations")
|
||
warehouse_location = models.ForeignKey("stock.WarehouseLocation", on_delete=models.CASCADE, related_name="user_accesses", verbose_name="Emplacement")
|
||
can_perform_inventory = models.BooleanField(default=False, verbose_name="Peut faire l'inventaire")
|
||
can_validate_inventory = models.BooleanField(default=False, verbose_name="Peut valider l'inventaire")
|
||
|
||
class Meta:
|
||
unique_together = ("user_config", "warehouse_location")
|
||
verbose_name = "Accès Emplacement"
|
||
verbose_name_plural = "Accès Emplacements"
|
||
|
||
def __str__(self):
|
||
return f"{self.user_config.user.username} - {self.warehouse_location.name}"
|
||
|
||
|
||
class UserAssetAccess(models.Model):
|
||
"""
|
||
Permissions explicites par instance d'asset ou de location pour un utilisateur.
|
||
|
||
Logique de priorité (identique aux contrats) :
|
||
- Si une ligne existe → ses valeurs (True OU False) sont DÉFINITIVES, sans fallback thématique.
|
||
- Si aucune ligne n'existe → on consulte AssetTypePermissionConfig.requires_explicit_permissions :
|
||
* True → l'utilisateur n'a PAS accès (droits thématiques insuffisants).
|
||
* False → fallback sur les droits thématiques habituels.
|
||
- Le rôle admin bypass toujours ces vérifications.
|
||
|
||
Le champ editable_fields, s'il est renseigné (liste non vide), REMPLACE complètement
|
||
les champs éditables déterminés par le rôle.
|
||
Le champ visible_fields, s'il est renseigné (liste non vide), REMPLACE complètement
|
||
les champs visibles par défaut (masque les champs absents de la liste).
|
||
"""
|
||
user_config = models.ForeignKey(
|
||
UserConfig,
|
||
on_delete=models.CASCADE,
|
||
related_name='asset_accesses',
|
||
verbose_name="Configuration utilisateur",
|
||
)
|
||
content_type = models.ForeignKey(
|
||
ContentType,
|
||
on_delete=models.CASCADE,
|
||
verbose_name="Type d'objet",
|
||
help_text="Le modèle Django de l'asset ou de la location concernée.",
|
||
)
|
||
object_id = models.PositiveIntegerField(
|
||
verbose_name="ID de l'objet",
|
||
)
|
||
content_object = GenericForeignKey('content_type', 'object_id')
|
||
|
||
can_view = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut visualiser",
|
||
)
|
||
can_edit = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut modifier",
|
||
)
|
||
can_delete = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut supprimer",
|
||
)
|
||
editable_fields = ArrayField(
|
||
models.CharField(max_length=200),
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Champs modifiables",
|
||
help_text=(
|
||
"Liste des noms de champs que cet utilisateur peut modifier sur cet objet. "
|
||
"Si renseignée (non vide), remplace complètement les champs autorisés par le rôle."
|
||
),
|
||
)
|
||
visible_fields = ArrayField(
|
||
models.CharField(max_length=200),
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Champs visibles",
|
||
help_text=(
|
||
"Liste des noms de champs que cet utilisateur peut voir sur cet objet. "
|
||
"Si renseignée (non vide), remplace complètement les champs affichés par défaut."
|
||
),
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = [('user_config', 'content_type', 'object_id')]
|
||
verbose_name = "Accès asset spécifique"
|
||
verbose_name_plural = "Accès assets spécifiques"
|
||
indexes = [
|
||
models.Index(fields=['content_type', 'object_id']),
|
||
]
|
||
|
||
def __str__(self):
|
||
return (
|
||
f"{self.user_config.user.username} → {self.content_type.model} #{self.object_id}"
|
||
)
|
||
|
||
|
||
class UserAssetTypeAccess(models.Model):
|
||
"""
|
||
Permissions explicites pour un utilisateur sur TOUS les objets d'un type d'asset donné.
|
||
|
||
Niveau intermédiaire entre UserAssetAccess (par instance) et les droits thématiques.
|
||
|
||
Logique de priorité :
|
||
- Plus prioritaire que les droits thématiques et que requires_explicit_permissions.
|
||
- Moins prioritaire qu'une ligne UserAssetAccess portant sur l'instance précise.
|
||
|
||
Le champ editable_fields, s'il est renseigné (liste non vide), remplace les champs
|
||
éditables déterminés par le rôle **pour tous les objets de ce type**, sauf si une
|
||
ligne UserAssetAccess spécifique à l'instance définit ses propres editable_fields.
|
||
Le champ visible_fields, s'il est renseigné (liste non vide), remplace les champs
|
||
visibles par défaut pour tous les objets de ce type. Peut être écrasé par instance.
|
||
"""
|
||
user_config = models.ForeignKey(
|
||
UserConfig,
|
||
on_delete=models.CASCADE,
|
||
related_name='asset_type_accesses',
|
||
verbose_name="Configuration utilisateur",
|
||
)
|
||
content_type = models.ForeignKey(
|
||
ContentType,
|
||
on_delete=models.CASCADE,
|
||
verbose_name="Type d'asset",
|
||
help_text="Le modèle Django concerné (s'applique à toutes ses instances).",
|
||
)
|
||
|
||
can_view = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut visualiser",
|
||
)
|
||
can_edit = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut modifier",
|
||
)
|
||
can_delete = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Peut supprimer",
|
||
)
|
||
editable_fields = ArrayField(
|
||
models.CharField(max_length=200),
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Champs modifiables",
|
||
help_text=(
|
||
"Liste des noms de champs modifiables pour tous les objets de ce type. "
|
||
"Si renseignée (non vide), remplace les champs autorisés par le rôle. "
|
||
"Peut être écrasée objet par objet via UserAssetAccess."
|
||
),
|
||
)
|
||
visible_fields = ArrayField(
|
||
models.CharField(max_length=200),
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Champs visibles",
|
||
help_text=(
|
||
"Liste des noms de champs visibles pour tous les objets de ce type. "
|
||
"Si renseignée (non vide), remplace les champs affichés par défaut. "
|
||
"Peut être écrasée objet par objet via UserAssetAccess."
|
||
),
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = [('user_config', 'content_type')]
|
||
verbose_name = "Accès par type d'asset"
|
||
verbose_name_plural = "Accès par type d'asset"
|
||
|
||
def __str__(self):
|
||
return (
|
||
f"{self.user_config.user.username} → {self.content_type.model} (tous les objets)"
|
||
)
|
||
|
||
|
||
class AssetTypePermissionConfig(models.Model):
|
||
"""
|
||
Configuration des permissions au niveau d'un type d'asset (modèle Django).
|
||
|
||
Lorsque requires_explicit_permissions=True, les droits thématiques ne sont pas
|
||
suffisants pour accéder aux instances de ce type : chaque utilisateur doit avoir
|
||
une ligne UserAssetAccess ou UserAssetTypeAccess explicite.
|
||
"""
|
||
content_type = models.OneToOneField(
|
||
ContentType,
|
||
on_delete=models.CASCADE,
|
||
related_name='asset_permission_config',
|
||
verbose_name="Type d'asset",
|
||
)
|
||
requires_explicit_permissions = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Droits explicites requis",
|
||
help_text=(
|
||
"Si coché, les droits thématiques seuls ne suffisent pas. "
|
||
"Chaque utilisateur doit avoir une entrée explicite dans "
|
||
"« Accès assets spécifiques » pour ce type d'objet."
|
||
),
|
||
)
|
||
|
||
class Meta:
|
||
verbose_name = "Config permissions par type d'asset"
|
||
verbose_name_plural = "Configs permissions par type d'asset"
|
||
|
||
def __str__(self):
|
||
flag = "⚠ droits explicites requis" if self.requires_explicit_permissions else "droits thématiques suffisants"
|
||
return f"{self.content_type.model} — {flag}"
|
||
|
||
|
||
class UserApprovalConfig(UserConfig):
|
||
"""Proxy model de UserConfig dédié à la gestion des approbateurs d'interventions."""
|
||
|
||
class Meta:
|
||
proxy = True
|
||
verbose_name = "Configuration approbateur"
|
||
verbose_name_plural = "Configurations approbateurs"
|
||
ordering = ['user__username']
|
||
|
||
|
||
|
||
DOCUMENT_STATUS_CHOICES = {
|
||
'active': 'Actif',
|
||
'archived': 'Archivé',
|
||
}
|
||
|
||
class Document(models.Model):
|
||
uploaded_by = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Ajouté par",
|
||
)
|
||
description = models.TextField(blank=True,null=True,verbose_name="Description du document")
|
||
upload_date = models.DateTimeField(auto_now=True,verbose_name="Date de téléversement")
|
||
status = models.CharField(choices=DOCUMENT_STATUS_CHOICES, default='active')
|
||
managed_document = models.ForeignKey(
|
||
"documents.ManagedDocument",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="%(app_label)s_%(class)s_sources",
|
||
)
|
||
|
||
@property
|
||
def filename(self):
|
||
return os.path.basename(self.file.name)
|
||
|
||
class Meta:
|
||
abstract = True
|
||
ordering = ["upload_date"]
|
||
|
||
def __str__(self):
|
||
return f"{self.file.name}"
|
||
|
||
|
||
DOCUMENT_MANAGEMENT_SYNC = False # Désactive la synchronisation par défaut
|
||
|
||
def get_document_context(self):
|
||
for field in self._meta.fields:
|
||
if not field.is_relation or not field.many_to_one:
|
||
continue
|
||
if field.name in {"uploaded_by", "managed_document"}:
|
||
continue
|
||
value = getattr(self, field.name, None)
|
||
if value is not None:
|
||
return value
|
||
return None
|
||
|
||
|
||
def get_document_title(self):
|
||
return getattr(self, "description", None) or self.filename
|
||
|
||
|
||
def rotate_image(self, angle=90):
|
||
"""Fait pivoter l'image de `angle` degrés dans le sens horaire (90, 180, 270) et régénère la miniature."""
|
||
if not self.file:
|
||
return False
|
||
try:
|
||
from PIL import Image, ImageOps
|
||
file_path = self.file.path
|
||
with Image.open(file_path) as img:
|
||
img = ImageOps.exif_transpose(img) or img
|
||
pil_angle = (360 - (angle % 360)) % 360
|
||
if pil_angle != 0:
|
||
img = img.rotate(pil_angle, expand=True)
|
||
fmt = img.format or "JPEG"
|
||
if fmt.upper() in ("JPEG", "JPG"):
|
||
if img.mode in ("RGBA", "P"):
|
||
img = img.convert("RGB")
|
||
img.save(file_path, "JPEG", quality=92)
|
||
else:
|
||
img.save(file_path, format=fmt)
|
||
|
||
self.generate_thumbnail()
|
||
return True
|
||
except Exception as e:
|
||
import logging
|
||
logging.getLogger(__name__).error(f"Error rotating image {self.file.name}: {e}")
|
||
return False
|
||
|
||
def generate_thumbnail(self):
|
||
"""Générer une vignette pour les fichiers PDF ou images."""
|
||
if not self.file:
|
||
return
|
||
if self.file.name.endswith(".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"
|
||
super().save(update_fields=['thumbnail'])
|
||
except Exception as e:
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
logger.error(f"Error generating PDF thumbnail for {self.file.name}: {e}")
|
||
else:
|
||
try:
|
||
from PIL import ImageOps
|
||
with Image.open(self.file.path) as img:
|
||
img = ImageOps.exif_transpose(img) or img
|
||
if img.mode in ("RGBA", "P"):
|
||
img = img.convert("RGB")
|
||
|
||
img.thumbnail((300, 300))
|
||
thumbnail_path = f"{self.file.path}_thumbnail.jpg"
|
||
img.save(thumbnail_path, "JPEG", quality=85)
|
||
self.thumbnail.name = f"{self.file.name}_thumbnail.jpg"
|
||
super().save(update_fields=['thumbnail'])
|
||
except Exception as e:
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
logger.error(f"Error generating image thumbnail for {self.file.name}: {e}")
|
||
|
||
|
||
def save(self, *args, **kwargs):
|
||
is_update = self.pk is not None
|
||
old_file_name = None
|
||
if is_update:
|
||
old_file_name = (
|
||
self.__class__.objects.filter(pk=self.pk)
|
||
.values_list("file", flat=True)
|
||
.first()
|
||
)
|
||
|
||
super().save(*args, **kwargs) # Save the file first
|
||
|
||
if self.file: # Vérifier qu'un fichier est bien présent
|
||
try:
|
||
from PIL import ImageOps
|
||
with Image.open(self.file.path) as img:
|
||
# Correction de l'orientation EXIF via ImageOps.exif_transpose
|
||
# (redresse les pixels et supprime le tag Orientation pour éviter la double rotation dans les navigateurs)
|
||
transposed_img = ImageOps.exif_transpose(img)
|
||
if transposed_img is not None:
|
||
fmt = img.format or "JPEG"
|
||
if fmt.upper() in ("JPEG", "JPG"):
|
||
if transposed_img.mode in ("RGBA", "P"):
|
||
transposed_img = transposed_img.convert("RGB")
|
||
transposed_img.save(self.file.path, "JPEG", quality=92)
|
||
else:
|
||
transposed_img.save(self.file.path, format=fmt)
|
||
except Exception:
|
||
pass # Fichier non-image (PDF, etc.)
|
||
|
||
if not self.thumbnail:
|
||
self.generate_thumbnail()
|
||
|
||
|
||
if not getattr(self, "DOCUMENT_MANAGEMENT_SYNC", False):
|
||
return
|
||
|
||
try:
|
||
from documents.models import ensure_document_for_legacy
|
||
|
||
ensure_document_for_legacy(self, old_file_name=old_file_name)
|
||
except Exception as exc:
|
||
print(f"Error while synchronising document management for {self}: {exc}")
|
||
|
||
|
||
|
||
|
||
class Municipality(models.Model):
|
||
fid = models.CharField(max_length=100)
|
||
name_fr = models.CharField(max_length=255, verbose_name="Name FR")
|
||
name_nl = models.CharField(max_length=255, verbose_name="Name NL")
|
||
niscode = models.CharField(max_length=30, verbose_name="Postal code", null=True, blank=True)
|
||
geom = models.MultiPolygonField(srid=3812, null=True, blank=True)
|
||
|
||
@property
|
||
def name(self):
|
||
"""Renvoie le nom dans la langue active."""
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl
|
||
return self.name_fr
|
||
|
||
objects = LanguageManager()
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
|
||
class AdrStreet(models.Model):
|
||
fid = models.CharField(max_length=100, unique=True)
|
||
name_fr = models.CharField(max_length=255, verbose_name="Nom")
|
||
name_nl = models.CharField(max_length=255, verbose_name="Naam")
|
||
municipality = models.ForeignKey(Municipality, on_delete=models.CASCADE, related_name="street_names", null=True, blank=True)
|
||
full_street_fr = models.CharField(max_length=2048, verbose_name="Nom complet FR", null=True, blank=True)
|
||
full_street_nl = models.CharField(max_length=2048, verbose_name="Nom complet NL", null=True, blank=True)
|
||
centroid = models.PointField(srid=3812, null=True, blank=True)
|
||
lon = models.FloatField(null=True, blank=True)
|
||
lat = models.FloatField(null=True, blank=True)
|
||
geom = models.MultiPolygonField(srid=3812, null=True, blank=True)
|
||
|
||
@property
|
||
def name(self):
|
||
"""Renvoie le nom dans la langue active."""
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
street_name = self.name_nl
|
||
else:
|
||
street_name = self.name_fr
|
||
|
||
return f"{street_name} ({self.municipality.name})" if self.municipality else street_name
|
||
|
||
def get_name(self):
|
||
"""Compatibilité avec les appels get_name()"""
|
||
return self.name
|
||
|
||
objects = LanguageManager()
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
|
||
|
||
class AdrPoint(models.Model):
|
||
fid = models.CharField(max_length=100, unique=True)
|
||
number = models.CharField(max_length=255, verbose_name="Numéro")
|
||
street_name = models.ForeignKey(AdrStreet, on_delete=models.CASCADE, related_name="addresses", null=True, blank=True)
|
||
zipcode = models.CharField(max_length=255, verbose_name="Code postal", null=True, blank=True)
|
||
municipality = models.ForeignKey(Municipality, on_delete=models.SET_NULL, related_name="addresses", null=True, blank=True)
|
||
full_address_fr = models.CharField(max_length=2048, verbose_name="Adresse complète FR", null=True, blank=True)
|
||
full_address_nl = models.CharField(max_length=2048, verbose_name="Adresse complète NL", null=True, blank=True)
|
||
geom = models.PointField(srid=3812, null=True, blank=True)
|
||
lon = models.FloatField(null=True, blank=True)
|
||
lat = models.FloatField(null=True, blank=True)
|
||
|
||
def __str__(self):
|
||
if self.street_name:
|
||
return f"{self.number} {self.street_name.name} ({self.municipality.name})" if self.municipality else f"{self.number} {self.street_name.name}"
|
||
return f"{self.number} ({self.municipality.name})" if self.municipality else self.number
|
||
|
||
|
||
LOG_LEVEL_CHOICES = {
|
||
'DEBUG': 'Debug',
|
||
'INFO': 'Info',
|
||
'WARNING': 'Warning',
|
||
'ERROR': 'Error',
|
||
'CRITICAL': 'Critical',
|
||
}
|
||
|
||
|
||
class RequestLog(models.Model):
|
||
"""
|
||
Modèle pour stocker les logs des requêtes et erreurs de l'application.
|
||
Permet de visualiser les erreurs dans l'admin Django.
|
||
"""
|
||
timestamp = models.DateTimeField(auto_now_add=True, verbose_name="Date et heure", db_index=True)
|
||
level = models.CharField(max_length=10, choices=LOG_LEVEL_CHOICES, default='ERROR', verbose_name="Niveau")
|
||
user = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Utilisateur",
|
||
related_name="request_logs"
|
||
)
|
||
username = models.CharField(max_length=150, blank=True, null=True, verbose_name="Nom d'utilisateur")
|
||
method = models.CharField(max_length=10, blank=True, null=True, verbose_name="Méthode HTTP")
|
||
path = models.CharField(max_length=2048, verbose_name="URL")
|
||
full_url = models.CharField(max_length=4096, blank=True, null=True, verbose_name="URL complète")
|
||
status_code = models.IntegerField(null=True, blank=True, verbose_name="Code HTTP")
|
||
ip_address = models.GenericIPAddressField(null=True, blank=True, verbose_name="Adresse IP")
|
||
user_agent = models.TextField(blank=True, null=True, verbose_name="User Agent")
|
||
referer = models.CharField(max_length=2048, blank=True, null=True, verbose_name="Referer")
|
||
exception_type = models.CharField(max_length=255, blank=True, null=True, verbose_name="Type d'exception")
|
||
exception_message = models.TextField(blank=True, null=True, verbose_name="Message d'erreur")
|
||
traceback = models.TextField(blank=True, null=True, verbose_name="Traceback")
|
||
request_data = models.TextField(blank=True, null=True, verbose_name="Données de la requête")
|
||
|
||
class Meta:
|
||
verbose_name = "Log de requête"
|
||
verbose_name_plural = "Logs de requêtes"
|
||
ordering = ["-timestamp"]
|
||
indexes = [
|
||
models.Index(fields=['level']),
|
||
models.Index(fields=['user']),
|
||
models.Index(fields=['status_code']),
|
||
models.Index(fields=['timestamp', 'level']),
|
||
]
|
||
|
||
def __str__(self):
|
||
return f"[{self.level}] {self.timestamp.strftime('%Y-%m-%d %H:%M:%S')} - {self.path}"
|
||
|
||
@classmethod
|
||
def log_request(cls, request, exception=None, level='ERROR', status_code=None, traceback_str=None):
|
||
"""
|
||
Crée une entrée de log à partir d'une requête HTTP.
|
||
"""
|
||
import json
|
||
|
||
# Récupérer l'adresse IP
|
||
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
|
||
if x_forwarded_for:
|
||
ip_address = x_forwarded_for.split(',')[0].strip()
|
||
else:
|
||
ip_address = request.META.get('REMOTE_ADDR')
|
||
|
||
# Récupérer les données de la requête (sans données sensibles)
|
||
request_data = None
|
||
try:
|
||
if request.method in ['POST', 'PUT', 'PATCH']:
|
||
data = dict(request.POST)
|
||
# Masquer les mots de passe et autres données sensibles
|
||
sensitive_keys = ['password', 'password1', 'password2', 'token', 'secret', 'csrfmiddlewaretoken']
|
||
for key in sensitive_keys:
|
||
if key in data:
|
||
data[key] = ['***MASKED***']
|
||
request_data = json.dumps(data, default=str)
|
||
except Exception:
|
||
pass
|
||
|
||
# Informations sur l'exception
|
||
exception_type = None
|
||
exception_message = None
|
||
if exception:
|
||
exception_type = type(exception).__name__
|
||
exception_message = str(exception)
|
||
|
||
# Créer le log
|
||
log_entry = cls.objects.create(
|
||
level=level,
|
||
user=request.user if request.user.is_authenticated else None,
|
||
username=request.user.username if request.user.is_authenticated else None,
|
||
method=request.method,
|
||
path=request.path,
|
||
full_url=request.build_absolute_uri(),
|
||
status_code=status_code,
|
||
ip_address=ip_address,
|
||
user_agent=request.META.get('HTTP_USER_AGENT', '')[:2000] if request.META.get('HTTP_USER_AGENT') else None,
|
||
referer=request.META.get('HTTP_REFERER', '')[:2048] if request.META.get('HTTP_REFERER') else None,
|
||
exception_type=exception_type,
|
||
exception_message=exception_message,
|
||
traceback=traceback_str,
|
||
request_data=request_data,
|
||
)
|
||
return log_entry
|
||
|
||
|
||
# Import des modèles de tâches planifiées
|
||
from .models_scheduled_tasks import ScheduledCommand, CommandExecutionLog
|