434 lines
17 KiB
Python
434 lines
17 KiB
Python
import json
|
||
import logging
|
||
import smtplib
|
||
import socket
|
||
import ssl
|
||
|
||
from django.conf import settings
|
||
from django.contrib.auth.decorators import login_required
|
||
from django.contrib.gis.geos import GEOSGeometry, MultiPolygon, Polygon
|
||
from django.core.mail import EmailMultiAlternatives
|
||
from django.shortcuts import render, redirect, get_object_or_404
|
||
from django.template.loader import render_to_string
|
||
from django.urls import reverse
|
||
from django.utils import translation
|
||
from django.utils.translation import gettext_lazy as _
|
||
from django.views.decorators.http import require_GET
|
||
from django.core.exceptions import PermissionDenied
|
||
from django.contrib import messages
|
||
from common.models import UserConfig
|
||
|
||
from .forms import (
|
||
AuthorizationForm,
|
||
AuthorizationDocumentFormSet,
|
||
AuthorizationDateFormSet,
|
||
)
|
||
from .models import RoadAuthorization, VerifiedEmail, EmailVerificationToken
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _is_smtp_available(timeout=5):
|
||
"""Test du serveur SMTP avant envoi d'emails."""
|
||
email_host = getattr(settings, 'EMAIL_HOST', None)
|
||
email_port = getattr(settings, 'EMAIL_PORT', 25)
|
||
use_ssl = bool(getattr(settings, 'EMAIL_USE_SSL', False))
|
||
use_tls = bool(getattr(settings, 'EMAIL_USE_TLS', False))
|
||
if not email_host:
|
||
return False
|
||
try:
|
||
email_port = int(email_port) if email_port else 25
|
||
except (ValueError, TypeError):
|
||
email_port = 25
|
||
if email_port == 465:
|
||
use_ssl = True
|
||
try:
|
||
if use_ssl:
|
||
ctx = ssl.create_default_context()
|
||
with smtplib.SMTP_SSL(host=email_host, port=email_port, timeout=timeout, context=ctx) as s:
|
||
s.ehlo()
|
||
else:
|
||
with smtplib.SMTP(host=email_host, port=email_port, timeout=timeout) as s:
|
||
s.ehlo()
|
||
if use_tls:
|
||
ctx = ssl.create_default_context()
|
||
s.starttls(context=ctx)
|
||
s.ehlo()
|
||
return True
|
||
except Exception as exc:
|
||
logger.warning("SMTP not available: %s", exc)
|
||
return False
|
||
|
||
|
||
def _resolve_recipients(addresses):
|
||
"""Applique l'override d'email si configuré."""
|
||
override = getattr(settings, 'EMAIL_OVERRIDE', None)
|
||
if override:
|
||
return [override]
|
||
return addresses
|
||
|
||
|
||
_VERIFY_SUBJECTS = {
|
||
'fr': "{code} – Confirmation de votre adresse email – {org}",
|
||
'nl': "{code} – Bevestiging van uw e-mailadres – {org}",
|
||
'en': "{code} – Confirmation of your email address – {org}",
|
||
}
|
||
|
||
_ACK_SUBJECTS = {
|
||
'fr': "{code} – Accusé de réception de votre demande – Occupation de l'espace public",
|
||
'nl': "{code} – Ontvangstbewijs van uw aanvraag – Inname van het openbaar domein",
|
||
'en': "{code} – Acknowledgement of your application – Use of public space",
|
||
}
|
||
|
||
_BCC_ADDRESS = getattr(settings, 'EMAIL_AUTHORIZATIONS_BCC', '')
|
||
|
||
|
||
def _send_verification_email(request, authorization, token_obj):
|
||
"""Envoie l'email de vérification avec le code à 6 chiffres, dans la langue de l'utilisateur."""
|
||
lang = authorization.language or 'fr'
|
||
verify_page_url = request.build_absolute_uri(
|
||
reverse('authorizations:verify_email_code', args=[authorization.pk])
|
||
)
|
||
context = {
|
||
'applicant_name': authorization.applicant_name,
|
||
'code': token_obj.code,
|
||
'verify_page_url': verify_page_url,
|
||
'validity_minutes': EmailVerificationToken.TOKEN_VALIDITY_MINUTES,
|
||
}
|
||
org_name = getattr(settings, 'ORGANIZATION_NAME', 'Loko')
|
||
org_name_nl = getattr(settings, 'ORGANIZATION_NAME_NL', 'Loko')
|
||
org = org_name_nl if lang == 'nl' else org_name
|
||
subject = _VERIFY_SUBJECTS.get(lang, _VERIFY_SUBJECTS['fr']).format(code=authorization.code, org=org)
|
||
body = render_to_string(f'authorizations/emails/{lang}/verify_email.txt', context)
|
||
recipients = _resolve_recipients([authorization.applicant_email])
|
||
if not _is_smtp_available():
|
||
logger.warning("SMTP non disponible – email de vérification non envoyé pour %s", authorization.applicant_email)
|
||
return
|
||
try:
|
||
msg = EmailMultiAlternatives(
|
||
subject=subject,
|
||
body=body,
|
||
from_email=settings.DEFAULT_FROM_EMAIL,
|
||
to=recipients,
|
||
)
|
||
msg.send()
|
||
logger.info("Email de vérification envoyé à %s", recipients)
|
||
except Exception as exc:
|
||
logger.exception("Erreur lors de l'envoi de l'email de vérification: %s", exc)
|
||
|
||
|
||
def _send_acknowledgment_email(request, authorization):
|
||
"""Envoie l'accusé de réception après confirmation de l'email, dans la langue de l'utilisateur."""
|
||
lang = authorization.language or 'fr'
|
||
context = {
|
||
'authorization': authorization,
|
||
'dates': authorization.dates.all(),
|
||
}
|
||
subject = _ACK_SUBJECTS.get(lang, _ACK_SUBJECTS['fr']).format(code=authorization.code)
|
||
body = render_to_string(f'authorizations/emails/{lang}/authorization_confirmation.txt', context)
|
||
recipients = _resolve_recipients([authorization.applicant_email])
|
||
# BCC uniquement en production (pas en mode override)
|
||
override = getattr(settings, 'EMAIL_OVERRIDE', None)
|
||
bcc = [] if override else ([_BCC_ADDRESS] if _BCC_ADDRESS else [])
|
||
if not _is_smtp_available():
|
||
logger.warning("SMTP non disponible – accusé de réception non envoyé pour %s", authorization.code)
|
||
return
|
||
try:
|
||
msg = EmailMultiAlternatives(
|
||
subject=subject,
|
||
body=body,
|
||
from_email=settings.DEFAULT_FROM_EMAIL,
|
||
to=recipients,
|
||
bcc=bcc,
|
||
)
|
||
msg.send()
|
||
logger.info("Accusé de réception envoyé à %s (bcc: %s) pour %s", recipients, bcc, authorization.code)
|
||
except Exception as exc:
|
||
logger.exception("Erreur lors de l'envoi de l'accusé de réception: %s", exc)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Public view: new authorization request
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def new_authorization(request):
|
||
"""
|
||
Public-facing form allowing citizens to submit a road-occupation request.
|
||
No login required.
|
||
"""
|
||
if request.method == 'POST':
|
||
form = AuthorizationForm(request.POST)
|
||
doc_formset = AuthorizationDocumentFormSet(request.POST, request.FILES, prefix='documents')
|
||
date_formset = AuthorizationDateFormSet(request.POST, prefix='dates')
|
||
|
||
if form.is_valid() and doc_formset.is_valid() and date_formset.is_valid():
|
||
# Require at least one begin_date
|
||
has_begin_date = any(
|
||
df.cleaned_data.get('begin_date')
|
||
for df in date_formset
|
||
if df.cleaned_data
|
||
)
|
||
if not has_begin_date:
|
||
date_error = True
|
||
else:
|
||
date_error = False
|
||
|
||
if not date_error:
|
||
authorization = form.save(commit=False)
|
||
|
||
# Convert GeoJSON polygon → MultiPolygon
|
||
geojson_str = form.cleaned_data['geom_geojson']
|
||
geom = GEOSGeometry(geojson_str, srid=4326)
|
||
if isinstance(geom, Polygon):
|
||
geom = MultiPolygon(geom)
|
||
authorization.geom = geom
|
||
|
||
# Check whitelist before saving
|
||
applicant_email = form.cleaned_data['applicant_email'].lower().strip()
|
||
email_already_verified = VerifiedEmail.objects.filter(email=applicant_email).exists()
|
||
authorization.is_email_verified = email_already_verified
|
||
|
||
# Language comes from the form field
|
||
authorization.language = form.cleaned_data.get('language') or 'fr'
|
||
|
||
authorization.save()
|
||
|
||
# Attach documents
|
||
doc_formset.instance = authorization
|
||
doc_formset.save()
|
||
|
||
# Attach dates (skip empty rows)
|
||
for date_form in date_formset:
|
||
if date_form.cleaned_data.get('begin_date'):
|
||
date_obj = date_form.save(commit=False)
|
||
date_obj.authorization = authorization
|
||
date_obj.save()
|
||
|
||
if email_already_verified:
|
||
# Email already in whitelist → send acknowledgment directly
|
||
_send_acknowledgment_email(request, authorization)
|
||
return redirect(
|
||
reverse('authorizations:authorization_success') + f'?code={authorization.code}'
|
||
)
|
||
else:
|
||
# Send verification email
|
||
token_obj = EmailVerificationToken.create_for_authorization(authorization)
|
||
_send_verification_email(request, authorization, token_obj)
|
||
return redirect(reverse('authorizations:verify_email_code', args=[authorization.pk]))
|
||
else:
|
||
date_error = False
|
||
else:
|
||
lang_param = request.GET.get('lang', '').lower()
|
||
initial = {'language': lang_param} if lang_param in ('fr', 'nl', 'en') else {}
|
||
form = AuthorizationForm(initial=initial)
|
||
doc_formset = AuthorizationDocumentFormSet(prefix='documents')
|
||
date_formset = AuthorizationDateFormSet(prefix='dates')
|
||
date_error = False
|
||
|
||
# Activate translation for template rendering based on ?lang= param (GET)
|
||
# or the language field already submitted (POST re-render)
|
||
lang_for_render = request.GET.get('lang', '').lower()
|
||
if lang_for_render not in ('fr', 'nl', 'en'):
|
||
# Try to read from POST data or form initial on error re-render
|
||
lang_for_render = request.POST.get('language', 'fr') if request.method == 'POST' else 'fr'
|
||
if lang_for_render in ('fr', 'nl', 'en'):
|
||
translation.activate(lang_for_render)
|
||
|
||
return render(request, [
|
||
'authorizations/new_authorization.html',
|
||
'authorizations/new_authorization_default.html',
|
||
], {
|
||
'form': form,
|
||
'doc_formset': doc_formset,
|
||
'date_formset': date_formset,
|
||
'date_error': date_error,
|
||
})
|
||
|
||
|
||
def verify_email_code(request, authorization_id):
|
||
"""Page de saisie du code de vérification à 6 chiffres."""
|
||
authorization = get_object_or_404(RoadAuthorization, pk=authorization_id)
|
||
|
||
try:
|
||
token_obj = authorization.verification_token
|
||
except EmailVerificationToken.DoesNotExist:
|
||
# Already verified or no token exists
|
||
return render(request, 'authorizations/email_verification_failed.html', {
|
||
'reason': 'invalid',
|
||
})
|
||
|
||
error = None
|
||
|
||
if request.method == 'POST':
|
||
entered_code = request.POST.get('code', '').strip()
|
||
|
||
if token_obj.is_locked():
|
||
error = 'locked'
|
||
elif token_obj.is_expired():
|
||
error = 'expired'
|
||
elif entered_code != token_obj.code:
|
||
token_obj.attempts += 1
|
||
token_obj.save(update_fields=['attempts'])
|
||
error = 'locked' if token_obj.is_locked() else 'wrong_code'
|
||
else:
|
||
# ✓ Code correct
|
||
authorization.is_email_verified = True
|
||
authorization.save(update_fields=['is_email_verified'])
|
||
VerifiedEmail.objects.get_or_create(email=authorization.applicant_email.lower().strip())
|
||
token_obj.delete()
|
||
_send_acknowledgment_email(request, authorization)
|
||
return redirect(
|
||
reverse('authorizations:authorization_success') + f'?code={authorization.code}&verified=1'
|
||
)
|
||
else:
|
||
if token_obj.is_locked():
|
||
error = 'locked'
|
||
elif token_obj.is_expired():
|
||
error = 'expired'
|
||
|
||
return render(request, 'authorizations/email_verify_code.html', {
|
||
'authorization': authorization,
|
||
'token_obj': token_obj,
|
||
'error': error,
|
||
})
|
||
|
||
|
||
def authorization_success(request):
|
||
"""Confirmation page displayed after a successful submission."""
|
||
auth_code = request.GET.get('code', '')
|
||
just_verified = request.GET.get('verified') == '1'
|
||
|
||
# Activate the language stored on the authorization so the page is
|
||
# rendered in the language the applicant chose.
|
||
if auth_code:
|
||
try:
|
||
auth_obj = RoadAuthorization.objects.only('language').get(code=auth_code)
|
||
if auth_obj.language in ('fr', 'nl', 'en'):
|
||
translation.activate(auth_obj.language)
|
||
except RoadAuthorization.DoesNotExist:
|
||
pass
|
||
|
||
return render(request, 'authorizations/authorization_success.html', {
|
||
'auth_code': auth_code,
|
||
'just_verified': just_verified,
|
||
})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Staff views
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@login_required
|
||
def authorization_list(request):
|
||
"""Staff list of all road-authorization requests."""
|
||
user_config = get_object_or_404(UserConfig, user=request.user)
|
||
if not user_config.can_access_view('authorizations'):
|
||
raise PermissionDenied(_("Vous n'avez pas accès à cette vue."))
|
||
|
||
from .forms import AuthorizationFilterForm
|
||
from datetime import datetime
|
||
from dateutil.relativedelta import relativedelta
|
||
|
||
filter_form = AuthorizationFilterForm(request.GET or None)
|
||
qs = RoadAuthorization.objects.select_related('agent').prefetch_related('dates', 'documents').order_by('-creation_date')
|
||
|
||
if filter_form.is_valid():
|
||
data = filter_form.cleaned_data
|
||
|
||
# Text search over description, location, applicant fields, code and agent
|
||
search = data.get('search_text', '').strip()
|
||
if search:
|
||
from django.db.models import Q
|
||
qs = qs.filter(
|
||
Q(description__icontains=search) |
|
||
Q(location__icontains=search) |
|
||
Q(applicant_name__icontains=search) |
|
||
Q(applicant_email__icontains=search) |
|
||
Q(code__icontains=search) |
|
||
Q(agent__first_name__icontains=search) |
|
||
Q(agent__last_name__icontains=search) |
|
||
Q(agent__username__icontains=search)
|
||
)
|
||
|
||
# Type filter
|
||
types = data.get('authorization_type')
|
||
if types:
|
||
qs = qs.filter(type__in=types)
|
||
|
||
# Agent filter
|
||
agents = data.get('agent')
|
||
if agents:
|
||
from django.db.models import Q
|
||
agent_q = Q()
|
||
if 'unassigned' in agents:
|
||
agent_q |= Q(agent__isnull=True)
|
||
user_agent_ids = [a for a in agents if a != 'unassigned']
|
||
if user_agent_ids:
|
||
agent_q |= Q(agent_id__in=user_agent_ids)
|
||
qs = qs.filter(agent_q)
|
||
|
||
# Archive filter: exclude entries whose latest end_date is > 2 years ago
|
||
if not data.get('include_archives'):
|
||
cutoff = datetime.now().astimezone() - relativedelta(years=2)
|
||
# Keep authorizations that have no dates or at least one recent date
|
||
from django.db.models import Max, Q as DQ
|
||
qs = qs.annotate(latest_end=Max('dates__end_date')).filter(
|
||
DQ(latest_end__isnull=True) | DQ(latest_end__gte=cutoff)
|
||
)
|
||
|
||
return render(request, 'authorizations/authorization_list.html', {
|
||
'authorizations': qs,
|
||
'filter_form': filter_form,
|
||
})
|
||
|
||
|
||
@login_required
|
||
def authorization_detail(request, pk):
|
||
"""Staff detail view for a single authorization with edit capabilities."""
|
||
user_config = get_object_or_404(UserConfig, user=request.user)
|
||
if not user_config.can_access_view('authorizations'):
|
||
raise PermissionDenied(_("Vous n'avez pas accès à cette vue."))
|
||
|
||
authorization = get_object_or_404(RoadAuthorization, pk=pk)
|
||
|
||
if request.method == 'POST':
|
||
action = request.POST.get('action')
|
||
if action == 'update_properties':
|
||
new_status = request.POST.get('status')
|
||
new_comment = request.POST.get('agent_comment', '').strip()
|
||
|
||
try:
|
||
authorization.status = int(new_status)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
authorization.agent_comment = new_comment
|
||
authorization.agent = request.user
|
||
authorization.save()
|
||
messages.success(request, _("La demande a été mise à jour avec succès."))
|
||
return redirect('authorizations:authorization_detail', pk=pk)
|
||
|
||
elif action == 'add_document':
|
||
doc_file = request.FILES.get('file')
|
||
description = request.POST.get('description', '').strip()
|
||
if doc_file and description:
|
||
from .models import RoadAuthorizationDocument
|
||
from django.core.exceptions import ValidationError
|
||
doc = RoadAuthorizationDocument(
|
||
authorization=authorization,
|
||
file=doc_file,
|
||
description=description
|
||
)
|
||
try:
|
||
doc.full_clean()
|
||
doc.save()
|
||
messages.success(request, _("Le document a été ajouté avec succès."))
|
||
except ValidationError as e:
|
||
messages.error(request, e.messages[0])
|
||
else:
|
||
messages.error(request, _("Veuillez fournir une description et un fichier."))
|
||
return redirect('authorizations:authorization_detail', pk=pk)
|
||
|
||
return render(request, 'authorizations/authorization_detail.html', {
|
||
'authorization': authorization,
|
||
'status_choices': RoadAuthorization.AUTH_STATUS,
|
||
})
|