590 lines
22 KiB
Python
590 lines
22 KiB
Python
from django.http import JsonResponse, FileResponse, Http404
|
||
from django.shortcuts import render, redirect
|
||
from django.views.decorators.http import require_GET, require_POST
|
||
from django.contrib.auth.decorators import login_required
|
||
from django.db.models import Q
|
||
from django.contrib import messages
|
||
from django.contrib.gis.db.models.functions import Distance
|
||
from django.contrib.gis.geos import Point
|
||
from django.contrib.auth import login as auth_login
|
||
from django.contrib.auth.decorators import login_not_required
|
||
from django.contrib.auth.forms import PasswordChangeForm
|
||
from django.contrib.auth import update_session_auth_hash
|
||
from django.utils.http import url_has_allowed_host_and_scheme
|
||
from django.utils.translation import gettext as _
|
||
from django.conf import settings
|
||
from django.urls import reverse
|
||
|
||
import re
|
||
import os
|
||
from urllib.parse import urlencode
|
||
|
||
from common.context_processors import (
|
||
SESSION_ALLOWED_IDS_KEY,
|
||
SESSION_ORIGIN_ID_KEY,
|
||
compute_initial_allowed_ids,
|
||
)
|
||
|
||
from common.models import AdrStreet, AdrPoint, UserConfig
|
||
|
||
from assets.models import AssetCategory
|
||
|
||
@login_not_required
|
||
@require_GET
|
||
def autocomplete_street(request):
|
||
query = request.GET.get('q', '').strip()
|
||
if len(query) < 2:
|
||
return JsonResponse([], safe=False)
|
||
|
||
# Regex patterns to detect number and street in different positions
|
||
patterns = [
|
||
("number_first", r'^(\d+\w*)[\s,]+(.+)$'), # ex: "12A rue de la Loi"
|
||
("street_first", r'^(.+)[\s,]+(\d+\w*)$'), # ex: "rue de la Loi 12A"
|
||
]
|
||
|
||
number_part = None
|
||
street_part = None
|
||
|
||
for mode, pat in patterns:
|
||
match = re.match(pat, query)
|
||
if match:
|
||
g1, g2 = match.groups()
|
||
if mode == "number_first":
|
||
number_part, street_part = g1.strip(), g2.strip()
|
||
else:
|
||
street_part, number_part = g1.strip(), g2.strip()
|
||
break
|
||
|
||
if number_part and street_part:
|
||
# Recherche par AdrPoint avec numéro ET nom de rue
|
||
points = (
|
||
AdrPoint.objects
|
||
.filter(number__iexact=number_part)
|
||
.filter(
|
||
Q(street_name__name_fr__icontains=street_part) |
|
||
Q(street_name__name_nl__icontains=street_part)
|
||
)
|
||
.select_related('street_name', 'municipality')[:10]
|
||
)
|
||
|
||
data = [{
|
||
"id": point.fid,
|
||
"label": f"{point.number} {point.street_name.name_fr} / {point.street_name.name_nl}"
|
||
+ (f" ({point.municipality.name})" if point.municipality else ""),
|
||
"lon": point.lon,
|
||
"lat": point.lat
|
||
} for point in points]
|
||
|
||
else:
|
||
# Recherche standard par rue uniquement
|
||
streets = (
|
||
AdrStreet.objects
|
||
.filter(Q(name_fr__icontains=query) | Q(name_nl__icontains=query))
|
||
.select_related('municipality')[:10]
|
||
)
|
||
|
||
data = [{
|
||
"id": street.fid,
|
||
"label": f"{street.name_fr} / {street.name_nl}"
|
||
+ (f" ({street.municipality.name})" if street.municipality else ""),
|
||
"lon": street.lon,
|
||
"lat": street.lat
|
||
} for street in streets]
|
||
|
||
return JsonResponse(data, safe=False)
|
||
|
||
|
||
@login_not_required
|
||
@require_GET
|
||
def get_nearest_address(request):
|
||
try:
|
||
lon = float(request.GET.get("lon"))
|
||
lat = float(request.GET.get("lat"))
|
||
except (TypeError, ValueError):
|
||
return JsonResponse({"error": _("Coordonnées invalides")}, status=400)
|
||
|
||
input_point = Point(lon, lat, srid=4326)
|
||
input_point.transform(3812)
|
||
|
||
# Étape 1 : Trouver la rue la plus proche
|
||
nearest_street = (
|
||
AdrStreet.objects
|
||
.filter(geom__isnull=False)
|
||
.annotate(distance=Distance("geom", input_point))
|
||
.filter(distance__lte=50) # filtre à 50 mètres
|
||
.order_by("distance")
|
||
.first()
|
||
)
|
||
|
||
if not nearest_street:
|
||
return JsonResponse({"error": _("Aucune rue trouvée")}, status=404)
|
||
|
||
# Étape 2 : Rechercher un point dans un rayon de 50 m pour cette rue
|
||
nearest_point = (
|
||
AdrPoint.objects
|
||
.filter(street_name=nearest_street, geom__isnull=False)
|
||
.annotate(distance=Distance("geom", input_point))
|
||
.filter(distance__lte=50) # filtre à 50 mètres
|
||
.order_by("distance")
|
||
.first()
|
||
)
|
||
|
||
# Étape 3 : Si un point est trouvé dans les 50 m, on le retourne
|
||
if nearest_point:
|
||
return JsonResponse({
|
||
"id": nearest_point.id,
|
||
"label": f"{nearest_point.number} {nearest_point.street_name.name_fr} / {nearest_point.street_name.name_nl}"
|
||
+ (f" ({nearest_point.municipality.name})" if nearest_point.municipality else "")
|
||
})
|
||
|
||
# Sinon : on retourne uniquement le nom de la rue
|
||
return JsonResponse({
|
||
"id": nearest_street.id,
|
||
"label": f"{nearest_street.name_fr} / {nearest_street.name_nl}"
|
||
+ (f" ({nearest_street.municipality.name})" if nearest_street.municipality else ""),
|
||
})
|
||
|
||
|
||
|
||
@login_not_required
|
||
@require_GET
|
||
def get_regional_roads_geojson(request):
|
||
file_path = os.path.join(settings.BASE_DIR, 'common', 'static', 'geojson', 'regional_roads_simplified_4326.geojson')
|
||
|
||
if not os.path.exists(file_path):
|
||
raise Http404("GeoJSON file not found")
|
||
|
||
return FileResponse(open(file_path, 'rb'), content_type='application/json')
|
||
|
||
|
||
|
||
def get_categories_for_thematics(request):
|
||
ids_string = request.GET.get('thematic_ids', '')
|
||
if ids_string:
|
||
try:
|
||
thematic_ids = [int(x) for x in ids_string.split(',') if x.strip().isdigit()]
|
||
except ValueError:
|
||
return JsonResponse({"results": []})
|
||
else:
|
||
thematic_ids = []
|
||
|
||
categories = AssetCategory.objects.filter(thematic__id__in=thematic_ids).distinct()
|
||
|
||
results = [
|
||
{"id": cat.id, "text": cat.get_name(), "parent_id": cat.parent_id} for cat in categories
|
||
]
|
||
return JsonResponse({"results": results})
|
||
|
||
|
||
|
||
|
||
def _safe_origin_next(request):
|
||
# 1) prendre 'next' si présent à l’arrivée
|
||
origin_next = request.GET.get("origin_next") or request.GET.get("next")
|
||
# 2) sinon, si c’est un POST (ex: retour form), reprendre la valeur transmise
|
||
origin_next = origin_next or request.POST.get("origin_next")
|
||
# 3) safe-guard (évite open redirect)
|
||
if origin_next and not url_has_allowed_host_and_scheme(
|
||
url=origin_next,
|
||
allowed_hosts={request.get_host()},
|
||
require_https=request.is_secure(),
|
||
):
|
||
origin_next = None
|
||
return origin_next
|
||
|
||
@login_required
|
||
def account_info(request):
|
||
origin_next = _safe_origin_next(request) # on capture une fois, puis on la véhicule
|
||
|
||
open_password_modal = False
|
||
if request.method == "POST" and "password_change" in request.POST:
|
||
form = PasswordChangeForm(request.user, request.POST)
|
||
if form.is_valid():
|
||
user = form.save()
|
||
update_session_auth_hash(request, user)
|
||
messages.success(request, _("Mot de passe mis à jour avec succès."))
|
||
# On redirige vers account_info en conservant origin_next + flag
|
||
params = {"password_changed": "1"}
|
||
if origin_next:
|
||
params["origin_next"] = origin_next
|
||
return redirect(f"{request.path}?{urlencode(params)}")
|
||
else:
|
||
open_password_modal = True
|
||
else:
|
||
form = PasswordChangeForm(request.user)
|
||
|
||
if request.GET.get("password_changed") == "1":
|
||
open_password_modal = True
|
||
|
||
return render(request, "common/account_info.html", {
|
||
"form": form,
|
||
"open_password_modal": open_password_modal,
|
||
"origin_next": origin_next, # <- clé : toujours renvoyer au template
|
||
})
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def switch_account(request):
|
||
origin_next = _safe_origin_next(request)
|
||
target_user_id = request.POST.get("target_user_id")
|
||
|
||
try:
|
||
current_config: UserConfig = request.user.config
|
||
except UserConfig.DoesNotExist:
|
||
messages.error(request, _("Aucun compte lié n'est disponible."))
|
||
return redirect(origin_next or reverse("home:index"))
|
||
|
||
if not target_user_id:
|
||
messages.error(request, _("Aucun compte cible n'a été sélectionné."))
|
||
return redirect(origin_next or reverse("home:index"))
|
||
|
||
if str(request.user.pk) == str(target_user_id):
|
||
messages.info(request, _("Ce compte est déjà actif."))
|
||
return redirect(origin_next or reverse("home:index"))
|
||
|
||
session = request.session
|
||
|
||
allowed_ids = session.get(SESSION_ALLOWED_IDS_KEY)
|
||
origin_user_id = session.get(SESSION_ORIGIN_ID_KEY)
|
||
|
||
if allowed_ids is None:
|
||
allowed_ids = compute_initial_allowed_ids(current_config)
|
||
|
||
if origin_user_id is None:
|
||
origin_user_id = request.user.pk
|
||
|
||
allowed_ids = list(dict.fromkeys(int(pk) for pk in allowed_ids))
|
||
|
||
if request.user.pk not in allowed_ids:
|
||
allowed_ids.append(request.user.pk)
|
||
|
||
try:
|
||
target_user_pk = int(target_user_id)
|
||
except (TypeError, ValueError):
|
||
messages.error(request, _("Vous ne pouvez pas accéder à ce compte."))
|
||
return redirect(origin_next or reverse("home:index"))
|
||
|
||
if target_user_pk not in allowed_ids:
|
||
messages.error(request, _("Vous ne pouvez pas accéder à ce compte."))
|
||
return redirect(origin_next or reverse("home:index"))
|
||
|
||
session[SESSION_ALLOWED_IDS_KEY] = allowed_ids
|
||
session[SESSION_ORIGIN_ID_KEY] = origin_user_id
|
||
|
||
try:
|
||
target_config = (
|
||
UserConfig.objects.select_related("user").get(user__pk=target_user_pk)
|
||
)
|
||
except UserConfig.DoesNotExist:
|
||
messages.error(request, _("Vous ne pouvez pas accéder à ce compte."))
|
||
return redirect(origin_next or reverse("home:index"))
|
||
|
||
target_user = target_config.user
|
||
target_user.backend = "django.contrib.auth.backends.ModelBackend"
|
||
auth_login(request, target_user)
|
||
messages.success(request, _("Changement de compte effectué."))
|
||
|
||
session[SESSION_ALLOWED_IDS_KEY] = allowed_ids
|
||
session[SESSION_ORIGIN_ID_KEY] = origin_user_id
|
||
|
||
return redirect(origin_next or reverse("home:index"))
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def save_intervention_list_layout(request):
|
||
"""API endpoint to save the user's preferred layout for the interventions list."""
|
||
import json
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
layout = data.get('layout')
|
||
except (json.JSONDecodeError, KeyError):
|
||
return JsonResponse({'error': _('Données invalides')}, status=400)
|
||
|
||
valid_layouts = ['both', 'table_only', 'map_only']
|
||
if layout not in valid_layouts:
|
||
return JsonResponse({'error': _('Layout invalide')}, status=400)
|
||
|
||
try:
|
||
user_config = request.user.config
|
||
user_config.intervention_list_layout = layout
|
||
user_config.save(update_fields=['intervention_list_layout'])
|
||
return JsonResponse({'success': True, 'layout': layout})
|
||
except UserConfig.DoesNotExist:
|
||
return JsonResponse({'error': _('Configuration utilisateur non trouvée')}, status=404)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def save_assets_list_layout(request):
|
||
"""API endpoint to save the user's preferred layout for the assets list."""
|
||
import json
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
layout = data.get('layout')
|
||
except (json.JSONDecodeError, KeyError):
|
||
return JsonResponse({'error': _('Données invalides')}, status=400)
|
||
|
||
valid_layouts = ['both', 'table_only', 'map_only']
|
||
if layout not in valid_layouts:
|
||
return JsonResponse({'error': _('Layout invalide')}, status=400)
|
||
|
||
try:
|
||
user_config = request.user.config
|
||
user_config.assets_list_layout = layout
|
||
user_config.save(update_fields=['assets_list_layout'])
|
||
return JsonResponse({'success': True, 'layout': layout})
|
||
except UserConfig.DoesNotExist:
|
||
return JsonResponse({'error': _('Configuration utilisateur non trouvée')}, status=404)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def save_intervention_list_preferences(request):
|
||
"""API endpoint to save the user's preferred filters and sort for the interventions list."""
|
||
import json
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
filters = data.get('filters', {})
|
||
sort = data.get('sort', {})
|
||
columns = data.get('columns', None)
|
||
except (json.JSONDecodeError, KeyError):
|
||
return JsonResponse({'error': _('Données invalides')}, status=400)
|
||
|
||
try:
|
||
user_config = request.user.config
|
||
|
||
# Validate filters structure (should be a dict with filter names as keys)
|
||
if not isinstance(filters, dict):
|
||
return JsonResponse({'error': _('Format de filtres invalide')}, status=400)
|
||
|
||
# Validate sort structure (should have sortName and sortOrder)
|
||
if not isinstance(sort, dict):
|
||
return JsonResponse({'error': _('Format de tri invalide')}, status=400)
|
||
|
||
update_fields = ['intervention_list_filters', 'intervention_list_sort']
|
||
user_config.intervention_list_filters = filters
|
||
user_config.intervention_list_sort = sort
|
||
|
||
if columns is not None:
|
||
if not isinstance(columns, list):
|
||
return JsonResponse({'error': _('Format de colonnes invalide')}, status=400)
|
||
user_config.intervention_list_columns = columns
|
||
update_fields.append('intervention_list_columns')
|
||
|
||
user_config.save(update_fields=update_fields)
|
||
|
||
return JsonResponse({'success': True, 'filters': filters, 'sort': sort})
|
||
except UserConfig.DoesNotExist:
|
||
return JsonResponse({'error': _('Configuration utilisateur non trouvée')}, status=404)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def save_observation_list_preferences(request):
|
||
"""API endpoint to save the user's preferred filters and sort for the observations list."""
|
||
import json
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
filters = data.get('filters', {})
|
||
sort = data.get('sort', {})
|
||
except (json.JSONDecodeError, KeyError):
|
||
return JsonResponse({'error': _('Données invalides')}, status=400)
|
||
|
||
try:
|
||
user_config = request.user.config
|
||
|
||
if not isinstance(filters, dict):
|
||
return JsonResponse({'error': _('Format de filtres invalide')}, status=400)
|
||
|
||
if not isinstance(sort, dict):
|
||
return JsonResponse({'error': _('Format de tri invalide')}, status=400)
|
||
|
||
user_config.observation_list_filters = filters
|
||
user_config.observation_list_sort = sort
|
||
user_config.save(update_fields=['observation_list_filters', 'observation_list_sort'])
|
||
|
||
return JsonResponse({'success': True, 'filters': filters, 'sort': sort})
|
||
except UserConfig.DoesNotExist:
|
||
return JsonResponse({'error': _('Configuration utilisateur non trouvée')}, status=404)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def save_assets_index_preferences(request):
|
||
"""API endpoint to save the user's preferences for the unified assets index page."""
|
||
import json
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
except json.JSONDecodeError:
|
||
return JsonResponse({'error': _('Données invalides')}, status=400)
|
||
|
||
thematics = data.get('thematics', [])
|
||
categories = data.get('categories', [])
|
||
models = data.get('models', [])
|
||
conditional_filters = data.get('conditional_filters', {})
|
||
visible_columns = data.get('visible_columns')
|
||
type_filter = data.get('type_filter', 'all')
|
||
status_filters = data.get('status_filters', [])
|
||
validation_status_filters = data.get('validation_status_filters', [])
|
||
include_no_geom = data.get('include_no_geom', True)
|
||
map_center = data.get('map_center')
|
||
map_zoom = data.get('map_zoom')
|
||
|
||
if not isinstance(thematics, list):
|
||
return JsonResponse({'error': _('Format de thématiques invalide')}, status=400)
|
||
if not isinstance(categories, list):
|
||
return JsonResponse({'error': _('Format de catégories invalide')}, status=400)
|
||
if not isinstance(models, list):
|
||
return JsonResponse({'error': _('Format de modèles invalide')}, status=400)
|
||
if not isinstance(conditional_filters, dict):
|
||
return JsonResponse({'error': _('Format de filtres thématiques invalide')}, status=400)
|
||
if visible_columns is not None and not isinstance(visible_columns, list):
|
||
return JsonResponse({'error': _('Format de colonnes visibles invalide')}, status=400)
|
||
if not isinstance(type_filter, str):
|
||
return JsonResponse({'error': _('Format de filtre type invalide')}, status=400)
|
||
if not isinstance(status_filters, list):
|
||
return JsonResponse({'error': _('Format de filtres statut invalide')}, status=400)
|
||
if not isinstance(validation_status_filters, list):
|
||
return JsonResponse({'error': _('Format de filtres statut de validation invalide')}, status=400)
|
||
|
||
try:
|
||
user_config, _ = UserConfig.objects.get_or_create(user=request.user)
|
||
existing_prefs = user_config.assets_index_preferences or {}
|
||
existing_visible_columns = existing_prefs.get('visible_columns', []) if isinstance(existing_prefs, dict) else []
|
||
if not isinstance(existing_visible_columns, list):
|
||
existing_visible_columns = []
|
||
|
||
user_config.assets_index_preferences = {
|
||
'thematics': thematics,
|
||
'categories': categories,
|
||
'models': models,
|
||
'conditional_filters': conditional_filters,
|
||
'visible_columns': visible_columns if visible_columns is not None else existing_visible_columns,
|
||
'type_filter': type_filter,
|
||
'status_filters': status_filters,
|
||
'validation_status_filters': validation_status_filters,
|
||
'include_no_geom': include_no_geom,
|
||
'map_center': map_center,
|
||
'map_zoom': map_zoom,
|
||
}
|
||
user_config.save(update_fields=['assets_index_preferences'])
|
||
return JsonResponse({'success': True})
|
||
except UserConfig.DoesNotExist:
|
||
return JsonResponse({'error': _('Configuration utilisateur non trouvée')}, status=404)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def save_calendar_preferences(request):
|
||
"""API endpoint to save the user's preferred filters, view_mode, zoom and backlog states for the calendar."""
|
||
import json
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
preferences = data.get('preferences', {})
|
||
except (json.JSONDecodeError, KeyError):
|
||
return JsonResponse({'error': _('Données invalides')}, status=400)
|
||
|
||
try:
|
||
user_config = request.user.config
|
||
|
||
if not isinstance(preferences, dict):
|
||
return JsonResponse({'error': _('Format de préférences invalide')}, status=400)
|
||
|
||
user_config.calendar_preferences = preferences
|
||
user_config.save(update_fields=['calendar_preferences'])
|
||
|
||
return JsonResponse({'success': True, 'preferences': preferences})
|
||
except UserConfig.DoesNotExist:
|
||
return JsonResponse({'error': _('Configuration utilisateur non trouvée')}, status=404)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def save_project_list_preferences(request):
|
||
"""API endpoint to save the user's preferred filters and sort for the projects list."""
|
||
import json
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
filters = data.get('filters', {})
|
||
sort = data.get('sort', {})
|
||
except (json.JSONDecodeError, KeyError):
|
||
return JsonResponse({'error': _('Données invalides')}, status=400)
|
||
|
||
try:
|
||
user_config = request.user.config
|
||
|
||
if not isinstance(filters, dict):
|
||
return JsonResponse({'error': _('Format de filtres invalide')}, status=400)
|
||
|
||
if not isinstance(sort, dict):
|
||
return JsonResponse({'error': _('Format de tri invalide')}, status=400)
|
||
|
||
user_config.project_list_filters = filters
|
||
user_config.project_list_sort = sort
|
||
user_config.save(update_fields=['project_list_filters', 'project_list_sort'])
|
||
|
||
return JsonResponse({'success': True, 'filters': filters, 'sort': sort})
|
||
except UserConfig.DoesNotExist:
|
||
return JsonResponse({'error': _('Configuration utilisateur non trouvée')}, status=404)
|
||
|
||
|
||
@login_required
|
||
@require_GET
|
||
def get_help_document(request, doc_name):
|
||
"""View to securely load and render Markdown help documents based on user roles and permissions."""
|
||
import markdown
|
||
from django.utils import translation
|
||
from django.http import HttpResponseForbidden
|
||
|
||
ALLOWED_DOCS = ['prise_en_main', 'assets', 'approuver_interventions', 'gerer_projets', 'statuts_intervention', 'creer_controle']
|
||
if doc_name not in ALLOWED_DOCS:
|
||
return JsonResponse({'error': _('Document non trouvé')}, status=404)
|
||
|
||
try:
|
||
user_config = request.user.config
|
||
except UserConfig.DoesNotExist:
|
||
return JsonResponse({'error': _('Configuration utilisateur non trouvée')}, status=404)
|
||
|
||
# Permission check for specific guides
|
||
if doc_name == 'approuver_interventions':
|
||
if not user_config.can_approve_any_intervention:
|
||
return HttpResponseForbidden(_('Accès interdit'))
|
||
elif doc_name in ['gerer_projets', 'creer_controle']:
|
||
has_project_role = user_config.roles.filter(name__in=["admin", "manager", "controller", "external_manager", "top_manager"]).exists()
|
||
if not (has_project_role or request.user.is_superuser):
|
||
return HttpResponseForbidden(_('Accès interdit'))
|
||
|
||
# Determine language directory
|
||
lang = translation.get_language()
|
||
if lang and lang.startswith('nl'):
|
||
lang_dir = 'nl'
|
||
else:
|
||
lang_dir = 'fr'
|
||
|
||
# Resolve file path
|
||
file_path = settings.BASE_DIR / 'common' / 'help_docs' / lang_dir / f'{doc_name}.md'
|
||
if not os.path.exists(file_path):
|
||
return JsonResponse({'error': _('Document non trouvé')}, status=404)
|
||
|
||
try:
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
md_content = f.read()
|
||
except Exception:
|
||
return JsonResponse({'error': _('Erreur lors de la lecture du document')}, status=500)
|
||
|
||
# Convert Markdown to HTML
|
||
html_content = markdown.markdown(md_content, extensions=['extra', 'nl2br'])
|
||
|
||
return JsonResponse({'html': html_content})
|
||
|
||
|
||
|