AdamThomasandClaude Sonnet 5
b268e308ab
- app/email/ : sender (smtplib, STARTTLS ou SSL, connexion réutilisée pour un
lot) + templates (rendu, textes dans locales/fr.json). EMAIL_ENABLED=false
(défaut) ⇒ journalisé seulement, sans destinataire ni corps (RGPD).
- Lien magique : POST .../auth/magic-link envoie désormais l'email
(PUBLIC_BASE_URL + magic_link_path + ?token=). Helper app/auth/magic_link.py
partagé. Sans SMTP, le jeton non nominatif reste journalisé pour le dev.
- Récapitulatif de planning : POST /editions/{id}/notifications/planning
(COORDINATION). subject/body sont des modèles ({prenom}, {nom}, {nb_creneaux},
{nb_heures}, {creneaux}, {lien_connexion} → lien magique personnel émis).
Cible tous les bénévoles avec email ou une liste. Envoi en BackgroundTasks.
- Config : PUBLIC_BASE_URL, EMAIL_ENABLED, SMTP_* ; .env.example,
docker-compose (dev + prod), deploy/README.
- 9 tests (sender activé/désactivé, rendu, notifications : contenu, gabarit par
défaut, lien magique embarqué, sous-ensemble, réservé COORDINATION).
137 tests au total ; ruff + alembic check verts.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017uUsRg7XzopkCPWimwg9ZM
121 lines
4.3 KiB
Python
121 lines
4.3 KiB
Python
"""Envoi d'emails d'information aux bénévoles (récapitulatif de planning)."""
|
|
|
|
import uuid
|
|
from collections import defaultdict
|
|
|
|
from fastapi import APIRouter, BackgroundTasks, Depends
|
|
from sqlmodel import Session, select
|
|
|
|
from app.api.deps import get_edition
|
|
from app.api.schemas import (
|
|
PlanningNotificationRequest,
|
|
PlanningNotificationResult,
|
|
)
|
|
from app.auth.deps import require_coordinator
|
|
from app.auth.magic_link import build_magic_link_url, issue_magic_link_token
|
|
from app.db import get_session
|
|
from app.email import render_planning, send_many
|
|
from app.i18n import t
|
|
from app.models import Account, Assignment, Edition, Position, Shift, Volunteer
|
|
|
|
router = APIRouter(
|
|
prefix="/editions/{edition_id}/notifications", tags=["notifications"]
|
|
)
|
|
|
|
_LOGIN_PLACEHOLDER = "{lien_connexion}"
|
|
|
|
|
|
def _login_link(session: Session, edition: Edition, volunteer: Volunteer) -> str:
|
|
"""Lien magique personnel si le bénévole a un compte, sinon l'accueil."""
|
|
account = session.exec(
|
|
select(Account)
|
|
.where(Account.edition_id == edition.id)
|
|
.where(Account.volunteer_id == volunteer.id)
|
|
.where(Account.password_hash.is_(None))
|
|
).first()
|
|
if account is None:
|
|
return build_magic_link_url("").split("?")[0]
|
|
return build_magic_link_url(issue_magic_link_token(session, account))
|
|
|
|
|
|
@router.post("/planning", response_model=PlanningNotificationResult)
|
|
def send_planning_notification(
|
|
payload: PlanningNotificationRequest,
|
|
background: BackgroundTasks,
|
|
edition: Edition = Depends(get_edition),
|
|
_guard: Account = Depends(require_coordinator),
|
|
session: Session = Depends(get_session),
|
|
) -> PlanningNotificationResult:
|
|
subject_template = payload.subject or t("email.planning.default_subject")
|
|
body_template = payload.body or t("email.planning.default_body")
|
|
wants_login_link = _LOGIN_PLACEHOLDER in subject_template or (
|
|
_LOGIN_PLACEHOLDER in body_template
|
|
)
|
|
|
|
stmt = select(Volunteer).where(Volunteer.edition_id == edition.id)
|
|
if payload.volunteer_ids is not None:
|
|
stmt = stmt.where(Volunteer.id.in_(payload.volunteer_ids))
|
|
volunteers = list(session.exec(stmt.order_by(Volunteer.last_name)))
|
|
|
|
positions = {
|
|
p.id: p
|
|
for p in session.exec(
|
|
select(Position).where(Position.edition_id == edition.id)
|
|
)
|
|
}
|
|
shifts_by_volunteer: dict[uuid.UUID, list[Shift]] = defaultdict(list)
|
|
for volunteer_id, shift in session.exec(
|
|
select(Assignment.volunteer_id, Shift)
|
|
.join(Shift, Shift.id == Assignment.shift_id)
|
|
.where(Assignment.edition_id == edition.id)
|
|
):
|
|
shifts_by_volunteer[volunteer_id].append(shift)
|
|
|
|
items: list[tuple[str, str, str]] = []
|
|
recipients: list[uuid.UUID] = []
|
|
skipped_no_email = 0
|
|
for volunteer in volunteers:
|
|
if not volunteer.email:
|
|
skipped_no_email += 1
|
|
continue
|
|
shifts = sorted(
|
|
shifts_by_volunteer.get(volunteer.id, []),
|
|
key=lambda s: s.starts_at,
|
|
)
|
|
minutes = sum(
|
|
(s.ends_at - s.starts_at).total_seconds() / 60 for s in shifts
|
|
)
|
|
shifts_text = "\n".join(
|
|
t(
|
|
"email.planning.shift_line",
|
|
position=positions[s.position_id].name,
|
|
start=s.starts_at.strftime("%d/%m %H:%M"),
|
|
end=s.ends_at.strftime("%d/%m %H:%M"),
|
|
)
|
|
for s in shifts
|
|
) or t("email.planning.no_shifts")
|
|
subject, body = render_planning(
|
|
subject_template=subject_template,
|
|
body_template=body_template,
|
|
first_name=volunteer.first_name,
|
|
last_name=volunteer.last_name,
|
|
shift_count=len(shifts),
|
|
hours=round(minutes / 60, 1),
|
|
shifts_text=shifts_text,
|
|
login_link=(
|
|
_login_link(session, edition, volunteer)
|
|
if wants_login_link
|
|
else ""
|
|
),
|
|
)
|
|
items.append((volunteer.email, subject, body))
|
|
recipients.append(volunteer.id)
|
|
|
|
session.commit() # persiste les éventuels MagicLinkToken émis
|
|
background.add_task(send_many, items)
|
|
return PlanningNotificationResult(
|
|
queued=len(items),
|
|
skipped_no_email=skipped_no_email,
|
|
recipients=recipients,
|
|
)
|