maxsoch/IT-Stock
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Résumé de stock dans les alertes + email récapitulatif planifié

Browse source 
L'alerte immédiate de stock bas inclut maintenant un résumé de tous les
articles actuellement sous leur seuil, pas seulement celui qui l'a
déclenchée. Ajout d'un second email, planifié via une expression cron
(RESUME_STOCK_CRON, APScheduler intégré à l'app — pas de cron système à
configurer) : stock bas en tête, puis le reste du stock groupé par
catégorie. scripts/envoyer_resume_stock.py permet de le déclencher
manuellement, ou de brancher l'envoi sur un cron système à la place.
This commit is contained in:
maxsoch 2026-07-14 19:28:39 +02:00
parent 5088182f20
commit a788bf4392
10 changed files with 295 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.env.example
View file

@ -10,3 +10,9 @@ SMTP_PORT=25
SMTP_USER=
SMTP_PASSWORD=
SMTP_FROM=stock-it@clinique.local
# Expression cron (minute heure jour mois jour_semaine) pour l'email
# récapitulatif périodique de l'état du stock. Par défaut : tous les
# jours à 8h. Le planificateur tourne dans le process de l'app (voir
# app/scheduler.py) — pas de cron système à configurer.
RESUME_STOCK_CRON=0 8 * * *

33
CLAUDE.md
View file

@ -64,12 +64,13 @@ l'absence de SMTP_HOST). Ne pas se limiter au commentaire minimal habituel.
```
app/
main.py point d'entrée FastAPI, montage des routers
config.py lecture des variables d'environnement (SMTP...)
main.py point d'entrée FastAPI, montage des routers, démarrage du planificateur
config.py lecture des variables d'environnement (SMTP, cron du résumé...)
database.py connexion SQLite + dependency get_session()
models.py modèles SQLModel : Categorie (arborescence auto-référencée), Materiel, DestinataireAlerte
categories_arbre.py parcours de l'arborescence (racines, aplatissement pour un <select> indenté)
email_alerts.py vérifie le seuil d'alerte et envoie/simule l'email
email_alerts.py alerte immédiate + résumé périodique de stock, envoie/simule l'email
scheduler.py planification (APScheduler) de l'email récapitulatif périodique
templates_engine.py instance Jinja2Templates partagée par tous les routers
routers/ une route FastAPI par ressource (categories, materiels, scan, destinataires)
templates/ pages HTML (Jinja2, formulaires HTML classiques)
@ -109,10 +110,26 @@ séparées pour ne jamais laisser d'ambiguïté sur ce qu'un nombre représente
« sens » (ajouter/retirer), `/scan/{id}/definir` fixe la quantité absolue
(utile après un inventaire physique).
## Emails d'alerte
## Emails de stock
Seuil configurable par matériel (`Materiel.seuil_alerte`), liste globale de
destinataires (`DestinataireAlerte`, gérée sur `/destinataires`). La logique
d'envoi est dans `app/email_alerts.py` : si `SMTP_HOST` n'est pas défini, rien
n'est envoyé, l'email est juste affiché dans la console du serveur — pratique
pour tester le déclenchement des alertes sans serveur mail réel.
destinataires (`DestinataireAlerte`, gérée sur `/destinataires`). Si
`SMTP_HOST` n'est pas défini, rien n'est envoyé : l'email est juste affiché
dans la console du serveur — pratique pour tester sans serveur mail réel.
Deux emails distincts, tous les deux dans `app/email_alerts.py` :
- **Alerte immédiate** (`verifier_et_alerter`) : déclenchée après toute
modification de quantité qui fait passer un matériel sous son seuil.
Contient ce matériel, puis un résumé de *tous* les articles actuellement
en stock bas (`materiels_stock_bas()`), pas seulement celui qui a
déclenché l'alerte.
- **Résumé périodique** (`envoyer_resume_stock`) : stock bas en premier,
puis le reste du stock groupé par catégorie. Planifié par
`app/scheduler.py` (APScheduler, tourne dans le process de l'app, pas de
cron système à configurer) via l'expression cron `RESUME_STOCK_CRON`
(`.env`, défaut `0 8 * * *` = tous les jours à 8h). Limite : ne se
déclenche que si le serveur tourne au moment prévu, pas de rattrapage.
`scripts/envoyer_resume_stock.py` permet de le déclencher manuellement
(utile pour tester le contenu, ou pour brancher l'envoi sur un cron
système / Planificateur de tâches Windows à la place si cette limite
devient un problème).

39
README.md
View file

@ -45,10 +45,11 @@ source .venv/bin/activate
ruff check .
```
## Configuration des emails d'alerte
## Configuration des emails
Les alertes de stock bas sont envoyées par email via le relais SMTP interne
de la clinique, configuré par variables d'environnement (voir `.env.example`) :
Les emails de stock (alerte immédiate + résumé périodique) sont envoyés via
le relais SMTP interne de la clinique, configuré par variables
d'environnement (voir `.env.example`) :
```
SMTP_HOST=...
@ -56,12 +57,27 @@ SMTP_PORT=25
SMTP_USER=...
SMTP_PASSWORD=...
SMTP_FROM=stock-it@clinique.local
RESUME_STOCK_CRON=0 8 * * *
```
Si `SMTP_HOST` n'est pas renseigné (cas du dev local), les alertes ne sont
pas réellement envoyées : elles sont affichées dans la console du serveur
Si `SMTP_HOST` n'est pas renseigné (cas du dev local), rien n'est réellement
envoyé : les emails sont affichés dans la console du serveur
(`--- [EMAIL SIMULÉ] ---`). Ça permet de développer et tester tout le flux
d'alerte sans avoir de vrai serveur mail sous la main.
sans avoir de vrai serveur mail sous la main.
Deux emails distincts :
- **Alerte immédiate** : envoyée dès qu'un matériel passe sous son seuil
d'alerte (`Materiel.seuil_alerte`). Contient ce matériel, puis un résumé
de tous les articles actuellement en stock bas.
- **Résumé périodique** : stock bas en premier, puis le reste du stock
groupé par catégorie. Planifié via `RESUME_STOCK_CRON` (expression cron
classique, ex. `0 8 * * *` = tous les jours à 8h) — le planificateur
tourne dans le process de l'app (APScheduler), rien à configurer côté
système. Pour le déclencher manuellement (test, ou pour le brancher sur
un cron système / Planificateur de tâches Windows à la place) :
`python scripts/envoyer_resume_stock.py`.
Les destinataires (plusieurs possibles) se gèrent sur `/destinataires`.
## Douchette code-barres
@ -74,15 +90,20 @@ soumet automatiquement le formulaire.
```
app/
main.py point d'entrée FastAPI
main.py point d'entrée FastAPI, démarrage du planificateur
config.py lecture des variables d'environnement
database.py connexion SQLite
models.py modèles SQLModel (Categorie, SousCategorie, Materiel, DestinataireAlerte)
email_alerts.py logique d'envoi/simulation des alertes email
models.py modèles SQLModel (Categorie en arborescence, Materiel, DestinataireAlerte)
categories_arbre.py parcours de l'arborescence de catégories
email_alerts.py alerte immédiate + résumé périodique de stock
scheduler.py planification (APScheduler) du résumé périodique
templates_engine.py instance Jinja2Templates partagée
routers/ une route FastAPI par ressource (categories, materiels, scan, destinataires)
templates/ pages HTML (Jinja2)
static/ CSS
scripts/
seed.py peuple la base avec des données de démo
envoyer_resume_stock.py déclenche manuellement l'email récapitulatif
tests/ tests pytest (client de test FastAPI + base SQLite en mémoire)
data/ base SQLite locale (ignorée par git)
```

