Planification du résumé de stock modifiable depuis l'interface

L'horaire de l'email récapitulatif se configure maintenant sur
/destinataires (raccourcis courants + expression cron personnalisée) au
lieu de .env, avec reprogrammation immédiate sans redémarrage. Stocké en
base (ParametrePlanification) plutôt que dans un fichier de config.
Défaut changé pour tous les lundis à 9h.

Les exemples et le code n'utilisent que des noms de jours (mon, tue, ...)
dans les expressions cron : vérifié empiriquement qu'APScheduler ne suit
pas la convention Unix pour les jours numériques (0=lundi chez lui, pas
0=dimanche), ce qui rendrait "0 9 * * 1" trompeur.
This commit is contained in:
maxsoch 2026-07-14 19:37:36 +02:00
parent a788bf4392
commit 11da706573
10 changed files with 302 additions and 37 deletions

View file

@ -11,8 +11,6 @@ SMTP_USER=
SMTP_PASSWORD= SMTP_PASSWORD=
SMTP_FROM=stock-it@clinique.local SMTP_FROM=stock-it@clinique.local
# Expression cron (minute heure jour mois jour_semaine) pour l'email # La planification de l'email récapitulatif de stock (tous les lundis à 9h
# récapitulatif périodique de l'état du stock. Par défaut : tous les # par défaut) se configure depuis l'interface, sur /destinataires — pas
# jours à 8h. Le planificateur tourne dans le process de l'app (voir # besoin de variable d'environnement pour ça. Voir app/scheduler.py.
# app/scheduler.py) — pas de cron système à configurer.
RESUME_STOCK_CRON=0 8 * * *

View file

@ -126,10 +126,21 @@ Deux emails distincts, tous les deux dans `app/email_alerts.py` :
- **Résumé périodique** (`envoyer_resume_stock`) : stock bas en premier, - **Résumé périodique** (`envoyer_resume_stock`) : stock bas en premier,
puis le reste du stock groupé par catégorie. Planifié par puis le reste du stock groupé par catégorie. Planifié par
`app/scheduler.py` (APScheduler, tourne dans le process de l'app, pas de `app/scheduler.py` (APScheduler, tourne dans le process de l'app, pas de
cron système à configurer) via l'expression cron `RESUME_STOCK_CRON` cron système à configurer), avec l'expression cron stockée en base
(`.env`, défaut `0 8 * * *` = tous les jours à 8h). Limite : ne se (`ParametrePlanification`, une seule ligne) et modifiable depuis
déclenche que si le serveur tourne au moment prévu, pas de rattrapage. l'interface sur `/destinataires` — pas de fichier à éditer ni de
`scripts/envoyer_resume_stock.py` permet de le déclencher manuellement redémarrage nécessaire. Défaut : `0 9 * * mon` (tous les lundis à 9h).
(utile pour tester le contenu, ou pour brancher l'envoi sur un cron Limite : ne se déclenche que si le serveur tourne au moment prévu, pas
système / Planificateur de tâches Windows à la place si cette limite de rattrapage. `scripts/envoyer_resume_stock.py` permet de le déclencher
devient un problème). 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).
**Piège APScheduler** : le champ "jour de la semaine" d'une expression
cron numérique NE SUIT PAS la convention Unix (0=dimanche). Chez
APScheduler, 0=lundi (convention `datetime.weekday()`), donc `0 9 * * 1`
tombe un **mardi**, pas un lundi. Pour éviter toute confusion, le code et
les exemples proposés à l'utilisateur (`EXEMPLES_CRON`) n'utilisent que
des noms de jours (`mon`, `tue`, ...), jamais de chiffres. Vérifié
empiriquement avec la version d'APScheduler installée (voir le
commentaire en tête de `app/scheduler.py`).

View file

