"""Permission helpers for the documents application.""" from __future__ import annotations from typing import Iterable from django.contrib.auth import get_user_model from django.contrib.auth.models import AnonymousUser from django.db import models from .models import ( DocumentFolder, DocumentFolderAttachment, ManagedDocument, ) from streetup import settings UserModel = get_user_model() _PERMISSION_ORDER = { ManagedDocument.PERMISSION_VIEW: 0, ManagedDocument.PERMISSION_COMMENT: 1, ManagedDocument.PERMISSION_EDIT: 2, ManagedDocument.PERMISSION_APPROVE: 3, } def _normalise_permission(permission: str) -> str: if permission not in _PERMISSION_ORDER: raise ValueError(f"Unknown permission '{permission}'") return permission def _permissions_at_least(permission: str) -> Iterable[str]: """Return every permission value equal or greater than ``permission``.""" level = _PERMISSION_ORDER[_normalise_permission(permission)] return [perm for perm, perm_level in _PERMISSION_ORDER.items() if perm_level >= level] def filter_documents_for_user( queryset: models.QuerySet[ManagedDocument], user: settings.AUTH_USER_MODEL | AnonymousUser, *, required_permission: str = ManagedDocument.PERMISSION_VIEW, ) -> models.QuerySet[ManagedDocument]: """Restrict the queryset to documents the user can access.""" if user is None or not getattr(user, "is_authenticated", False): return queryset.none() if getattr(user, "is_superuser", False): return queryset allowed_permissions = list(_permissions_at_least(required_permission)) filter_q = models.Q(created_by=user) filter_q |= models.Q(shares__user=user, shares__permission__in=allowed_permissions) filter_q |= models.Q(folders__shares__user=user, folders__shares__permission__in=allowed_permissions) # Ajouter les documents des folders accessibles via les thématiques directes (M2M sur le folder) accessible_thematics = _get_accessible_thematic_ids(user) if accessible_thematics: filter_q |= models.Q(folders__thematics__pk__in=accessible_thematics) # Ajouter les documents des folders accessibles via les objets attachés accessible_folder_ids = _get_folder_ids_from_accessible_objects(user) if accessible_folder_ids: filter_q |= models.Q(folders__pk__in=accessible_folder_ids) # Ajouter les documents directement attachés via DocumentAttachment à un objet accessible # (couvre les documents importés sans folder ou dont le folder n'a pas de DocumentFolderAttachment) accessible_ct_ids = _get_accessible_content_type_ids(user) if accessible_ct_ids: filter_q |= models.Q(attachments__content_type_id__in=accessible_ct_ids) return queryset.filter(filter_q).distinct() def user_has_document_permission( user: settings.AUTH_USER_MODEL | AnonymousUser, document: ManagedDocument, *, required_permission: str = ManagedDocument.PERMISSION_VIEW, ) -> bool: """Return ``True`` if ``user`` has ``required_permission`` on ``document``.""" if user is None or not getattr(user, "is_authenticated", False): return False if getattr(user, "is_superuser", False): return True if document.created_by_id and document.created_by_id == getattr(user, "pk", None): return True allowed_permissions = list(_permissions_at_least(required_permission)) if document.shares.filter(user=user, permission__in=allowed_permissions).exists(): return True if document.folders.filter( shares__user=user, shares__permission__in=allowed_permissions ).exists(): return True # Pour les vérifications EDIT+, on exige can_edit_assets=True sur la thématique. # Pour VIEW/COMMENT, can_view_assets=True suffit. is_edit_check = _PERMISSION_ORDER.get(required_permission, 0) >= _PERMISSION_ORDER[ManagedDocument.PERMISSION_EDIT] # Collecter les IDs des dossiers du document et de tous leurs ancêtres doc_folder_ids = set(document.folders.values_list("pk", flat=True)) all_doc_folder_ids = _collect_ancestor_ids(doc_folder_ids) if doc_folder_ids else set() # Vérifier si l'utilisateur a accès via les thématiques directes des folders ou de leurs ancêtres accessible_thematics = _get_editable_thematic_ids(user) if is_edit_check else _get_accessible_thematic_ids(user) if accessible_thematics and all_doc_folder_ids: if DocumentFolder.objects.filter(pk__in=all_doc_folder_ids, thematics__pk__in=accessible_thematics).exists(): return True # Vérifier si l'utilisateur a accès via un folder (ou sous-dossier) lié à un objet accessible accessible_folder_ids = _get_folder_ids_from_editable_objects(user) if is_edit_check else _get_folder_ids_from_accessible_objects(user) if accessible_folder_ids and all_doc_folder_ids: if all_doc_folder_ids & accessible_folder_ids: return True # Vérifier si le document est directement attaché à un objet accessible via DocumentAttachment # (couvre les documents importés sans folder ou dont le folder n'a pas de DocumentFolderAttachment) accessible_ct_ids = _get_editable_content_type_ids(user) if is_edit_check else _get_accessible_content_type_ids(user) if accessible_ct_ids: if document.attachments.filter(content_type_id__in=accessible_ct_ids).exists(): return True return False def user_is_documents_admin(user) -> bool: if user is None or not getattr(user, "is_authenticated", False): return False if getattr(user, "is_superuser", False) or getattr(user, "is_staff", False): return True config = getattr(user, "config", None) if config is None: return False # Admins et top_managers peuvent tout voir if config.roles.filter(name__in=["admin", "top_manager"]).exists(): return True return False def user_is_internal_manager(user) -> bool: if user is None or not getattr(user, "is_authenticated", False): return False config = getattr(user, "config", None) if config is None or not getattr(config, "is_intern", False): return False return config.roles.filter(name="manager").exists() def document_has_project_or_intervention_attachment(document: ManagedDocument) -> bool: return ( document.attachments.filter(content_type__app_label="projects", content_type__model="project").exists() or document.attachments.filter(content_type__app_label="interventions", content_type__model="intervention").exists() ) def _get_accessible_thematic_ids(user) -> set[int]: config = getattr(user, "config", None) if not config: return set() return set( config.userthematics.filter(can_view_assets=True).values_list( "thematic_id", flat=True ) ) def _get_editable_thematic_ids(user) -> set[int]: """Thématiques pour lesquelles l'utilisateur a le droit d'édition des assets.""" config = getattr(user, "config", None) if not config: return set() return set( config.userthematics.filter(can_edit_assets=True).values_list( "thematic_id", flat=True ) ) def _get_user_contract_ids(user) -> set[int]: config = getattr(user, "config", None) if not config: return set() return set(config.contracts.values_list("pk", flat=True)) def _folder_contract_ids(folder_id: int) -> set[int]: """Récupère les contract IDs associés à un folder (avec cache).""" attachments = ( DocumentFolderAttachment.objects.filter(folder_id=folder_id) .select_related("content_type") .all() ) contract_ids: set[int] = set() # Grouper les attachments par content_type pour optimiser les requêtes from collections import defaultdict attachments_by_type = defaultdict(list) for attachment in attachments: content_type = attachment.content_type if content_type.app_label == "contracts" and content_type.model == "contract": contract_ids.add(attachment.object_id) else: attachments_by_type[content_type.id].append(attachment.object_id) # Pour chaque type de contenu, faire une seule requête pour tous les objets from django.contrib.contenttypes.models import ContentType for content_type_id, object_ids in attachments_by_type.items(): try: content_type = ContentType.objects.get(pk=content_type_id) model_class = content_type.model_class() if model_class is None: continue # Récupérer tous les objets d'un coup avec leurs contrats objects = model_class.objects.filter(pk__in=object_ids) # Vérifier si le modèle a un champ contract ou contract_id if hasattr(model_class, 'contract_id'): objects = objects.values_list('contract_id', flat=True) contract_ids.update(cid for cid in objects if cid) elif hasattr(model_class, 'contract'): objects = objects.select_related('contract').values_list('contract__pk', flat=True) contract_ids.update(cid for cid in objects if cid) except Exception: # pragma: no cover - defensive continue return contract_ids def _filter_ids_by_contracts(folder_ids: set[int], user) -> set[int]: config = getattr(user, "config", None) if not config or not config.limit_assets_to_contracts: return folder_ids allowed_contract_ids = _get_user_contract_ids(user) if not allowed_contract_ids: return set() # Optimisation: traiter tous les folders en batch from collections import defaultdict from django.contrib.contenttypes.models import ContentType # Récupérer tous les attachments pour tous les folders en une seule requête attachments = DocumentFolderAttachment.objects.filter( folder_id__in=folder_ids ).select_related('content_type').values( 'folder_id', 'content_type_id', 'content_type__app_label', 'content_type__model', 'object_id' ) folder_contracts = defaultdict(set) attachments_by_type = defaultdict(list) # Première passe: identifier les contrats directs et grouper les autres par type for att in attachments: folder_id = att['folder_id'] if att['content_type__app_label'] == 'contracts' and att['content_type__model'] == 'contract': folder_contracts[folder_id].add(att['object_id']) else: attachments_by_type[(att['content_type_id'], att['content_type__app_label'], att['content_type__model'])].append({ 'folder_id': folder_id, 'object_id': att['object_id'] }) # Deuxième passe: pour chaque type de contenu, récupérer les contrats en batch for (ct_id, app_label, model_name), att_list in attachments_by_type.items(): try: content_type = ContentType.objects.get(pk=ct_id) model_class = content_type.model_class() if model_class is None: continue object_ids = [att['object_id'] for att in att_list] # Récupérer les contract_ids en une seule requête if hasattr(model_class, 'contract_id'): objects_with_contracts = model_class.objects.filter( pk__in=object_ids ).values('pk', 'contract_id') obj_to_contract = {obj['pk']: obj['contract_id'] for obj in objects_with_contracts if obj['contract_id']} elif hasattr(model_class, 'contract'): objects_with_contracts = model_class.objects.filter( pk__in=object_ids ).select_related('contract').values('pk', 'contract__pk') obj_to_contract = {obj['pk']: obj['contract__pk'] for obj in objects_with_contracts if obj['contract__pk']} else: obj_to_contract = {} # Associer les contrats aux folders for att in att_list: contract_id = obj_to_contract.get(att['object_id']) if contract_id: folder_contracts[att['folder_id']].add(contract_id) except Exception: # pragma: no cover - defensive continue # Filtrer les folders qui ont au moins un contrat autorisé allowed: set[int] = set() for folder_id in folder_ids: contract_ids = folder_contracts.get(folder_id, set()) if not contract_ids: # Pas de contrats = accessible allowed.add(folder_id) elif contract_ids & allowed_contract_ids: # Au moins un contrat autorisé allowed.add(folder_id) return allowed # Mapping centralisé des modèles par thématique (utilisé dans plusieurs fonctions) _THEMATIC_MODEL_MAPPING: dict[str, list[str]] = { 'trafficlights': [ 'trafficlightintersection', 'trafficlightpole', 'trafficlightcontroller', 'trafficlightcontrollerhardware', 'trafficlightcable', 'trafficlightlantern', 'trafficlightdetector', 'trafficlightnetworkhardware', 'trafficlightvoltagehardware', 'trafficlightelectricalcabinet', 'trafficlightelectricalcabinetcontent', 'trafficlightaccessory', 'trafficlightprogramming', ], 'structures': [ 'structure', 'structurelocation', 'structuregeoasset', ], 'publiclighting': [ 'publiclightingstreet', 'publiclightingasset', ], 'greensurfaces': [ 'greensurface', ], 'controls': [ 'control', ], } # Inverse: model_name -> thematic_code _MODEL_TO_THEMATIC: dict[str, str] = { model: thematic for thematic, models in _THEMATIC_MODEL_MAPPING.items() for model in models } def _user_can_access_content_object(user, content_object) -> bool: """ Vérifie si l'utilisateur peut accéder à l'objet de contenu (intersection, structure, etc.). Args: user: L'utilisateur content_object: L'objet attaché au folder Returns: bool: True si l'utilisateur peut accéder à l'objet """ if content_object is None: return False config = getattr(user, "config", None) if not config: return False # Les admins et operators peuvent tout voir if config.roles.filter(name__in=['admin', 'operator']).exists(): return True # Déterminer la thématique de l'objet from django.contrib.contenttypes.models import ContentType from common.models import Thematic, UserThematics content_type = ContentType.objects.get_for_model(content_object) model_name = content_type.model.lower() thematic_code = _MODEL_TO_THEMATIC.get(model_name) if not thematic_code: return False # Vérifier si l'utilisateur a accès à la thématique thematic = Thematic.objects.filter(code=thematic_code).first() if not thematic: return False has_access = UserThematics.objects.filter( user_config=config, thematic=thematic, can_view_assets=True ).exists() if not has_access: return False # Vérifier les restrictions de contrat si applicable if config.limit_assets_to_contracts: allowed_contract_ids = _get_user_contract_ids(user) if allowed_contract_ids: # Vérifier si l'objet est lié à un contrat autorisé contract_id = getattr(content_object, "contract_id", None) if contract_id and contract_id not in allowed_contract_ids: return False contract = getattr(content_object, "contract", None) if contract is not None: pk = getattr(contract, "pk", None) if pk and pk not in allowed_contract_ids: return False return True def _get_accessible_content_type_ids(user) -> list[int]: """ Retourne les IDs des ContentType correspondant aux modèles accessibles par l'utilisateur via ses thématiques (can_view_assets=True). Résultat mis en cache sur le user pour la durée de la requête. """ return _get_content_type_ids_for_user(user, edit_only=False) def _get_editable_content_type_ids(user) -> list[int]: """ Retourne les IDs des ContentType correspondant aux modèles pour lesquels l'utilisateur a le droit d'édition (can_edit_assets=True). """ return _get_content_type_ids_for_user(user, edit_only=True) def _get_content_type_ids_for_user(user, *, edit_only: bool) -> list[int]: """Implémentation commune pour view et edit content-type IDs.""" cache_key = f'_{"editable" if edit_only else "accessible"}_ct_ids_{getattr(user, "pk", None)}' if hasattr(user, '_folder_cache') and cache_key in user._folder_cache: return user._folder_cache[cache_key] result: list[int] = [] config = getattr(user, "config", None) if not config or user_is_documents_admin(user): return result from common.models import UserThematics from django.contrib.contenttypes.models import ContentType filter_kwargs = {'user_config': config, 'can_edit_assets': True} if edit_only else {'user_config': config, 'can_view_assets': True} accessible_thematics = UserThematics.objects.filter( **filter_kwargs ).select_related('thematic') for ut in accessible_thematics: thematic_code = ut.thematic.code model_names = _THEMATIC_MODEL_MAPPING.get(thematic_code, []) for model_name in model_names: try: ct = ContentType.objects.get(model=model_name) result.append(ct.id) except ContentType.DoesNotExist: continue if not hasattr(user, '_folder_cache'): user._folder_cache = {} user._folder_cache[cache_key] = result return result def _get_folder_ids_from_accessible_objects(user) -> set[int]: """ Récupère les IDs des folders auxquels l'utilisateur a accès (visualisation) via les objets attachés. """ return _get_folder_ids_from_objects(user, edit_only=False) def _get_folder_ids_from_editable_objects(user) -> set[int]: """ Récupère les IDs des folders pour lesquels l'utilisateur a le droit d'édition via les objets attachés. """ return _get_folder_ids_from_objects(user, edit_only=True) def _get_folder_ids_from_objects(user, *, edit_only: bool) -> set[int]: """Implémentation commune pour view et edit folder IDs via objets attachés.""" cache_key = f'_{"editable" if edit_only else "accessible"}_folder_ids_{getattr(user, "pk", None)}' if hasattr(user, '_folder_cache') and cache_key in user._folder_cache: return user._folder_cache[cache_key] accessible_folder_ids: set[int] = set() config = getattr(user, "config", None) if not config: return accessible_folder_ids if user_is_documents_admin(user): return accessible_folder_ids accessible_content_types = _get_editable_content_type_ids(user) if edit_only else _get_accessible_content_type_ids(user) if not accessible_content_types: # Pas de content_types thématiques — mais on continue pour les projets pass else: # Récupérer les folder_ids associés aux content_types accessibles attachments = DocumentFolderAttachment.objects.filter( content_type_id__in=accessible_content_types ).values_list('folder_id', flat=True).distinct() accessible_folder_ids = set(attachments) # Filtrer par contrats si nécessaire if config.limit_assets_to_contracts: allowed_contract_ids = _get_user_contract_ids(user) if allowed_contract_ids: accessible_folder_ids = _filter_ids_by_contracts(accessible_folder_ids, user) # ── Folders de projets ────────────────────────────────────────────── # Pour les projets, on ne distingue pas view/edit (pas de notion can_edit_projects). # On ne les inclut que pour la vérification de visualisation. if not edit_only: from django.contrib.contenttypes.models import ContentType from django.db.models import Q try: project_ct = ContentType.objects.get(app_label='projects', model='project') except ContentType.DoesNotExist: project_ct = None if project_ct: from projects.models import Project, ProjectUserAccess from common.models import UserThematics thematic_project_ids: set[int] = set() accessible_thematics = UserThematics.objects.filter( user_config=config, can_view_projects=True ) if config.is_intern and accessible_thematics.exists(): accessible_thematic_objs = list(accessible_thematics.values_list('thematic_id', flat=True)) thematic_project_ids = set( Project.objects.filter( thematics__pk__in=accessible_thematic_objs ).values_list('pk', flat=True) ) member_project_ids = set( ProjectUserAccess.objects.filter(user=user, can_view=True).values_list('project_id', flat=True) ) accessible_project_ids = thematic_project_ids | member_project_ids if accessible_project_ids: project_folder_ids = set( DocumentFolderAttachment.objects.filter( content_type=project_ct, object_id__in=accessible_project_ids, ).values_list('folder_id', flat=True).distinct() ) accessible_folder_ids |= project_folder_ids # Récupérer l'ensemble des sous-dossiers (descendants) des dossiers d'assets accessibles accessible_folder_ids = _collect_descendant_ids(accessible_folder_ids) if not hasattr(user, '_folder_cache'): user._folder_cache = {} user._folder_cache[cache_key] = accessible_folder_ids return accessible_folder_ids def _collect_ancestor_ids(folder_ids: set[int]) -> set[int]: if not folder_ids: return set() seen = set(folder_ids) stack = list(folder_ids) # Optimisation: récupérer toutes les relations parent en une seule requête # au lieu de faire une requête par folder all_parent_relations = {} if stack: from .models import DocumentFolder # Récupérer toutes les relations parent_folders pour tous les folders en une seule requête relations = DocumentFolder.child_folders.through.objects.filter( from_documentfolder_id__in=folder_ids ).values_list('from_documentfolder_id', 'to_documentfolder_id') for child_id, parent_id in relations: if child_id not in all_parent_relations: all_parent_relations[child_id] = [] all_parent_relations[child_id].append(parent_id) # Parcours itératif en utilisant les relations pré-chargées while stack: current_id = stack.pop() parent_ids = all_parent_relations.get(current_id, []) # Si on découvre de nouveaux parents, on doit aussi charger leurs relations new_parents = [pid for pid in parent_ids if pid not in seen] if new_parents: new_relations = DocumentFolder.child_folders.through.objects.filter( from_documentfolder_id__in=new_parents ).values_list('from_documentfolder_id', 'to_documentfolder_id') for child_id, parent_id in new_relations: if child_id not in all_parent_relations: all_parent_relations[child_id] = [] if parent_id not in all_parent_relations[child_id]: all_parent_relations[child_id].append(parent_id) for parent_id in parent_ids: if parent_id and parent_id not in seen: seen.add(parent_id) stack.append(parent_id) return seen def _collect_descendant_ids(folder_ids: set[int]) -> set[int]: """Récupère récursivement tous les sous-dossiers (descendants) des dossier_ids transmis.""" if not folder_ids: return set() seen = set(folder_ids) stack = list(folder_ids) from .models import DocumentFolder while stack: batch = stack stack = [] children = DocumentFolder.parent_folders.through.objects.filter( to_documentfolder_id__in=batch ).values_list("from_documentfolder_id", flat=True) for cid in children: if cid not in seen: seen.add(cid) stack.append(cid) return seen def filter_folders_for_user( queryset: models.QuerySet[DocumentFolder], user: settings.AUTH_USER_MODEL | AnonymousUser, *, required_permission: str = ManagedDocument.PERMISSION_VIEW, include_ancestors: bool = False, ) -> models.QuerySet[DocumentFolder]: if user is None or not getattr(user, "is_authenticated", False): return queryset.none() if user_is_documents_admin(user): return queryset allowed_permissions = list(_permissions_at_least(required_permission)) owned_ids = set( queryset.filter(created_by=user).values_list("pk", flat=True) ) shared_ids = set( queryset.filter( shares__user=user, shares__permission__in=allowed_permissions ).values_list("pk", flat=True) ) thematic_ids = set() accessible_thematics = _get_accessible_thematic_ids(user) if accessible_thematics: thematic_ids = set( queryset.filter(thematics__pk__in=accessible_thematics).values_list( "pk", flat=True ) ) thematic_ids = _collect_descendant_ids(thematic_ids) thematic_ids = _filter_ids_by_contracts(thematic_ids, user) # Ajouter les folders accessibles via les objets attachés (avec leurs sous-dossiers) attached_object_ids = _get_folder_ids_from_accessible_objects(user) visible_ids = owned_ids | shared_ids | thematic_ids | attached_object_ids if include_ancestors: visible_ids = _collect_ancestor_ids(visible_ids) if not visible_ids: return queryset.none() return queryset.filter(pk__in=visible_ids).distinct() def user_can_browse_folder(user, folder: DocumentFolder) -> bool: """ Vérifie si l'utilisateur peut naviguer dans le module documents vers ce folder. Cette fonction est utilisée pour déterminer si le lien vers le répertoire principal doit être affiché. L'utilisateur doit avoir accès au folder via: - Être admin/superuser - Être interne (is_intern=True) ET avoir le folder dans ses folders visibles - Être externe avec can_access_view('documents') ET avoir le folder partagé Args: user: L'utilisateur folder: Le folder à vérifier Returns: bool: True si l'utilisateur peut naviguer vers le folder """ if user is None or not getattr(user, "is_authenticated", False): return False config = getattr(user, "config", None) if not config: return False # Les admins peuvent toujours accéder if user_is_documents_admin(user): return True # Les utilisateurs internes (managers, controllers, etc.) ont accès # si le folder est dans leurs folders visibles (filtrage par thématiques/partages) if config.is_intern: visible_ids = get_user_visible_folder_ids(user) return folder.pk in visible_ids # Les utilisateurs externes doivent avoir l'accès explicite à l'app documents if not config.can_access_view('documents'): return False # Vérifier si le folder a été partagé avec l'utilisateur allowed_permissions = list(_permissions_at_least(ManagedDocument.PERMISSION_VIEW)) if folder.shares.filter(user=user, permission__in=allowed_permissions).exists(): return True # Vérifier si l'utilisateur est propriétaire du folder if folder.created_by_id and folder.created_by_id == getattr(user, "pk", None): return True return False def get_user_visible_folder_ids(user) -> set[int]: """Récupère les IDs des folders visibles par l'utilisateur (avec cache).""" # Utiliser un cache au niveau de la requête cache_key = '_visible_folder_ids' if hasattr(user, '_folder_cache') and cache_key in user._folder_cache: return user._folder_cache[cache_key] queryset = DocumentFolder.objects.all() visible_qs = filter_folders_for_user( queryset, user, include_ancestors=True ) visible_ids = set(visible_qs.values_list("pk", flat=True)) # Mettre en cache if not hasattr(user, '_folder_cache'): user._folder_cache = {} user._folder_cache[cache_key] = visible_ids return visible_ids __all__ = [ "filter_documents_for_user", "filter_folders_for_user", "get_user_visible_folder_ids", "user_is_documents_admin", "user_has_document_permission", "user_can_browse_folder", ]