5fdef77e2b
Chaque utilisateur a désormais son propre fichier logs.json dans son
dossier users/<id>/, alimenté automatiquement quand un rappel ntfy est
envoyé (ou échoue) et quand Tasker signale une arrivée ou un départ.
Une nouvelle page /logs, accessible depuis la nav, affiche ces entrées
avec filtres (Tout / Notifications / Présence), et le journal tourne à
500 entrées pour rester compact.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
"""Reminders: nag via ntfy when present-but-unpointed or departed-but-not-depointed."""
|
|
import asyncio
|
|
import logging
|
|
import urllib.request
|
|
from datetime import datetime
|
|
|
|
from models import list_users, load_notif_config, load_presence, load_week_pointages, append_log
|
|
|
|
logger = logging.getLogger("pointeuse.notifications")
|
|
|
|
CHECK_INTERVAL_SECONDS = 300
|
|
|
|
|
|
def send_ntfy(server: str, topic: str, message: str, title: str, user_id: str | None = None):
|
|
req = urllib.request.Request(
|
|
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: server=%r topic=%r title=%r", server, topic, title)
|
|
if user_id:
|
|
append_log(user_id, "notif", "envoyée", f"{title} — {message}")
|
|
except Exception:
|
|
logger.exception("échec de l'envoi ntfy: server=%r topic=%r title=%r", server, topic, title)
|
|
if user_id:
|
|
append_log(user_id, "notif", "échec", f"{title} — {message}")
|
|
# best-effort: a missed reminder isn't worth crashing the loop
|
|
|
|
|
|
def _todays_pointage(user_id: str, today: datetime) -> dict:
|
|
iso = today.isocalendar()
|
|
pointages = load_week_pointages(iso.year, iso.week, user_id)
|
|
date_str = today.strftime("%Y-%m-%d")
|
|
return next((p for p in pointages if p["date"] == date_str), {})
|
|
|
|
|
|
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 (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)
|
|
jour = _todays_pointage(user_id, now)
|
|
logger.debug(
|
|
"check_user(%r): presence=%s matin_entree=%r aprem_sortie=%r",
|
|
user_id, presence.get("present"), jour.get("matin_entree"), jour.get("aprem_sortie"),
|
|
)
|
|
|
|
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, server, topic,
|
|
"Tu es dans les locaux mais pas encore pointé ce matin.",
|
|
"⏰ Pense à pointer", user_id,
|
|
)
|
|
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, server, topic,
|
|
"Tu as quitté les locaux sans dépointer la sortie.",
|
|
"⏰ Pense à dépointer", user_id,
|
|
)
|
|
|
|
|
|
async def reminder_loop():
|
|
logger.info("boucle de rappels démarrée (intervalle=%ss)", CHECK_INTERVAL_SECONDS)
|
|
while True:
|
|
now = datetime.now()
|
|
users = list_users()
|
|
logger.debug("cycle de vérification pour %d utilisateur(s)", len(users))
|
|
for user_id in users:
|
|
await check_user(user_id, now)
|
|
await asyncio.sleep(CHECK_INTERVAL_SECONDS)
|