1035 lines
37 KiB
Python
1035 lines
37 KiB
Python
# --------------------
|
|
# Asset Permissions
|
|
# --------------------
|
|
|
|
from functools import wraps
|
|
from django.shortcuts import get_object_or_404
|
|
from django.core.exceptions import PermissionDenied
|
|
from django.http import HttpResponseForbidden, JsonResponse
|
|
from django.utils.translation import gettext as _
|
|
|
|
from common.models import UserConfig, UserThematics, UserAssetAccess, UserAssetTypeAccess, AssetTypePermissionConfig
|
|
|
|
|
|
# Fields editable by role for assets
|
|
ASSET_EDITABLE_FIELDS_BY_ROLE = {
|
|
'admin': [
|
|
'code', 'external_reference', 'name_fr', 'name_nl', 'status', 'model', 'category',
|
|
'serial_number', 'installation_date', 'brand', 'warranty_duration',
|
|
'last_inspection_date', 'funding_program', 'geom', 'lon', 'lat',
|
|
'location_id', 'intersection_id',
|
|
],
|
|
'manager': [
|
|
'code', 'external_reference', 'name_fr', 'name_nl', 'status', 'model', 'category',
|
|
'serial_number', 'installation_date', 'brand', 'warranty_duration',
|
|
'last_inspection_date', 'funding_program', 'geom', 'lon', 'lat',
|
|
'location_id', 'intersection_id',
|
|
],
|
|
'controller': [
|
|
'code', 'external_reference', 'name_fr', 'name_nl', 'status', 'model', 'category',
|
|
'serial_number', 'installation_date', 'brand', 'warranty_duration',
|
|
'last_inspection_date', 'funding_program', 'geom', 'lon', 'lat',
|
|
'location_id', 'intersection_id',
|
|
],
|
|
'external_manager': [
|
|
'name_fr', 'name_nl', 'external_reference', 'category', 'brand', 'serial_number', 'funding_program'
|
|
],
|
|
'technician': ['name_fr', 'name_nl', 'external_reference', 'category', 'brand', 'serial_number', 'funding_program'],
|
|
'operator': [],
|
|
'observer': [],
|
|
'viewer': [],
|
|
'editor': [
|
|
'code', 'external_reference', 'name_fr', 'name_nl', 'status', 'model', 'category',
|
|
'serial_number', 'installation_date', 'brand', 'warranty_duration',
|
|
'last_inspection_date', 'funding_program', 'geom', 'lon', 'lat',
|
|
'location_id', 'intersection_id',
|
|
],
|
|
}
|
|
|
|
|
|
# Actions allowed by role for assets
|
|
ASSET_ACTIONS_BY_ROLE = {
|
|
'admin': ['edit', 'create', 'archive', 'replace', 'bulk_archive'],
|
|
'manager': ['edit', 'create', 'archive', 'replace', 'bulk_archive'],
|
|
'controller': ['edit', 'create', 'archive', 'replace', 'bulk_archive'],
|
|
'external_manager': ['edit', 'create', 'replace'],
|
|
'operator': [],
|
|
'technician': ['edit', 'create', 'replace'],
|
|
'observer': [],
|
|
'viewer': [],
|
|
'editor': ['edit', 'create', 'archive', 'replace'],
|
|
}
|
|
|
|
|
|
def get_user_asset_thematic_access(user, thematic):
|
|
"""
|
|
Check if user has access to view/edit assets for a given thematic.
|
|
Returns a tuple (can_view, can_edit).
|
|
"""
|
|
if getattr(user, 'is_superuser', False):
|
|
return True, True
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return False, False
|
|
|
|
if user_config.roles.filter(name__in=['admin', 'operator']).exists():
|
|
return True, True
|
|
|
|
try:
|
|
user_thematic = UserThematics.objects.get(
|
|
user_config=user_config,
|
|
thematic=thematic
|
|
)
|
|
return user_thematic.can_view_assets, user_thematic.can_edit_assets
|
|
except UserThematics.DoesNotExist:
|
|
return False, False
|
|
|
|
|
|
|
|
def _check_instance_permission(user, obj, permission_field, thematic_fallback_fn=None):
|
|
"""
|
|
Shared logic for per-instance permission checks (assets and locations).
|
|
|
|
Priority order (mirrors contract permission logic):
|
|
1. Admin role → always True.
|
|
2. Explicit UserAssetAccess row exists → use its value (True or False, no fallback).
|
|
3. No row exists:
|
|
a. AssetTypePermissionConfig.requires_explicit_permissions=True → False.
|
|
b. Otherwise → call thematic_fallback_fn() (defaults to False if None).
|
|
|
|
Args:
|
|
user: Django User instance.
|
|
obj: Asset or Location model instance.
|
|
permission_field: 'can_view', 'can_edit', or 'can_delete'.
|
|
thematic_fallback_fn: callable() → bool; used when no explicit row exists
|
|
and the type does not require explicit permissions.
|
|
"""
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
|
|
# Step 1: admin bypass
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return False
|
|
|
|
if user_config.roles.filter(name='admin').exists():
|
|
return True
|
|
|
|
# Step 2: explicit per-instance row
|
|
ct = ContentType.objects.get_for_model(obj)
|
|
try:
|
|
access = UserAssetAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
object_id=obj.pk,
|
|
)
|
|
return getattr(access, permission_field, False)
|
|
except UserAssetAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Step 3: per-user type-level access (applies to all instances of this content type)
|
|
try:
|
|
type_access = UserAssetTypeAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
)
|
|
return getattr(type_access, permission_field, False)
|
|
except UserAssetTypeAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Step 4a: type-level restriction
|
|
try:
|
|
type_config = AssetTypePermissionConfig.objects.get(content_type=ct)
|
|
if type_config.requires_explicit_permissions:
|
|
return False
|
|
except AssetTypePermissionConfig.DoesNotExist:
|
|
pass
|
|
|
|
# Step 4b: thematic fallback
|
|
if thematic_fallback_fn is not None:
|
|
return thematic_fallback_fn()
|
|
return False
|
|
|
|
|
|
def get_asset_contract_ids(asset):
|
|
"""
|
|
Retourne l'ensemble des IDs de contrats associés à une instance d'asset ou de localisation,
|
|
ou None si le modèle d'asset ne gère pas de contrats.
|
|
"""
|
|
if asset is None:
|
|
return None
|
|
|
|
contract_ids = set()
|
|
has_contract_relation = False
|
|
|
|
# 1. Clés étrangères directes
|
|
if hasattr(asset, 'maintenance_contract_id') and asset.maintenance_contract_id:
|
|
contract_ids.add(asset.maintenance_contract_id)
|
|
has_contract_relation = True
|
|
elif hasattr(asset, 'maintenance_contract'):
|
|
has_contract_relation = True
|
|
|
|
if hasattr(asset, 'controller_maintenance_contract_id') and asset.controller_maintenance_contract_id:
|
|
contract_ids.add(asset.controller_maintenance_contract_id)
|
|
has_contract_relation = True
|
|
elif hasattr(asset, 'controller_maintenance_contract'):
|
|
has_contract_relation = True
|
|
|
|
if hasattr(asset, 'contract_id') and asset.contract_id:
|
|
contract_ids.add(asset.contract_id)
|
|
has_contract_relation = True
|
|
elif hasattr(asset, 'contract'):
|
|
has_contract_relation = True
|
|
|
|
# 2. Relations ManyToMany ou Reverse managers (contracts, etc.)
|
|
if hasattr(asset, 'contracts'):
|
|
has_contract_relation = True
|
|
try:
|
|
model = asset.contracts.model
|
|
if model.__name__ == 'Contract':
|
|
for cid in asset.contracts.values_list('pk', flat=True):
|
|
contract_ids.add(cid)
|
|
elif hasattr(model, 'contract_id'):
|
|
qs = asset.contracts.all()
|
|
if hasattr(model, 'status'):
|
|
qs = qs.filter(status='active')
|
|
for cid in qs.values_list('contract_id', flat=True):
|
|
contract_ids.add(cid)
|
|
except Exception:
|
|
pass
|
|
|
|
for rel_name in [
|
|
'trafficlight_contracts', 'structure_contracts', 'its_locations_contracts',
|
|
'its_assets_contracts', 'controlcenters_contracts', 'controlcenters_assets_contracts',
|
|
'clean_location_contracts'
|
|
]:
|
|
if hasattr(asset, rel_name):
|
|
has_contract_relation = True
|
|
try:
|
|
manager = getattr(asset, rel_name)
|
|
if hasattr(manager.model, 'status'):
|
|
cids = manager.filter(status='active').values_list('contract_id', flat=True)
|
|
else:
|
|
cids = manager.values_list('contract_id', flat=True)
|
|
contract_ids.update(cids)
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. Assets enfants liés à un parent (intersection, location, structure, control_center, pole, etc.)
|
|
for parent_attr in ['intersection', 'location', 'structure', 'control_center', 'building', 'pole']:
|
|
if hasattr(asset, parent_attr):
|
|
parent = getattr(asset, parent_attr)
|
|
if parent:
|
|
parent_cids = get_asset_contract_ids(parent)
|
|
if parent_cids is not None:
|
|
has_contract_relation = True
|
|
contract_ids.update(parent_cids)
|
|
|
|
if not has_contract_relation:
|
|
return None
|
|
return contract_ids
|
|
|
|
|
|
def can_view_asset(user, asset):
|
|
"""Check if user can view a specific asset."""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
user_config = None
|
|
|
|
if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False):
|
|
return True
|
|
if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')):
|
|
return True
|
|
|
|
def _fallback():
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
can_view, _ = get_user_asset_thematic_access(user, thematic)
|
|
if not can_view:
|
|
return False
|
|
|
|
# Vérifier la restriction par contrat et commune pour les utilisateurs externes ou limités aux contrats
|
|
is_external = (user_config is None) or (not user_config.is_intern)
|
|
must_limit = is_external or (user_config and user_config.limit_assets_to_contracts)
|
|
if must_limit and user_config:
|
|
asset_cids = get_asset_contract_ids(asset)
|
|
if asset_cids is not None:
|
|
from common.models import UserContractAccess
|
|
accessible_cids = set(
|
|
UserContractAccess.objects.filter(
|
|
user_config=user_config,
|
|
can_view_assets=True
|
|
).values_list('contract_id', flat=True)
|
|
)
|
|
if user_config.default_contract_id:
|
|
accessible_cids.add(user_config.default_contract_id)
|
|
if not (asset_cids & accessible_cids):
|
|
return False
|
|
else:
|
|
# Vérifier la restriction par commune pour les assets sans contrat direct
|
|
accessible_munis = get_user_accessible_municipalities(user)
|
|
accessible_muni_ids = set(accessible_munis.values_list('id', flat=True))
|
|
if accessible_muni_ids:
|
|
muni_id = getattr(asset, 'municipality_id', None)
|
|
if not muni_id and hasattr(asset, 'location') and asset.location:
|
|
muni_id = getattr(asset.location, 'municipality_id', None)
|
|
if not muni_id and hasattr(asset, 'street') and asset.street:
|
|
muni_id = getattr(asset.street, 'municipality_id', None)
|
|
if not muni_id and hasattr(asset, 'pole') and asset.pole:
|
|
muni_id = getattr(asset.pole, 'municipality_id', None)
|
|
if not muni_id and hasattr(asset.pole, 'street') and asset.pole.street:
|
|
muni_id = getattr(asset.pole.street, 'municipality_id', None)
|
|
if not muni_id and hasattr(asset, 'intersection') and asset.intersection:
|
|
muni_id = getattr(asset.intersection, 'municipality_id', None)
|
|
if not muni_id and hasattr(asset, 'building') and asset.building:
|
|
muni_id = getattr(asset.building, 'municipality_id', None)
|
|
if muni_id is not None:
|
|
if muni_id not in accessible_muni_ids:
|
|
return False
|
|
elif getattr(asset, 'geom', None):
|
|
from common.models import Municipality
|
|
if not Municipality.objects.filter(id__in=accessible_muni_ids, geom__intersects=asset.geom).exists():
|
|
return False
|
|
return True
|
|
|
|
return _check_instance_permission(user, asset, 'can_view', _fallback)
|
|
|
|
|
|
def can_edit_asset(user, asset):
|
|
"""Check if user can edit a specific asset."""
|
|
def _fallback():
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
_, can_edit = get_user_asset_thematic_access(user, thematic)
|
|
return can_edit
|
|
|
|
return _check_instance_permission(user, asset, 'can_edit', _fallback)
|
|
|
|
|
|
def can_delete_asset(user, asset):
|
|
"""Check if user can delete (archive/remove) a specific asset."""
|
|
def _fallback():
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
_, can_edit = get_user_asset_thematic_access(user, thematic)
|
|
# By default, delete follows edit rights at the thematic level
|
|
return can_edit
|
|
|
|
return _check_instance_permission(user, asset, 'can_delete', _fallback)
|
|
|
|
|
|
def can_validate_asset(user, asset):
|
|
"""Check if user has permission to validate a specific asset."""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return False
|
|
|
|
if user_config.roles.filter(name='admin').exists():
|
|
return True
|
|
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
|
|
try:
|
|
ut = UserThematics.objects.get(user_config=user_config, thematic=thematic)
|
|
return ut.can_validate_assets
|
|
except UserThematics.DoesNotExist:
|
|
return False
|
|
|
|
|
|
def can_inspect_asset(user, asset):
|
|
"""Check if user has permission to perform an inspection on a specific asset."""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
if not hasattr(user, 'config') or not user.config:
|
|
return False
|
|
if user.config.roles.filter(name='admin').exists():
|
|
return True
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False
|
|
inspectable = user.config.get_inspectable_thematics()
|
|
return inspectable.filter(pk=thematic.pk).exists() if inspectable is not None else False
|
|
|
|
|
|
def can_view_location(user, location):
|
|
"""Check if user can view a specific location."""
|
|
def _fallback():
|
|
thematic = _get_location_thematic(location)
|
|
if not thematic:
|
|
return False
|
|
can_view, _ = get_user_asset_thematic_access(user, thematic)
|
|
return can_view
|
|
|
|
return _check_instance_permission(user, location, 'can_view', _fallback)
|
|
|
|
|
|
def can_edit_location(user, location):
|
|
"""Check if user can edit a specific location."""
|
|
def _fallback():
|
|
thematic = _get_location_thematic(location)
|
|
if not thematic:
|
|
return False
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
ut = UserThematics.objects.get(user_config=user_config, thematic=thematic)
|
|
return ut.can_edit_locations
|
|
except Exception:
|
|
return False
|
|
|
|
return _check_instance_permission(user, location, 'can_edit', _fallback)
|
|
|
|
|
|
def can_delete_location(user, location):
|
|
"""Check if user can delete a specific location."""
|
|
def _fallback():
|
|
thematic = _get_location_thematic(location)
|
|
if not thematic:
|
|
return False
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
ut = UserThematics.objects.get(user_config=user_config, thematic=thematic)
|
|
return ut.can_edit_locations
|
|
except Exception:
|
|
return False
|
|
|
|
return _check_instance_permission(user, location, 'can_delete', _fallback)
|
|
|
|
|
|
def _get_asset_thematic(asset, thematics_by_code=None):
|
|
"""Get the thematic for an asset instance or model class."""
|
|
if asset is None:
|
|
return None
|
|
|
|
def _resolve_thematic(code):
|
|
if thematics_by_code is not None:
|
|
return thematics_by_code.get(code)
|
|
from common.models import Thematic
|
|
return Thematic.objects.filter(code=code).first()
|
|
|
|
thematic_mappings = {
|
|
'naturerwiz': 'water',
|
|
'naturerwiasset': 'water',
|
|
'trafficlight': 'trafficlights',
|
|
'publiclighting': 'publiclighting',
|
|
'road': 'roads',
|
|
'artwork': 'artworks',
|
|
'monument': 'artworks',
|
|
'fountain': 'artworks',
|
|
'structure': 'structures',
|
|
'nature': 'nature',
|
|
'green': 'nature',
|
|
'sign': 'sign',
|
|
'controlcenter': 'controlcenters',
|
|
'its': 'its',
|
|
'building': 'publicbuildings',
|
|
'publicbuilding': 'publicbuildings',
|
|
'parking': 'parking',
|
|
'parkingspot': 'parking',
|
|
'parkinglocation': 'parking',
|
|
'clean': 'clean',
|
|
'cleanlocation': 'clean',
|
|
'cleanlitterbin': 'clean',
|
|
'cleanglasscontainer': 'clean',
|
|
}
|
|
|
|
if isinstance(asset, type):
|
|
class_name = asset.__name__.lower()
|
|
for prefix, thematic_code in thematic_mappings.items():
|
|
if class_name.startswith(prefix):
|
|
th = _resolve_thematic(thematic_code)
|
|
if th:
|
|
return th
|
|
# Fallback: check AssetCategory via ContentType
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from assets.models import AssetCategory
|
|
ct = ContentType.objects.filter(model=class_name).first()
|
|
if ct:
|
|
cat = AssetCategory.objects.filter(allowed_models=ct).first()
|
|
if cat and cat.thematic:
|
|
return cat.thematic
|
|
return None
|
|
|
|
# Try from category
|
|
try:
|
|
if hasattr(asset, 'category') and asset.category and hasattr(asset.category, 'thematic'):
|
|
return asset.category.thematic
|
|
except Exception:
|
|
pass
|
|
|
|
# Try direct thematic attribute
|
|
try:
|
|
if hasattr(asset, 'thematic') and asset.thematic and hasattr(asset.thematic, 'code'):
|
|
return asset.thematic
|
|
except Exception:
|
|
pass
|
|
|
|
# Try from parent building category
|
|
try:
|
|
if hasattr(asset, 'building') and asset.building and hasattr(asset.building, 'category') and asset.building.category:
|
|
return asset.building.category.thematic
|
|
except Exception:
|
|
pass
|
|
|
|
# Try from model's category
|
|
try:
|
|
if hasattr(asset, 'model') and asset.model and hasattr(asset.model, 'category') and asset.model.category:
|
|
return asset.model.category.thematic
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback: deduce thematic from asset class name
|
|
class_name = asset.__class__.__name__.lower()
|
|
for prefix, thematic_code in thematic_mappings.items():
|
|
if class_name.startswith(prefix):
|
|
th = _resolve_thematic(thematic_code)
|
|
if th:
|
|
return th
|
|
|
|
return None
|
|
|
|
|
|
def _get_location_thematic(location):
|
|
"""Deduce the thematic for a location instance."""
|
|
from common.models import Thematic
|
|
|
|
# RoadStreet, NatureLocation, etc. have a `thematic` attribute via their linked assets/categories,
|
|
# but locations are usually tied to a thematic via class name.
|
|
class_name = location.__class__.__name__.lower()
|
|
thematic_mappings = {
|
|
'naturerwiz': 'water',
|
|
'roadstreet': 'roads',
|
|
'trafficlightintersection': 'trafficlights',
|
|
'naturelocation': 'nature',
|
|
'structurelocation': 'structures',
|
|
'controlcenter': 'controlcenters',
|
|
'itslocation': 'its',
|
|
'cleanlocation': 'clean',
|
|
'publiclightingstreet': 'publiclighting',
|
|
'signstreet': 'sign',
|
|
'publicbuilding': 'publicbuildings',
|
|
'buildinglocation': 'publicbuildings',
|
|
}
|
|
thematic_code = thematic_mappings.get(class_name)
|
|
if not thematic_code:
|
|
for prefix, code in thematic_mappings.items():
|
|
if class_name.startswith(prefix[:5]):
|
|
thematic_code = code
|
|
break
|
|
if not thematic_code:
|
|
return None
|
|
try:
|
|
return Thematic.objects.get(code=thematic_code)
|
|
except Thematic.DoesNotExist:
|
|
return None
|
|
|
|
|
|
def get_allowed_update_fields_for_asset(user, asset):
|
|
"""
|
|
Get the list of fields the user is allowed to update for a specific asset.
|
|
|
|
Priority:
|
|
1. If a UserAssetAccess row with non-empty editable_fields exists, that list is returned
|
|
as-is (completely replaces role-based logic).
|
|
2. Otherwise uses the existing role-based logic (ASSET_EDITABLE_FIELDS_BY_ROLE).
|
|
"""
|
|
if not can_edit_asset(user, asset):
|
|
return []
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return []
|
|
|
|
# Check for per-instance editable_fields override
|
|
from django.contrib.contenttypes.models import ContentType
|
|
ct = ContentType.objects.get_for_model(asset)
|
|
try:
|
|
access = UserAssetAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
object_id=asset.pk,
|
|
)
|
|
if access.editable_fields: # non-null, non-empty list → use it
|
|
return list(access.editable_fields)
|
|
except UserAssetAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Check for per-type editable_fields override (applies to all instances)
|
|
try:
|
|
type_access = UserAssetTypeAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
)
|
|
if type_access.editable_fields: # non-null, non-empty list → use it
|
|
return list(type_access.editable_fields)
|
|
except UserAssetTypeAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Fall back to role-based logic
|
|
allowed_fields = set()
|
|
for role in user_config.roles.all():
|
|
role_fields = ASSET_EDITABLE_FIELDS_BY_ROLE.get(role.name, [])
|
|
allowed_fields.update(role_fields)
|
|
|
|
# Check validation permission
|
|
if can_validate_asset(user, asset):
|
|
allowed_fields.add('validation_status')
|
|
|
|
# Also add asset-specific fields (not in base models) when the user already has some edit rights
|
|
if allowed_fields and asset is not None:
|
|
from django.db.models import ForeignKey, OneToOneField, ManyToManyField, AutoField
|
|
from django.contrib.gis.db.models import GeometryField
|
|
from assets.models import AbstractAsset, AbstractGeoAsset
|
|
|
|
excluded_types = (ForeignKey, OneToOneField, ManyToManyField, AutoField, GeometryField)
|
|
base_field_names = {f.name for f in AbstractAsset._meta.fields}
|
|
base_field_names |= {f.name for f in AbstractGeoAsset._meta.fields}
|
|
base_field_names.add('geojson')
|
|
|
|
for field in type(asset)._meta.fields:
|
|
if field.name in base_field_names:
|
|
continue
|
|
if isinstance(field, excluded_types):
|
|
continue
|
|
allowed_fields.add(field.name)
|
|
|
|
return list(allowed_fields)
|
|
|
|
|
|
def get_visible_fields_for_asset(user, asset):
|
|
"""
|
|
Get the list of fields visible to the user for a specific asset.
|
|
|
|
Priority (mirrors get_allowed_update_fields_for_asset):
|
|
1. UserAssetAccess.visible_fields (non-empty) → return it.
|
|
2. UserAssetTypeAccess.visible_fields (non-empty) → return it.
|
|
3. No restriction defined → return None (all fields visible).
|
|
"""
|
|
if not can_view_asset(user, asset):
|
|
return []
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return None
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
ct = ContentType.objects.get_for_model(asset)
|
|
|
|
# Check per-instance override
|
|
try:
|
|
access = UserAssetAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
object_id=asset.pk,
|
|
)
|
|
if access.visible_fields:
|
|
return list(access.visible_fields)
|
|
except UserAssetAccess.DoesNotExist:
|
|
pass
|
|
|
|
# Check per-type override
|
|
try:
|
|
type_access = UserAssetTypeAccess.objects.get(
|
|
user_config=user_config,
|
|
content_type=ct,
|
|
)
|
|
if type_access.visible_fields:
|
|
return list(type_access.visible_fields)
|
|
except UserAssetTypeAccess.DoesNotExist:
|
|
pass
|
|
|
|
# No restriction → all fields visible
|
|
return None
|
|
|
|
|
|
def get_allowed_actions_for_asset(user, asset):
|
|
"""
|
|
Get the list of actions the user is allowed to perform on assets.
|
|
"""
|
|
if not can_edit_asset(user, asset):
|
|
return []
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return []
|
|
|
|
allowed_actions = set()
|
|
for role in user_config.roles.all():
|
|
role_actions = ASSET_ACTIONS_BY_ROLE.get(role.name, [])
|
|
allowed_actions.update(role_actions)
|
|
|
|
return list(allowed_actions)
|
|
|
|
|
|
def get_allowed_actions_for_thematic(user, thematic):
|
|
"""
|
|
Get the list of actions the user is allowed to perform for a thematic.
|
|
"""
|
|
can_view, can_edit = get_user_asset_thematic_access(user, thematic)
|
|
if not can_edit:
|
|
return []
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return []
|
|
|
|
allowed_actions = set()
|
|
for role in user_config.roles.all():
|
|
role_actions = ASSET_ACTIONS_BY_ROLE.get(role.name, [])
|
|
allowed_actions.update(role_actions)
|
|
|
|
return list(allowed_actions)
|
|
|
|
|
|
def asset_edit_permission_required(view_func):
|
|
"""
|
|
Decorator that checks if user has permission to edit assets.
|
|
Expects asset_id and asset_model as URL parameters.
|
|
"""
|
|
@wraps(view_func)
|
|
def wrapper(request, *args, **kwargs):
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
asset_model = kwargs.get('asset_model')
|
|
asset_id = kwargs.get('asset_id')
|
|
|
|
if not asset_model or not asset_id:
|
|
return HttpResponseForbidden(_("Missing asset information"))
|
|
|
|
try:
|
|
cts = ContentType.objects.filter(model=asset_model.lower())
|
|
if not cts.exists():
|
|
return HttpResponseForbidden(_("Invalid asset type"))
|
|
cts_sorted = sorted(cts, key=lambda ct: (0 if ct.app_label == 'assets' else 1))
|
|
model_class = None
|
|
for ct in cts_sorted:
|
|
mc = ct.model_class()
|
|
if mc is not None:
|
|
model_class = mc
|
|
break
|
|
if model_class is None:
|
|
return HttpResponseForbidden(_("Invalid asset type"))
|
|
asset = get_object_or_404(model_class, pk=asset_id)
|
|
except Exception:
|
|
return HttpResponseForbidden(_("Invalid asset type"))
|
|
|
|
if not can_edit_asset(request.user, asset):
|
|
return HttpResponseForbidden(_("You do not have permission to edit this asset"))
|
|
|
|
return view_func(request, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def can_edit_location_for_thematic(user, thematic_code):
|
|
"""
|
|
Check if the user has permission to create/edit/delete locations for a given thematic.
|
|
Relies on UserThematics.can_edit_locations.
|
|
"""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
from common.models import Thematic
|
|
thematic = Thematic.objects.get(code=thematic_code)
|
|
user_thematic = UserThematics.objects.get(user_config=user_config, thematic=thematic)
|
|
return user_thematic.can_edit_locations
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def location_asset_edit_permission_required(thematic_code):
|
|
"""
|
|
Decorator that checks if user has permission to edit assets for a thematic.
|
|
Used for location-based asset operations.
|
|
"""
|
|
def decorator(view_func):
|
|
@wraps(view_func)
|
|
def wrapper(request, *args, **kwargs):
|
|
from common.models import Thematic
|
|
|
|
try:
|
|
thematic = Thematic.objects.get(code=thematic_code)
|
|
except Thematic.DoesNotExist:
|
|
return HttpResponseForbidden(_("Invalid thematic"))
|
|
|
|
can_view, can_edit = get_user_asset_thematic_access(request.user, thematic)
|
|
if not can_edit:
|
|
return HttpResponseForbidden(_("You do not have permission to edit assets for this thematic"))
|
|
|
|
return view_func(request, *args, **kwargs)
|
|
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def can_view_exceptional_transport(user):
|
|
"""
|
|
Vérifie si l'utilisateur a le droit de visualiser les données et documents de transport exceptionnel sur les ouvrages.
|
|
"""
|
|
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 not user_config:
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except Exception:
|
|
return False
|
|
if user_config.has_role('admin') or user_config.has_role('top_manager'):
|
|
return True
|
|
return bool(getattr(user_config, 'has_exceptional_transport_profile', False))
|
|
|
|
|
|
def can_edit_exceptional_transport(user):
|
|
"""
|
|
Vérifie si l'utilisateur a le droit de modifier les données de transport exceptionnel sur les ouvrages.
|
|
"""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
if getattr(user, 'is_superuser', False):
|
|
return True
|
|
user_config = getattr(user, 'config', None)
|
|
if not user_config:
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except Exception:
|
|
return False
|
|
if user_config.has_role('admin'):
|
|
return True
|
|
return bool(
|
|
getattr(user_config, 'has_exceptional_transport_profile', False)
|
|
and getattr(user_config, 'exceptional_transport_can_edit', False)
|
|
)
|
|
|
|
|
|
|
|
def get_user_accessible_municipalities(user):
|
|
"""
|
|
Retourne le queryset des communes (Municipality) auxquelles l'utilisateur a accès.
|
|
- Si superuser ou (utilisateur interne sans limitation aux contrats) :
|
|
retourne toutes les communes (Municipality.objects.all()).
|
|
- Si utilisateur externe ou avec limit_assets_to_contracts=True :
|
|
retourne uniquement les communes associées à ses contrats accessibles
|
|
(via Contract.municipalities ou localisations liées).
|
|
"""
|
|
from common.models import Municipality, UserConfig, UserContractAccess
|
|
from django.utils.translation import get_language
|
|
|
|
if not user or not user.is_authenticated:
|
|
return Municipality.objects.none()
|
|
|
|
if getattr(user, 'is_superuser', False):
|
|
lang = get_language() or 'fr'
|
|
order_field = 'name_nl' if lang.startswith('nl') else 'name_fr'
|
|
return Municipality.objects.all().order_by(order_field)
|
|
|
|
user_config = getattr(user, 'config', None)
|
|
if not user_config:
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return Municipality.objects.none()
|
|
|
|
if user_config.has_role('admin') or user_config.has_role('top_manager'):
|
|
lang = get_language() or 'fr'
|
|
order_field = 'name_nl' if lang.startswith('nl') else 'name_fr'
|
|
return Municipality.objects.all().order_by(order_field)
|
|
|
|
is_external = not user_config.is_intern
|
|
must_limit = is_external or user_config.limit_assets_to_contracts
|
|
|
|
if not must_limit:
|
|
lang = get_language() or 'fr'
|
|
order_field = 'name_nl' if lang.startswith('nl') else 'name_fr'
|
|
return Municipality.objects.all().order_by(order_field)
|
|
|
|
# Utilisateur restreint par contrats :
|
|
accessible_contract_ids = set(
|
|
UserContractAccess.objects.filter(
|
|
user_config=user_config,
|
|
can_view_assets=True
|
|
).values_list('contract_id', flat=True)
|
|
)
|
|
if user_config.default_contract_id:
|
|
accessible_contract_ids.add(user_config.default_contract_id)
|
|
|
|
if not accessible_contract_ids:
|
|
return Municipality.objects.none()
|
|
|
|
# Communes associées via ManyToMany Contract.municipalities
|
|
municipality_ids = set(
|
|
Municipality.objects.filter(
|
|
contracts__id__in=accessible_contract_ids
|
|
).values_list('id', flat=True)
|
|
)
|
|
|
|
# Communes associées via les localisations liées aux contrats
|
|
try:
|
|
from assets.models import CleanLocationContract
|
|
clean_muni_ids = CleanLocationContract.objects.filter(
|
|
contract_id__in=accessible_contract_ids,
|
|
location__municipality__isnull=False
|
|
).values_list('location__municipality_id', flat=True)
|
|
municipality_ids.update(clean_muni_ids)
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
from assets.models import TrafficLightContract
|
|
tl_muni_ids = TrafficLightContract.objects.filter(
|
|
contract_id__in=accessible_contract_ids,
|
|
intersection__municipality__isnull=False
|
|
).values_list('intersection__municipality_id', flat=True)
|
|
municipality_ids.update(tl_muni_ids)
|
|
except Exception:
|
|
pass
|
|
|
|
if not municipality_ids:
|
|
return Municipality.objects.none()
|
|
|
|
lang = get_language() or 'fr'
|
|
order_field = 'name_nl' if lang.startswith('nl') else 'name_fr'
|
|
return Municipality.objects.filter(id__in=municipality_ids).order_by(order_field)
|
|
|
|
|
|
# ─── Management Zones Permissions ─────────────────────────────────────────────
|
|
|
|
def can_view_management_zones(user, thematic=None):
|
|
"""
|
|
Vérifie si l'utilisateur a le droit de visualiser les zones de gestion.
|
|
Si thematic est fourni (instance Thematic ou code str), vérifie pour cette thématique.
|
|
"""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
if getattr(user, 'is_superuser', False) or user.has_perm('assets.view_managementzone'):
|
|
return True
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return False
|
|
|
|
if user_config.roles.filter(name__in=['admin', 'operator', 'manager']).exists():
|
|
return True
|
|
|
|
if thematic:
|
|
thematic_obj = thematic
|
|
if isinstance(thematic, str):
|
|
from common.models import Thematic
|
|
thematic_obj = Thematic.objects.filter(code=thematic).first()
|
|
if not thematic_obj:
|
|
return False
|
|
|
|
try:
|
|
user_thematic = UserThematics.objects.get(user_config=user_config, thematic=thematic_obj)
|
|
return user_thematic.can_view_assets
|
|
except UserThematics.DoesNotExist:
|
|
return False
|
|
|
|
return UserThematics.objects.filter(user_config=user_config, can_view_assets=True).exists()
|
|
|
|
|
|
def can_edit_management_zones(user, thematic=None):
|
|
"""
|
|
Vérifie si l'utilisateur a le droit de créer ou modifier les zones de gestion.
|
|
Autorisé si:
|
|
- Superutilisateur ou permission Django explicite (assets.change_managementzone / assets.add_managementzone)
|
|
- Rôle 'admin' ou 'manager'
|
|
- Ou UserThematics.can_edit_management_zones == True (ou fallback can_edit_assets)
|
|
"""
|
|
from django.db import models as db_models
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
if getattr(user, 'is_superuser', False):
|
|
return True
|
|
if user.has_perm('assets.change_managementzone') or user.has_perm('assets.add_managementzone'):
|
|
return True
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return False
|
|
|
|
if user_config.roles.filter(name__in=['admin', 'manager']).exists():
|
|
return True
|
|
|
|
if thematic:
|
|
thematic_obj = thematic
|
|
if isinstance(thematic, str):
|
|
from common.models import Thematic
|
|
thematic_obj = Thematic.objects.filter(code=thematic).first()
|
|
if not thematic_obj:
|
|
return False
|
|
|
|
try:
|
|
user_thematic = UserThematics.objects.get(user_config=user_config, thematic=thematic_obj)
|
|
return bool(
|
|
getattr(user_thematic, 'can_edit_management_zones', False) or
|
|
user_thematic.can_edit_assets
|
|
)
|
|
except UserThematics.DoesNotExist:
|
|
return False
|
|
|
|
return UserThematics.objects.filter(
|
|
user_config=user_config
|
|
).filter(
|
|
db_models.Q(can_edit_management_zones=True) | db_models.Q(can_edit_assets=True)
|
|
).exists()
|
|
|
|
|
|
def can_delete_management_zone(user, zone):
|
|
"""Vérifie si l'utilisateur a le droit de supprimer une zone de gestion."""
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
if getattr(user, 'is_superuser', False) or user.has_perm('assets.delete_managementzone'):
|
|
return True
|
|
return can_edit_management_zones(user, thematic=zone.thematic)
|
|
|
|
|
|
def get_editable_thematics_for_management_zones(user):
|
|
"""
|
|
Retourne le queryset des thématiques pour lesquelles l'utilisateur peut créer/éditer des zones de gestion.
|
|
"""
|
|
from common.models import Thematic
|
|
from django.db import models as db_models
|
|
if not user or not user.is_authenticated:
|
|
return Thematic.objects.none()
|
|
if getattr(user, 'is_superuser', False) or user.has_perm('assets.change_managementzone'):
|
|
return Thematic.objects.all()
|
|
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
except UserConfig.DoesNotExist:
|
|
return Thematic.objects.none()
|
|
|
|
if user_config.roles.filter(name__in=['admin', 'manager']).exists():
|
|
return Thematic.objects.all()
|
|
|
|
return Thematic.objects.filter(
|
|
userthematics__user_config=user_config
|
|
).filter(
|
|
db_models.Q(userthematics__can_edit_management_zones=True) |
|
|
db_models.Q(userthematics__can_edit_assets=True)
|
|
).distinct()
|