diff --git a/loko/assets/tests_exceptional_transport.py b/loko/assets/tests_exceptional_transport.py index 932d2c6..24913ee 100644 --- a/loko/assets/tests_exceptional_transport.py +++ b/loko/assets/tests_exceptional_transport.py @@ -237,6 +237,7 @@ class ExceptionalTransportTests(TestCase): title="Note de calcul convoi 240t", created_by=self.superuser ) + doc.thematics.add(self.thematic_documents) doc.folders.add(te_folder) user_no_te = User.objects.get(pk=self.user_no_te.pk) diff --git a/loko/documents/forms.py b/loko/documents/forms.py index cbc2688..0c1a5db 100644 --- a/loko/documents/forms.py +++ b/loko/documents/forms.py @@ -34,16 +34,23 @@ class TagField(forms.CharField): class DocumentUpdateForm(forms.ModelForm): class Meta: model = ManagedDocument - fields = ["title", "description"] + fields = ["title", "description", "visibility", "thematics"] widgets = { "title": forms.TextInput(attrs={"class": "form-control"}), "description": forms.Textarea(attrs={"class": "form-control", "rows": 4}), + "visibility": forms.Select(attrs={"class": "form-select"}), + "thematics": forms.SelectMultiple(attrs={"class": "form-select"}), } def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.fields["title"].widget.attrs.setdefault("class", "form-control") self.fields["description"].widget.attrs.setdefault("class", "form-control") + self.fields["visibility"].widget.attrs.setdefault("class", "form-select") + self.fields["visibility"].required = False + if "thematics" in self.fields: + self.fields["thematics"].queryset = Thematic.objects.order_by("name_fr") + self.fields["thematics"].required = False class ManagedDocumentForm(forms.ModelForm): @@ -53,9 +60,11 @@ class ManagedDocumentForm(forms.ModelForm): class Meta: model = ManagedDocument - fields = ["title", "description", "types", "tags", "folders"] + fields = ["title", "description", "visibility", "thematics", "types", "tags", "folders"] widgets = { "description": forms.Textarea(attrs={"rows": 3}), + "visibility": forms.Select(attrs={"class": "form-select"}), + "thematics": forms.SelectMultiple(attrs={"class": "form-select"}), "types": forms.SelectMultiple(attrs={"class": "form-select"}), "tags": forms.SelectMultiple(attrs={"class": "form-select"}), "folders": forms.SelectMultiple(attrs={"class": "form-select"}), @@ -65,6 +74,7 @@ class ManagedDocumentForm(forms.ModelForm): user = kwargs.pop("user", None) initial_folder = kwargs.pop("initial_folder", None) super().__init__(*args, **kwargs) + self.fields["visibility"].required = False # Limiter les folders accessibles à l'utilisateur if user: @@ -335,6 +345,7 @@ class DocumentFolderForm(forms.ModelForm): fields = [ "name", "description", + "visibility", "application_label", "parent_folders", "thematics", @@ -343,6 +354,7 @@ class DocumentFolderForm(forms.ModelForm): widgets = { "name": forms.TextInput(attrs={"class": "form-control"}), "description": forms.Textarea(attrs={"class": "form-control", "rows": 3}), + "visibility": forms.Select(attrs={"class": "form-select"}), "application_label": forms.TextInput(attrs={"class": "form-control"}), "parent_folders": forms.SelectMultiple(attrs={"class": "form-select"}), "thematics": forms.SelectMultiple(attrs={"class": "form-select"}), @@ -352,6 +364,8 @@ class DocumentFolderForm(forms.ModelForm): def __init__(self, *args, **kwargs): self.request_user = kwargs.pop("user", None) super().__init__(*args, **kwargs) + if "visibility" in self.fields: + self.fields["visibility"].required = False instance = getattr(self, "instance", None) if instance and instance.pk: diff --git a/loko/documents/migrations/0003_documentfolder_visibility_manageddocument_thematics_and_more.py b/loko/documents/migrations/0003_documentfolder_visibility_manageddocument_thematics_and_more.py new file mode 100644 index 0000000..60cb296 --- /dev/null +++ b/loko/documents/migrations/0003_documentfolder_visibility_manageddocument_thematics_and_more.py @@ -0,0 +1,29 @@ +# Generated by Django 6.0.7 on 2026-08-29 16:35 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('common', '0014_userconfig_inspection_list_filters_and_more'), + ('documents', '0002_documentattachment_documents_d_content_5921e5_idx_and_more'), + ] + + operations = [ + migrations.AddField( + model_name='documentfolder', + name='visibility', + field=models.CharField(choices=[('private', 'Private (Creator only)'), ('restricted', 'Shared with specific users'), ('internal', 'Internal (Thematics & Contracts)'), ('scoped', 'All qualified users (Internal & External)')], db_index=True, default='internal', max_length=20, verbose_name='Visibility'), + ), + migrations.AddField( + model_name='manageddocument', + name='thematics', + field=models.ManyToManyField(blank=True, related_name='managed_documents', to='common.thematic', verbose_name='Thematics'), + ), + migrations.AddField( + model_name='manageddocument', + name='visibility', + field=models.CharField(choices=[('private', 'Private (Creator only)'), ('restricted', 'Shared with specific users'), ('internal', 'Internal (Thematics & Contracts)'), ('scoped', 'All qualified users (Internal & External)')], db_index=True, default='internal', max_length=20, verbose_name='Visibility'), + ), + ] diff --git a/loko/documents/models.py b/loko/documents/models.py index ef92d16..8b5cc23 100644 --- a/loko/documents/models.py +++ b/loko/documents/models.py @@ -111,11 +111,25 @@ class DocumentTag(models.Model): super().save(*args, **kwargs) +class VisibilityScope(models.TextChoices): + PRIVATE = "private", _("Private (Creator only)") + RESTRICTED = "restricted", _("Shared with specific users") + INTERNAL = "internal", _("Internal (Thematics & Contracts)") + SCOPED = "scoped", _("All qualified users (Internal & External)") + + class DocumentFolder(models.Model): uuid = models.UUIDField(default=uuid.uuid4, unique=True, editable=False) name = models.CharField(max_length=255, verbose_name=_("Name")) slug = models.SlugField(max_length=150, unique=True, verbose_name=_("Slug")) description = models.TextField(blank=True, verbose_name=_("Description")) + visibility = models.CharField( + max_length=20, + choices=VisibilityScope.choices, + default=VisibilityScope.INTERNAL, + verbose_name=_("Visibility"), + db_index=True, + ) is_predefined = models.BooleanField(default=False, verbose_name=_("Predefined")) application_label = models.CharField( max_length=100, blank=True, verbose_name=_("Application label") @@ -182,6 +196,13 @@ class ManagedDocument(models.Model): uuid = models.UUIDField(default=uuid.uuid4, unique=True, editable=False) title = models.CharField(max_length=255, verbose_name=_("Title")) description = models.TextField(blank=True, verbose_name=_("Description")) + visibility = models.CharField( + max_length=20, + choices=VisibilityScope.choices, + default=VisibilityScope.INTERNAL, + verbose_name=_("Visibility"), + db_index=True, + ) created_by = models.ForeignKey( User, on_delete=models.SET_NULL, @@ -221,6 +242,12 @@ class ManagedDocument(models.Model): types = models.ManyToManyField(DocumentType, blank=True, related_name="documents") tags = models.ManyToManyField(DocumentTag, blank=True, related_name="documents") + thematics = models.ManyToManyField( + "common.Thematic", + blank=True, + related_name="managed_documents", + verbose_name=_("Thematics"), + ) folders = models.ManyToManyField( DocumentFolder, related_name="documents", diff --git a/loko/documents/permissions.py b/loko/documents/permissions.py index 1f6357e..91b5fd2 100644 --- a/loko/documents/permissions.py +++ b/loko/documents/permissions.py @@ -4,6 +4,7 @@ from __future__ import annotations from typing import Iterable +from django.conf import settings from django.contrib.auth import get_user_model from django.contrib.auth.models import AnonymousUser from django.db import models @@ -11,9 +12,10 @@ from django.db import models from .models import ( DocumentFolder, DocumentFolderAttachment, + DocumentAttachment, ManagedDocument, + VisibilityScope, ) -from django.conf import settings UserModel = get_user_model() @@ -35,7 +37,6 @@ def _normalise_permission(permission: str) -> str: 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] @@ -46,35 +47,47 @@ def filter_documents_for_user( *, required_permission: str = ManagedDocument.PERMISSION_VIEW, ) -> models.QuerySet[ManagedDocument]: - """Restrict the queryset to documents the user can access.""" + """Restrict the queryset to documents the user can access based on document-level permissions.""" if user is None or not getattr(user, "is_authenticated", False): return queryset.none() - if getattr(user, "is_superuser", False): + if getattr(user, "is_superuser", False) or user_is_documents_admin(user): return queryset + config = getattr(user, "config", None) + is_intern = bool(config and getattr(config, "is_intern", False)) + allowed_permissions = list(_permissions_at_least(required_permission)) + is_edit_check = _PERMISSION_ORDER.get(required_permission, 0) >= _PERMISSION_ORDER[ManagedDocument.PERMISSION_EDIT] - 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) + # 1. Accès direct : créateur ou partage direct (DocumentShare) + base_q = models.Q(created_by=user) + base_q |= models.Q(shares__user=user, 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) + # 2. Documents qualifiés par thématiques & contrats (attachments et direct thematics) + attached_doc_ids = _get_document_ids_from_objects(user, edit_only=is_edit_check) + + accessible_thematics = _get_editable_thematic_ids(user) if is_edit_check else _get_accessible_thematic_ids(user) + thematic_doc_ids: set[int] = set() if accessible_thematics: - filter_q |= models.Q(folders__thematics__pk__in=accessible_thematics) + thematic_doc_ids = set( + queryset.filter(thematics__pk__in=accessible_thematics).values_list("pk", flat=True) + ) + thematic_doc_ids = _filter_document_ids_by_contracts(thematic_doc_ids, user) - # 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) + qualif_doc_ids = attached_doc_ids | thematic_doc_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) + if qualif_doc_ids: + internal_q = ( + models.Q(visibility=VisibilityScope.INTERNAL, pk__in=qualif_doc_ids) + if is_intern + else models.Q(pk__in=[]) + ) + scoped_q = models.Q(visibility=VisibilityScope.SCOPED, pk__in=qualif_doc_ids) + filter_q = base_q | internal_q | scoped_q + else: + filter_q = base_q result_qs = queryset.filter(filter_q).distinct() @@ -117,61 +130,105 @@ def user_has_document_permission( if user is None or not getattr(user, "is_authenticated", False): return False - if getattr(user, "is_superuser", False): + if getattr(user, "is_superuser", False) or user_is_documents_admin(user): return True from assets.permissions import can_view_exceptional_transport if not can_view_exceptional_transport(user): - if ( - document.tags.filter(slug__in=['exceptional_transport', 'exceptional-transport', 'transports-exceptionnels']).exists() - or document.folders.filter( - models.Q(slug__icontains='exceptional-transport') - | models.Q(slug__icontains='transport-exceptionnel') - | models.Q(slug__icontains='transports-exceptionnels') - | models.Q(name__icontains='Exceptional transport') - | models.Q(name__icontains='Transport exceptionnel') - ).exists() - ): + if document.tags.filter(slug__in=['exceptional_transport', 'exceptional-transport', 'transports-exceptionnels']).exists(): + return False + if any(_is_exceptional_transport_folder(f) for f in document.folders.all()): return False 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(): + # PRIVATE et RESTRICTED ne sont accessibles que via créateur ou partage direct + if document.visibility in [VisibilityScope.PRIVATE, VisibilityScope.RESTRICTED]: + return False + + config = getattr(user, "config", None) + is_intern = bool(config and getattr(config, "is_intern", False)) + + # INTERNAL exige is_intern=True + if document.visibility == VisibilityScope.INTERNAL and not is_intern: + return False + + is_edit_check = _PERMISSION_ORDER.get(required_permission, 0) >= _PERMISSION_ORDER[ManagedDocument.PERMISSION_EDIT] + accessible_thematics = _get_editable_thematic_ids(user) if is_edit_check else _get_accessible_thematic_ids(user) + + # 1. Vérification thématiques directes + if accessible_thematics and document.thematics.filter(pk__in=accessible_thematics).exists(): + if config and config.limit_assets_to_contracts: + doc_contracts = _document_contract_ids(document.pk) + allowed_contract_ids = _get_user_contract_ids(user) + if not doc_contracts or (doc_contracts & allowed_contract_ids): + return True + else: + return True + + # 2. Vérification objets attachés (assets, projets, contrats, etc.) + for attachment in document.attachments.all(): + content_object = attachment.content_object + if content_object and _user_can_access_content_object(user, content_object, edit_only=is_edit_check): + return True + + return False + + +def user_has_folder_permission( + user: settings.AUTH_USER_MODEL | AnonymousUser, + folder: DocumentFolder, + *, + required_permission: str = ManagedDocument.PERMISSION_VIEW, +) -> bool: + """Return ``True`` if ``user`` has ``required_permission`` on ``folder``.""" + + if user is None or not getattr(user, "is_authenticated", False): + return False + + if getattr(user, "is_superuser", False) or user_is_documents_admin(user): 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. + from assets.permissions import can_view_exceptional_transport + if not can_view_exceptional_transport(user) and _is_exceptional_transport_folder(folder): + return False + + if folder.created_by_id and folder.created_by_id == getattr(user, "pk", None): + return True + + allowed_permissions = list(_permissions_at_least(required_permission)) + if folder.shares.filter(user=user, permission__in=allowed_permissions).exists(): + return True + + if folder.visibility in [VisibilityScope.PRIVATE, VisibilityScope.RESTRICTED]: + return False + + config = getattr(user, "config", None) + is_intern = bool(config and getattr(config, "is_intern", False)) + + if folder.visibility == VisibilityScope.INTERNAL and not is_intern: + return False + 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(): + + if accessible_thematics and folder.thematics.filter(pk__in=accessible_thematics).exists(): + if config and config.limit_assets_to_contracts: + folder_contracts = _folder_contract_ids(folder.pk) + allowed_contract_ids = _get_user_contract_ids(user) + if not folder_contracts or (folder_contracts & allowed_contract_ids): + return True + else: 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(): + for attachment in folder.attachments.all(): + content_object = attachment.content_object + if content_object and _user_can_access_content_object(user, content_object, edit_only=is_edit_check): return True return False @@ -185,8 +242,7 @@ def user_is_documents_admin(user) -> bool: config = getattr(user, "config", None) if config is None: return False - # Seuls les admins ont tous les droits d'administration sur les répertoires et documents - if config.roles.filter(name="admin").exists(): + if config.roles.filter(name__in=["admin", "operator"]).exists(): return True return False @@ -214,8 +270,6 @@ def user_can_delete_document( 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: @@ -254,19 +308,17 @@ def _folder_contract_ids(folder_id: int) -> set[int]: .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: @@ -274,20 +326,57 @@ def _folder_contract_ids(folder_id: int) -> set[int]: 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 + except Exception: continue - + + return contract_ids + + +def _document_contract_ids(document_id: int) -> set[int]: + """Récupère les contract IDs associés à un document (via ses attachments).""" + attachments = ( + DocumentAttachment.objects.filter(document_id=document_id) + .select_related("content_type") + .all() + ) + contract_ids: set[int] = set() + + 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) + + 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 + + objects = model_class.objects.filter(pk__in=object_ids) + 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: + continue + return contract_ids @@ -297,25 +386,20 @@ def _filter_ids_by_contracts(folder_ids: set[int], user) -> set[int]: 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': @@ -325,18 +409,16 @@ def _filter_ids_by_contracts(folder_ids: set[int], user) -> set[int]: '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 @@ -349,24 +431,92 @@ def _filter_ids_by_contracts(folder_ids: set[int], user) -> set[int]: 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 + except Exception: 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 + if not contract_ids: allowed.add(folder_id) - elif contract_ids & allowed_contract_ids: # Au moins un contrat autorisé + elif contract_ids & allowed_contract_ids: allowed.add(folder_id) - + + return allowed + + +def _filter_document_ids_by_contracts(doc_ids: set[int], user) -> set[int]: + config = getattr(user, "config", None) + if not config or not config.limit_assets_to_contracts: + return doc_ids + + allowed_contract_ids = _get_user_contract_ids(user) + + from collections import defaultdict + from django.contrib.contenttypes.models import ContentType + + attachments = DocumentAttachment.objects.filter( + document_id__in=doc_ids + ).select_related('content_type').values( + 'document_id', 'content_type_id', 'content_type__app_label', + 'content_type__model', 'object_id' + ) + + doc_contracts = defaultdict(set) + attachments_by_type = defaultdict(list) + + for att in attachments: + doc_id = att['document_id'] + if att['content_type__app_label'] == 'contracts' and att['content_type__model'] == 'contract': + doc_contracts[doc_id].add(att['object_id']) + else: + attachments_by_type[(att['content_type_id'], att['content_type__app_label'], att['content_type__model'])].append({ + 'document_id': doc_id, + 'object_id': att['object_id'] + }) + + 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] + + 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 = {} + + for att in att_list: + contract_id = obj_to_contract.get(att['object_id']) + if contract_id: + doc_contracts[att['document_id']].add(contract_id) + except Exception: + continue + + allowed: set[int] = set() + for doc_id in doc_ids: + contract_ids = doc_contracts.get(doc_id, set()) + if not contract_ids: + allowed.add(doc_id) + elif contract_ids & allowed_contract_ids: + allowed.add(doc_id) + return allowed @@ -412,58 +562,60 @@ _MODEL_TO_THEMATIC: dict[str, str] = { } -def _user_can_access_content_object(user, content_object) -> bool: +def _user_can_access_content_object(user, content_object, edit_only: bool = False) -> 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: + # Cas des projets ou contrats ou autres modèles génériques + if content_type.app_label == 'projects' and model_name == 'project': + from projects.models import ProjectUserAccess + if ProjectUserAccess.objects.filter(user=user, project=content_object, can_view=True).exists(): + return True + if config.is_intern: + user_them_ids = _get_accessible_thematic_ids(user) + if content_object.thematics.filter(pk__in=user_them_ids).exists(): + return True + return False + + if content_type.app_label == 'contracts' and model_name == 'contract': + if not config.limit_assets_to_contracts: + return True + allowed_cids = _get_user_contract_ids(user) + return getattr(content_object, 'pk', None) in allowed_cids + 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() - + + filter_kwargs = {'user_config': config, 'thematic': thematic, 'can_edit_assets': True} if edit_only else {'user_config': config, 'thematic': thematic, 'can_view_assets': True} + has_access = UserThematics.objects.filter(**filter_kwargs).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 @@ -472,30 +624,19 @@ def _user_can_access_content_object(user, content_object) -> bool: 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] @@ -530,22 +671,8 @@ def _get_content_type_ids_for_user(user, *, edit_only: bool) -> list[int]: 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.""" + """Folder IDs via DocumentFolderAttachment attachés à des objets accessibles.""" 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] @@ -553,37 +680,26 @@ def _get_folder_ids_from_objects(user, *, edit_only: bool) -> set[int]: accessible_folder_ids: set[int] = set() config = getattr(user, "config", None) - if not config: - return accessible_folder_ids - - if user_is_documents_admin(user): + if not config or 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 + if accessible_content_types: 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. + # Folders de projets 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: @@ -618,9 +734,6 @@ def _get_folder_ids_from_objects(user, *, edit_only: bool) -> set[int]: ) 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 @@ -628,6 +741,76 @@ def _get_folder_ids_from_objects(user, *, edit_only: bool) -> set[int]: return accessible_folder_ids +def _get_document_ids_from_objects(user, *, edit_only: bool) -> set[int]: + """Document IDs via DocumentAttachment attachés à des objets accessibles.""" + cache_key = f'_{"editable" if edit_only else "accessible"}_doc_ids_{getattr(user, "pk", None)}' + if hasattr(user, '_folder_cache') and cache_key in user._folder_cache: + return user._folder_cache[cache_key] + + accessible_doc_ids: set[int] = set() + + config = getattr(user, "config", None) + if not config or user_is_documents_admin(user): + return accessible_doc_ids + + accessible_content_types = _get_editable_content_type_ids(user) if edit_only else _get_accessible_content_type_ids(user) + + if accessible_content_types: + attachments = DocumentAttachment.objects.filter( + content_type_id__in=accessible_content_types + ).values_list('document_id', flat=True).distinct() + + accessible_doc_ids = set(attachments) + + if config.limit_assets_to_contracts: + allowed_contract_ids = _get_user_contract_ids(user) + if allowed_contract_ids: + accessible_doc_ids = _filter_document_ids_by_contracts(accessible_doc_ids, user) + + # Documents de projets + if not edit_only: + from django.contrib.contenttypes.models import ContentType + 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_doc_ids = set( + DocumentAttachment.objects.filter( + content_type=project_ct, + object_id__in=accessible_project_ids, + ).values_list('document_id', flat=True).distinct() + ) + accessible_doc_ids |= project_doc_ids + + if not hasattr(user, '_folder_cache'): + user._folder_cache = {} + user._folder_cache[cache_key] = accessible_doc_ids + + return accessible_doc_ids + + def _collect_ancestor_ids(folder_ids: set[int]) -> set[int]: if not folder_ids: return set() @@ -635,12 +818,9 @@ def _collect_ancestor_ids(folder_ids: set[int]) -> set[int]: 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') @@ -650,12 +830,10 @@ def _collect_ancestor_ids(folder_ids: set[int]) -> set[int]: 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( @@ -708,38 +886,45 @@ def filter_folders_for_user( if user is None or not getattr(user, "is_authenticated", False): return queryset.none() - if user_is_documents_admin(user): + if getattr(user, "is_superuser", False) or user_is_documents_admin(user): return queryset + config = getattr(user, "config", None) + is_intern = bool(config and getattr(config, "is_intern", False)) + allowed_permissions = list(_permissions_at_least(required_permission)) + is_edit_check = _PERMISSION_ORDER.get(required_permission, 0) >= _PERMISSION_ORDER[ManagedDocument.PERMISSION_EDIT] - 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) - ) + # 1. Base : créateur ou partage direct (DocumentFolderShare) + base_q = models.Q(created_by=user) + base_q |= models.Q(shares__user=user, shares__permission__in=allowed_permissions) - thematic_ids = set() - accessible_thematics = _get_accessible_thematic_ids(user) + # 2. Scope qualifié : thématiques directes ou objets attachés (DocumentFolderAttachment) + attached_folder_ids = _get_folder_ids_from_objects(user, edit_only=is_edit_check) + accessible_thematics = _get_editable_thematic_ids(user) if is_edit_check else _get_accessible_thematic_ids(user) + + thematic_folder_ids: set[int] = set() if accessible_thematics: - thematic_ids = set( - queryset.filter(thematics__pk__in=accessible_thematics).values_list( - "pk", flat=True - ) + thematic_folder_ids = set( + queryset.filter(thematics__pk__in=accessible_thematics).values_list("pk", flat=True) ) - thematic_ids = _collect_descendant_ids(thematic_ids) + thematic_folder_ids = _filter_ids_by_contracts(thematic_folder_ids, user) - thematic_ids = _filter_ids_by_contracts(thematic_ids, user) + qualif_folder_ids = attached_folder_ids | thematic_folder_ids - # Ajouter les folders accessibles via les objets attachés (avec leurs sous-dossiers) - attached_object_ids = _get_folder_ids_from_accessible_objects(user) + if qualif_folder_ids: + internal_q = ( + models.Q(visibility=VisibilityScope.INTERNAL, pk__in=qualif_folder_ids) + if is_intern + else models.Q(pk__in=[]) + ) + scoped_q = models.Q(visibility=VisibilityScope.SCOPED, pk__in=qualif_folder_ids) + filter_q = base_q | internal_q | scoped_q + else: + filter_q = base_q - visible_ids = owned_ids | shared_ids | thematic_ids | attached_object_ids - - if include_ancestors: + visible_ids = set(queryset.filter(filter_q).values_list("pk", flat=True)) + if include_ancestors and visible_ids: visible_ids = _collect_ancestor_ids(visible_ids) if not visible_ids: @@ -763,75 +948,30 @@ def filter_folders_for_user( 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 - from assets.permissions import can_view_exceptional_transport - if not can_view_exceptional_transport(user) and _is_exceptional_transport_folder(folder): - 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 + + return user_has_folder_permission(user, folder, required_permission=ManagedDocument.PERMISSION_VIEW) 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_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 @@ -841,5 +981,8 @@ __all__ = [ "get_user_visible_folder_ids", "user_is_documents_admin", "user_has_document_permission", + "user_has_folder_permission", "user_can_browse_folder", + "user_can_delete_document", + "user_is_internal_manager", ] \ No newline at end of file diff --git a/loko/documents/templates/documents/document_detail.html b/loko/documents/templates/documents/document_detail.html index a741023..db3239f 100644 --- a/loko/documents/templates/documents/document_detail.html +++ b/loko/documents/templates/documents/document_detail.html @@ -21,6 +21,15 @@