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', 'StreetUp') org_name_nl = getattr(settings, 'ORGANIZATION_NAME_NL', 'StreetUp') 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.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 and code 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) ) # Type filter types = data.get('authorization_type') if types: qs = qs.filter(type__in=types) # 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, })