@ -57,7 +57,6 @@ SMTP_PORT=25
SMTP_USER=... SMTP_USER=...
SMTP_PASSWORD=... SMTP_PASSWORD=...
SMTP_FROM=stock-it@clinique.local SMTP_FROM=stock-it@clinique.local
RESUME_STOCK_CRON=0 8 * * *
``` ```
Si `SMTP_HOST` n'est pas renseigné (cas du dev local), rien n'est réellement Si `SMTP_HOST` n'est pas renseigné (cas du dev local), rien n'est réellement
@ -70,14 +69,16 @@ Deux emails distincts :
d'alerte (`Materiel.seuil_alerte`). Contient ce matériel, puis un résumé d'alerte (`Materiel.seuil_alerte`). Contient ce matériel, puis un résumé
de tous les articles actuellement en stock bas. de tous les articles actuellement en stock bas.
- **Résumé périodique** : stock bas en premier, puis le reste du stock - **Résumé périodique** : stock bas en premier, puis le reste du stock
groupé par catégorie. Planifié via `RESUME_STOCK_CRON` (expression cron groupé par catégorie. L'horaire se configure **depuis l'interface**, sur
classique, ex. `0 8 * * *` = tous les jours à 8h) — le planificateur `/destinataires` (raccourcis courants + expression cron personnalisée) —
tourne dans le process de l'app (APScheduler), rien à configurer côté pas de fichier à éditer, le changement s'applique immédiatement sans
système. Pour le déclencher manuellement (test, ou pour le brancher sur redémarrer le serveur. Défaut : tous les lundis à 9h. Le planificateur
un cron système / Planificateur de tâches Windows à la place) : (APScheduler) tourne dans le process de l'app, rien à configurer côté
`python scripts/envoyer_resume_stock.py`. système d'exploitation. Pour déclencher l'envoi 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`. Les destinataires (plusieurs possibles) se gèrent aussi sur `/destinataires`.
## Douchette code-barres ## Douchette code-barres

View file

@ -28,10 +28,5 @@ class Settings:
# affichées dans la console au lieu d'être envoyées. Voir email_alerts.py. # affichées dans la console au lieu d'être envoyées. Voir email_alerts.py.
smtp_configured: bool = bool(smtp_host) 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() settings = Settings()

View file

@ -80,3 +80,16 @@ class DestinataireAlerte(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True) id: Optional[int] = Field(default=None, primary_key=True)
email: str = Field(unique=True) email: str = Field(unique=True)
class ParametrePlanification(SQLModel, table=True):
"""Réglage de la planification de l'email récapitulatif de stock.
Une seule ligne (id=1, upsert), modifiable depuis l'interface
(/destinataires) sans éditer .env ni redémarrer le serveur voir
app/scheduler.py qui lit/écrit cette table et reprogramme la tâche en
direct à chaque changement.
"""
id: Optional[int] = Field(default=None, primary_key=True)
expression_cron: str = Field(default="0 9 * * mon")

View file

@ -1,4 +1,7 @@
"""Routes pour gérer la liste des emails qui reçoivent les alertes de stock bas.""" """
Routes pour gérer la liste des emails qui reçoivent les alertes de stock
bas, et la planification de l'email récapitulatif périodique.
"""
from fastapi import APIRouter, Depends, Form from fastapi import APIRouter, Depends, Form
from fastapi.requests import Request from fastapi.requests import Request
@ -7,20 +10,41 @@ from sqlmodel import Session, select
from app.database import get_session from app.database import get_session
from app.models import DestinataireAlerte from app.models import DestinataireAlerte
from app.scheduler import (
EXEMPLES_CRON,
definir_expression_cron,
obtenir_expression_cron,
prochaine_execution,
valider_expression_cron,
)
from app.templates_engine import templates from app.templates_engine import templates
router = APIRouter() router = APIRouter()
def _contexte_planification(session: Session, **kwargs) -> dict:
contexte = {
"expression_cron": obtenir_expression_cron(session),
"exemples_cron": EXEMPLES_CRON,
"prochaine_execution": prochaine_execution(),
"erreur_cron": None,
}
contexte.update(kwargs)
return contexte
@router.get("/destinataires") @router.get("/destinataires")
def lister_destinataires(request: Request, q: str = "", session: Session = Depends(get_session)): def lister_destinataires(request: Request, q: str = "", session: Session = Depends(get_session)):
"""Affiche la liste des emails configurés pour recevoir les alertes, """Affiche la liste des emails configurés pour recevoir les alertes,
filtrée par ?q=... si fourni (même comportement que /materiels, pour filtrée par ?q=... si fourni (même comportement que /materiels, pour
rester cohérent si la liste grandit).""" rester cohérent si la liste grandit), ainsi que la planification de
l'email récapitulatif périodique."""
tous = session.exec(select(DestinataireAlerte)).all() tous = session.exec(select(DestinataireAlerte)).all()
destinataires = [d for d in tous if q.lower() in d.email.lower()] if q else tous destinataires = [d for d in tous if q.lower() in d.email.lower()] if q else tous
return templates.TemplateResponse( return templates.TemplateResponse(
request, "destinataires.html", {"destinataires": destinataires, "q": q} request,
"destinataires.html",
{"destinataires": destinataires, "q": q, **_contexte_planification(session)},
) )
@ -40,3 +64,40 @@ def supprimer_destinataire(destinataire_id: int, session: Session = Depends(get_
session.delete(destinataire) session.delete(destinataire)
session.commit() session.commit()
return RedirectResponse(url="/destinataires", status_code=303) return RedirectResponse(url="/destinataires", status_code=303)
@router.post("/destinataires/planification")
def modifier_planification(
request: Request, expression_cron: str = Form(...), session: Session = Depends(get_session)
):
"""Change l'horaire de l'email récapitulatif périodique et reprogramme
la tâche immédiatement (pas besoin de redémarrer le serveur).
Pas de redirection en cas d'expression cron invalide : on réaffiche la
page avec un message d'erreur et la valeur saisie, pour ne pas la
perdre (contrairement au reste du formulaire, elle n'est pas
enregistrée tant qu'elle n'est pas valide).
"""
expression_cron = expression_cron.strip()
if not valider_expression_cron(expression_cron):
tous = session.exec(select(DestinataireAlerte)).all()
return templates.TemplateResponse(
request,
"destinataires.html",
{
"destinataires": tous,
"q": "",
**_contexte_planification(
session,
expression_cron=expression_cron,
erreur_cron=(
f"Expression cron invalide : « {expression_cron} ». "
"Format attendu : minute heure jour mois jour_semaine "
"(ex: 0 9 * * mon)."
),
),
},
)
definir_expression_cron(session, expression_cron)
return RedirectResponse(url="/destinataires", status_code=303)

View file

@ -3,10 +3,20 @@ Planification de l'email récapitulatif de stock (voir
email_alerts.envoyer_resume_stock). email_alerts.envoyer_resume_stock).
Choix : un planificateur qui tourne dans le process FastAPI lui-même Choix : un planificateur qui tourne dans le process FastAPI lui-même
(APScheduler), configuré par une expression cron classique (RESUME_STOCK_CRON (APScheduler), configuré par une expression cron classique stockée en base
dans .env, ex: "0 8 * * *" = tous les jours à 8h). Rien à configurer côté (ParametrePlanification, modifiable depuis /destinataires) plutôt que dans
système d'exploitation du serveur — tout reste dans ce dossier, comme le .env : ça permet à l'utilisateur de changer l'horaire depuis l'interface,
reste du projet. sans éditer de fichier ni redémarrer le serveur. Rien à configurer côté
système d'exploitation.
Piège à connaître : le champ "jour de la semaine" d'APScheduler NE SUIT
PAS la convention cron Unix (0=dimanche). Chez APScheduler, un jour
numérique suit datetime.weekday() (0=lundi), donc "0 9 * * 1" avec le
chiffre 1 tombe un MARDI, pas un lundi ! Pour éviter toute ambiguïté, ce
module n'utilise jamais de jour numérique : uniquement les noms
(mon/tue/wed/thu/fri/sat/sun), y compris dans les exemples proposés à
l'utilisateur. Vérifié empiriquement avec la version d'APScheduler
installée ici avant d'écrire ce commentaire.
Limite à connaître : la tâche ne se déclenche que si le process est en 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 cours d'exécution au moment prévu, sans rattrapage si le serveur était
@ -16,15 +26,30 @@ devient un problème, scripts/envoyer_resume_stock.py peut être branché
sur un cron/Planificateur de tâches Windows à la place. sur un cron/Planificateur de tâches Windows à la place.
""" """
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.cron import CronTrigger
from sqlmodel import Session from sqlmodel import Session
from app.config import settings
from app.database import engine from app.database import engine
from app.email_alerts import envoyer_resume_stock from app.email_alerts import envoyer_resume_stock
from app.models import ParametrePlanification
EXPRESSION_PAR_DEFAUT = "0 9 * * mon" # tous les lundis à 9h
# Proposés dans le formulaire de /destinataires comme raccourcis courants ;
# l'utilisateur peut aussi saisir n'importe quelle expression cron valide.
EXEMPLES_CRON: list[tuple[str, str]] = [
("0 9 * * mon", "Tous les lundis à 9h"),
("0 8 * * *", "Tous les jours à 8h"),
("0 8 * * mon-fri", "Les jours ouvrés à 8h"),
("0 */6 * * *", "Toutes les 6 heures"),
("0 8 1 * *", "Le 1er de chaque mois à 8h"),
]
_scheduler = BackgroundScheduler() _scheduler = BackgroundScheduler()
_JOB_ID = "resume_stock"
def _tache_resume_stock() -> None: def _tache_resume_stock() -> None:
@ -35,11 +60,57 @@ def _tache_resume_stock() -> None:
envoyer_resume_stock(session) envoyer_resume_stock(session)
def obtenir_expression_cron(session: Session) -> str:
"""Lit l'expression cron actuelle depuis la base, en créant la ligne
de réglage avec la valeur par défaut si c'est le tout premier appel."""
parametre = session.get(ParametrePlanification, 1)
if parametre is None:
parametre = ParametrePlanification(id=1, expression_cron=EXPRESSION_PAR_DEFAUT)
session.add(parametre)
session.commit()
session.refresh(parametre)
return parametre.expression_cron
def valider_expression_cron(expression: str) -> bool:
"""Vérifie qu'une expression cron (minute heure jour mois
jour_semaine) est syntaxiquement valide, sans l'appliquer."""
try:
CronTrigger.from_crontab(expression)
except ValueError:
return False
return True
def definir_expression_cron(session: Session, expression: str) -> None:
"""Enregistre une nouvelle expression cron et reprogramme
immédiatement la tâche planifiée, sans redémarrer le serveur.
L'appelant doit avoir validé l'expression au préalable avec
valider_expression_cron() : ici on suppose qu'elle est valide.
"""
parametre = session.get(ParametrePlanification, 1)
if parametre is None:
parametre = ParametrePlanification(id=1, expression_cron=expression)
else:
parametre.expression_cron = expression
session.add(parametre)
session.commit()
_programmer(expression)
def _programmer(expression: str) -> None:
trigger = CronTrigger.from_crontab(expression)
_scheduler.add_job(_tache_resume_stock, trigger, id=_JOB_ID, replace_existing=True)
def demarrer_planification() -> None: def demarrer_planification() -> None:
"""Démarre la tâche planifiée. Appelé une fois au démarrage de l'app """Démarre le planificateur au lancement de l'app, avec l'expression
(voir lifespan dans main.py).""" actuellement enregistrée en base (ou la valeur par défaut au tout
trigger = CronTrigger.from_crontab(settings.resume_stock_cron) premier démarrage)."""
_scheduler.add_job(_tache_resume_stock, trigger, id="resume_stock", replace_existing=True) with Session(engine) as session:
expression = obtenir_expression_cron(session)
_programmer(expression)
if not _scheduler.running: if not _scheduler.running:
_scheduler.start() _scheduler.start()
@ -48,3 +119,10 @@ def arreter_planification() -> None:
"""Arrête proprement le planificateur (appelé à l'extinction de l'app).""" """Arrête proprement le planificateur (appelé à l'extinction de l'app)."""
if _scheduler.running: if _scheduler.running:
_scheduler.shutdown(wait=False) _scheduler.shutdown(wait=False)
def prochaine_execution() -> datetime | None:
"""Prochaine date d'exécution de la tâche planifiée, pour l'affichage
dans l'interface. None si le planificateur n'est pas démarré."""
job = _scheduler.get_job(_JOB_ID)
return job.next_run_time if job else None

View file

@ -207,6 +207,19 @@ details[open] > summary .noeud-chevron {
white-space: nowrap; white-space: nowrap;
} }
/* Planification de l'email récapitulatif (/destinataires) : un select de
raccourcis courants et un champ libre pour une expression cron
personnalisée, tous deux visibles en même temps plutôt que de basculer
l'un/l'autre en JavaScript. */
.formulaire-planification select {
min-width: 260px;
}
.champ-cron {
font-family: var(--police-donnee);
min-width: 200px;
}
/* --------------------------------------------------------------------- /* ---------------------------------------------------------------------
Arborescence de catégories : profondeur illimitée, repliée par défaut. Arborescence de catégories : profondeur illimitée, repliée par défaut.
------------------------------------------------------------------ */ ------------------------------------------------------------------ */

View file

@ -27,4 +27,44 @@
<input type="email" name="email" placeholder="email@clinique.local" required> <input type="email" name="email" placeholder="email@clinique.local" required>
<button type="submit">Ajouter</button> <button type="submit">Ajouter</button>
</form> </form>
<h2>Résumé périodique du stock</h2>
<p class="etiquette-sous-titre">
Un email récapitulatif (stock bas en premier, puis le reste du stock) est
envoyé automatiquement selon l'horaire ci-dessous, aux destinataires listés
plus haut.
{% if prochaine_execution %}
Prochain envoi : {{ prochaine_execution.strftime("%d/%m/%Y à %H:%M") }}.
{% endif %}
</p>
{% if erreur_cron %}
<p class="message-erreur">{{ erreur_cron }}</p>
{% endif %}
{% set exemples_valeurs = exemples_cron | map(attribute=0) | list %}
<form method="post" action="/destinataires/planification" class="formulaire-inline formulaire-planification">
<select name="expression_cron">
{% for expression, description in exemples_cron %}
<option value="{{ expression }}" {{ "selected" if expression == expression_cron else "" }}>
{{ description }} ({{ expression }})
</option>
{% endfor %}
{% if expression_cron not in exemples_valeurs %}
<option value="{{ expression_cron }}" selected>Personnalisé ({{ expression_cron }})</option>
{% endif %}
</select>
<button type="submit">Appliquer</button>
</form>
<form method="post" action="/destinataires/planification" class="formulaire-inline">
<input type="text" name="expression_cron" value="{{ expression_cron }}" class="champ-cron" placeholder="0 9 * * mon">
<button type="submit">Enregistrer une expression personnalisée</button>
</form>
<p class="etiquette-sous-titre">
Format cron : minute (0-59) heure (0-23) jour (1-31) mois (1-12) jour de la
semaine. Utilisez les noms de jours (mon, tue, wed, thu, fri, sat, sun),
pas de chiffres — leur numérotation ne suit pas le cron Unix habituel et
prête à confusion.
</p>
{% endblock %} {% endblock %}

View file

@ -0,0 +1,55 @@
"""
Tests de la planification de l'email récapitulatif, modifiable depuis
/destinataires : lecture/écriture de l'expression cron en base, et
validation (APScheduler a sa propre convention de jour de la semaine, donc
la validation doit accepter les noms de jours comme "mon" sans ambiguïté).
"""
from app.scheduler import EXPRESSION_PAR_DEFAUT, obtenir_expression_cron, valider_expression_cron
def test_expression_par_defaut_est_tous_les_lundis(session):
assert obtenir_expression_cron(session) == EXPRESSION_PAR_DEFAUT == "0 9 * * mon"
def test_valider_expression_cron_accepte_les_noms_de_jours():
assert valider_expression_cron("0 9 * * mon") is True
assert valider_expression_cron("0 8 * * mon-fri") is True
assert valider_expression_cron("0 */6 * * *") is True
def test_valider_expression_cron_rejette_une_expression_invalide():
assert valider_expression_cron("pas une expression cron") is False
assert valider_expression_cron("0 25 * * *") is False # heure hors plage
def test_modifier_planification_avec_une_expression_valide(client, session):
reponse = client.post(
"/destinataires/planification",
data={"expression_cron": "0 8 * * *"},
follow_redirects=False,
)
assert reponse.status_code == 303
assert obtenir_expression_cron(session) == "0 8 * * *"
def test_modifier_planification_avec_une_expression_invalide_ne_sauvegarde_pas(client, session):
expression_avant = obtenir_expression_cron(session)
reponse = client.post(
"/destinataires/planification",
data={"expression_cron": "n'importe quoi"},
)
assert reponse.status_code == 200
assert "invalide" in reponse.text.lower()
assert obtenir_expression_cron(session) == expression_avant
def test_page_destinataires_affiche_les_exemples_cron(client, session):
reponse = client.get("/destinataires")
assert reponse.status_code == 200
assert "Tous les lundis à 9h" in reponse.text
assert "0 9 * * mon" in reponse.text