AdamThomasandClaude Opus 5.5
440466ccf9
L'email de test reprend le modèle en cours d'édition, balises remplies avec le planning d'un bénévole au choix (premier avec un créneau par défaut), option lien de connexion sans vrai jeton, sujet « [Test] … » — même fonction de rendu que l'envoi réel. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
223 lines
8.3 KiB
Python
223 lines
8.3 KiB
Python
"""Envoi d'emails d'information aux bénévoles (récapitulatif de planning)."""
|
|
|
|
import logging
|
|
import uuid
|
|
from collections import defaultdict
|
|
|
|
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, send_many
|
|
from app.email.sender import build_message, deliver_test
|
|
from app.i18n import t
|
|
from app.models import Account, Assignment, Edition, Position, Shift, Volunteer
|
|
|
|
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))
|
|
|
|
|
|
def _planning_email(
|
|
*,
|
|
subject_template: str,
|
|
body_template: str,
|
|
first_name: str,
|
|
last_name: str,
|
|
shifts: list[Shift],
|
|
positions: dict[uuid.UUID, Position],
|
|
login_link: str,
|
|
) -> tuple[str, str]:
|
|
"""Sujet et corps du récapitulatif de planning d'un bénévole — partagé
|
|
par l'envoi réel et l'email de test."""
|
|
shifts = sorted(shifts, 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")
|
|
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,
|
|
)
|
|
|
|
|
|
@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"})
|
|
|
|
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)
|
|
|
|
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,
|
|
)
|
|
subject, body = _planning_email(
|
|
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 [],
|
|
positions=positions,
|
|
login_link=build_magic_link_url("").split("?")[0],
|
|
)
|
|
message = build_message(
|
|
email,
|
|
t("email.test.subject_prefix", subject=subject),
|
|
body,
|
|
logo=edition_logo(session, edition.id),
|
|
)
|
|
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)))
|
|
|
|
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
|
|
subject, body = _planning_email(
|
|
subject_template=subject_template,
|
|
body_template=body_template,
|
|
first_name=volunteer.first_name,
|
|
last_name=volunteer.last_name,
|
|
shifts=shifts_by_volunteer.get(volunteer.id, []),
|
|
positions=positions,
|
|
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
|
|
logger.info(
|
|
"Récapitulatif de planning (édition %s) : %d emails en file, %d bénévoles sans email",
|
|
edition.id,
|
|
len(items),
|
|
skipped_no_email,
|
|
)
|
|
background.add_task(send_many, items, logo=edition_logo(session, edition.id))
|
|
return PlanningNotificationResult(
|
|
queued=len(items),
|
|
skipped_no_email=skipped_no_email,
|
|
recipients=recipients,
|
|
)
|