loko/streetup/assets/views/clean.py
2026-07-22 14:48:40 +02:00

632 lines
22 KiB
Python

from django.shortcuts import render, get_object_or_404
from django.http import JsonResponse, Http404
from django.views.decorators.http import require_POST
from django.db import transaction
from django.utils import timezone
from django.utils.dateparse import parse_datetime
from django.contrib.gis.geos import GEOSGeometry, Point
from django.contrib.gis.db.models.functions import Distance
from django.contrib.contenttypes.models import ContentType
from django.utils.translation import get_language, gettext as _
from django.db.models import F, Q
from django.db.models.functions import Lower
from django.urls import reverse
import json
import unicodedata
from collections import defaultdict
from common.utils import check_thematic_access
from common.models import UserThematics
from assets.models import CleanLocation, CleanLitterBin, CleanLitterBinModel, CleanLitterBinObservation, CleanGlassContainer, CleanGlassContainerModel, AssetCategory
from assets.permissions import can_edit_asset
from interventions.models import InterventionLocation
from interventions.views.drafts import InterventionDraftManager
MAX_ASSET_FEATURES = 5000
# Ordered list of (keyword_variants, canonical_color_name).
# The canonical name must match the keys used in the front-end 'match' expression.
_COLOR_KEYWORDS = [
(['anthracite'], 'gris anthracite'),
(['bordeaux'], 'bordeaux'),
(['violet', 'violette'], 'violet'),
(['beige'], 'beige'),
(['marron', 'brun', 'brune'], 'marron'),
(['orange'], 'orange'),
(['jaune'], 'jaune'),
(['rouge'], 'rouge'),
(['bleu', 'bleue'], 'bleu'),
(['vert', 'verte'], 'vert'),
(['noir', 'noire'], 'noir'),
(['blanc', 'blanche'], 'blanc'),
(['gris', 'grise'], 'gris'),
]
def _color_from_text(text: str | None) -> str:
"""
Return a canonical lowercase colour name by scanning *text* for colour
keywords (word-boundary aware). Returns '' if nothing is recognised.
"""
import re
import unicodedata
if not text:
return ''
# Normalise to ASCII-folded lowercase so accents don't matter
normalised = unicodedata.normalize('NFD', text.lower())
normalised = ''.join(c for c in normalised if unicodedata.category(c) != 'Mn')
for variants, canonical in _COLOR_KEYWORDS:
for kw in variants:
if re.search(r'(?<![a-z])' + re.escape(kw) + r'(?![a-z])', normalised):
return canonical
return ''
# ---------------------
# Assets autocomplete
# ---------------------
def clean_assets_autocomplete(request):
raw_query = (request.GET.get('q') or '').strip()
qn = unicodedata.normalize('NFD', raw_query).encode('ascii', 'ignore').decode('ascii').lower()
has_search_term = len(qn) > 0
location_id = request.GET.get('location_id')
category_id = request.GET.get('asset_category')
try:
category_id = int(category_id)
except (TypeError, ValueError):
category_id = None
def base_qs(model_class):
qs = model_class.objects.annotate(
code_u=Lower('code'),
name_fr_u=Lower('name_fr'),
name_nl_u=Lower('name_nl'),
).filter(status='active')
if location_id:
qs = qs.filter(location_id=location_id)
if category_id is not None:
qs = qs.filter(
Q(category_id=category_id) |
Q(category__isnull=True, model__category_id=category_id)
)
if has_search_term:
qs = qs.filter(
Q(code_u__contains=qn) |
Q(name_fr_u__contains=qn) |
Q(name_nl_u__contains=qn)
)
return qs.select_related('location', 'model')[:100]
items = []
for model_class in (CleanLitterBin, CleanGlassContainer):
for a in base_qs(model_class):
code_u = getattr(a, 'code_u', '') or ''
name_fr_u = getattr(a, 'name_fr_u', '') or ''
name_nl_u = getattr(a, 'name_nl_u', '') or ''
if has_search_term:
if code_u == qn:
prio = 0
elif code_u.startswith(qn):
prio = 1
elif name_fr_u.startswith(qn) or name_nl_u.startswith(qn):
prio = 2
elif qn in code_u or qn in name_fr_u or qn in name_nl_u:
prio = 4
else:
prio = 7
else:
prio = 5
location = a.location
label = " - ".join(p for p in [a.code, a.get_name()] if p)
equipment_ct_id = ContentType.objects.get_for_model(a).id
location_ct_id = ContentType.objects.get_for_model(location).id if location else None
items.append({
"prio": prio,
"code": a.code or "",
"id": a.id,
"label": label,
"equipment_id": a.id,
"equipment_label": label,
"equipment_type": a.model.get_name() if a.model else "",
"equipment_ct": equipment_ct_id,
"location_id": location.id if location else None,
"location_label": f"{location.code} - {location.get_name()}" if location else None,
"location_ct": location_ct_id,
"lon": a.lon,
"lat": a.lat,
})
items.sort(key=lambda x: (x["prio"], x["code"]))
return JsonResponse(items[:100], safe=False)
# ---------------------
# Street detail
# ---------------------
@check_thematic_access('clean')
def clean_location_detail(request, location_id):
location = get_object_or_404(CleanLocation, pk=location_id)
ct = ContentType.objects.get_for_model(CleanLocation)
interventions = [
obj.intervention
for obj in InterventionLocation.objects.filter(content_type=ct, object_id=location.id).select_related('intervention').order_by('-intervention__creation_time')
]
assets = location.get_active_assets()
all_assets = location.get_all_assets()
geom = None
if location.geom:
geom = json.loads(location.geom.transform(4326, clone=True).geojson)
return render(request, "assets/clean_location_detail.html", {
"location": location,
"interventions": interventions,
"assets": assets,
"all_assets": all_assets,
"geom": geom,
})
# ---------------------
# Index
# ---------------------
@check_thematic_access('clean')
def clean_index(request):
thematic = 'clean'
lang = get_language()
statuses = request.GET.getlist('status')
selected_categories = request.GET.getlist('category')
selected_models = request.GET.getlist('model')
preventive_draft_id = request.GET.get('preventive_draft_id')
preventive_element_type = ''
selected_assets = []
selected_locations_ids = []
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),
}
categories = sorted(
AssetCategory.objects.filter(thematic__code='clean'),
key=lambda m: m.get_name().lower()
)
category_choices = [(m.id, m.get_name()) for m in categories]
models = sorted(
list(CleanLitterBinModel.objects.all()) + list(CleanGlassContainerModel.objects.all()),
key=lambda m: (m.get_name() or '').lower()
)
model_choices = [(m.id, m.get_name()) for m in models]
category_model_map = defaultdict(list)
for model in CleanLitterBinModel.objects.all():
if model.category_id:
category_model_map[str(model.category_id)].append({"key": str(model.id), "label": model.get_name()})
for model in CleanGlassContainerModel.objects.all():
if model.category_id:
category_model_map[str(model.category_id)].append({"key": str(model.id), "label": model.get_name()})
context = {
'thematics': thematics,
'category_choices': category_choices,
'model_choices': model_choices,
'category_model_map': dict(category_model_map),
'selected_thematic': selected_thematic,
'selected_statuses': statuses,
'selected_categories': selected_categories,
'selected_models': selected_models,
'preventive_draft_id': preventive_draft_id,
'preventive_element_type': preventive_element_type,
'selected_locations_ids': selected_locations_ids,
'selected_assets_json': json.dumps(selected_assets),
'locations_tab_title': _("Locations"),
'locations_assets_title': _("Assets"),
}
return render(request, "assets/clean_index.html", context)
# ---------------------
# Locations GeoJSON
# ---------------------
@check_thematic_access('clean')
def clean_locations_geojson(request):
if request.method == "POST":
try:
data = json.loads(request.body)
except json.JSONDecodeError:
return JsonResponse({"error": "Invalid JSON"}, status=400)
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:
filter_geom = None
qs = CleanLocation.objects.all()
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)
def obj_to_feature(obj):
geometry = json.loads(obj.geojson) if obj.geojson else None
return {
"type": "Feature",
"id": obj.id,
"geometry": geometry,
"properties": {
"code": obj.code,
"name": obj.get_name(),
"lon": obj.lon,
"lat": obj.lat,
"detail_url": obj.get_absolute_url() if hasattr(obj, 'get_absolute_url') else None,
"element_type": "Zone propreté",
"thematic": "clean",
"is_location": True,
}
}
features = [obj_to_feature(location) for location in qs]
return JsonResponse({"type": "FeatureCollection", "features": features}, safe=False)
# ---------------------
# Assets GeoJSON
# ---------------------
@check_thematic_access('clean')
def clean_assets_geojson(request):
lang = get_language() if get_language() in ('fr', 'nl') else 'fr'
if request.method == "POST":
try:
data = json.loads(request.body)
except json.JSONDecodeError:
return JsonResponse({"error": "Invalid JSON"}, status=400)
locations_ids = data.get('locations_ids', [])
statuses = data.get('asset_status', [])
categories = data.get('category') or []
models = data.get('model') or []
if isinstance(categories, str):
categories = [categories]
if isinstance(models, str):
models = [models]
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:
location_id = request.GET.get('location')
locations_ids = [location_id] if location_id else []
statuses = request.GET.getlist('asset_status')
categories = request.GET.getlist('category', [])
models = request.GET.getlist('model', [])
filter_geom = None
_search_q = (data.get('q') or '').strip() if request.method == 'POST' else ''
if not request.GET.get('unified') and not locations_ids and not categories and not models and not filter_geom and not _search_q:
return JsonResponse({"type": "FeatureCollection", "features": []})
def build_qs(model_class, asset_type_str):
qs = model_class.objects.all()
if locations_ids:
qs = qs.filter(location__id__in=locations_ids)
if categories:
qs = qs.filter(category__id__in=categories)
if models:
qs = qs.filter(model__id__in=models)
if filter_geom:
filter_geom_3812 = filter_geom.transform(3812, clone=True)
qs = qs.filter(
Q(geom__intersects=filter_geom_3812) |
Q(geom__isnull=True, location__geom__intersects=filter_geom_3812)
)
if statuses:
qs = qs.filter(status__in=statuses)
else:
qs = qs.exclude(status='removed')
# Recherche plein texte
if request.method == 'POST':
from assets.views.core import apply_search
qs = apply_search(qs, data)
qs = (
qs.annotate(
name=F(f'name_{lang}'),
category_name=F(f'category__name_{lang}'),
model_name=F(f'model__name_{lang}'),
location_lon=F('location__lon'),
location_lat=F('location__lat'),
)
.values('id', 'code', 'name', 'status', 'validation_status', 'lon', 'lat', 'geojson',
'category_name', 'model_name', 'category_id', 'model_id', 'color',
'location_lon', 'location_lat')
.order_by('code')
)[:MAX_ASSET_FEATURES + 1]
return qs
from assets.views.core import get_element_type_label
def obj_to_feature(obj, asset_type):
lon = obj['lon']
lat = obj['lat']
if lon is None:
lon = obj.get('location_lon')
if lat is None:
lat = obj.get('location_lat')
if obj['geojson']:
geometry = json.loads(obj['geojson'])
elif lon is not None and lat is not None:
geometry = {"type": "Point", "coordinates": [lon, lat]}
else:
geometry = None
detail_url = reverse('assets:clean_assets_detail', kwargs={'asset_model': asset_type, 'asset_id': obj['id']})
return {
"type": "Feature",
"id": obj['id'],
"geometry": geometry,
"properties": {
"code": obj['code'],
"name": obj['name'],
"category": obj['category_name'],
"model": obj['model_name'],
"category_id": obj.get('category_id'),
"model_id": obj.get('model_id'),
"status": obj['status'],
"validation_status": obj.get('validation_status'),
"lon": lon,
"lat": lat,
"color": obj.get('color') or '',
"display_color": _color_from_text(obj.get('color')) or _color_from_text(obj.get('model_name')),
"asset_type": asset_type,
"detail_url": detail_url,
"element_type": get_element_type_label(asset_type),
"thematic": "clean",
"is_location": False,
}
}
litter_bin_qs = build_qs(CleanLitterBin, 'cleanlitterbin')
glass_container_qs = build_qs(CleanGlassContainer, 'cleanglasscontainer')
total = litter_bin_qs.count() + glass_container_qs.count()
if total > MAX_ASSET_FEATURES:
return JsonResponse(
{"error": f"Too many objects to display (> {MAX_ASSET_FEATURES}). Please narrow your selection."},
status=413
)
features = []
for obj in litter_bin_qs:
features.append(obj_to_feature(obj, 'cleanlitterbin'))
for obj in glass_container_qs:
features.append(obj_to_feature(obj, 'cleanglasscontainer'))
return JsonResponse({"type": "FeatureCollection", "features": features}, safe=False)
# ---------------------
# Asset detail
# ---------------------
@check_thematic_access('clean')
def clean_assets_detail(request, asset_model, asset_id):
try:
content_type = ContentType.objects.get(model=asset_model)
model_class = content_type.model_class()
except ContentType.DoesNotExist:
raise Http404(_("Unknown model: %(asset_model)s") % {'asset_model': asset_model})
asset = get_object_or_404(model_class, id=asset_id)
from django.template.loader import select_template
template = select_template([
f"assets/{asset_model}_detail.html",
"assets/detail_generic.html",
])
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))
context['thematic_code'] = 'clean'
return render(request, template.template.name, context)
# ---------------------
# Observations API
# ---------------------
@require_POST
@check_thematic_access('clean')
def update_litter_bin_observations(request, asset_id):
"""
Create / update / delete observations for a litter bin.
Expects JSON: { "observations": [ {id?, ...fields}, ... ] }
Observations currently in DB but absent from the payload are deleted.
"""
asset = get_object_or_404(CleanLitterBin, pk=asset_id)
if not can_edit_asset(request.user, asset):
return JsonResponse({'success': False, 'error': _(
"Vous n'avez pas la permission de modifier cet asset"
)}, status=403)
try:
data = json.loads(request.body)
except json.JSONDecodeError:
return JsonResponse({'success': False, 'error': _("Format JSON invalide")}, status=400)
observations_data = data.get('observations', [])
BOOL_FIELDS = [
'sol_pente_trop_forte', 'sol_instable', 'sol_reflechissant', 'sol_glissant',
'sol_fentes', 'sol_entrave', 'sol_dalles_podo',
'acc_approche_2m', 'acc_voirie_a_traverser', 'acc_poubelle_isolee',
'acc_gene_flux_routier', 'acc_gene_flux_pieton', 'acc_chemin_logique',
'acc_contraste', 'acc_indications_tri',
]
FLOAT_FIELDS = ['sol_pente_transversale', 'sol_pente_longitudinale']
TEXT_FIELDS = [
'observer_note', 'sol_remarque',
'acc_type_quartier', 'acc_densite_flux_pieton',
'acc_agent_vehicule', 'acc_agent_pied',
'acc_point_interet_autre', 'acc_point_interet_remarque', 'acc_remarque',
]
JSON_FIELDS = ['acc_point_interet', 'acc_pertinence_placement', 'acc_entrave']
with transaction.atomic():
submitted_ids = []
for obs_data in observations_data:
obs_id = obs_data.get('id')
if obs_id:
try:
obs = CleanLitterBinObservation.objects.get(pk=obs_id, litter_bin=asset)
except CleanLitterBinObservation.DoesNotExist:
obs = CleanLitterBinObservation(litter_bin=asset)
else:
obs = CleanLitterBinObservation(litter_bin=asset)
# datetime
dt_str = obs_data.get('observation_datetime') or ''
if dt_str:
dt = parse_datetime(dt_str)
if dt and timezone.is_naive(dt):
dt = timezone.make_aware(dt)
obs.observation_datetime = dt
else:
obs.observation_datetime = None
# boolean fields
for f in BOOL_FIELDS:
if f in obs_data:
setattr(obs, f, bool(obs_data[f]))
# float fields
for f in FLOAT_FIELDS:
val = obs_data.get(f, '')
if val == '' or val is None:
setattr(obs, f, None)
else:
try:
setattr(obs, f, float(val))
except (ValueError, TypeError):
setattr(obs, f, None)
# text / choice fields
for f in TEXT_FIELDS:
val = obs_data.get(f, '')
setattr(obs, f, str(val).strip() if val else None)
# JSON list fields
for f in JSON_FIELDS:
if f in obs_data:
raw = obs_data[f]
setattr(obs, f, raw if isinstance(raw, list) else [])
obs.save()
submitted_ids.append(obs.pk)
# Delete any observation not in the submitted payload
CleanLitterBinObservation.objects.filter(
litter_bin=asset
).exclude(pk__in=submitted_ids).delete()
return JsonResponse({'success': True})
# ---------------------
# Get nearest location
# ---------------------
def get_nearest_clean_location(request):
try:
lon = float(request.GET.get('lon'))
lat = float(request.GET.get('lat'))
except (TypeError, ValueError):
return JsonResponse({"error": "Invalid coordinates"}, status=400)
user_point = Point(lon, lat, srid=4326)
user_point.transform(3812)
nearest = (
CleanLocation.objects
.filter(geom__isnull=False)
.annotate(distance=Distance('geom', user_point))
.filter(distance__lte=50) # within 50 metres
.order_by('distance')
.first()
)
if nearest:
location_ct_id = ContentType.objects.get_for_model(nearest).id
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": _("No location found nearby")})