Rend les plages horaires et les rappels ntfy configurables, supprime le
classement, ajoute une suite de tests pytest - calcul.py: matin_debut/fin, aprem_debut/fin et pause_dejeuner_fin sont paramétrables via config.json > plages (moteur de calcul de l'heure de sortie optimale inclus, pas seulement l'affichage). Défauts identiques à l'ancien comportement, validé par la suite de tests. - stats.py: les seuils de conformité utilisent la même config. - main.py/templates: warn de saisie et suffixes de cible dynamiques. - Rappels ntfy: serveur, topic et plage horaire (rappel_debut_h/fin_h) configurables par utilisateur depuis /settings, plus en dur dans le code. - Page classement supprimée (pas de compétition entre utilisateurs). - Nouvelle suite pytest (app/tests/), à lancer via `docker compose run --rm pointeuse pytest -v`.
This commit is contained in:
@ -8,24 +8,21 @@ from models import list_users, load_notif_config, load_presence, load_week_point
|
||||
|
||||
logger = logging.getLogger("pointeuse.notifications")
|
||||
|
||||
NTFY_BASE = "https://ntfy.sh"
|
||||
CHECK_INTERVAL_SECONDS = 300
|
||||
RAPPEL_DEBUT_H = 7
|
||||
RAPPEL_FIN_H = 20
|
||||
|
||||
|
||||
def send_ntfy(topic: str, message: str, title: str):
|
||||
def send_ntfy(server: str, topic: str, message: str, title: str):
|
||||
req = urllib.request.Request(
|
||||
f"{NTFY_BASE}/{topic}",
|
||||
f"{server}/{topic}",
|
||||
data=message.encode("utf-8"),
|
||||
headers={"Title": title.encode("utf-8"), "Priority": "high"},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
urllib.request.urlopen(req, timeout=10).close()
|
||||
logger.info("notif ntfy envoyée: topic=%r title=%r", topic, title)
|
||||
logger.info("notif ntfy envoyée: server=%r topic=%r title=%r", server, topic, title)
|
||||
except Exception:
|
||||
logger.exception("échec de l'envoi ntfy: topic=%r title=%r", topic, title)
|
||||
logger.exception("échec de l'envoi ntfy: server=%r topic=%r title=%r", server, topic, title)
|
||||
# best-effort: a missed reminder isn't worth crashing the loop
|
||||
|
||||
|
||||
@ -39,11 +36,14 @@ def _todays_pointage(user_id: str, today: datetime) -> dict:
|
||||
async def check_user(user_id: str, now: datetime):
|
||||
notif = load_notif_config(user_id)
|
||||
topic = notif.get("ntfy_topic")
|
||||
server = notif.get("ntfy_server", "https://ntfy.sh")
|
||||
debut_h = notif.get("rappel_debut_h", 7)
|
||||
fin_h = notif.get("rappel_fin_h", 20)
|
||||
if not topic:
|
||||
logger.debug("check_user(%r): pas de topic ntfy configuré, ignoré", user_id)
|
||||
return
|
||||
if now.weekday() >= 5 or not (RAPPEL_DEBUT_H <= now.hour < RAPPEL_FIN_H):
|
||||
logger.debug("check_user(%r): hors plage de rappel (%s)", user_id, now)
|
||||
if now.weekday() >= 5 or not (debut_h <= now.hour < fin_h):
|
||||
logger.debug("check_user(%r): hors plage de rappel (%s, plage %sh-%sh)", user_id, now, debut_h, fin_h)
|
||||
return
|
||||
|
||||
presence = load_presence(user_id)
|
||||
@ -55,10 +55,10 @@ async def check_user(user_id: str, now: datetime):
|
||||
|
||||
if presence.get("present") and not jour.get("matin_entree"):
|
||||
logger.info("rappel 'pense à pointer' pour %r", user_id)
|
||||
await asyncio.to_thread(send_ntfy, topic, "Tu es dans les locaux mais pas encore pointé ce matin.", "⏰ Pense à pointer")
|
||||
await asyncio.to_thread(send_ntfy, server, topic, "Tu es dans les locaux mais pas encore pointé ce matin.", "⏰ Pense à pointer")
|
||||
elif not presence.get("present") and jour.get("matin_entree") and not jour.get("aprem_sortie"):
|
||||
logger.info("rappel 'pense à dépointer' pour %r", user_id)
|
||||
await asyncio.to_thread(send_ntfy, topic, "Tu as quitté les locaux sans dépointer la sortie.", "⏰ Pense à dépointer")
|
||||
await asyncio.to_thread(send_ntfy, server, topic, "Tu as quitté les locaux sans dépointer la sortie.", "⏰ Pense à dépointer")
|
||||
|
||||
|
||||
async def reminder_loop():
|
||||
|
||||
Reference in New Issue
Block a user