787 lines
No EOL
29 KiB
Python
787 lines
No EOL
29 KiB
Python
"""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 user_can_delete_document(
|
|
user: settings.AUTH_USER_MODEL | AnonymousUser,
|
|
document: ManagedDocument,
|
|
) -> bool:
|
|
"""Return ``True`` if ``user`` can delete ``document``."""
|
|
if user is None or not getattr(user, "is_authenticated", False):
|
|
return False
|
|
if user_is_documents_admin(user):
|
|
return True
|
|
if document.created_by_id and document.created_by_id == getattr(user, "pk", None):
|
|
return True
|
|
return user_has_document_permission(user, document, required_permission=ManagedDocument.PERMISSION_EDIT)
|
|
|
|
|
|
|
|
|
|
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",
|
|
] |