811 lines
No EOL
31 KiB
Python
811 lines
No EOL
31 KiB
Python
from django.shortcuts import render, get_object_or_404, redirect
|
||
from django.http import JsonResponse, Http404
|
||
from django.urls import reverse
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from django.contrib.gis.db.models.functions import Transform
|
||
from django.contrib.gis.geos import Point, GEOSGeometry
|
||
from django.contrib.gis.db.models.functions import Distance
|
||
from django.views.decorators.http import require_GET, require_POST
|
||
from django.template.loader import render_to_string
|
||
from django.db.models import Value, OuterRef, Subquery, Q, Case, When
|
||
from django.db.models import IntegerField, CharField, BooleanField
|
||
from django.db.models import Func, F
|
||
from django.db.models.functions import Lower
|
||
from django.utils.translation import gettext as _, get_language
|
||
|
||
import json
|
||
import unicodedata
|
||
|
||
from common.models import Thematic, UserConfig, UserContractAccess, UserThematics, Municipality
|
||
from common.utils import check_thematic_access
|
||
|
||
from assets.models import RoadStreet, RoadStreetDocument, Roadway, RoadSidewalk, RoadFurniture, RoadFurnitureModel, AssetCategory
|
||
from assets.forms import RoadStreetDocumentForm
|
||
from assets.models.roads import QUALITY_LEVEL_CHOICES, RoadDrain, RoadConcreteBlockSet
|
||
|
||
from interventions.models import InterventionLocation
|
||
from interventions.views.drafts import InterventionDraftManager
|
||
|
||
|
||
@check_thematic_access('roads')
|
||
def roads_index(request):
|
||
thematic = 'roads'
|
||
lang = get_language()
|
||
|
||
# Get filters from URL
|
||
statuses = request.GET.getlist('status')
|
||
selected_municipalities = request.GET.getlist('municipality')
|
||
selected_categories = request.GET.getlist('category')
|
||
selected_models = request.GET.getlist('model')
|
||
selected_levels = request.GET.getlist('quality_level')
|
||
|
||
# Get preventive draft ID (if any) for selecting assets
|
||
preventive_draft_id = request.GET.get('preventive_draft_id')
|
||
if preventive_draft_id:
|
||
drafts = InterventionDraftManager(request.session)
|
||
draft_data = drafts.get(preventive_draft_id)
|
||
form_data = draft_data.get('form_data', {})
|
||
preventive_element_type = form_data.get('type_of_selected_elements')
|
||
if preventive_element_type == 'assets':
|
||
selected_assets = draft_data.get('assets', []) or []
|
||
else:
|
||
selected_locations_ids = draft_data.get('locations', [])
|
||
|
||
thematic_choices_qs = (
|
||
UserThematics.objects
|
||
.filter(user_config=request.user.config, can_view_assets=True)
|
||
.select_related('thematic')
|
||
.order_by(f'thematic__{"name_nl" if lang == "nl" else "name_fr"}')
|
||
)
|
||
thematics = [tc.thematic for tc in thematic_choices_qs]
|
||
|
||
selected_thematic = {
|
||
"code": thematic,
|
||
"name": next((t.name for t in thematics if t.code == thematic), None),
|
||
"icon": next((t.icon for t in thematics if t.code == thematic), None),
|
||
}
|
||
|
||
from assets.permissions import get_user_accessible_municipalities
|
||
municipalities = list(get_user_accessible_municipalities(request.user))
|
||
municipalities.sort(key=lambda m: m.name.lower())
|
||
municipality_choices = [(m.id, m.name) for m in municipalities] if len(municipalities) > 1 else []
|
||
|
||
categories = list(AssetCategory.objects.filter(thematic__code = 'roads'))
|
||
categories.sort(key=lambda m: m.get_name().lower())
|
||
category_choices = [(m.id, m.get_name()) for m in categories]
|
||
|
||
models = list(RoadFurnitureModel.objects.all())
|
||
models.sort(key=lambda m: (m.get_name() or "").lower())
|
||
model_choices = [(m.id, m.get_name()) for m in models]
|
||
|
||
quality_levels = QUALITY_LEVEL_CHOICES
|
||
quality_levels_choices = [(level[0], level[1]) for level in quality_levels]
|
||
|
||
context = {
|
||
'thematics': thematics,
|
||
'municipality_choices':municipality_choices,
|
||
'category_choices':category_choices,
|
||
'model_choices':model_choices,
|
||
'quality_levels_choices':quality_levels_choices,
|
||
'selected_thematic': selected_thematic,
|
||
'selected_statuses': statuses,
|
||
'selected_municipalities': selected_municipalities,
|
||
'selected_categories': selected_categories,
|
||
'selected_models': selected_models,
|
||
'selected_levels': selected_levels,
|
||
'preventive_draft_id': preventive_draft_id,
|
||
'preventive_element_type': preventive_element_type if 'preventive_element_type' in locals() else '',
|
||
'selected_locations_ids': selected_locations_ids if 'selected_locations_ids' in locals() else [],
|
||
"selected_assets_json": json.dumps(selected_assets) if 'selected_assets' in locals() else [],
|
||
'locations_tab_title': _("Voiries"),
|
||
'locations_assets_title': "Assets",
|
||
}
|
||
|
||
return render(request, "assets/roads_index.html", context)
|
||
|
||
|
||
|
||
def roads_streets(request):
|
||
pass
|
||
|
||
def roads_assets(request):
|
||
pass
|
||
|
||
|
||
def ensure_list(value):
|
||
if isinstance(value, list):
|
||
return value
|
||
if value is None:
|
||
return []
|
||
return [value]
|
||
|
||
|
||
@check_thematic_access('roads')
|
||
def roads_streets_geojson(request):
|
||
if request.method == "POST":
|
||
try:
|
||
data = json.loads(request.body)
|
||
except json.JSONDecodeError:
|
||
return JsonResponse({"error": "Invalid JSON"}, status=400)
|
||
|
||
municipalities = ensure_list(data.get('municipality'))
|
||
|
||
filter_geom_raw = data.get("filter_geom")
|
||
if filter_geom_raw:
|
||
try:
|
||
filter_geom = GEOSGeometry(json.dumps(filter_geom_raw['geometry']), srid=4326)
|
||
filter_geom.transform(3812)
|
||
except Exception as e:
|
||
return JsonResponse({"error": f"Invalid geometry: {str(e)}"}, status=400)
|
||
else:
|
||
filter_geom = None
|
||
else:
|
||
municipalities = request.GET.getlist('municipality')
|
||
filter_geom = None
|
||
|
||
from assets.permissions import get_user_accessible_municipalities
|
||
user = request.user
|
||
user_config = getattr(user, 'config', None)
|
||
is_admin = getattr(user, 'is_superuser', False) or (user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')))
|
||
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)
|
||
|
||
# --------- Point d'entrée du queryset
|
||
qs = RoadStreet.objects.all()
|
||
|
||
# --------- Filtre par communes et restriction contractuelle
|
||
if not is_admin and must_limit:
|
||
accessible_munis = get_user_accessible_municipalities(user)
|
||
accessible_muni_ids = list(accessible_munis.values_list('id', flat=True))
|
||
if municipalities:
|
||
valid_munis = [mid for mid in municipalities if (int(mid) if str(mid).isdigit() else mid) in accessible_muni_ids]
|
||
qs = qs.filter(municipality_id__in=valid_munis)
|
||
else:
|
||
qs = qs.filter(municipality_id__in=accessible_muni_ids)
|
||
elif municipalities:
|
||
qs = qs.filter(municipality_id__in=municipalities)
|
||
|
||
# --------- Filtre géométrique
|
||
if filter_geom:
|
||
qs = qs.filter(geom__intersects=filter_geom)
|
||
|
||
# Recherche plein texte
|
||
if request.method == 'POST':
|
||
from assets.views.core import apply_search
|
||
qs = apply_search(qs, data)
|
||
|
||
lang = get_language() or "fr"
|
||
name_field = 'name_nl' if lang == 'nl' else 'name_fr'
|
||
name_expr = F(name_field)
|
||
|
||
qs = (
|
||
qs.annotate(
|
||
name=name_expr,
|
||
municipality_name=F('municipality__name'),
|
||
)
|
||
.values('id', 'code', 'lon', 'lat', 'name', 'municipality_name', 'geojson')
|
||
.order_by('name')
|
||
)
|
||
|
||
# --------- Conversion en GeoJSON
|
||
def obj_to_feature(r):
|
||
geom = json.loads(r['geojson']) if r['geojson'] else None
|
||
return {
|
||
"type": "Feature",
|
||
"id": r['id'],
|
||
"geometry": geom,
|
||
"properties": {
|
||
"code": r['code'],
|
||
"municipality": r['municipality_name'] or "-",
|
||
"name": r['name'],
|
||
"lon": r['lon'],
|
||
"lat": r['lat'],
|
||
"detail_url": reverse('assets:roads_streets_detail', kwargs={'street_id': r['id']}),
|
||
"element_type": "Voirie",
|
||
"thematic": "roads",
|
||
"is_location": True,
|
||
}
|
||
}
|
||
|
||
|
||
features = []
|
||
for location in qs:
|
||
features.append(obj_to_feature(location))
|
||
|
||
geojson_data = {
|
||
"type": "FeatureCollection",
|
||
"features": features,
|
||
}
|
||
return JsonResponse(geojson_data, safe=False)
|
||
|
||
|
||
|
||
|
||
MAX_ASSET_FEATURES = 5000 # limite imposée
|
||
|
||
@check_thematic_access('roads')
|
||
def roads_assets_geojson(request):
|
||
if request.method == "POST":
|
||
try:
|
||
data = json.loads(request.body)
|
||
except json.JSONDecodeError:
|
||
return JsonResponse({"error": "Invalid JSON"}, status=400)
|
||
|
||
roads_location_ids = data.get('locations_ids', [])
|
||
statuses = data.get('asset_status', [])
|
||
begin_date = data.get('asset_begin_date')
|
||
categories = ensure_list(data.get('category'))
|
||
models = ensure_list(data.get('model'))
|
||
municipalities = ensure_list(data.get('municipality'))
|
||
quality_levels = ensure_list(data.get('quality_level'))
|
||
|
||
# Récupération et parsing de la géométrie
|
||
filter_geom_raw = data.get("filter_geom")
|
||
if filter_geom_raw:
|
||
try:
|
||
filter_geom = GEOSGeometry(json.dumps(filter_geom_raw['geometry']), srid=4326)
|
||
filter_geom.transform(3812)
|
||
except Exception as e:
|
||
return JsonResponse({"error": f"Invalid geometry: {str(e)}"}, status=400)
|
||
else:
|
||
filter_geom = None
|
||
|
||
else:
|
||
roads_location_id = request.GET.get('location')
|
||
roads_location_ids = [roads_location_id] if roads_location_id else []
|
||
statuses = request.GET.getlist('asset_status')
|
||
begin_date = request.GET.get('asset_begin_date')
|
||
categories = request.GET.getlist('category', [])
|
||
models = request.GET.getlist('model', [])
|
||
municipalities = request.GET.getlist('municipality', [])
|
||
quality_levels = request.GET.getlist('quality_level', [])
|
||
filter_geom = None
|
||
|
||
# Pas de paramètres = pas de carte (comportement existant conservé)
|
||
_search_q = (data.get('q') or '').strip() if request.method == 'POST' else ''
|
||
if not request.GET.get('unified') and not roads_location_ids and not categories and not models and not municipalities and not filter_geom and not quality_levels and not _search_q:
|
||
geojson_data = {"type": "FeatureCollection", "features": []}
|
||
return JsonResponse(geojson_data)
|
||
|
||
from assets.views.core import get_element_type_label
|
||
|
||
def build_geometry_and_coords(obj):
|
||
geometry = None
|
||
if obj.geojson:
|
||
try:
|
||
geometry = json.loads(obj.geojson)
|
||
except (TypeError, ValueError, json.JSONDecodeError):
|
||
geometry = None
|
||
|
||
lon = obj.lon
|
||
lat = obj.lat
|
||
|
||
if lon is None or lat is None:
|
||
street = getattr(obj, 'street', None)
|
||
if street is not None:
|
||
if lon is None:
|
||
lon = getattr(street, 'lon', None)
|
||
if lat is None:
|
||
lat = getattr(street, 'lat', None)
|
||
|
||
if geometry is None and lon is not None and lat is not None:
|
||
geometry = {
|
||
"type": "Point",
|
||
"coordinates": [lon, lat],
|
||
}
|
||
|
||
return geometry, lon, lat
|
||
|
||
def obj_to_feature(obj, asset_type=None):
|
||
geometry, lon, lat = build_geometry_and_coords(obj)
|
||
model = getattr(obj, 'model', None)
|
||
return {
|
||
"type": "Feature",
|
||
"id": obj.id,
|
||
"geometry": geometry,
|
||
"properties": {
|
||
"code": obj.code,
|
||
"name": obj.get_name() if hasattr(obj, "get_name") else (getattr(obj, "name_fr", None) or obj.code),
|
||
"category": obj.category.get_name() if obj.category else None,
|
||
"model": model.get_name() if model else None,
|
||
"quality_level": obj.get_quality_level_display() if hasattr(obj, 'get_quality_level_display') else None,
|
||
"quality_level_value": getattr(obj, 'quality_level', None),
|
||
"status": obj.status,
|
||
"validation_status": obj.validation_status,
|
||
"lon": lon,
|
||
"lat": lat,
|
||
"asset_type": asset_type,
|
||
"area": obj.area if hasattr(obj, 'area') else None,
|
||
"length": obj.length if hasattr(obj, 'length') else None,
|
||
"detail_url": obj.get_absolute_url() if hasattr(obj, 'get_absolute_url') else None,
|
||
"element_type": get_element_type_label(asset_type or ''),
|
||
"thematic": "roads",
|
||
"is_location": False,
|
||
},
|
||
}
|
||
|
||
def add_assets_to_features(queryset, asset_type, features, feature_count):
|
||
for obj in queryset:
|
||
if feature_count >= MAX_ASSET_FEATURES:
|
||
return feature_count, True # early stop
|
||
features.append(obj_to_feature(obj, asset_type=asset_type))
|
||
feature_count += 1
|
||
return feature_count, False
|
||
|
||
# --------- Query de base + optimisations
|
||
roadways_qs = (
|
||
Roadway.objects
|
||
.select_related('street', 'category', 'model')
|
||
.all()
|
||
)
|
||
sidewalks_qs = (
|
||
RoadSidewalk.objects
|
||
.select_related('street', 'category', 'model')
|
||
.all()
|
||
)
|
||
furnitures_qs = (
|
||
RoadFurniture.objects
|
||
.select_related('street', 'category', 'model')
|
||
.all()
|
||
)
|
||
drains_qs = (
|
||
RoadDrain.objects
|
||
.select_related('street', 'model')
|
||
.all()
|
||
)
|
||
concrete_block_sets_qs = (
|
||
RoadConcreteBlockSet.objects
|
||
.select_related('category')
|
||
.all()
|
||
)
|
||
|
||
# --------- Filtres métier existants
|
||
if roads_location_ids:
|
||
roadways_qs = roadways_qs.filter(street__id__in=roads_location_ids)
|
||
sidewalks_qs = sidewalks_qs.filter(street__id__in=roads_location_ids)
|
||
furnitures_qs = furnitures_qs.filter(street__id__in=roads_location_ids)
|
||
drains_qs = drains_qs.filter(street__id__in=roads_location_ids)
|
||
concrete_block_sets_qs = concrete_block_sets_qs.none()
|
||
|
||
from assets.permissions import get_user_accessible_municipalities
|
||
user = request.user
|
||
user_config = getattr(user, 'config', None)
|
||
is_admin = getattr(user, 'is_superuser', False) or (user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')))
|
||
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 not is_admin and must_limit:
|
||
accessible_munis = get_user_accessible_municipalities(user)
|
||
accessible_muni_ids = list(accessible_munis.values_list('id', flat=True))
|
||
if municipalities:
|
||
valid_munis = [mid for mid in municipalities if (int(mid) if str(mid).isdigit() else mid) in accessible_muni_ids]
|
||
roadways_qs = roadways_qs.filter(street__municipality_id__in=valid_munis)
|
||
sidewalks_qs = sidewalks_qs.filter(street__municipality_id__in=valid_munis)
|
||
furnitures_qs = furnitures_qs.filter(street__municipality_id__in=valid_munis)
|
||
drains_qs = drains_qs.filter(street__municipality_id__in=valid_munis)
|
||
else:
|
||
roadways_qs = roadways_qs.filter(street__municipality_id__in=accessible_muni_ids)
|
||
sidewalks_qs = sidewalks_qs.filter(street__municipality_id__in=accessible_muni_ids)
|
||
furnitures_qs = furnitures_qs.filter(street__municipality_id__in=accessible_muni_ids)
|
||
drains_qs = drains_qs.filter(street__municipality_id__in=accessible_muni_ids)
|
||
concrete_block_sets_qs = concrete_block_sets_qs.none()
|
||
elif municipalities:
|
||
roadways_qs = roadways_qs.filter(street__municipality_id__in=municipalities)
|
||
sidewalks_qs = sidewalks_qs.filter(street__municipality_id__in=municipalities)
|
||
furnitures_qs = furnitures_qs.filter(street__municipality_id__in=municipalities)
|
||
drains_qs = drains_qs.filter(street__municipality_id__in=municipalities)
|
||
concrete_block_sets_qs = concrete_block_sets_qs.none()
|
||
|
||
if models:
|
||
roadways_qs = roadways_qs.filter(model__id__in=models)
|
||
sidewalks_qs = sidewalks_qs.filter(model__id__in=models)
|
||
furnitures_qs = furnitures_qs.filter(model__id__in=models)
|
||
drains_qs = drains_qs.filter(model__id__in=models)
|
||
concrete_block_sets_qs = concrete_block_sets_qs.none()
|
||
|
||
if categories:
|
||
roadways_qs = roadways_qs.filter(category__id__in=categories)
|
||
sidewalks_qs = sidewalks_qs.filter(category__id__in=categories)
|
||
furnitures_qs = furnitures_qs.filter(category__id__in=categories)
|
||
drains_qs = drains_qs.filter(category__id__in=categories)
|
||
concrete_block_sets_qs = concrete_block_sets_qs.filter(category__id__in=categories)
|
||
|
||
if filter_geom:
|
||
roadways_qs = roadways_qs.filter(
|
||
Q(geom__intersects=filter_geom) |
|
||
Q(geom__isnull=True, street__geom__intersects=filter_geom)
|
||
)
|
||
sidewalks_qs = sidewalks_qs.filter(
|
||
Q(geom__intersects=filter_geom) |
|
||
Q(geom__isnull=True, street__geom__intersects=filter_geom)
|
||
)
|
||
furnitures_qs = furnitures_qs.filter(
|
||
Q(geom__intersects=filter_geom) |
|
||
Q(geom__isnull=True, street__geom__intersects=filter_geom)
|
||
)
|
||
drains_qs = drains_qs.filter(
|
||
Q(geom__intersects=filter_geom) |
|
||
Q(geom__isnull=True, street__geom__intersects=filter_geom)
|
||
)
|
||
concrete_block_sets_qs = concrete_block_sets_qs.filter(geom__intersects=filter_geom)
|
||
|
||
if quality_levels:
|
||
roadways_qs = roadways_qs.filter(quality_level__in=quality_levels)
|
||
sidewalks_qs = sidewalks_qs.filter(quality_level__in=quality_levels)
|
||
furnitures_qs = furnitures_qs.none() # pas de quality_level sur furnitures
|
||
drains_qs = drains_qs.none() # pas de quality_level sur les avaloirs
|
||
concrete_block_sets_qs = concrete_block_sets_qs.none() # pas de quality_level sur les blocs béton
|
||
|
||
# Recherche plein texte
|
||
if request.method == 'POST':
|
||
from assets.views.core import apply_search
|
||
roadways_qs = apply_search(roadways_qs, data)
|
||
sidewalks_qs = apply_search(sidewalks_qs, data)
|
||
furnitures_qs = apply_search(furnitures_qs, data)
|
||
drains_qs = apply_search(drains_qs, data)
|
||
concrete_block_sets_qs = apply_search(concrete_block_sets_qs, data, fields=['code', 'name_fr', 'name_nl', 'address'])
|
||
|
||
# --------- Cap de sécurité
|
||
roadways_qs = roadways_qs[0:MAX_ASSET_FEATURES+1]
|
||
if len(roadways_qs) >= MAX_ASSET_FEATURES:
|
||
return JsonResponse({"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES})"}, status=413)
|
||
|
||
sidewalks_qs = sidewalks_qs[0:MAX_ASSET_FEATURES+1]
|
||
if len(sidewalks_qs) >= MAX_ASSET_FEATURES:
|
||
return JsonResponse({"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES})"}, status=413)
|
||
|
||
furnitures_qs = furnitures_qs[0:MAX_ASSET_FEATURES+1]
|
||
if len(furnitures_qs) >= MAX_ASSET_FEATURES:
|
||
return JsonResponse({"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES})"}, status=413)
|
||
|
||
drains_qs = drains_qs[0:MAX_ASSET_FEATURES+1]
|
||
if len(drains_qs) >= MAX_ASSET_FEATURES:
|
||
return JsonResponse({"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES})"}, status=413)
|
||
|
||
concrete_block_sets_qs = concrete_block_sets_qs[0:MAX_ASSET_FEATURES+1]
|
||
if len(concrete_block_sets_qs) >= MAX_ASSET_FEATURES:
|
||
return JsonResponse({"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES})"}, status=413)
|
||
|
||
|
||
# --------- Construction GeoJSON
|
||
features = []
|
||
feature_count = 0
|
||
|
||
for qs, asset_type in [
|
||
(roadways_qs, "roadway"),
|
||
]:
|
||
feature_count, limit_reached = add_assets_to_features(qs, asset_type, features, feature_count)
|
||
if limit_reached:
|
||
return JsonResponse({
|
||
"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES}). {_("Limitez votre sélection.")}"
|
||
}, status=413)
|
||
|
||
for qs, asset_type in [
|
||
(sidewalks_qs, "sidewalk"),
|
||
]:
|
||
feature_count, limit_reached = add_assets_to_features(qs, asset_type, features, feature_count)
|
||
if limit_reached:
|
||
return JsonResponse({
|
||
"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES}). {_("Limitez votre sélection.")}"
|
||
}, status=413)
|
||
|
||
for qs, asset_type in [
|
||
(furnitures_qs, "streetfurniture"),
|
||
]:
|
||
feature_count, limit_reached = add_assets_to_features(qs, asset_type, features, feature_count)
|
||
if limit_reached:
|
||
return JsonResponse({
|
||
"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES}). {_("Limitez votre sélection.")}"
|
||
}, status=413)
|
||
|
||
for qs, asset_type in [
|
||
(drains_qs, "drain"),
|
||
]:
|
||
feature_count, limit_reached = add_assets_to_features(qs, asset_type, features, feature_count)
|
||
if limit_reached:
|
||
return JsonResponse({
|
||
"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES}). {_("Limitez votre sélection.")}"
|
||
}, status=413)
|
||
|
||
for qs, asset_type in [
|
||
(concrete_block_sets_qs, "roadconcreteblockset"),
|
||
]:
|
||
feature_count, limit_reached = add_assets_to_features(qs, asset_type, features, feature_count)
|
||
if limit_reached:
|
||
return JsonResponse({
|
||
"error": f"{_("Trop d'objets à afficher")} (> {MAX_ASSET_FEATURES}). {_("Limitez votre sélection.")}"
|
||
}, status=413)
|
||
|
||
|
||
features = sorted(features, key=lambda f: f["properties"]["code"])
|
||
|
||
geojson_data = {
|
||
"type": "FeatureCollection",
|
||
"features": features,
|
||
}
|
||
return JsonResponse(geojson_data, safe=False)
|
||
|
||
|
||
|
||
|
||
|
||
@check_thematic_access('roads')
|
||
def roads_streets_detail(request, street_id):
|
||
street = get_object_or_404(RoadStreet, pk=street_id)
|
||
|
||
ct_roads_street = ContentType.objects.get_for_model(RoadStreet)
|
||
interventions = [obj.intervention for obj in InterventionLocation.objects.filter(content_type=ct_roads_street, object_id=street.id).select_related('intervention').order_by('-intervention__creation_time')]
|
||
|
||
assets = street.get_active_assets()
|
||
|
||
as_built_doc = street.documents.filter(document_type='as_built').filter(status='active').order_by('-id').first()
|
||
other_docs = street.documents.filter(document_type='other').filter(status='active').order_by('-id')
|
||
|
||
street_geom_4326 = street.geom.transform(4326, clone=True)
|
||
geom = json.loads(street_geom_4326.geojson)
|
||
|
||
return render(request, "assets/roads_street_detail.html",
|
||
{
|
||
"street":street,
|
||
"interventions":interventions,
|
||
"assets":assets,
|
||
'main_documents': [as_built_doc,],
|
||
"other_documents": other_docs,
|
||
'add_document_form': RoadStreetDocumentForm(),
|
||
"geom": geom
|
||
})
|
||
|
||
|
||
|
||
@check_thematic_access('roads')
|
||
def roads_assets_detail(request, asset_model, asset_id):
|
||
content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first()
|
||
if not content_type:
|
||
raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model})
|
||
model_class = content_type.model_class()
|
||
if not model_class:
|
||
raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model})
|
||
|
||
# Récupérer l’objet asset
|
||
asset = get_object_or_404(model_class, id=asset_id)
|
||
|
||
# Choisir dynamiquement le template à utiliser
|
||
template_name = f"roads/detail_{asset_model}.html"
|
||
|
||
# Si le template spécifique n’existe pas, tu peux fallback vers un template générique :
|
||
from django.template.loader import select_template
|
||
try:
|
||
template = select_template([template_name, "assets/detail_generic.html"])
|
||
except:
|
||
raise Http404(_("Aucun template trouvé pour ce type d’asset."))
|
||
|
||
# Get edit context
|
||
from assets.views.core import get_asset_edit_context
|
||
context = {
|
||
"asset": asset,
|
||
"content_type": content_type,
|
||
}
|
||
context.update(get_asset_edit_context(request, asset, content_type))
|
||
|
||
# Override thematic_code for roads
|
||
context['thematic_code'] = 'roads'
|
||
|
||
return render(request, template.template.name, context)
|
||
|
||
|
||
|
||
|
||
# Appelle la fonction PostgreSQL unaccent(...)
|
||
class Unaccent(Func):
|
||
function = 'public.unaccent_immutable'
|
||
output_field = CharField()
|
||
|
||
def strip_accents_lower(s: str) -> str:
|
||
if not s:
|
||
return ''
|
||
s = ''.join(c for c in unicodedata.normalize('NFD', s) if unicodedata.category(c) != 'Mn')
|
||
return s.lower()
|
||
|
||
|
||
|
||
|
||
@check_thematic_access('roads', edit_permission_required=True)
|
||
@require_POST
|
||
def roads_streets_add_document(request, street_id):
|
||
roadsstreet = get_object_or_404(RoadStreet, pk=street_id)
|
||
count = int(request.POST.get('file_count', 0))
|
||
files = request.FILES.getlist('files')
|
||
|
||
for i in range(count):
|
||
file = files[i]
|
||
document_type = request.POST.get(f'document_type_{i}', 'other')
|
||
|
||
RoadStreetDocument.objects.create(
|
||
street=roadsstreet,
|
||
file=file,
|
||
document_type=document_type,
|
||
uploaded_by = request.user
|
||
)
|
||
return redirect('assets:roads_streets_detail', street_id=street_id)
|
||
|
||
@check_thematic_access('roads', edit_permission_required=True)
|
||
@require_POST
|
||
def roads_streets_delete_document(request, document_id):
|
||
document = get_object_or_404(RoadStreetDocument, id=document_id)
|
||
street_id = document.street.id # pour rediriger vers la bonne page
|
||
|
||
# Suppression du fichier si nécessaire (en plus de l'objet DB)
|
||
if document.file:
|
||
document.file.delete(save=False)
|
||
if document.thumbnail:
|
||
document.thumbnail.delete(save=False)
|
||
|
||
document.delete()
|
||
#messages.success(request, "Le document a été supprimé avec succès.")
|
||
return redirect('assets:roads_streets_detail', structure_id=street_id)
|
||
|
||
|
||
@check_thematic_access('roads')
|
||
def fetch_document_history(request):
|
||
street_id = request.GET.get('street_id')
|
||
document_type = request.GET.get('document_type')
|
||
|
||
documents = RoadStreetDocument.objects.filter(
|
||
street_id=street_id,
|
||
document_type=document_type
|
||
).order_by('-id')
|
||
|
||
html = render_to_string('assets/document_history_list.html', {'documents': documents})
|
||
return JsonResponse({'html': html})
|
||
|
||
|
||
|
||
|
||
@require_GET
|
||
def get_nearest_roads_street(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)
|
||
|
||
user_point = Point(lon, lat, srid=4326)
|
||
user_point.transform(3812)
|
||
|
||
nearest = (
|
||
RoadStreet.objects
|
||
.filter(geom__isnull=False)
|
||
.annotate(distance=Distance('geom', user_point))
|
||
.filter(distance__lte=50) # filtre à 50 mètres
|
||
.order_by('distance')
|
||
.first()
|
||
)
|
||
|
||
location_ct_id = ContentType.objects.get_for_model(nearest).id
|
||
|
||
if nearest:
|
||
return JsonResponse({
|
||
"location_id": nearest.id,
|
||
"location_ct": location_ct_id,
|
||
"location_label": f"{nearest.code} - {nearest.get_name()}",
|
||
"lon": nearest.lon, "lat": nearest.lat
|
||
})
|
||
else:
|
||
return JsonResponse({"error": _("Aucune rue trouvée")})
|
||
|
||
|
||
@require_GET
|
||
def roads_assets_autocomplete(request):
|
||
raw_query_original = (request.GET.get('q') or '')
|
||
raw_query = raw_query_original.strip()
|
||
|
||
qn = strip_accents_lower(raw_query)
|
||
has_search_term = len(qn) > 0
|
||
|
||
location_id = request.GET.get('location_id') # = id de la rue
|
||
category_id = request.GET.get('asset_category')
|
||
try:
|
||
category_id = int(category_id)
|
||
except (TypeError, ValueError):
|
||
category_id = None
|
||
|
||
# Prépare filtre "dans cette rue uniquement" si location_id fourni
|
||
street_filter = {}
|
||
if location_id:
|
||
street_filter = {'street_id': location_id}
|
||
|
||
|
||
def base_qs(model, extra_filter=None):
|
||
ann = {
|
||
'code_u': Lower(Unaccent('code')),
|
||
'name_fr_u': Lower(Unaccent('name_fr')),
|
||
'name_nl_u': Lower(Unaccent('name_nl')),
|
||
}
|
||
q = model.objects.annotate(**ann)
|
||
|
||
# Si un terme de recherche est fourni, filtrer dessus
|
||
if has_search_term:
|
||
q = q.filter(
|
||
Q(code_u__contains=qn) |
|
||
Q(name_fr_u__contains=qn) |
|
||
Q(name_nl_u__contains=qn)
|
||
)
|
||
|
||
# Filtrer uniquement les non supprimés
|
||
q = q.exclude(status='removed')
|
||
|
||
if extra_filter:
|
||
q = q.filter(**extra_filter)
|
||
|
||
if category_id is not None:
|
||
q = q.filter(
|
||
Q(category_id=category_id) |
|
||
Q(category__isnull=True, model__category_id=category_id)
|
||
)
|
||
|
||
q = q.select_related('street', 'model')
|
||
|
||
return q
|
||
|
||
# Récupère QS pour chaque type (limités grossièrement pour éviter de charger trop, puis triés/merge côté Python)
|
||
roadways_qs = base_qs(Roadway, extra_filter=street_filter)[:100]
|
||
sidewalks_qs = base_qs(RoadSidewalk, extra_filter=street_filter)[:100]
|
||
furnitures_qs = base_qs(RoadFurniture, extra_filter=street_filter)[:100]
|
||
|
||
# Concatène en mémoire (on priorise ensuite)
|
||
items = []
|
||
for a in list(roadways_qs) + list(sidewalks_qs) + list(furnitures_qs):
|
||
# priorisation locale (recalcule ici car on merge des modèles différents)
|
||
code_u = getattr(a, 'code_u', None)
|
||
name_fr_u = getattr(a, 'name_fr_u', None)
|
||
name_nl_u = getattr(a, 'name_nl_u', None)
|
||
|
||
prio = 7
|
||
# Priorisation uniquement si un terme de recherche est fourni
|
||
if has_search_term:
|
||
if code_u == qn:
|
||
prio = 0
|
||
elif code_u and code_u.startswith(qn):
|
||
prio = 1
|
||
elif (name_fr_u and name_fr_u.startswith(qn)) or (name_nl_u and name_nl_u.startswith(qn)):
|
||
prio = 2
|
||
elif (code_u and qn in code_u) or (name_fr_u and qn in name_fr_u) or (name_nl_u and qn in name_nl_u):
|
||
prio = 4
|
||
else:
|
||
# Sans terme de recherche, on trie par code
|
||
prio = 5
|
||
code = getattr(a, 'code', '') or ''
|
||
label = equipment_label
|
||
lon, lat = [getattr(a, 'lon', None), getattr(a, 'lat', None)]
|
||
|
||
location_id_val = street.id if street else None
|
||
location_label = (f"{street.code} - {street.get_name()}" if street else None)
|
||
|
||
equipment_ct_id = ContentType.objects.get_for_model(a).id
|
||
location_ct_id = ContentType.objects.get_for_model(street).id if street else None
|
||
|
||
items.append({
|
||
"prio": prio,
|
||
"code": code or "",
|
||
"id": getattr(a, 'id', None),
|
||
"label": label,
|
||
|
||
# attendus par ton autocomplete équipements
|
||
"equipment_id": getattr(a, 'id', None),
|
||
"equipment_label": equipment_label,
|
||
"equipment_type": equipment_type,
|
||
"equipment_ct": equipment_ct_id,
|
||
|
||
"location_id": location_id_val,
|
||
"location_label": location_label,
|
||
"location_ct": location_ct_id,
|
||
|
||
"lon": lon,
|
||
"lat": lat,
|
||
})
|
||
|
||
# Tri global par priorité puis code, coupe à 100
|
||
items.sort(key=lambda x: (x["prio"], x["code"]))
|
||
data = items[:100]
|
||
|
||
return JsonResponse(data, safe=False) |