5
app/config.py
View file

@ -28,5 +28,10 @@ class Settings:
# affichées dans la console au lieu d'être envoyées. Voir email_alerts.py.
smtp_configured: bool = bool(smtp_host)
# Expression cron classique (minute heure jour mois jour_semaine) pour
# l'email récapitulatif périodique de l'état du stock. Par défaut :
# tous les jours à 8h. Voir app/scheduler.py.
resume_stock_cron: str = os.getenv("RESUME_STOCK_CRON", "0 8 * * *")
settings = Settings()

83
app/email_alerts.py
View file

@ -1,14 +1,17 @@
"""
Envoi des emails d'alerte de stock bas.
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), les alertes ne sont pas réellement envoyées : elles sont
juste affichées 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.
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
@ -16,11 +19,25 @@ 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 à tous les destinataires configurés si le stock vient
de passer sous le seuil d'alerte défini sur ce 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
@ -35,11 +52,63 @@ def verifier_et_alerter(materiel: Materiel, session: Session) -> None:
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"
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

7
app/main.py
View file

@ -11,13 +11,18 @@ from fastapi.staticfiles import StaticFiles
from app.database import init_db
from app.routers import categories, destinataires, materiels, scan
from app.scheduler import arreter_planification, demarrer_planification
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Crée les tables SQLite au démarrage si elles n'existent pas encore."""
"""Crée les tables SQLite au démarrage si elles n'existent pas encore,
puis démarre la planification de l'email récapitulatif de stock
(voir app/scheduler.py) — arrêtée proprement à l'extinction."""
init_db()
demarrer_planification()
yield
arreter_planification()
app = FastAPI(title="Gestion de stock IT", lifespan=lifespan)

50
app/scheduler.py Normal file
View file

