"""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é ; - publics (**aucune authentification** — les seuls endpoints de toute l'API dans ce cas) : consultation de la configuration et soumission, protégés par hCaptcha (`app/forms/captcha.py`) et un filet de sécurité global (`app/auth/rate_limit.py:form_submission_limiter`). 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 uuid from fastapi import ( APIRouter, Depends, File, Form, HTTPException, Response, UploadFile, status, ) from sqlalchemy import delete from sqlmodel import Session, select from app.api._common import set_artists, set_wished_positions from app.api.deps import get_edition from app.api.schemas import ( FormFieldsSet, FormSubmissionResult, FormTemplateRead, FormTemplateUpdate, PublicArtistOption, PublicFormConfig, PublicPositionOption, ) from app.auth.deps import require_coordinator from app.auth.rate_limit import form_submission_limiter from app.db import get_session from app.forms import verify_hcaptcha from app.models import ( Account, Artist, Edition, FormField, FormFieldType, FormTemplate, Position, Volunteer, ) 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 _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, yapla_url=template.yapla_url, ) @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("yapla_url"): if not changes["yapla_url"].startswith(("http://", "https://")): raise HTTPException( status_code=422, detail={"code": "form.invalid_yapla_url"} ) for field, value in changes.items(): setattr(template, field, value) session.add(template) session.commit() return _template_read(template, _enabled_fields(session, edition.id)) 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) ) ] return PublicFormConfig( is_open=template.is_open, fields=fields, positions=positions, artists=artists, has_logo=template.logo is not None, description=template.description, yapla_url=template.yapla_url, ) 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), 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) 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 ), willing_to_pay_fee=( willing_to_pay_fee 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) return FormSubmissionResult(volunteer_id=volunteer.id)