"""Formulaire public d'inscription des bénévoles. Deux familles d'endpoints sous le même routeur : - configuration (COORDINATION) : quels champs sont activés, formulaire ouvert ou fermé, identifiants SumUp de paiement de la cotisation ; - publics (**aucune authentification**) : consultation de la configuration et soumission (protégées par hCaptcha, `app/forms/captcha.py`, et un filet de sécurité global, `app/auth/rate_limit.py:form_submission_limiter`), et le webhook SumUp (`/public/sumup-webhook`, appelé par SumUp lui-même — pas de hCaptcha possible serveur à serveur, voir `sumup_webhook` plus bas). Nom et prénom sont toujours collectés (indispensables à la création du bénévole) : ils ne font pas partie des champs activables, contrairement aux neuf autres (`FormFieldType`). Les vœux de postes/artistes réutilisent la validation de `app/api/_common.py` (partagée avec `volunteers.py`), avec deux règles propres au formulaire : au plus `_MAX_POSITION_WISHES` postes, et au plus un artiste souhaité par jour de festival (`performance_date`, voir `_check_one_artist_wish_per_day`). Le vœu de compagnon est un texte libre (`companion_wish_note`) résolu ensuite manuellement par la coordination, faute de savoir si la personne visée est déjà inscrite (voir CLAUDE.md § 2, `Volunteer`).""" import json import logging import uuid from fastapi import ( APIRouter, Depends, File, Form, HTTPException, Response, UploadFile, status, ) from sqlalchemy import delete from sqlalchemy.exc import IntegrityError from sqlmodel import Session, select from app.api._common import owned_or_404, set_artists, set_wished_positions from app.api.deps import get_edition from app.api.schemas import ( FormFieldsSet, FormPaymentOptionCreate, FormPaymentOptionRead, FormPaymentOptionUpdate, FormSubmissionResult, FormTemplateRead, FormTemplateUpdate, PublicArtistOption, PublicFormConfig, PublicPaymentOption, PublicPositionOption, SumUpWebhookPayload, ) from app.auth.deps import require_coordinator from app.auth.rate_limit import form_submission_limiter from app.config import get_settings from app.db import get_session from app.forms import verify_hcaptcha from app.models import ( Account, Artist, Edition, FeePaymentStatus, FormField, FormFieldType, FormPaymentOption, FormTemplate, Position, Volunteer, local_now, ) from app.sumup import create_checkout, get_checkout logger = logging.getLogger("app.forms") router = APIRouter(prefix="/editions/{edition_id}/form", tags=["form"]) _MAX_PHOTO_BYTES = 8 * 1024 * 1024 _ALLOWED_PHOTO_TYPES = {"image/png", "image/jpeg"} # Vœux de poste : au plus deux, pour éviter que tout le monde coche tous les # postes « au cas où » — force une vraie priorisation dès le formulaire. _MAX_POSITION_WISHES = 2 # --------------------------------------------------------------------------- # # Configuration (COORDINATION) # # --------------------------------------------------------------------------- # def _get_or_create_template(session: Session, edition_id: uuid.UUID) -> FormTemplate: template = session.exec( select(FormTemplate).where(FormTemplate.edition_id == edition_id) ).first() if template is None: template = FormTemplate(edition_id=edition_id) session.add(template) session.commit() session.refresh(template) return template def _enabled_fields(session: Session, edition_id: uuid.UUID) -> list[FormFieldType]: return list( session.exec( select(FormField.field_type) .where(FormField.edition_id == edition_id) .order_by(FormField.field_type) ) ) def _payment_options(session: Session, edition_id: uuid.UUID) -> list[FormPaymentOption]: return list( session.exec( select(FormPaymentOption) .where(FormPaymentOption.edition_id == edition_id) .order_by(FormPaymentOption.order, FormPaymentOption.label) ) ) def _sumup_enabled(template: FormTemplate, options: list[FormPaymentOption]) -> bool: return bool(template.sumup_merchant_code and template.sumup_api_key and options) def _template_read(template: FormTemplate, fields: list[FormFieldType]) -> FormTemplateRead: return FormTemplateRead( edition_id=template.edition_id, is_open=template.is_open, fields=fields, has_logo=template.logo is not None, description=template.description, sumup_merchant_code=template.sumup_merchant_code, sumup_api_key_set=bool(template.sumup_api_key), ) @router.get("", response_model=FormTemplateRead) def get_form( edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> FormTemplateRead: template = _get_or_create_template(session, edition.id) return _template_read(template, _enabled_fields(session, edition.id)) @router.put("/fields", response_model=FormTemplateRead) def set_fields( payload: FormFieldsSet, edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> FormTemplateRead: template = _get_or_create_template(session, edition.id) session.execute(delete(FormField).where(FormField.edition_id == edition.id)) for field_type in dict.fromkeys(payload.fields): # dé-doublonne, garde l'ordre session.add(FormField(edition_id=edition.id, field_type=field_type)) session.commit() return _template_read(template, _enabled_fields(session, edition.id)) @router.patch("", response_model=FormTemplateRead) def update_form( payload: FormTemplateUpdate, edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> FormTemplateRead: template = _get_or_create_template(session, edition.id) changes = payload.model_dump(exclude_unset=True) if changes.get("sumup_api_key") == "": changes["sumup_api_key"] = None for field, value in changes.items(): setattr(template, field, value) session.add(template) session.commit() return _template_read(template, _enabled_fields(session, edition.id)) # --------------------------------------------------------------------------- # # Types de paiement (COORDINATION) — voir `FormPaymentOption` # # --------------------------------------------------------------------------- # @router.get("/payment-options", response_model=list[FormPaymentOptionRead]) def list_payment_options( edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> list[FormPaymentOption]: return _payment_options(session, edition.id) @router.post( "/payment-options", response_model=FormPaymentOptionRead, status_code=status.HTTP_201_CREATED, ) def create_payment_option( payload: FormPaymentOptionCreate, edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> FormPaymentOption: if payload.amount <= 0: raise HTTPException( status_code=422, detail={"code": "form.invalid_sumup_amount"} ) option = FormPaymentOption( edition_id=edition.id, label=payload.label.strip(), amount=payload.amount, order=payload.order, ) session.add(option) try: session.commit() except IntegrityError as exc: session.rollback() raise HTTPException( status_code=409, detail={"code": "form.duplicate_payment_option_label"} ) from exc session.refresh(option) return option @router.patch("/payment-options/{option_id}", response_model=FormPaymentOptionRead) def update_payment_option( option_id: uuid.UUID, payload: FormPaymentOptionUpdate, edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> FormPaymentOption: option = owned_or_404(session.get(FormPaymentOption, option_id), edition.id) changes = payload.model_dump(exclude_unset=True) if changes.get("amount") is not None and changes["amount"] <= 0: raise HTTPException( status_code=422, detail={"code": "form.invalid_sumup_amount"} ) if changes.get("label") is not None: changes["label"] = changes["label"].strip() for field, value in changes.items(): setattr(option, field, value) session.add(option) try: session.commit() except IntegrityError as exc: session.rollback() raise HTTPException( status_code=409, detail={"code": "form.duplicate_payment_option_label"} ) from exc session.refresh(option) return option @router.delete("/payment-options/{option_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_payment_option( option_id: uuid.UUID, edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> Response: option = owned_or_404(session.get(FormPaymentOption, option_id), edition.id) # Les bénévoles ayant déjà choisi cette option gardent leur copie figée # (`Volunteer.fee_payment_label`/`fee_amount`) — rien à faire côté eux. session.delete(option) session.commit() return Response(status_code=status.HTTP_204_NO_CONTENT) async def _read_logo(file: UploadFile) -> tuple[bytes, str]: content_type = file.content_type or "" if content_type not in _ALLOWED_PHOTO_TYPES: raise HTTPException( status_code=422, detail={"code": "form.logo_type_not_supported"} ) content = await file.read() if not content: raise HTTPException(status_code=400, detail={"code": "form.logo_empty"}) if len(content) > _MAX_PHOTO_BYTES: raise HTTPException(status_code=413, detail={"code": "form.logo_too_large"}) return content, content_type @router.put("/logo", response_model=FormTemplateRead) async def set_logo( file: UploadFile = File(...), edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> FormTemplateRead: template = _get_or_create_template(session, edition.id) content, content_type = await _read_logo(file) template.logo = content template.logo_content_type = content_type session.add(template) session.commit() return _template_read(template, _enabled_fields(session, edition.id)) @router.get("/logo") def get_logo( edition: Edition = Depends(get_edition), session: Session = Depends(get_session), ) -> Response: """Public (aucune authentification) : affiché sur `/inscription`, comme `GET .../form/public` — voir le commentaire du module.""" template = _get_or_create_template(session, edition.id) if template.logo is None: raise HTTPException(status_code=404, detail={"code": "error.not_found"}) return Response( content=template.logo, media_type=template.logo_content_type or "application/octet-stream", ) # --------------------------------------------------------------------------- # # Formulaire public (aucune authentification) # # --------------------------------------------------------------------------- # @router.get("/public", response_model=PublicFormConfig) def get_public_form( edition: Edition = Depends(get_edition), session: Session = Depends(get_session), ) -> PublicFormConfig: template = _get_or_create_template(session, edition.id) fields = _enabled_fields(session, edition.id) positions: list[PublicPositionOption] = [] if FormFieldType.POSITION_WISHES in fields: positions = [ PublicPositionOption(id=p.id, name=p.name) for p in session.exec( select(Position) .where(Position.edition_id == edition.id) .order_by(Position.name) ) ] artists: list[PublicArtistOption] = [] if FormFieldType.ARTIST_WISHES in fields: artists = [ PublicArtistOption( id=a.id, name=a.name, performance_date=a.performance_date ) for a in session.exec( select(Artist) .where(Artist.edition_id == edition.id) .order_by(Artist.performance_date, Artist.name) ) ] options = _payment_options(session, edition.id) return PublicFormConfig( is_open=template.is_open, fields=fields, positions=positions, artists=artists, has_logo=template.logo is not None, description=template.description, sumup_enabled=_sumup_enabled(template, options), payment_options=[ PublicPaymentOption(id=o.id, label=o.label, amount=o.amount) for o in options ], ) def _parse_id_list(raw: str | None) -> list[uuid.UUID]: """`position_wish_ids`/`artist_wish_ids` arrivent en JSON dans un champ `multipart/form-data` (impossible d'y envoyer un tableau natif).""" if not raw: return [] try: values = json.loads(raw) if not isinstance(values, list): raise ValueError return [uuid.UUID(str(v)) for v in values] except (ValueError, TypeError) as exc: raise HTTPException( status_code=422, detail={"code": "form.invalid_payload"} ) from exc def _check_one_artist_wish_per_day( session: Session, edition_id: uuid.UUID, artist_ids: list[uuid.UUID] ) -> None: """Un seul vœu d'artiste par soir de festival — `set_artists` (validation d'appartenance à l'édition) s'occupe du reste.""" dates = list( session.exec( select(Artist.performance_date).where( Artist.edition_id == edition_id, Artist.id.in_(artist_ids) ) ) ) if len(dates) != len(set(dates)): raise HTTPException( status_code=422, detail={"code": "form.too_many_artists_same_day"} ) @router.post( "/public/submit", response_model=FormSubmissionResult, status_code=status.HTTP_201_CREATED, ) async def submit_public_form( edition: Edition = Depends(get_edition), session: Session = Depends(get_session), first_name: str = Form(...), last_name: str = Form(...), hcaptcha_token: str = Form(...), email: str | None = Form(None), phone: str | None = Form(None), comment: str | None = Form(None), companion_wish_note: str | None = Form(None), available: bool = Form(False), willing_to_pay_fee: bool = Form(False), payment_option_id: str | None = Form(None), position_wish_ids: str | None = Form(None), artist_wish_ids: str | None = Form(None), photo: UploadFile | None = File(None), ) -> FormSubmissionResult: # Filet de sécurité global (par édition) : hCaptcha est la défense # principale, voir `app/auth/rate_limit.py`. form_submission_limiter.hit(str(edition.id)) template = _get_or_create_template(session, edition.id) if not template.is_open: raise HTTPException(status_code=422, detail={"code": "form.closed"}) if not verify_hcaptcha(hcaptcha_token): raise HTTPException(status_code=422, detail={"code": "form.captcha_failed"}) if not first_name.strip() or not last_name.strip(): raise HTTPException( status_code=422, detail={ "code": "import.row.missing_required_field", "params": {"field": "first_name"}, }, ) fields = set(_enabled_fields(session, edition.id)) position_ids = ( _parse_id_list(position_wish_ids) if FormFieldType.POSITION_WISHES in fields else [] ) if len(position_ids) > _MAX_POSITION_WISHES: raise HTTPException( status_code=422, detail={"code": "form.too_many_position_wishes"} ) artist_ids = ( _parse_id_list(artist_wish_ids) if FormFieldType.ARTIST_WISHES in fields else [] ) if artist_ids: _check_one_artist_wish_per_day(session, edition.id, artist_ids) payment_option: FormPaymentOption | None = None if FormFieldType.WILLING_TO_PAY_FEE in fields and payment_option_id: try: option_uuid = uuid.UUID(payment_option_id) except ValueError: option_uuid = None payment_option = ( session.get(FormPaymentOption, option_uuid) if option_uuid else None ) if payment_option is None or payment_option.edition_id != edition.id: raise HTTPException( status_code=422, detail={"code": "form.unknown_payment_option"} ) volunteer = Volunteer( edition_id=edition.id, first_name=first_name.strip().title(), last_name=last_name.strip().title(), email=((email or "").strip() or None) if FormFieldType.EMAIL in fields else None, phone=((phone or "").strip() or None) if FormFieldType.PHONE in fields else None, comment=( (comment or "").strip() or None if FormFieldType.COMMENT in fields else None ), companion_wish_note=( (companion_wish_note or "").strip() or None if FormFieldType.COMPANION_WISH in fields else None ), self_declared_available=( available if FormFieldType.AVAILABLE in fields else False ), # Choisir un type de paiement vaut « je participe » même sans cocher # la case (le formulaire public n'affiche d'ailleurs plus qu'un choix # unique dès qu'au moins un type est configuré, voir PublicSignupPage). willing_to_pay_fee=( (willing_to_pay_fee or payment_option is not None) if FormFieldType.WILLING_TO_PAY_FEE in fields else False ), ) if FormFieldType.PHOTO in fields and photo is not None and photo.filename: content_type = photo.content_type or "" if content_type not in _ALLOWED_PHOTO_TYPES: raise HTTPException( status_code=422, detail={"code": "form.photo_type_not_supported"} ) content = await photo.read() if len(content) > _MAX_PHOTO_BYTES: raise HTTPException( status_code=413, detail={"code": "form.photo_too_large"} ) if content: volunteer.photo = content volunteer.photo_content_type = content_type session.add(volunteer) session.flush() if FormFieldType.POSITION_WISHES in fields: set_wished_positions(session, volunteer, position_ids) if FormFieldType.ARTIST_WISHES in fields: set_artists(session, volunteer, artist_ids) session.commit() session.refresh(volunteer) checkout_url = None if ( payment_option is not None and template.sumup_merchant_code and template.sumup_api_key ): checkout_url = _create_fee_checkout( session, edition, template, volunteer, payment_option ) return FormSubmissionResult(volunteer_id=volunteer.id, sumup_checkout_url=checkout_url) def _create_fee_checkout( session: Session, edition: Edition, template: FormTemplate, volunteer: Volunteer, option: FormPaymentOption, ) -> str | None: """Crée le checkout SumUp du type de paiement choisi — best effort (comme l'envoi d'emails) : un échec de l'appel API ne remet jamais en cause l'inscription déjà enregistrée. `option.label`/`option.amount` sont recopiés sur le bénévole (`fee_payment_label`/`fee_amount`) avant même l'appel, pour garder une trace du choix même si l'appel échoue.""" volunteer.fee_payment_label = option.label volunteer.fee_amount = option.amount base = get_settings().public_base_url.rstrip("/") checkout = create_checkout( api_key=template.sumup_api_key, merchant_code=template.sumup_merchant_code, amount=option.amount, currency="EUR", checkout_reference=str(volunteer.id), description=f"{option.label} — édition {edition.year}", return_url=f"{base}/api/editions/{edition.id}/form/public/sumup-webhook", redirect_url=f"{base}/inscription?paiement=ok", ) if checkout is not None: volunteer.fee_checkout_id = checkout.id volunteer.fee_payment_status = FeePaymentStatus.PENDING session.add(volunteer) session.commit() return checkout.hosted_checkout_url if checkout is not None else None @router.post("/public/sumup-webhook", status_code=status.HTTP_200_OK) def sumup_webhook( payload: SumUpWebhookPayload, edition: Edition = Depends(get_edition), session: Session = Depends(get_session), ) -> Response: """Appelé par SumUp (aucune authentification possible côté SumUp) à chaque changement de statut d'un checkout. Le payload ne donne qu'un identifiant : jamais fait confiance sans relire l'état réel via l'API (voir `app/sumup/client.py:get_checkout`). Répond toujours `200`, y compris sur un événement inconnu — SumUp réessaie sinon inutilement.""" template = _get_or_create_template(session, edition.id) if template.sumup_api_key: checkout = get_checkout(api_key=template.sumup_api_key, checkout_id=payload.id) volunteer = ( session.exec( select(Volunteer).where( Volunteer.edition_id == edition.id, Volunteer.fee_checkout_id == payload.id, ) ).first() if checkout is not None else None ) if checkout is not None and volunteer is not None: try: new_status = FeePaymentStatus(checkout.status) except ValueError: new_status = None if new_status is not None: volunteer.fee_payment_status = new_status if new_status == FeePaymentStatus.PAID and volunteer.fee_paid_at is None: volunteer.fee_paid_at = local_now() session.add(volunteer) session.commit() elif checkout is None: logger.warning("Webhook SumUp : relecture du checkout impossible.") return Response(status_code=status.HTTP_200_OK)