@ -0,0 +1,50 @@
"""
Planification de l'email récapitulatif de stock (voir
email_alerts.envoyer_resume_stock).
Choix : un planificateur qui tourne dans le process FastAPI lui-même
(APScheduler), configuré par une expression cron classique (RESUME_STOCK_CRON
dans .env, ex: "0 8 * * *" = tous les jours à 8h). Rien à configurer côté
système d'exploitation du serveur — tout reste dans ce dossier, comme le
reste du projet.
Limite à connaître : la tâche ne se déclenche que si le process est en
cours d'exécution au moment prévu, sans rattrapage si le serveur était
arrêté à ce moment-là. Pour un outil interne redémarré rarement, c'est un
compromis raisonnable face à la complexité d'un vrai cron système. Si ça
devient un problème, scripts/envoyer_resume_stock.py peut être branché
sur un cron/Planificateur de tâches Windows à la place.
"""
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from sqlmodel import Session
from app.config import settings
from app.database import engine
from app.email_alerts import envoyer_resume_stock
_scheduler = BackgroundScheduler()
def _tache_resume_stock() -> None:
"""Ouvre sa propre session DB : cette fonction est appelée par le
thread du planificateur, en dehors du cycle requête/réponse FastAPI où
la session habituelle (get_session) n'existe pas."""
with Session(engine) as session:
envoyer_resume_stock(session)
def demarrer_planification() -> None:
"""Démarre la tâche planifiée. Appelé une fois au démarrage de l'app
(voir lifespan dans main.py)."""
trigger = CronTrigger.from_crontab(settings.resume_stock_cron)
_scheduler.add_job(_tache_resume_stock, trigger, id="resume_stock", replace_existing=True)
if not _scheduler.running:
_scheduler.start()
def arreter_planification() -> None:
"""Arrête proprement le planificateur (appelé à l'extinction de l'app)."""
if _scheduler.running:
_scheduler.shutdown(wait=False)

1
requirements.txt
View file

@ -4,3 +4,4 @@ sqlmodel>=0.0.22
jinja2>=3.1
python-dotenv>=1.0
python-multipart>=0.0.12
apscheduler>=3.10,<4.0

25
scripts/envoyer_resume_stock.py Normal file
View file

@ -0,0 +1,25 @@
"""
Déclenche manuellement l'envoi de l'email récapitulatif de stock (stock
bas en premier, puis le reste du stock groupé par catégorie), sans
attendre la prochaine exécution planifiée. Utile pour tester le contenu
de l'email localement, ou pour brancher l'envoi sur un cron système /
Planificateur de tâches Windows au lieu du planificateur intégré à l'app
(voir app/scheduler.py pour le compromis entre les deux approches).
Usage : python scripts/envoyer_resume_stock.py
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from sqlmodel import Session # noqa: E402
from app.database import engine, init_db # noqa: E402
from app.email_alerts import envoyer_resume_stock # noqa: E402
if __name__ == "__main__":
init_db()
with Session(engine) as session:
envoyer_resume_stock(session)

71
tests/test_email_alerts.py Normal file
View file

@ -0,0 +1,71 @@
"""
Tests du contenu des emails : le résumé des articles en stock bas ajouté à
l'alerte immédiate, et l'email récapitulatif périodique (stock bas puis
reste du stock groupé par catégorie).
"""
from sqlmodel import Session
from app.email_alerts import envoyer_resume_stock, verifier_et_alerter
from app.models import Categorie, DestinataireAlerte, Materiel
def _categorie(session: Session, nom: str) -> Categorie:
categorie = Categorie(nom=nom)
session.add(categorie)
session.commit()
session.refresh(categorie)
return categorie
def test_alerte_immediate_inclut_le_resume_des_stocks_bas(session: Session, capsys):
categorie = _categorie(session, "Ordinateurs")
# Un autre matériel déjà en stock bas, pour vérifier qu'il apparaît
# dans le résumé même s'il n'est pas celui qui déclenche l'alerte.
session.add(Materiel(nom="Souris USB", categorie_id=categorie.id, quantite=1, seuil_alerte=5))
materiel = Materiel(nom="PC Dell XPS 13", categorie_id=categorie.id, quantite=2, seuil_alerte=5)
session.add(materiel)
session.add(DestinataireAlerte(email="it-dept@clinique.local"))
session.commit()
session.refresh(materiel)
verifier_et_alerter(materiel, session)
sortie = capsys.readouterr().out
assert "Stock bas : PC Dell XPS 13" in sortie
assert "Tous les articles actuellement en stock bas" in sortie
assert "Souris USB" in sortie
assert "PC Dell XPS 13" in sortie
def test_resume_stock_separe_stock_bas_et_reste_du_stock(session: Session, capsys):
categorie = _categorie(session, "Réseau")
session.add(
Materiel(nom="Câble RJ45 2m", categorie_id=categorie.id, quantite=1, seuil_alerte=10)
)
session.add(Materiel(nom="Switch 8 ports", categorie_id=categorie.id, quantite=6))
session.add(DestinataireAlerte(email="it-dept@clinique.local"))
session.commit()
envoyer_resume_stock(session)
sortie = capsys.readouterr().out
assert "=== Stock bas ===" in sortie
assert "=== Reste du stock ===" in sortie
position_bas = sortie.index("=== Stock bas ===")
position_reste = sortie.index("=== Reste du stock ===")
position_cable = sortie.index("Câble RJ45 2m")
position_switch = sortie.index("Switch 8 ports")
# Le câble (en stock bas) doit apparaître dans la première section,
# le switch (pas en stock bas) dans la seconde.
assert position_bas < position_cable < position_reste
assert position_reste < position_switch
def test_resume_stock_sans_destinataire_ne_leve_pas_derreur(session: Session, capsys):
_categorie(session, "Réseau")
envoyer_resume_stock(session)
sortie = capsys.readouterr().out
assert "aucun destinataire" in sortie.lower()