""" Envoi des emails liés au stock : alerte immédiate quand un matériel passe sous son seuil, et résumé périodique de l'état complet du stock (voir app/scheduler.py pour la planification de ce second email). En local, si aucun SMTP_HOST n'est configuré (voir .env.example et app/config.py), rien n'est réellement envoyé : les emails sont juste affichés dans la console. Ça permet de développer et tester tout le flux d'alerte sans avoir besoin d'un vrai serveur mail sous la main. """ import smtplib from email.message import EmailMessage from itertools import groupby from sqlmodel import Session, select from app.config import settings from app.models import DestinataireAlerte, Materiel def materiels_stock_bas(session: Session) -> list[Materiel]: """Tous les matériels actuellement sous leur seuil d'alerte, triés par catégorie puis nom. Partagé entre l'alerte immédiate (qui y ajoute le matériel qui vient de déclencher l'alerte, en contexte) et le résumé périodique (qui l'utilise comme section "Stock bas").""" tous = session.exec(select(Materiel)).all() bas = [m for m in tous if m.seuil_alerte is not None and m.quantite < m.seuil_alerte] bas.sort(key=lambda m: (m.categorie.chemin().lower(), m.nom.lower())) return bas def verifier_et_alerter(materiel: Materiel, session: Session) -> None: """À appeler après toute modification de la quantité d'un matériel. Envoie un email si le stock de ce matériel vient de passer sous son seuil d'alerte. Le corps contient le matériel concerné, puis en dessous un résumé de tous les articles actuellement en stock bas (pas seulement celui-ci) : ça évite d'avoir à recouper plusieurs emails pour savoir où en est le stock dans son ensemble. """ if materiel.seuil_alerte is None: return if materiel.quantite >= materiel.seuil_alerte: return destinataires = session.exec(select(DestinataireAlerte)).all() if not destinataires: return sujet = f"[Stock IT] Stock bas : {materiel.nom}" corps = ( f"Le stock de « {materiel.nom} » est passé sous le seuil d'alerte.\n\n" f"Quantité actuelle : {materiel.quantite}\n" f"Seuil d'alerte : {materiel.seuil_alerte}\n\n" "--- Tous les articles actuellement en stock bas ---\n" f"{_bloc_stock_bas(session)}\n" ) _envoyer_email(sujet, corps, [d.email for d in destinataires]) def envoyer_resume_stock(session: Session) -> None: """Email périodique (voir app/scheduler.py) : l'état des stocks bas en tête du message, puis le reste du stock en dessous, regroupé par catégorie. Contrairement à verifier_et_alerter(), n'est pas déclenché par une action de l'utilisateur mais par la planification cron.""" destinataires = session.exec(select(DestinataireAlerte)).all() if not destinataires: print("Résumé de stock non envoyé : aucun destinataire configuré.") return bas = materiels_stock_bas(session) ids_bas = {m.id for m in bas} sujet = f"[Stock IT] Résumé du stock — {len(bas)} article(s) en stock bas" corps = ( "=== Stock bas ===\n" f"{_bloc_stock_bas(session)}\n\n" "=== Reste du stock ===\n" f"{_bloc_reste_du_stock(session, ids_bas)}\n" ) _envoyer_email(sujet, corps, [d.email for d in destinataires]) def _bloc_stock_bas(session: Session) -> str: bas = materiels_stock_bas(session) if not bas: return "Aucun article en dessous de son seuil d'alerte." return "\n".join( f"- {m.nom} ({m.categorie.chemin()}) : {m.quantite} en stock (seuil {m.seuil_alerte})" for m in bas ) def _bloc_reste_du_stock(session: Session, ids_a_exclure: set[int]) -> str: """Tous les matériels qui ne sont pas déjà listés dans la section "Stock bas", regroupés par catégorie pour rester lisible sur un stock avec beaucoup d'articles.""" tous = session.exec(select(Materiel)).all() reste = [m for m in tous if m.id not in ids_a_exclure] if not reste: return "Aucun autre article." reste.sort(key=lambda m: (m.categorie.chemin().lower(), m.nom.lower())) blocs = [] for chemin, groupe in groupby(reste, key=lambda m: m.categorie.chemin()): lignes = "\n".join(f" - {m.nom} : {m.quantite} en stock" for m in groupe) blocs.append(f"{chemin}\n{lignes}") return "\n\n".join(blocs) def _envoyer_email(sujet: str, corps: str, destinataires: list[str]) -> None: if not settings.smtp_configured: # Pas de SMTP configuré (typiquement en dev local) : on affiche # l'email dans la console plutôt que de l'envoyer, pour pouvoir # tester le déclenchement des alertes sans serveur mail réel. print( "--- [EMAIL SIMULÉ] ---\n" f"À: {', '.join(destinataires)}\n" f"Sujet: {sujet}\n\n" f"{corps}" "----------------------" ) return message = EmailMessage() message["Subject"] = sujet message["From"] = settings.smtp_from message["To"] = ", ".join(destinataires) message.set_content(corps) with smtplib.SMTP(settings.smtp_host, settings.smtp_port) as smtp: if settings.smtp_user and settings.smtp_password: smtp.starttls() smtp.login(settings.smtp_user, settings.smtp_password) smtp.send_message(message)