304 lines
13 KiB
Python
304 lines
13 KiB
Python
from functools import reduce
|
|
from operator import or_
|
|
from collections import defaultdict
|
|
from django.conf import settings
|
|
from django.db.models import Q
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
|
|
def is_user_internal_or_admin(user):
|
|
"""
|
|
Détermine si un utilisateur est un utilisateur interne ou un administrateur ayant un accès global.
|
|
"""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False):
|
|
return True
|
|
user_config = getattr(user, 'config', None)
|
|
if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager') or user_config.is_intern):
|
|
return True
|
|
return False
|
|
|
|
|
|
def can_edit_inspection(user, inspection):
|
|
"""
|
|
Vérifie si un utilisateur a le droit de modifier une inspection.
|
|
Les auteurs (inspecteurs de l'inspection) et les administrateurs ont le droit de modifier.
|
|
"""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
|
|
# Administrateur Django superuser ou staff
|
|
if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False):
|
|
return True
|
|
|
|
# Administrateur au sens configuration de rôle
|
|
user_config = getattr(user, 'config', None)
|
|
if user_config and hasattr(user_config, 'has_role'):
|
|
if user_config.has_role('admin') or user_config.has_role('top_manager'):
|
|
return True
|
|
|
|
# Auteur de l'inspection (l'inspecteur créateur)
|
|
if inspection and inspection.inspector_id == user.id:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def can_view_inspection(user, inspection):
|
|
"""
|
|
Vérifie si un utilisateur a le droit de visualiser une inspection.
|
|
- Administrateurs et utilisateurs internes (is_intern=True) : accès complet.
|
|
- Utilisateurs externes : seules les inspections qui les concernent sont visibles :
|
|
1. L'utilisateur est l'auteur/inspecteur de l'inspection.
|
|
2. L'inspection est liée à une mission d'intervention visible par l'utilisateur.
|
|
3. L'inspection est liée à une intervention contrôlée visible par l'utilisateur.
|
|
4. L'inspection est liée à un asset ou une intervention cible visible par l'utilisateur.
|
|
"""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
|
|
if is_user_internal_or_admin(user):
|
|
return True
|
|
|
|
if not inspection:
|
|
return False
|
|
|
|
# 1. Auteur / Inspecteur de l'inspection
|
|
if inspection.inspector_id == user.id:
|
|
return True
|
|
|
|
# 2. Mission d'intervention
|
|
if inspection.mission_intervention_id:
|
|
from interventions.permissions import can_view_intervention
|
|
if can_view_intervention(user, inspection.mission_intervention):
|
|
return True
|
|
|
|
# 3. Intervention de travaux contrôlée
|
|
if inspection.checked_intervention_id:
|
|
from interventions.permissions import can_view_intervention
|
|
if can_view_intervention(user, inspection.checked_intervention):
|
|
return True
|
|
|
|
# 4. Cible (Asset ou Intervention)
|
|
if inspection.asset_content_type_id and inspection.asset_object_id:
|
|
if inspection.asset_content_type.app_label == 'interventions':
|
|
from interventions.permissions import can_view_intervention
|
|
target = inspection.asset
|
|
if target and can_view_intervention(user, target):
|
|
return True
|
|
else:
|
|
from assets.permissions import can_view_asset
|
|
target = inspection.asset
|
|
if target and can_view_asset(user, target):
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids):
|
|
"""
|
|
Retourne un QuerySet filtré des instances de ModelClass liées aux contrats autorisés,
|
|
ou None si ModelClass n'est pas un modèle gérant des contrats.
|
|
"""
|
|
class_name = ModelClass.__name__.lower()
|
|
|
|
# Structures
|
|
if class_name == 'structure':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(contracts__status='active', contracts__contract_id__in=accessible_contract_ids)
|
|
elif class_name == 'structurelocation':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(structure__contracts__status='active', structure__contracts__contract_id__in=accessible_contract_ids)
|
|
elif class_name == 'structuregeoasset':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(location__structure__contracts__status='active', location__structure__contracts__contract_id__in=accessible_contract_ids)
|
|
|
|
# ITS
|
|
elif class_name == 'itslocation':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(its_locations_contracts__status='active', its_locations_contracts__contract_id__in=accessible_contract_ids)
|
|
elif class_name == 'itsgeoasset':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(its_assets_contracts__status='active', its_assets_contracts__contract_id__in=accessible_contract_ids)
|
|
|
|
# Control Centers
|
|
elif class_name == 'controlcenter':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(controlcenters_contracts__status='active', controlcenters_contracts__contract_id__in=accessible_contract_ids)
|
|
elif class_name == 'controlcenterlocation':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(control_center__controlcenters_contracts__status='active', control_center__controlcenters_contracts__contract_id__in=accessible_contract_ids)
|
|
elif class_name == 'controlcentergeoasset':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(controlcenters_assets_contracts__status='active', controlcenters_assets_contracts__contract_id__in=accessible_contract_ids)
|
|
|
|
# Traffic Lights
|
|
elif class_name == 'trafficlightintersection':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(trafficlight_contracts__status='active', trafficlight_contracts__contract_id__in=accessible_contract_ids)
|
|
elif class_name in ('trafficlightpole', 'trafficlightlantern', 'trafficlightcable', 'trafficlightdetector',
|
|
'trafficlightcontroller', 'trafficlightelectricalcabinet', 'trafficlightradar',
|
|
'trafficlightaccessory'):
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(intersection__trafficlight_contracts__status='active', intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids)
|
|
elif class_name == 'trafficlightradarpole':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(pole__intersection__trafficlight_contracts__status='active', pole__intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids)
|
|
|
|
# Clean
|
|
elif class_name == 'cleanlocation':
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(clean_location_contracts__contract_id__in=accessible_contract_ids)
|
|
elif class_name in ('cleanlitterbin', 'cleanglasscontainer'):
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(location__clean_location_contracts__contract_id__in=accessible_contract_ids)
|
|
|
|
# Artworks
|
|
elif class_name in ('artwork', 'monument', 'fountain'):
|
|
if not accessible_contract_ids:
|
|
return ModelClass.objects.none()
|
|
return ModelClass.objects.filter(maintenance_contract_id__in=accessible_contract_ids)
|
|
|
|
# Modèles sans notion de contrat (ex: ParkingSpot, NatureRWIZ, Road, PublicLighting, Sign, etc.)
|
|
return None
|
|
|
|
|
|
def filter_viewable_inspections_for_user(user, qs=None):
|
|
"""
|
|
Filtre un QuerySet d'inspections pour ne renvoyer que celles visibles par l'utilisateur.
|
|
Pour les internes et administrateurs : toutes les inspections (qs complet).
|
|
Pour les externes : seules les inspections qui les concernent :
|
|
- Inspections créées par l'utilisateur (inspecteur)
|
|
- Inspections liées à une mission d'intervention ou intervention contrôlée visible
|
|
- Inspections dont l'asset ou intervention cible est visible
|
|
"""
|
|
from inspections.models import Inspection
|
|
|
|
if qs is None:
|
|
qs = Inspection.objects.all()
|
|
|
|
if not user or not user.is_authenticated:
|
|
return qs.none()
|
|
|
|
if is_user_internal_or_admin(user):
|
|
return qs
|
|
|
|
user_config = getattr(user, 'config', None)
|
|
conditions = []
|
|
|
|
# 1. Auteur / Inspecteur de l'inspection
|
|
conditions.append(Q(inspector=user))
|
|
|
|
# 2. Interventions liées visibles (missions d'intervention, contrôles post-travaux, ou cible intervention)
|
|
from interventions.permissions import filter_viewable_interventions_for_user
|
|
viewable_interventions = filter_viewable_interventions_for_user(user)
|
|
|
|
conditions.append(Q(mission_intervention__in=viewable_interventions))
|
|
conditions.append(Q(checked_intervention__in=viewable_interventions))
|
|
|
|
interv_ct = ContentType.objects.filter(app_label='interventions', model='intervention').first()
|
|
if interv_ct:
|
|
conditions.append(Q(asset_content_type=interv_ct, asset_object_id__in=viewable_interventions.values('pk')))
|
|
|
|
# 3. Assets cibles visibles
|
|
if user_config:
|
|
from common.models import UserThematics, UserContractAccess, UserAssetAccess, UserAssetTypeAccess, AssetTypePermissionConfig
|
|
from assets.permissions import _get_asset_thematic
|
|
|
|
viewable_thematic_ids = set(
|
|
UserThematics.objects.filter(user_config=user_config, can_view_assets=True).values_list('thematic_id', flat=True)
|
|
)
|
|
must_limit = not user_config.is_intern and user_config.limit_assets_to_contracts
|
|
accessible_contract_ids = list(
|
|
UserContractAccess.objects.filter(user_config=user_config, can_view_assets=True).values_list('contract_id', flat=True)
|
|
) if must_limit else []
|
|
|
|
explicit_denies = defaultdict(set)
|
|
for row in UserAssetAccess.objects.filter(user_config=user_config, can_view=False).values_list('content_type_id', 'object_id'):
|
|
explicit_denies[row[0]].add(row[1])
|
|
|
|
explicit_allows = defaultdict(set)
|
|
for row in UserAssetAccess.objects.filter(user_config=user_config, can_view=True).values_list('content_type_id', 'object_id'):
|
|
explicit_allows[row[0]].add(row[1])
|
|
|
|
# ContentTypes présents dans les inspections ciblées
|
|
ct_ids = set(qs.values_list('asset_content_type_id', flat=True).distinct())
|
|
|
|
for ct_id in ct_ids:
|
|
if not ct_id:
|
|
continue
|
|
try:
|
|
ct = ContentType.objects.get(pk=ct_id)
|
|
except ContentType.DoesNotExist:
|
|
continue
|
|
|
|
if ct.app_label == 'interventions':
|
|
continue
|
|
|
|
ModelClass = ct.model_class()
|
|
if not ModelClass:
|
|
continue
|
|
|
|
# Vérifier si AssetTypePermissionConfig exige des permissions explicites
|
|
requires_explicit = AssetTypePermissionConfig.objects.filter(content_type=ct, requires_explicit_permissions=True).exists()
|
|
type_access = UserAssetTypeAccess.objects.filter(user_config=user_config, content_type=ct).first()
|
|
|
|
if type_access and not type_access.can_view:
|
|
allowed_ids = explicit_allows.get(ct.id, set())
|
|
if allowed_ids:
|
|
conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids))
|
|
continue
|
|
|
|
if requires_explicit:
|
|
allowed_ids = explicit_allows.get(ct.id, set())
|
|
if allowed_ids:
|
|
conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids))
|
|
continue
|
|
|
|
# Vérifier l'accès thématique
|
|
th = _get_asset_thematic(ModelClass)
|
|
if th and th.id not in viewable_thematic_ids:
|
|
allowed_ids = explicit_allows.get(ct.id, set())
|
|
if allowed_ids:
|
|
conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids))
|
|
continue
|
|
|
|
# Thématique autorisée (ou sans thématique assignée)
|
|
if must_limit:
|
|
filtered_asset_qs = get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids)
|
|
if filtered_asset_qs is not None:
|
|
ct_q = Q(asset_content_type=ct, asset_object_id__in=filtered_asset_qs.values('pk'))
|
|
else:
|
|
ct_q = Q(asset_content_type=ct)
|
|
else:
|
|
ct_q = Q(asset_content_type=ct)
|
|
|
|
denied_ids = explicit_denies.get(ct.id, set())
|
|
if denied_ids:
|
|
ct_q &= ~Q(asset_object_id__in=denied_ids)
|
|
|
|
allowed_ids = explicit_allows.get(ct.id, set())
|
|
if allowed_ids:
|
|
ct_q |= Q(asset_content_type=ct, asset_object_id__in=allowed_ids)
|
|
|
|
conditions.append(ct_q)
|
|
|
|
if not conditions:
|
|
return qs.none()
|
|
|
|
q = reduce(or_, conditions)
|
|
return qs.filter(q).distinct()
|