{hello}
" "Ton professeur t'a inscrit·e à UQO-Chat, le tuteur IA des cours " f"IMM1003 et IMM1033 ({html.escape(term)}). Il explique les notes de cours, corrige tes calculs " "d'évaluation immobilière et produit des classeurs Excel.
" f"Pour activer ton compte {html.escape(to)}, choisis un mot de passe. " "Tu l'utiliseras ensuite avec ton courriel à chaque connexion.
") footer = (f"Ce lien est valide {ttl_days} jours et ne sert qu'une fois. " "S'il a expiré, clique sur « Première connexion ou mot de passe oublié » sur la page de connexion.
") text = (f"{hello.replace(',', ',')}\n\nTon professeur t'a inscrit·e à UQO-Chat, le tuteur IA des cours IMM1003 et " f"IMM1033 ({term}).\n\nPour activer ton compte {to}, choisis un mot de passe ici (valide {ttl_days} jours) :\n\n" f"{link}\n\nTu l'utiliseras ensuite avec ton courriel à chaque connexion. Si le lien a expiré, utilise " "« Première connexion ou mot de passe oublié » sur la page de connexion.\n\n— UQO-Chat") return Mail(to, title, text, _layout(title, intro, "Choisir mon mot de passe", link, footer)) def reset_mail(to: str, link: str, ttl_minutes: int, first_time: bool) -> Mail: if first_time: title = "Choisis ton mot de passe UQO-Chat" intro = ("Bonjour,
" f"Tu as demandé à activer ton compte UQO-Chat {html.escape(to)}. " "Choisis un mot de passe pour te connecter.
") cta = "Choisir mon mot de passe" else: title = "Réinitialisation de ton mot de passe UQO-Chat" intro = ("Bonjour,
" f"Une réinitialisation du mot de passe a été demandée pour " f"{html.escape(to)}. Clique ci-dessous pour en choisir un nouveau.
") cta = "Choisir un nouveau mot de passe" footer = (f"Ce lien est valide {ttl_minutes} minutes et ne sert qu'une fois. " "Si tu n'as rien demandé, ignore ce courriel : ton mot de passe actuel reste inchangé.
") text = (f"Bonjour,\n\n{'Tu as demandé à activer ton compte' if first_time else 'Une réinitialisation du mot de passe a été demandée pour'} " f"UQO-Chat {to}.\n\nChoisis ton mot de passe ici (valide {ttl_minutes} minutes) :\n\n{link}\n\n" "Si tu n'as rien demandé, ignore ce courriel.\n\n— UQO-Chat") return Mail(to, title, text, _layout(title, intro, cta, link, footer)) def login_link_mail(to: str, link: str) -> Mail: title = "Connexion à UQO-Chat" intro = f"Voici ton lien de connexion à UQO-Chat pour {html.escape(to)}.
" footer = "Si tu n'as pas demandé ce lien, ignore ce courriel.
" text = f"Bonjour,\n\nVoici ton lien de connexion à UQO-Chat :\n\n{link}\n\nSi tu n'as pas demandé ce lien, ignore ce courriel.\n\n— UQO-Chat" return Mail(to, title, text, _layout(title, intro, "Me connecter", link, footer)) # ------------------------------------------------------------------ transport def _resend_payload(settings: Settings, m: Mail) -> dict: payload = {"from": settings.MAIL_FROM, "to": [m.to], "subject": m.subject, "text": m.text, "html": m.html} if settings.MAIL_REPLY_TO: payload["reply_to"] = settings.MAIL_REPLY_TO return payload def _resend_headers(settings: Settings) -> dict[str, str]: return {"Authorization": f"Bearer {settings.RESEND_API_KEY.get_secret_value()}", "Content-Type": "application/json"} async def _send_smtp(settings: Settings, m: Mail) -> bool: import aiosmtplib msg = EmailMessage() msg["From"] = settings.SMTP_FROM msg["To"] = m.to msg["Subject"] = m.subject msg.set_content(m.text) msg.add_alternative(m.html, subtype="html") await aiosmtplib.send( msg, hostname=settings.SMTP_HOST, port=settings.SMTP_PORT, username=settings.SMTP_USER or None, password=settings.SMTP_PASSWORD.get_secret_value() or None, start_tls=settings.SMTP_PORT == 587) return True async def send(settings: Settings, m: Mail) -> bool: """Send one message. Returns False (and logs) when no transport is configured or it fails.""" if settings.resend_enabled: try: async with httpx.AsyncClient(timeout=20) as client: r = await client.post(f"{settings.RESEND_BASE_URL}/emails", headers=_resend_headers(settings), json=_resend_payload(settings, m)) if r.status_code < 300: log.info("mail_sent", provider="resend", to_domain=m.to.split("@")[-1]) return True log.warning("mail_failed", provider="resend", status=r.status_code, body=r.text[:300]) except httpx.HTTPError as exc: log.warning("mail_failed", provider="resend", error=str(exc)) return False if settings.smtp_enabled: try: return await _send_smtp(settings, m) except Exception as exc: # noqa: BLE001 log.warning("mail_failed", provider="smtp", error=str(exc)) return False log.info("mail_not_sent_no_transport", to_domain=m.to.split("@")[-1]) return False async def send_batch(settings: Settings, mails: list[Mail]) -> dict[str, bool]: """Send many messages; Resend batch endpoint (100/call), otherwise one by one.""" result: dict[str, bool] = {} if not mails: return result if settings.resend_enabled: async with httpx.AsyncClient(timeout=30) as client: for i in range(0, len(mails), 100): chunk = mails[i:i + 100] try: r = await client.post(f"{settings.RESEND_BASE_URL}/emails/batch", headers=_resend_headers(settings), json=[_resend_payload(settings, m) for m in chunk]) ok = r.status_code < 300 if not ok: log.warning("mail_batch_failed", status=r.status_code, body=r.text[:300]) except httpx.HTTPError as exc: ok = False log.warning("mail_batch_failed", error=str(exc)) if ok: for m in chunk: result[m.to] = True else: # degrade to unit sends so one bad address does not block the cohort for m in chunk: result[m.to] = await send(settings, m) log.info("mail_batch_sent", n=sum(result.values()), total=len(mails)) return result for m in mails: result[m.to] = await send(settings, m) return result # ------------------------------------------------------------------ convenience async def send_magic_link(settings: Settings, to: str, link: str) -> bool: return await send(settings, login_link_mail(to, link))