import asyncio import logging import uuid from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import urlparse import httpx from fastapi import ( BackgroundTasks, Depends, FastAPI, File, HTTPException, Query, Request, UploadFile, status, ) from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, RedirectResponse, Response from fastapi.staticfiles import StaticFiles from starlette.middleware.gzip import GZipMiddleware from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware from starlette.middleware.sessions import SessionMiddleware from starlette.middleware.trustedhost import TrustedHostMiddleware from geoalchemy2.elements import WKTElement from sqlalchemy import case, exists, func, select, text from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import selectinload from .auth import generate_csrf_token, hash_password, verify_password from .config import get_settings from .database import SessionLocal, engine, get_db from .email_service import NotificationUnavailable, send_email from .image_service import InvalidImage, normalise_image, safe_original_name from .logging_config import configure_logging from .models import ( Base, Intervention, InterventionQuote, InterventionRequest, InterventionTariff, NotificationLog, Observation, Commune, User, UserRole, AuditLog, ObservationStatus, Photo, QuoteStatus, RequestStatus, StatusHistory, ) from .pdf_service import build_quote_pdf, build_request_pdf from .schemas import ( AuthOut, CommuneOut, UserCreate, UserUpdate, UserOut, PasswordReset, GeocodingResult, InterventionOut, LoginRequest, InterventionUpsert, QuoteOut, QuoteUpsert, TariffOut, TariffUpdate, MapItemOut, ObservationCreate, ObservationOut, ObservationStatusUpdate, PhotoOut, PublicConfigOut, RequestCreate, RequestOut, StatsOut, StatusUpdate, TreatmentUpdate, ) from .security import ( require_user, require_technical_user, require_technical_csrf, require_admin_user, require_admin_csrf as require_database_admin_csrf, session_is_authenticated, session_is_technical, session_is_admin, ) settings = get_settings() settings.upload_path.mkdir(parents=True, exist_ok=True) settings.log_path.mkdir(parents=True, exist_ok=True) configure_logging() logger = logging.getLogger(__name__) docs_url = "/docs" if settings.docs_enabled else None redoc_url = "/redoc" if settings.docs_enabled else None openapi_url = "/openapi.json" if settings.docs_enabled else None app = FastAPI( title=settings.app_name, version="0.2.6", docs_url=docs_url, redoc_url=redoc_url, openapi_url=openapi_url, ) # Signed cookie sessions protect the technical portal. In production, the # startup validation requires a long random SESSION_SECRET_KEY. session_secret = settings.session_secret_key or "development-only-change-me" app.add_middleware( SessionMiddleware, secret_key=session_secret, session_cookie=settings.session_cookie_name, max_age=settings.session_max_age_seconds, same_site="lax", https_only=settings.session_cookie_secure, ) if settings.trusted_hosts and settings.trusted_hosts != ["*"]: app.add_middleware(TrustedHostMiddleware, allowed_hosts=settings.trusted_hosts, www_redirect=False) if settings.force_https: app.add_middleware(HTTPSRedirectMiddleware) app.add_middleware(GZipMiddleware, minimum_size=1000) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, allow_credentials=True, allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"], allow_headers=["Content-Type", "X-CSRF-Token"], ) frontend_path = Path(__file__).resolve().parents[2] / "frontend" # Chemin public expose par Apache. Les routes FastAPI restent sans prefixe, # car ProxyPass retire /ternois/frelons/ avant de transmettre la requete. PUBLIC_BASE_PATH = urlparse(settings.app_base_url).path.rstrip("/") def public_path(path: str) -> str: """Construit une URL publique en tenant compte du sous-repertoire Apache.""" normalised = "/" + path.lstrip("/") return f"{PUBLIC_BASE_PATH}{normalised}" if PUBLIC_BASE_PATH else normalised # Compatibilite des noms utilises par les routes techniques existantes. require_admin = require_technical_user require_admin_csrf = require_technical_csrf OBSERVATION_HISTORY_STATUSES = { ObservationStatus.non_confirme, ObservationStatus.transforme, ObservationStatus.archive, } REQUEST_HISTORY_STATUSES = { RequestStatus.annulee, RequestStatus.refusee, RequestStatus.archive, } REQUEST_ARCHIVABLE_STATUSES = { RequestStatus.terminee, RequestStatus.annulee, RequestStatus.refusee, } def remember_submission(request: Request, entity_id: uuid.UUID) -> None: submissions = [str(value) for value in request.session.get("submissions", [])][-19:] value = str(entity_id) if value not in submissions: submissions.append(value) request.session["submissions"] = submissions def can_upload_to_submission(request: Request, entity_id: uuid.UUID) -> bool: if session_is_authenticated(request): return True return str(entity_id) in {str(value) for value in request.session.get("submissions", [])} def auth_out(user: User, request: Request) -> AuthOut: csrf = str(request.session.get("csrf_token") or "") if not csrf: csrf = generate_csrf_token() request.session["csrf_token"] = csrf full_name = " ".join(value for value in [user.prenom, user.nom] if value).strip() return AuthOut( authenticated=True, username=full_name or user.email, email=user.email, role=user.role, commune_code=user.commune_code, commune_nom=user.commune.nom if user.commune else None, csrf_token=csrf, ) @app.post("/api/auth/login", response_model=AuthOut, include_in_schema=False) async def login( payload: LoginRequest, request: Request, db: AsyncSession = Depends(get_db), ) -> AuthOut: email = payload.username.strip().lower() statement = ( select(User) .options(selectinload(User.commune)) .where(func.lower(User.email) == email, User.actif.is_(True)) ) user = await db.scalar(statement) now = datetime.now(timezone.utc) if user and user.verrouille_jusqu_a and user.verrouille_jusqu_a > now: logger.warning("Compte temporairement verrouille : %s", email[:120]) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Identifiants invalides") valid = bool(user and verify_password(payload.password, user.password_hash)) if not valid: if user: user.tentatives_echouees = (user.tentatives_echouees or 0) + 1 if user.tentatives_echouees >= 5: user.verrouille_jusqu_a = now + timedelta(minutes=15) await db.commit() logger.warning("Echec de connexion pour %s", email[:120]) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Identifiants invalides") user.tentatives_echouees = 0 user.verrouille_jusqu_a = None user.derniere_connexion = now request.session.clear() request.session.update({ "user_id": str(user.id), "email": user.email, "role": user.role.value, "commune_code": user.commune_code, "csrf_token": generate_csrf_token(), }) db.add(AuditLog( utilisateur_id=user.id, action="connexion", type_objet="utilisateur", objet_id=user.id, adresse_ip=request.client.host if request.client else None, )) await db.commit() return auth_out(user, request) @app.get("/api/auth/me", response_model=AuthOut, include_in_schema=False) async def auth_me(request: Request, user: User = Depends(require_user)) -> AuthOut: return auth_out(user, request) @app.post("/api/auth/logout", include_in_schema=False) async def logout( request: Request, user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> dict[str, bool]: from .security import verify_csrf verify_csrf(request) db.add(AuditLog( utilisateur_id=user.id, action="deconnexion", type_objet="utilisateur", objet_id=user.id, adresse_ip=request.client.host if request.client else None, )) await db.commit() request.session.clear() return {"ok": True} @app.on_event("startup") async def startup() -> None: errors = settings.validate_production() if errors: raise RuntimeError("Configuration de production invalide : " + " | ".join(errors)) if settings.auto_create_tables: async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Application démarrée en environnement %s", settings.app_env) def make_reference(prefix: str) -> str: stamp = datetime.now(timezone.utc).strftime("%Y%m%d") return f"{prefix}-{stamp}-{uuid.uuid4().hex[:6].upper()}" def photo_out(photo: Photo) -> PhotoOut: return PhotoOut( id=photo.id, url=public_path(f"/uploads/{photo.filename}"), original_name=photo.original_name, mime_type=photo.mime_type, size_bytes=photo.size_bytes, ) def obs_out(row, include_private: bool = True) -> ObservationOut: obj, longitude, latitude = row return ObservationOut( id=obj.id, reference=obj.reference, commune=obj.commune, adresse=obj.adresse, type_observation=obj.type_observation, support=obj.support, hauteur_estimee=obj.hauteur_estimee, niveau_danger=obj.niveau_danger, description=obj.description, contact_nom=obj.contact_nom if include_private else None, contact_telephone=obj.contact_telephone if include_private else None, contact_email=obj.contact_email if include_private else None, status=obj.status, longitude=longitude, latitude=latitude, created_at=obj.created_at, updated_at=obj.updated_at, photos=[photo_out(photo) for photo in getattr(obj, "photos", [])], ) def request_out(row, include_private: bool = True) -> RequestOut: obj, longitude, latitude = row return RequestOut( id=obj.id, reference=obj.reference, observation_id=obj.observation_id, type_demande=obj.type_demande, status=obj.status, commune=obj.commune, adresse=obj.adresse, niveau_urgence=obj.niveau_urgence, demandeur_nom=obj.demandeur_nom if include_private else None, demandeur_fonction=obj.demandeur_fonction if include_private else None, demandeur_email=obj.demandeur_email if include_private else None, demandeur_telephone=obj.demandeur_telephone if include_private else None, description_risque=obj.description_risque, hauteur_estimee=obj.hauteur_estimee, support=obj.support, difficulte_acces=obj.difficulte_acces, proprietaire_nom=obj.proprietaire_nom if include_private else None, proprietaire_adresse=obj.proprietaire_adresse if include_private else None, proprietaire_telephone=obj.proprietaire_telephone if include_private else None, proprietaire_email=obj.proprietaire_email if include_private else None, accord_proprietaire=obj.accord_proprietaire if include_private else None, justification_danger_imminent=obj.justification_danger_imminent, assigned_to=obj.assigned_to, intervention_date=obj.intervention_date, longitude=longitude, latitude=latitude, created_at=obj.created_at, updated_at=obj.updated_at, photos=[photo_out(photo) for photo in getattr(obj, "photos", [])], ) def tariff_out(obj: InterventionTariff) -> TariffOut: return TariffOut( code=obj.code, libelle=obj.libelle, description=obj.description, prix=obj.prix, actif=obj.actif, ordre=obj.ordre, ) def quote_out(obj: InterventionQuote) -> QuoteOut: return QuoteOut( id=obj.id, demande_id=obj.demande_id, reference=obj.reference, statut=obj.statut, type_intervention=obj.type_intervention, libelle=obj.libelle_snapshot, prix_unitaire=obj.prix_unitaire, quantite=obj.quantite, montant_total=obj.montant_total, commune_facturee=obj.commune_facturee, date_devis=obj.date_devis, validite_jusqu_au=obj.validite_jusqu_au, observations=obj.observations, created_at=obj.created_at, updated_at=obj.updated_at, ) async def pdf_tariff_rows(db: AsyncSession) -> list[tuple[str, str, object]]: rows = (await db.scalars( select(InterventionTariff) .where(InterventionTariff.actif.is_(True)) .order_by(InterventionTariff.ordre, InterventionTariff.libelle) )).all() return [(row.code, row.libelle, row.prix) for row in rows] async def _record_notification( entity_type: str, entity_id: uuid.UUID, recipients: list[str], subject: str, status: str, error: str | None = None, ) -> None: try: async with SessionLocal() as db: for recipient in recipients or ["non-configure"]: db.add( NotificationLog( entity_type=entity_type, entity_id=entity_id, recipient=recipient, subject=subject, status=status, error=error, ) ) await db.commit() except Exception: # La création du dossier ne doit jamais échouer à cause du journal SMTP. return async def notify_observation(payload: dict) -> None: recipients = list(settings.notification_recipients) subject = f"Nouveau signalement {payload['reference']} - {payload['commune']}" body = ( f"Un nouveau signalement de frelon a été enregistré.\n\n" f"Référence : {payload['reference']}\n" f"Commune : {payload['commune']}\n" f"Adresse : {payload.get('adresse') or 'Non renseignée'}\n" f"Niveau de danger : {payload.get('niveau_danger') or 'Non renseigné'}\n" f"Coordonnées : {payload['latitude']}, {payload['longitude']}\n\n" f"Portail technique : {settings.app_base_url.rstrip('/')}/admin.html" ) try: await send_email(recipients, subject, body) await _record_notification("observation", payload["id"], recipients, subject, "sent") except (NotificationUnavailable, OSError, RuntimeError) as exc: await _record_notification("observation", payload["id"], recipients, subject, "skipped", str(exc)) contact_email = payload.get("contact_email") if contact_email: confirmation_subject = f"Votre signalement {payload['reference']} a bien été enregistré" confirmation_body = ( f"Bonjour,\n\nVotre signalement concernant la commune de {payload['commune']} a été enregistré " f"sous la référence {payload['reference']}.\n\nLes services compétents pourront vous recontacter si nécessaire." ) try: await send_email([contact_email], confirmation_subject, confirmation_body) await _record_notification("observation", payload["id"], [contact_email], confirmation_subject, "sent") except (NotificationUnavailable, OSError, RuntimeError) as exc: await _record_notification("observation", payload["id"], [contact_email], confirmation_subject, "skipped", str(exc)) async def notify_request(payload: dict) -> None: recipients = list(settings.notification_recipients) subject = f"Nouvelle demande d'intervention {payload['reference']} - {payload['commune']}" body = ( f"Une nouvelle demande d'intervention a été transmise.\n\n" f"Référence : {payload['reference']}\n" f"Type : {payload['type_demande']}\n" f"Commune : {payload['commune']}\n" f"Adresse : {payload['adresse']}\n" f"Urgence : {payload['niveau_urgence']}\n" f"Demandeur : {payload['demandeur_nom']}\n\n" f"Portail technique : {settings.app_base_url.rstrip('/')}/admin.html" ) pdf_bytes = build_request_pdf(payload["pdf_object"], payload["longitude"], payload["latitude"], tariffs=payload.get("pdf_tariffs")) try: await send_email( recipients, subject, body, attachment_bytes=pdf_bytes, attachment_name=f"{payload['reference']}.pdf", ) await _record_notification("demande", payload["id"], recipients, subject, "sent") except (NotificationUnavailable, OSError, RuntimeError) as exc: await _record_notification("demande", payload["id"], recipients, subject, "skipped", str(exc)) confirmation_subject = f"Demande {payload['reference']} transmise" confirmation_body = ( f"Bonjour,\n\nVotre demande d'intervention a été enregistrée sous la référence " f"{payload['reference']}. Les services de TernoisCom pourront vous recontacter pour la qualification et la programmation." ) try: await send_email( [payload["demandeur_email"]], confirmation_subject, confirmation_body, attachment_bytes=pdf_bytes, attachment_name=f"{payload['reference']}.pdf", ) await _record_notification("demande", payload["id"], [payload["demandeur_email"]], confirmation_subject, "sent") except (NotificationUnavailable, OSError, RuntimeError) as exc: await _record_notification("demande", payload["id"], [payload["demandeur_email"]], confirmation_subject, "skipped", str(exc)) def user_out(user: User) -> UserOut: return UserOut( id=user.id, email=user.email, nom=user.nom, prenom=user.prenom, role=user.role, commune_code=user.commune_code, commune_nom=user.commune.nom if getattr(user, "commune", None) else None, actif=user.actif, doit_changer_mot_de_passe=user.doit_changer_mot_de_passe, derniere_connexion=user.derniere_connexion, cree_le=user.cree_le, ) async def resolve_commune( db: AsyncSession, *, current_user: User, requested_name: str | None, ) -> Commune: if current_user.role == UserRole.elu: if not current_user.commune_code: raise HTTPException(status_code=403, detail="Aucune commune n'est rattachée à ce compte élu.") commune = await db.get(Commune, current_user.commune_code) else: value = (requested_name or "").strip() commune = await db.scalar( select(Commune).where( Commune.actif.is_(True), func.lower(Commune.nom) == value.lower(), ) ) if not commune: raise HTTPException(status_code=422, detail="Commune du Ternois introuvable.") return commune @app.get("/api/communes", response_model=list[CommuneOut]) async def list_communes( _user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> list[CommuneOut]: rows = (await db.scalars( select(Commune).where(Commune.actif.is_(True)).order_by(Commune.nom) )).all() return [CommuneOut(insee_com=row.insee_com, nom=row.nom, population=row.population) for row in rows] @app.get("/api/tarifs", response_model=list[TariffOut]) async def list_tariffs( _user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> list[TariffOut]: rows = (await db.scalars( select(InterventionTariff) .where(InterventionTariff.actif.is_(True)) .order_by(InterventionTariff.ordre, InterventionTariff.libelle) )).all() return [tariff_out(row) for row in rows] @app.get("/api/tech/tarifs", response_model=list[TariffOut]) async def list_technical_tariffs( _user: User = Depends(require_technical_user), db: AsyncSession = Depends(get_db), ) -> list[TariffOut]: rows = (await db.scalars( select(InterventionTariff).order_by(InterventionTariff.ordre, InterventionTariff.libelle) )).all() return [tariff_out(row) for row in rows] @app.patch("/api/admin/tarifs/{code}", response_model=TariffOut) async def update_tariff( code: str, payload: TariffUpdate, request: Request, admin: User = Depends(require_database_admin_csrf), db: AsyncSession = Depends(get_db), ) -> TariffOut: obj = await db.get(InterventionTariff, code) if not obj: raise HTTPException(status_code=404, detail="Tarif introuvable.") changes = payload.model_dump(exclude_unset=True) for field, value in changes.items(): setattr(obj, field, value) db.add(AuditLog( utilisateur_id=admin.id, action="tarif_modifie", type_objet="tarif_intervention", details={"code": code, "modifications": {k: str(v) for k, v in changes.items()}}, adresse_ip=request.client.host if request.client else None, )) await db.commit() await db.refresh(obj) return tariff_out(obj) @app.get("/api/admin/users", response_model=list[UserOut]) async def list_users( _admin: User = Depends(require_admin_user), db: AsyncSession = Depends(get_db), ) -> list[UserOut]: rows = (await db.scalars( select(User).options(selectinload(User.commune)).order_by(User.nom, User.prenom, User.email) )).all() return [user_out(row) for row in rows] @app.post("/api/admin/users", response_model=UserOut, status_code=201) async def create_user( payload: UserCreate, request: Request, admin: User = Depends(require_database_admin_csrf), db: AsyncSession = Depends(get_db), ) -> UserOut: email = str(payload.email).strip().lower() if await db.scalar(select(User.id).where(func.lower(User.email) == email)): raise HTTPException(status_code=409, detail="Cette adresse électronique existe déjà.") commune = None if payload.commune_code: commune = await db.get(Commune, payload.commune_code) if not commune or not commune.actif: raise HTTPException(status_code=422, detail="Commune invalide.") if payload.role == UserRole.elu and not commune: raise HTTPException(status_code=422, detail="Une commune est obligatoire pour un élu.") obj = User( email=email, nom=payload.nom.strip(), prenom=(payload.prenom or "").strip() or None, password_hash=hash_password(payload.password), role=payload.role, commune_code=commune.insee_com if commune else None, actif=payload.actif, ) db.add(obj) await db.flush() db.add(AuditLog( utilisateur_id=admin.id, action="creation_utilisateur", type_objet="utilisateur", objet_id=obj.id, adresse_ip=request.client.host if request.client else None, )) await db.commit() statement = select(User).options(selectinload(User.commune)).where(User.id == obj.id) return user_out(await db.scalar(statement)) @app.patch("/api/admin/users/{user_id}", response_model=UserOut) async def update_user( user_id: uuid.UUID, payload: UserUpdate, request: Request, admin: User = Depends(require_database_admin_csrf), db: AsyncSession = Depends(get_db), ) -> UserOut: obj = await db.scalar(select(User).options(selectinload(User.commune)).where(User.id == user_id)) if not obj: raise HTTPException(status_code=404, detail="Utilisateur introuvable.") values = payload.model_dump(exclude_unset=True) if "email" in values: values["email"] = str(values["email"]).strip().lower() duplicate = await db.scalar(select(User.id).where(func.lower(User.email) == values["email"], User.id != user_id)) if duplicate: raise HTTPException(status_code=409, detail="Cette adresse électronique existe déjà.") role = values.get("role", obj.role) commune_code = values.get("commune_code", obj.commune_code) if role == UserRole.elu and not commune_code: raise HTTPException(status_code=422, detail="Une commune est obligatoire pour un élu.") if commune_code and not await db.get(Commune, commune_code): raise HTTPException(status_code=422, detail="Commune invalide.") if obj.id == admin.id and values.get("actif") is False: raise HTTPException(status_code=422, detail="Vous ne pouvez pas désactiver votre propre compte.") for key, value in values.items(): setattr(obj, key, value) db.add(AuditLog( utilisateur_id=admin.id, action="modification_utilisateur", type_objet="utilisateur", objet_id=obj.id, adresse_ip=request.client.host if request.client else None, )) await db.commit() statement = select(User).options(selectinload(User.commune)).where(User.id == obj.id) return user_out(await db.scalar(statement)) @app.post("/api/admin/users/{user_id}/password", response_model=UserOut) async def reset_user_password( user_id: uuid.UUID, payload: PasswordReset, request: Request, admin: User = Depends(require_database_admin_csrf), db: AsyncSession = Depends(get_db), ) -> UserOut: obj = await db.scalar(select(User).options(selectinload(User.commune)).where(User.id == user_id)) if not obj: raise HTTPException(status_code=404, detail="Utilisateur introuvable.") obj.password_hash = hash_password(payload.password) obj.doit_changer_mot_de_passe = payload.doit_changer_mot_de_passe obj.tentatives_echouees = 0 obj.verrouille_jusqu_a = None db.add(AuditLog( utilisateur_id=admin.id, action="reinitialisation_mot_de_passe", type_objet="utilisateur", objet_id=obj.id, adresse_ip=request.client.host if request.client else None, )) await db.commit() return user_out(obj) @app.get("/api/health/live", include_in_schema=False) async def health_live() -> dict[str, str]: return {"status": "ok", "app": settings.app_name, "version": "0.2.6"} @app.get("/api/health") @app.get("/api/health/ready", include_in_schema=False) async def health(db: AsyncSession = Depends(get_db)) -> dict[str, str | int]: try: await db.execute(text("SELECT 1")) table_count = await db.scalar( text("SELECT count(*) FROM information_schema.tables WHERE table_schema = 'frelons'") ) except Exception as exc: logger.exception("Échec du contrôle de disponibilité PostgreSQL") raise HTTPException( status_code=503, detail="Connexion PostgreSQL indisponible. Vérifiez les variables PGHOST, PGPORT, PGDATABASE, PGUSER et PGPASSWORD ainsi que les droits du schéma frelons.", ) from exc return { "status": "ok", "database": "connected", "tables": int(table_count or 0), "app": settings.app_name, "version": "0.2.6", } @app.get("/api/config/public", response_model=PublicConfigOut) async def public_config() -> PublicConfigOut: return PublicConfigOut( app_name=settings.app_name, portal_title=settings.public_portal_title, support_email=settings.support_email, support_phone=settings.support_phone, environment=settings.app_env, notifications_enabled=settings.notifications_enabled, ) @app.get("/api/geocodage", response_model=list[GeocodingResult]) async def geocode( q: str = Query(min_length=3, max_length=180), _user: User = Depends(require_user), ) -> list[GeocodingResult]: params = {"q": q, "index": "address", "limit": 6} try: async with httpx.AsyncClient(timeout=8) as client: response = await client.get(settings.geocoding_url, params=params) response.raise_for_status() data = response.json() except (httpx.HTTPError, ValueError) as exc: raise HTTPException(status_code=502, detail="Le service de géocodage est temporairement indisponible.") from exc results: list[GeocodingResult] = [] for feature in data.get("features", []): coordinates = feature.get("geometry", {}).get("coordinates", []) properties = feature.get("properties", {}) if len(coordinates) < 2: continue label = properties.get("label") or properties.get("name") or q if isinstance(label, list): label = " - ".join(str(item) for item in label) results.append( GeocodingResult( label=str(label), city=properties.get("city"), postcode=properties.get("postcode"), longitude=float(coordinates[0]), latitude=float(coordinates[1]), ) ) return results @app.get("/api/carte", response_model=list[MapItemOut]) async def map_items( user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> list[MapItemOut]: linked_request_exists = exists( select(1).where(InterventionRequest.observation_id == Observation.id) ) obs_stmt = ( select(Observation, func.ST_X(Observation.geom), func.ST_Y(Observation.geom)) .where( Observation.status.notin_(OBSERVATION_HISTORY_STATUSES), ~linked_request_exists, ) .order_by(Observation.created_at.desc()) .limit(500) ) req_stmt = ( select(InterventionRequest, func.ST_X(InterventionRequest.geom), func.ST_Y(InterventionRequest.geom)) .where(InterventionRequest.status.notin_(REQUEST_HISTORY_STATUSES)) .order_by(InterventionRequest.created_at.desc()) .limit(500) ) if user.role == UserRole.elu: obs_stmt = obs_stmt.where(Observation.commune_code == user.commune_code) req_stmt = req_stmt.where(InterventionRequest.commune_code == user.commune_code) obs_rows = (await db.execute(obs_stmt)).all() req_rows = (await db.execute(req_stmt)).all() items: list[MapItemOut] = [] for obj, longitude, latitude in obs_rows: items.append( MapItemOut( id=obj.id, reference=obj.reference, kind="observation", commune=obj.commune, adresse=obj.adresse, status=obj.status.value, type_observation=obj.type_observation, niveau_danger=obj.niveau_danger, description=obj.description, longitude=longitude, latitude=latitude, created_at=obj.created_at, ) ) for obj, longitude, latitude in req_rows: items.append( MapItemOut( id=obj.id, reference=obj.reference, kind="request", commune=obj.commune, adresse=obj.adresse, status=obj.status.value, type_demande=obj.type_demande.value, niveau_urgence=obj.niveau_urgence, description=obj.description_risque, longitude=longitude, latitude=latitude, created_at=obj.created_at, ) ) return items async def _observation_rows(db: AsyncSession, user: User | None = None): statement = ( select(Observation, func.ST_X(Observation.geom), func.ST_Y(Observation.geom)) .options(selectinload(Observation.photos)) .order_by(Observation.created_at.desc()) .limit(500) ) if user and user.role == UserRole.elu: statement = statement.where(Observation.commune_code == user.commune_code) return (await db.execute(statement)).all() async def _observation_row(observation_id: uuid.UUID, db: AsyncSession, user: User | None = None): statement = ( select(Observation, func.ST_X(Observation.geom), func.ST_Y(Observation.geom)) .options(selectinload(Observation.photos)) .where(Observation.id == observation_id) ) if user and user.role == UserRole.elu: statement = statement.where(Observation.commune_code == user.commune_code) row = (await db.execute(statement)).one_or_none() if not row: raise HTTPException(status_code=404, detail="Observation introuvable") return row @app.get("/api/observations", response_model=list[ObservationOut]) async def list_public_observations( user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> list[ObservationOut]: return [obs_out(row, include_private=False) for row in await _observation_rows(db, user)] @app.get("/api/tech/observations", response_model=list[ObservationOut]) async def list_technical_observations( _admin: str = Depends(require_admin), db: AsyncSession = Depends(get_db), ) -> list[ObservationOut]: return [obs_out(row, include_private=True) for row in await _observation_rows(db)] @app.get("/api/observations/{observation_id}", response_model=ObservationOut) async def get_public_observation( observation_id: uuid.UUID, user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> ObservationOut: return obs_out(await _observation_row(observation_id, db, user), include_private=False) @app.get("/api/tech/observations/{observation_id}", response_model=ObservationOut) async def get_technical_observation( observation_id: uuid.UUID, _admin: str = Depends(require_admin), db: AsyncSession = Depends(get_db), ) -> ObservationOut: return obs_out(await _observation_row(observation_id, db), include_private=True) @app.post("/api/observations", response_model=ObservationOut, status_code=201) async def create_observation( payload: ObservationCreate, background_tasks: BackgroundTasks, request: Request, current_user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> ObservationOut: commune = await resolve_commune(db, current_user=current_user, requested_name=payload.commune) obj = Observation( reference=make_reference("OBS"), commune=commune.nom, commune_code=commune.insee_com, cree_par=current_user.id, adresse=payload.adresse, type_observation=payload.type_observation, support=payload.support, hauteur_estimee=payload.hauteur_estimee, niveau_danger=payload.niveau_danger, description=payload.description, contact_nom=payload.contact_nom, contact_telephone=payload.contact_telephone, contact_email=str(payload.contact_email) if payload.contact_email else None, geom=WKTElement(f"POINT({payload.longitude} {payload.latitude})", srid=4326), ) db.add(obj) await db.commit() statement = ( select(Observation, func.ST_X(Observation.geom), func.ST_Y(Observation.geom)) .options(selectinload(Observation.photos)) .where(Observation.id == obj.id) ) row = (await db.execute(statement)).one() result = obs_out(row) remember_submission(request, obj.id) background_tasks.add_task( notify_observation, { "id": obj.id, "reference": obj.reference, "commune": obj.commune, "adresse": obj.adresse, "niveau_danger": obj.niveau_danger, "contact_email": obj.contact_email, "longitude": result.longitude, "latitude": result.latitude, }, ) return result @app.patch("/api/tech/observations/{observation_id}/status", response_model=ObservationOut) async def update_observation_status( observation_id: uuid.UUID, payload: ObservationStatusUpdate, _admin: str = Depends(require_admin_csrf), db: AsyncSession = Depends(get_db), ) -> ObservationOut: obj = await db.get(Observation, observation_id) if not obj: raise HTTPException(status_code=404, detail="Observation introuvable") obj.status = payload.status await db.commit() statement = ( select(Observation, func.ST_X(Observation.geom), func.ST_Y(Observation.geom)) .options(selectinload(Observation.photos)) .where(Observation.id == obj.id) ) return obs_out((await db.execute(statement)).one()) @app.post("/api/demandes", response_model=RequestOut, status_code=201) async def create_request( payload: RequestCreate, background_tasks: BackgroundTasks, request: Request, current_user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> RequestOut: observation: Observation | None = None request_geom = WKTElement(f"POINT({payload.longitude} {payload.latitude})", srid=4326) if payload.observation_id: observation = await db.get(Observation, payload.observation_id) if not observation: raise HTTPException(status_code=404, detail="Observation introuvable.") if current_user.role == UserRole.elu and observation.commune_code != current_user.commune_code: raise HTTPException(status_code=403, detail="Cette observation appartient à une autre commune.") if observation.status in OBSERVATION_HISTORY_STATUSES: raise HTTPException(status_code=409, detail="Cette observation est déjà clôturée ou archivée.") existing_request = await db.scalar( select(InterventionRequest.id).where( InterventionRequest.observation_id == observation.id ) ) if existing_request: raise HTTPException(status_code=409, detail="Une demande existe déjà pour cette observation.") commune = await db.get(Commune, observation.commune_code) if observation.commune_code else None if not commune: commune = await resolve_commune( db, current_user=current_user, requested_name=observation.commune, ) # Une conversion conserve exactement le point de l'observation : # aucun doublon spatial n'est créé sur la carte. request_geom = observation.geom else: commune = await resolve_commune( db, current_user=current_user, requested_name=payload.commune, ) obj = InterventionRequest( reference=make_reference("INT"), observation_id=payload.observation_id, type_demande=payload.type_demande, commune=commune.nom, commune_code=commune.insee_com, cree_par=current_user.id, demandeur_nom=payload.demandeur_nom, demandeur_fonction=payload.demandeur_fonction, demandeur_email=str(payload.demandeur_email), demandeur_telephone=payload.demandeur_telephone, adresse=payload.adresse, niveau_urgence=payload.niveau_urgence, description_risque=payload.description_risque, hauteur_estimee=payload.hauteur_estimee, support=payload.support, difficulte_acces=payload.difficulte_acces, proprietaire_nom=payload.proprietaire_nom, proprietaire_adresse=payload.proprietaire_adresse, proprietaire_telephone=payload.proprietaire_telephone, proprietaire_email=str(payload.proprietaire_email) if payload.proprietaire_email else None, accord_proprietaire=payload.accord_proprietaire, justification_danger_imminent=payload.justification_danger_imminent, geom=request_geom, ) db.add(obj) await db.flush() db.add( StatusHistory( demande_id=obj.id, ancien_statut=None, nouveau_statut=obj.status.value, auteur=payload.demandeur_nom, ) ) if observation: observation.status = ObservationStatus.transforme try: await db.commit() except IntegrityError as exc: await db.rollback() if observation: raise HTTPException( status_code=409, detail="Une demande existe déjà pour cette observation.", ) from exc raise statement = ( select(InterventionRequest, func.ST_X(InterventionRequest.geom), func.ST_Y(InterventionRequest.geom)) .options(selectinload(InterventionRequest.photos)) .where(InterventionRequest.id == obj.id) ) row = (await db.execute(statement)).one() result = request_out(row) remember_submission(request, obj.id) tariffs_for_pdf = await pdf_tariff_rows(db) background_tasks.add_task( notify_request, { "id": obj.id, "reference": obj.reference, "type_demande": obj.type_demande.value, "commune": obj.commune, "adresse": obj.adresse, "niveau_urgence": obj.niveau_urgence, "demandeur_nom": obj.demandeur_nom, "demandeur_email": obj.demandeur_email, "longitude": result.longitude, "latitude": result.latitude, "pdf_object": obj, "pdf_tariffs": tariffs_for_pdf, }, ) return result async def _request_rows( db: AsyncSession, status: RequestStatus | None = None, commune: str | None = None, user: User | None = None, ): statement = ( select(InterventionRequest, func.ST_X(InterventionRequest.geom), func.ST_Y(InterventionRequest.geom)) .options(selectinload(InterventionRequest.photos)) ) if status: statement = statement.where(InterventionRequest.status == status) if user and user.role == UserRole.elu: statement = statement.where(InterventionRequest.commune_code == user.commune_code) elif commune: statement = statement.where(func.lower(InterventionRequest.commune) == commune.lower()) statement = statement.order_by(InterventionRequest.created_at.desc()).limit(500) return (await db.execute(statement)).all() async def _request_row(request_id: uuid.UUID, db: AsyncSession, user: User | None = None): statement = ( select(InterventionRequest, func.ST_X(InterventionRequest.geom), func.ST_Y(InterventionRequest.geom)) .options(selectinload(InterventionRequest.photos)) .where(InterventionRequest.id == request_id) ) if user and user.role == UserRole.elu: statement = statement.where(InterventionRequest.commune_code == user.commune_code) row = (await db.execute(statement)).one_or_none() if not row: raise HTTPException(status_code=404, detail="Demande introuvable") return row @app.get("/api/demandes", response_model=list[RequestOut]) async def list_public_requests( status: RequestStatus | None = None, commune: str | None = None, user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> list[RequestOut]: rows = await _request_rows(db, status=status, commune=commune, user=user) return [request_out(row, include_private=False) for row in rows] @app.get("/api/tech/demandes", response_model=list[RequestOut]) async def list_technical_requests( status: RequestStatus | None = None, commune: str | None = None, _admin: str = Depends(require_admin), db: AsyncSession = Depends(get_db), ) -> list[RequestOut]: rows = await _request_rows(db, status=status, commune=commune) return [request_out(row, include_private=True) for row in rows] @app.get("/api/demandes/{request_id}", response_model=RequestOut) async def get_public_request( request_id: uuid.UUID, user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> RequestOut: return request_out(await _request_row(request_id, db, user), include_private=False) @app.get("/api/tech/demandes/{request_id}", response_model=RequestOut) async def get_technical_request( request_id: uuid.UUID, _admin: str = Depends(require_admin), db: AsyncSession = Depends(get_db), ) -> RequestOut: return request_out(await _request_row(request_id, db), include_private=True) @app.patch("/api/tech/demandes/{request_id}/status", response_model=RequestOut) async def update_request_status( request_id: uuid.UUID, payload: StatusUpdate, _admin: str = Depends(require_admin_csrf), db: AsyncSession = Depends(get_db), ) -> RequestOut: return await apply_treatment_update( request_id, TreatmentUpdate(status=payload.status, commentaire=payload.commentaire, auteur=_admin.email), db, ) @app.patch("/api/tech/demandes/{request_id}/traitement", response_model=RequestOut) async def update_request_treatment( request_id: uuid.UUID, payload: TreatmentUpdate, _admin: str = Depends(require_admin_csrf), db: AsyncSession = Depends(get_db), ) -> RequestOut: secured_payload = payload.model_copy(update={"auteur": _admin.email}) return await apply_treatment_update(request_id, secured_payload, db) async def apply_treatment_update(request_id: uuid.UUID, payload: TreatmentUpdate, db: AsyncSession) -> RequestOut: obj = await db.get(InterventionRequest, request_id) if not obj: raise HTTPException(status_code=404, detail="Demande introuvable") old_status = obj.status if payload.status == RequestStatus.archive and old_status not in REQUEST_ARCHIVABLE_STATUSES: raise HTTPException( status_code=422, detail="Une demande doit être terminée, refusée ou annulée avant son archivage.", ) if old_status == RequestStatus.archive and payload.status != RequestStatus.archive: raise HTTPException(status_code=422, detail="Une demande archivée ne peut pas être rouverte depuis cette interface.") old = old_status.value obj.status = payload.status obj.assigned_to = payload.assigned_to obj.intervention_date = payload.intervention_date db.add( StatusHistory( demande_id=obj.id, ancien_statut=old, nouveau_statut=payload.status.value, commentaire=payload.commentaire, auteur=payload.auteur, ) ) await db.commit() statement = ( select(InterventionRequest, func.ST_X(InterventionRequest.geom), func.ST_Y(InterventionRequest.geom)) .options(selectinload(InterventionRequest.photos)) .where(InterventionRequest.id == obj.id) ) return request_out((await db.execute(statement)).one()) async def _get_quote(request_id: uuid.UUID, db: AsyncSession) -> InterventionQuote | None: return await db.scalar( select(InterventionQuote).where(InterventionQuote.demande_id == request_id) ) @app.get("/api/tech/demandes/{request_id}/devis", response_model=QuoteOut | None) async def get_technical_quote( request_id: uuid.UUID, _user: User = Depends(require_technical_user), db: AsyncSession = Depends(get_db), ) -> QuoteOut | None: if not await db.get(InterventionRequest, request_id): raise HTTPException(status_code=404, detail="Demande introuvable.") obj = await _get_quote(request_id, db) return quote_out(obj) if obj else None @app.get("/api/demandes/{request_id}/devis", response_model=QuoteOut | None) async def get_public_quote( request_id: uuid.UUID, user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> QuoteOut | None: await _request_row(request_id, db, user) obj = await _get_quote(request_id, db) if not obj or obj.statut == QuoteStatus.brouillon: return None return quote_out(obj) @app.put("/api/tech/demandes/{request_id}/devis", response_model=QuoteOut) async def upsert_quote( request_id: uuid.UUID, payload: QuoteUpsert, request: Request, technical_user: User = Depends(require_technical_csrf), db: AsyncSession = Depends(get_db), ) -> QuoteOut: request_obj = await db.get(InterventionRequest, request_id) if not request_obj: raise HTTPException(status_code=404, detail="Demande introuvable.") tariff = await db.get(InterventionTariff, payload.type_intervention) if not tariff or not tariff.actif: raise HTTPException(status_code=422, detail="Le type d'intervention ou son tarif est indisponible.") obj = await _get_quote(request_id, db) created = obj is None if created: obj = InterventionQuote( demande_id=request_id, reference=make_reference("DEV"), date_devis=datetime.now(timezone.utc).date(), cree_par=technical_user.id, ) db.add(obj) obj.statut = payload.statut obj.type_intervention = tariff.code obj.libelle_snapshot = tariff.libelle obj.prix_unitaire = tariff.prix obj.quantite = 1 obj.montant_total = tariff.prix obj.commune_facturee = request_obj.commune obj.validite_jusqu_au = payload.validite_jusqu_au or (datetime.now(timezone.utc) + timedelta(days=30)).date() obj.observations = payload.observations await db.flush() db.add(AuditLog( utilisateur_id=technical_user.id, action="devis_cree" if created else "devis_modifie", type_objet="devis_intervention", objet_id=obj.id, details={ "demande_id": str(request_id), "reference": obj.reference, "type_intervention": tariff.code, "montant_total": str(obj.montant_total), "statut": obj.statut.value, }, adresse_ip=request.client.host if request.client else None, )) await db.commit() await db.refresh(obj) return quote_out(obj) @app.get("/api/tech/demandes/{request_id}/devis/pdf") async def technical_quote_pdf( request_id: uuid.UUID, _user: User = Depends(require_technical_user), db: AsyncSession = Depends(get_db), ) -> Response: request_obj = await db.get(InterventionRequest, request_id) quote = await _get_quote(request_id, db) if not request_obj or not quote: raise HTTPException(status_code=404, detail="Devis introuvable.") content = build_quote_pdf(request_obj, quote) headers = {"Content-Disposition": f'attachment; filename="{quote.reference}.pdf"'} return Response(content=content, media_type="application/pdf", headers=headers) @app.get("/api/demandes/{request_id}/devis/pdf") async def public_quote_pdf( request_id: uuid.UUID, user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> Response: request_obj, _longitude, _latitude = await _request_row(request_id, db, user) quote = await _get_quote(request_id, db) if not quote or quote.statut == QuoteStatus.brouillon: raise HTTPException(status_code=404, detail="Devis indisponible.") content = build_quote_pdf(request_obj, quote) headers = {"Content-Disposition": f'attachment; filename="{quote.reference}.pdf"'} return Response(content=content, media_type="application/pdf", headers=headers) @app.get("/api/tech/demandes/{request_id}/intervention", response_model=InterventionOut | None) async def get_intervention( request_id: uuid.UUID, _admin: str = Depends(require_admin), db: AsyncSession = Depends(get_db), ) -> InterventionOut | None: obj = await db.scalar(select(Intervention).where(Intervention.demande_id == request_id)) if not obj: return None return InterventionOut( id=obj.id, demande_id=obj.demande_id, agent_nom=obj.agent_nom, debut_intervention=obj.debut_intervention, fin_intervention=obj.fin_intervention, methode=obj.methode, resultat=obj.resultat, cout_ht=obj.cout_ht, compte_rendu=obj.compte_rendu, signature_proprietaire_apres=obj.signature_proprietaire_apres, ) @app.put("/api/tech/demandes/{request_id}/intervention", response_model=InterventionOut) async def upsert_intervention( request_id: uuid.UUID, payload: InterventionUpsert, _admin: str = Depends(require_admin_csrf), db: AsyncSession = Depends(get_db), ) -> InterventionOut: request_obj = await db.get(InterventionRequest, request_id) if not request_obj: raise HTTPException(status_code=404, detail="Demande introuvable") obj = await db.scalar(select(Intervention).where(Intervention.demande_id == request_id)) if not obj: obj = Intervention(demande_id=request_id) db.add(obj) values = payload.model_dump() quote = await _get_quote(request_id, db) method = values.get("methode") or (quote.type_intervention if quote else None) if quote and method and method != quote.type_intervention: raise HTTPException( status_code=422, detail="Le type d'intervention doit correspondre au devis. Modifiez d'abord le devis.", ) if method: tariff = await db.get(InterventionTariff, method) if not tariff: raise HTTPException(status_code=422, detail="Type d'intervention inconnu.") values["methode"] = method values["cout_ht"] = quote.montant_total if quote else tariff.prix for field, value in values.items(): setattr(obj, field, value) await db.commit() await db.refresh(obj) return InterventionOut( id=obj.id, demande_id=obj.demande_id, agent_nom=obj.agent_nom, debut_intervention=obj.debut_intervention, fin_intervention=obj.fin_intervention, methode=obj.methode, resultat=obj.resultat, cout_ht=obj.cout_ht, compte_rendu=obj.compte_rendu, signature_proprietaire_apres=obj.signature_proprietaire_apres, ) @app.post("/api/{entity}/{entity_id}/photos", status_code=201) async def upload_photo( entity: str, entity_id: uuid.UUID, request: Request, file: UploadFile = File(...), _user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> dict[str, str]: if entity not in {"observations", "demandes"}: raise HTTPException(status_code=400, detail="Type d'objet invalide") if not can_upload_to_submission(request, entity_id): raise HTTPException(status_code=403, detail="Ce navigateur n'est pas autorisé à joindre une photo à ce dossier.") parent = await db.get(Observation if entity == "observations" else InterventionRequest, entity_id) if not parent: raise HTTPException(status_code=404, detail="Dossier introuvable") if _user.role == UserRole.elu and parent.commune_code != _user.commune_code: raise HTTPException(status_code=403, detail="Ce dossier appartient à une autre commune.") max_bytes = settings.max_upload_mb * 1024 * 1024 content = await file.read(max_bytes + 1) await file.close() if len(content) > max_bytes: raise HTTPException(status_code=413, detail=f"Image trop volumineuse (maximum {settings.max_upload_mb} Mo).") try: normalised, suffix, mime = await asyncio.to_thread( normalise_image, content, max_width=settings.image_max_width, max_height=settings.image_max_height, jpeg_quality=settings.image_jpeg_quality, ) except InvalidImage as exc: raise HTTPException(status_code=415, detail=str(exc)) from exc filename = f"{uuid.uuid4().hex}{suffix}" path = settings.upload_path / filename await asyncio.to_thread(path.write_bytes, normalised) photo = Photo( observation_id=entity_id if entity == "observations" else None, demande_id=entity_id if entity == "demandes" else None, filename=filename, original_name=safe_original_name(file.filename), mime_type=mime, size_bytes=len(normalised), ) try: db.add(photo) await db.commit() except Exception: path.unlink(missing_ok=True) raise return {"id": str(photo.id), "url": public_path(f"/uploads/{filename}")} @app.get("/api/demandes/{request_id}/pdf") async def public_request_pdf( request_id: uuid.UUID, user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ) -> Response: obj, longitude, latitude = await _request_row(request_id, db, user) quote = await _get_quote(request_id, db) intervention = await db.scalar(select(Intervention).where(Intervention.demande_id == request_id)) tariffs_for_pdf = await pdf_tariff_rows(db) content = build_request_pdf(obj, longitude, latitude, quote=quote, intervention=intervention, tariffs=tariffs_for_pdf) headers = {"Content-Disposition": f'attachment; filename="{obj.reference}.pdf"'} return Response(content=content, media_type="application/pdf", headers=headers) @app.get("/api/tech/demandes/{request_id}/pdf") async def request_pdf( request_id: uuid.UUID, _admin: str = Depends(require_admin), db: AsyncSession = Depends(get_db), ) -> Response: statement = select(InterventionRequest, func.ST_X(InterventionRequest.geom), func.ST_Y(InterventionRequest.geom)).where(InterventionRequest.id == request_id) row = (await db.execute(statement)).one_or_none() if not row: raise HTTPException(status_code=404, detail="Demande introuvable") obj, longitude, latitude = row quote = await _get_quote(request_id, db) intervention = await db.scalar(select(Intervention).where(Intervention.demande_id == request_id)) tariffs_for_pdf = await pdf_tariff_rows(db) content = build_request_pdf(obj, longitude, latitude, quote=quote, intervention=intervention, tariffs=tariffs_for_pdf) headers = {"Content-Disposition": f'attachment; filename="{obj.reference}.pdf"'} return Response(content=content, media_type="application/pdf", headers=headers) @app.get("/api/stats", response_model=StatsOut) async def stats( _user: User = Depends(require_technical_user), db: AsyncSession = Depends(get_db), ) -> StatsOut: obs_count = await db.scalar(select(func.count()).select_from(Observation).where(Observation.status == ObservationStatus.a_qualifier)) statement = select( func.sum(case((InterventionRequest.niveau_urgence == "urgent", 1), else_=0)), func.sum(case((InterventionRequest.status.in_([RequestStatus.transmise, RequestStatus.a_qualifier]), 1), else_=0)), func.sum(case((InterventionRequest.status == RequestStatus.programmee, 1), else_=0)), func.sum(case((InterventionRequest.status == RequestStatus.terminee, 1), else_=0)), ) urgent, to_qualify, planned, done = (await db.execute(statement)).one() return StatsOut( observations_a_qualifier=obs_count or 0, demandes_urgentes=urgent or 0, demandes_a_qualifier=to_qualify or 0, interventions_programmees=planned or 0, interventions_terminees=done or 0, ) @app.middleware("http") async def security_headers(request: Request, call_next): response = await call_next(request) response.headers.setdefault("X-Content-Type-Options", "nosniff") response.headers.setdefault("X-Frame-Options", "SAMEORIGIN") response.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin") response.headers.setdefault( "Permissions-Policy", "camera=(self), geolocation=(self), microphone=()", ) if request.url.path.startswith(("/api/tech", "/api/auth")) or request.url.path in { "/admin.html", "/users.html", "/login.html", }: response.headers.setdefault("Cache-Control", "no-store") if settings.is_production and request.url.scheme == "https": response.headers.setdefault("Strict-Transport-Security", "max-age=31536000; includeSubDomains") return response @app.get("/", include_in_schema=False) @app.get("/index.html", include_in_schema=False) async def index_page(request: Request): if not session_is_authenticated(request): return RedirectResponse(url=public_path("/login.html"), status_code=303) return FileResponse(frontend_path / "index.html", headers={"Cache-Control": "no-store"}) @app.get("/admin.html", include_in_schema=False) async def admin_page(request: Request): if not session_is_technical(request): return RedirectResponse(url=public_path("/login.html"), status_code=303) return FileResponse(frontend_path / "admin.html", headers={"Cache-Control": "no-store"}) @app.get("/users.html", include_in_schema=False) async def users_page(request: Request): if not session_is_admin(request): return RedirectResponse(url=public_path("/login.html"), status_code=303) return FileResponse(frontend_path / "users.html", headers={"Cache-Control": "no-store"}) @app.get("/login.html", include_in_schema=False) async def login_page(request: Request): if session_is_authenticated(request): destination = "/admin.html" if session_is_technical(request) else "/index.html" return RedirectResponse(url=public_path(destination), status_code=303) return FileResponse(frontend_path / "login.html", headers={"Cache-Control": "no-store"}) @app.get("/uploads/{filename}", include_in_schema=False) async def protected_upload( filename: str, user: User = Depends(require_user), db: AsyncSession = Depends(get_db), ): safe_name = Path(filename).name if safe_name != filename: raise HTTPException(status_code=400, detail="Nom de fichier invalide") photo = await db.scalar(select(Photo).where(Photo.filename == safe_name)) if not photo: raise HTTPException(status_code=404, detail="Photo introuvable") if user.role == UserRole.elu: allowed = False if photo.observation_id: parent = await db.get(Observation, photo.observation_id) allowed = bool(parent and parent.commune_code == user.commune_code) elif photo.demande_id: parent = await db.get(InterventionRequest, photo.demande_id) allowed = bool(parent and parent.commune_code == user.commune_code) if not allowed: raise HTTPException(status_code=403, detail="Photo inaccessible pour cette commune.") path = settings.upload_path / safe_name if not path.is_file(): raise HTTPException(status_code=404, detail="Photo introuvable") return FileResponse( path, media_type=photo.mime_type or "image/jpeg", headers={"Cache-Control": "private, max-age=3600"}, ) app.mount("/", StaticFiles(directory=frontend_path, html=True), name="frontend")