AdamThomasandClaude Opus 5.5
8d6781e6ce
{creneaux} devient un tableau par jour de festival (horaire, poste,
référent avec son téléphone), en HTML dans l'email et en lignes lisibles
dans la version texte — envoi réel et email de test.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
316 lines
12 KiB
Python
316 lines
12 KiB
Python
"""Envoi d'emails d'information aux bénévoles (récapitulatif de planning)."""
|
||
|
||
import datetime as dt
|
||
import logging
|
||
import uuid
|
||
from collections import defaultdict
|
||
from dataclasses import dataclass
|
||
from email.message import EmailMessage
|
||
|
||
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
|
||
from sqlmodel import Session, select
|
||
|
||
from app.api.deps import get_edition
|
||
from app.api.schemas import (
|
||
PlanningNotificationRequest,
|
||
PlanningNotificationResult,
|
||
TestEmailRequest,
|
||
TestEmailResult,
|
||
)
|
||
from app.auth.deps import require_coordinator
|
||
from app.auth.magic_link import build_magic_link_url, issue_magic_link_token
|
||
from app.config import get_settings
|
||
from app.db import get_session
|
||
from app.email import edition_logo, render_planning
|
||
from app.email.html import render_planning_table
|
||
from app.email.sender import build_message, deliver_test, send_messages
|
||
from app.i18n import t
|
||
from app.models import Account, Assignment, DayPart, Edition, Position, Shift, Volunteer
|
||
from app.validation.engine import day_boundary, festival_day
|
||
|
||
logger = logging.getLogger("app.notifications")
|
||
|
||
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))
|
||
|
||
|
||
# Repère du tableau des créneaux dans la version HTML (caractère d'usage
|
||
# privé : jamais saisi dans un modèle), remplacé par le vrai tableau.
|
||
_SHIFTS_MARKER = "\ue000creneaux\ue000"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _PlanningContext:
|
||
"""Ce qui sert à composer le récapitulatif de chaque bénévole, chargé
|
||
une fois par envoi."""
|
||
|
||
positions: dict[uuid.UUID, Position]
|
||
# Nom + téléphone du référent de chaque poste qui en a un.
|
||
referents: dict[uuid.UUID, str]
|
||
shifts_by_volunteer: dict[uuid.UUID, list[Shift]]
|
||
boundary: dt.time
|
||
|
||
|
||
def _load_context(session: Session, edition: Edition) -> _PlanningContext:
|
||
positions = {
|
||
p.id: p for p in session.exec(select(Position).where(Position.edition_id == edition.id))
|
||
}
|
||
referents: dict[uuid.UUID, str] = {}
|
||
for position_id, referent in session.exec(
|
||
select(Position.id, Volunteer)
|
||
.join(Account, Account.id == Position.lead_account_id)
|
||
.join(Volunteer, Volunteer.id == Account.volunteer_id)
|
||
.where(Position.edition_id == edition.id)
|
||
):
|
||
name = f"{referent.first_name} {referent.last_name}".strip()
|
||
referents[position_id] = f"{name} · {referent.phone}" if referent.phone else name
|
||
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)
|
||
boundary = day_boundary(session.exec(select(DayPart).where(DayPart.edition_id == edition.id)))
|
||
return _PlanningContext(positions, referents, shifts_by_volunteer, boundary)
|
||
|
||
|
||
def _day_label(day: dt.date) -> str:
|
||
"""« Samedi 11 » — même en-tête de jour que le tableau des badges."""
|
||
return t("badge.table.day_label", weekday=t(f"badge.weekday.{day.weekday()}"), day=day.day)
|
||
|
||
|
||
def _shift_sections(
|
||
shifts: list[Shift], ctx: _PlanningContext
|
||
) -> list[tuple[str, list[tuple[str, str, str]]]]:
|
||
"""Créneaux regroupés par journée de festival (une nuit reste sur le
|
||
jour où elle a commencé), comme le tableau des badges :
|
||
`[(jour, [(horaire, poste, référent)])]`."""
|
||
by_day: dict[dt.date, list[tuple[str, str, str]]] = defaultdict(list)
|
||
for s in sorted(shifts, key=lambda s: s.starts_at):
|
||
by_day[festival_day(s.starts_at, ctx.boundary)].append(
|
||
(
|
||
f"{s.starts_at:%H:%M}–{s.ends_at:%H:%M}",
|
||
ctx.positions[s.position_id].name,
|
||
ctx.referents.get(s.position_id) or t("email.planning.table.no_referent"),
|
||
)
|
||
)
|
||
return [(_day_label(day), rows) for day, rows in sorted(by_day.items())]
|
||
|
||
|
||
def _shifts_text(sections: list[tuple[str, list[tuple[str, str, str]]]]) -> str:
|
||
"""Version texte du tableau (clients mail sans HTML) : le jour, puis
|
||
une ligne par créneau."""
|
||
if not sections:
|
||
return t("email.planning.no_shifts")
|
||
return "\n\n".join(
|
||
day
|
||
+ "\n"
|
||
+ "\n".join(
|
||
t("email.planning.table.text_line", time=time, position=position, referent=referent)
|
||
for time, position, referent in rows
|
||
)
|
||
for day, rows in sections
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _PlanningEmail:
|
||
subject: str
|
||
body: str
|
||
# Corps de la version HTML : `{creneaux}` y devient un vrai tableau.
|
||
html_body: str
|
||
html_blocks: dict[str, str]
|
||
|
||
|
||
def _planning_email(
|
||
*,
|
||
ctx: _PlanningContext,
|
||
subject_template: str,
|
||
body_template: str,
|
||
first_name: str,
|
||
last_name: str,
|
||
shifts: list[Shift],
|
||
login_link: str,
|
||
) -> _PlanningEmail:
|
||
"""Récapitulatif de planning d'un bénévole — partagé par l'envoi réel
|
||
et l'email de test. `{creneaux}` : tableau par jour (horaire, poste,
|
||
référent avec son téléphone), en HTML et en texte."""
|
||
minutes = sum((s.ends_at - s.starts_at).total_seconds() / 60 for s in shifts)
|
||
sections = _shift_sections(shifts, ctx)
|
||
|
||
def render(shifts_text: str) -> tuple[str, str]:
|
||
return render_planning(
|
||
subject_template=subject_template,
|
||
body_template=body_template,
|
||
first_name=first_name,
|
||
last_name=last_name,
|
||
shift_count=len(shifts),
|
||
hours=round(minutes / 60, 1),
|
||
shifts_text=shifts_text,
|
||
login_link=login_link,
|
||
)
|
||
|
||
subject, body = render(_shifts_text(sections))
|
||
if not sections:
|
||
return _PlanningEmail(subject, body, body, {})
|
||
_, html_body = render(_SHIFTS_MARKER)
|
||
table = render_planning_table(
|
||
sections,
|
||
(
|
||
t("email.planning.table.time"),
|
||
t("email.planning.table.position"),
|
||
t("email.planning.table.referent"),
|
||
),
|
||
)
|
||
return _PlanningEmail(subject, body, html_body, {_SHIFTS_MARKER: table})
|
||
|
||
|
||
@router.post("/test", response_model=TestEmailResult)
|
||
def send_test_email(
|
||
payload: TestEmailRequest,
|
||
edition: Edition = Depends(get_edition),
|
||
_guard: Account = Depends(require_coordinator),
|
||
session: Session = Depends(get_session),
|
||
) -> TestEmailResult:
|
||
"""Envoie tout de suite à l'adresse saisie le récapitulatif de planning
|
||
**tel que les bénévoles le recevront** : le modèle en cours d'édition
|
||
(`subject`/`body`), variables remplies avec les données d'un vrai
|
||
bénévole, même mise en forme et même logo — sujet préfixé « [Test] ».
|
||
Renvoie le résultat réel du serveur SMTP, pour vérifier à la fois le
|
||
contenu et la configuration d'envoi depuis l'écran « Messages ».
|
||
`{lien_connexion}` y est remplacé par l'adresse de la page de connexion,
|
||
**sans jeton** : un vrai lien magique ouvrirait le compte du bénévole à
|
||
quiconque lit la boîte de test."""
|
||
email = payload.email.strip()
|
||
if "@" not in email or email.startswith("@") or email.endswith("@") or " " in email:
|
||
raise HTTPException(status_code=422, detail={"code": "notifications.invalid_email"})
|
||
|
||
ctx = _load_context(session, edition)
|
||
shifts_by_volunteer = ctx.shifts_by_volunteer
|
||
volunteers = list(
|
||
session.exec(
|
||
select(Volunteer)
|
||
.where(Volunteer.edition_id == edition.id)
|
||
.order_by(Volunteer.last_name, Volunteer.first_name)
|
||
)
|
||
)
|
||
example = next((v for v in volunteers if v.id == payload.volunteer_id), None) or next(
|
||
(v for v in volunteers if shifts_by_volunteer.get(v.id)),
|
||
volunteers[0] if volunteers else None,
|
||
)
|
||
composed = _planning_email(
|
||
ctx=ctx,
|
||
subject_template=payload.subject or t("email.planning.default_subject"),
|
||
body_template=payload.body or t("email.planning.default_body"),
|
||
first_name=example.first_name if example else t("email.test.example_first_name"),
|
||
last_name=example.last_name if example else t("email.test.example_last_name"),
|
||
shifts=shifts_by_volunteer.get(example.id, []) if example else [],
|
||
login_link=build_magic_link_url("").split("?")[0],
|
||
)
|
||
message = build_message(
|
||
email,
|
||
t("email.test.subject_prefix", subject=composed.subject),
|
||
composed.body,
|
||
logo=edition_logo(session, edition.id),
|
||
html_body=composed.html_body,
|
||
html_blocks=composed.html_blocks,
|
||
)
|
||
status, detail = deliver_test(message)
|
||
settings = get_settings()
|
||
return TestEmailResult(
|
||
status=status,
|
||
detail=detail,
|
||
smtp_host=f"{settings.smtp_host}:{settings.smtp_port}",
|
||
smtp_from=settings.smtp_from,
|
||
example_volunteer_id=example.id if example else None,
|
||
example_volunteer_name=(
|
||
f"{example.first_name} {example.last_name}".strip() if example else ""
|
||
),
|
||
)
|
||
|
||
|
||
@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)))
|
||
|
||
ctx = _load_context(session, edition)
|
||
logo = edition_logo(session, edition.id)
|
||
|
||
messages: list[EmailMessage] = []
|
||
recipients: list[uuid.UUID] = []
|
||
skipped_no_email = 0
|
||
for volunteer in volunteers:
|
||
if not volunteer.email:
|
||
skipped_no_email += 1
|
||
continue
|
||
composed = _planning_email(
|
||
ctx=ctx,
|
||
subject_template=subject_template,
|
||
body_template=body_template,
|
||
first_name=volunteer.first_name,
|
||
last_name=volunteer.last_name,
|
||
shifts=ctx.shifts_by_volunteer.get(volunteer.id, []),
|
||
login_link=(
|
||
_login_link(session, edition, volunteer)
|
||
if wants_login_link
|
||
else ""
|
||
),
|
||
)
|
||
messages.append(
|
||
build_message(
|
||
volunteer.email,
|
||
composed.subject,
|
||
composed.body,
|
||
logo=logo,
|
||
html_body=composed.html_body,
|
||
html_blocks=composed.html_blocks,
|
||
)
|
||
)
|
||
recipients.append(volunteer.id)
|
||
|
||
session.commit() # persiste les éventuels MagicLinkToken émis
|
||
logger.info(
|
||
"Récapitulatif de planning (édition %s) : %d emails en file, %d bénévoles sans email",
|
||
edition.id,
|
||
len(messages),
|
||
skipped_no_email,
|
||
)
|
||
background.add_task(send_messages, messages)
|
||
return PlanningNotificationResult(
|
||
queued=len(messages),
|
||
skipped_no_email=skipped_no_email,
|
||
recipients=recipients,
|
||
)
|