Ajoute les rappels de pointage via ntfy et détection de présence Tasker
Notifie l'utilisateur toutes les 5 min s'il est présent dans les locaux sans avoir pointé le matin, ou parti sans avoir dépointé la sortie.
This commit is contained in:
56
app/notifications.py
Normal file
56
app/notifications.py
Normal file
@ -0,0 +1,56 @@
|
||||
"""Reminders: nag via ntfy when present-but-unpointed or departed-but-not-depointed."""
|
||||
import asyncio
|
||||
import urllib.request
|
||||
from datetime import datetime
|
||||
|
||||
from models import list_users, load_notif_config, load_presence, load_week_pointages
|
||||
|
||||
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):
|
||||
req = urllib.request.Request(
|
||||
f"{NTFY_BASE}/{topic}",
|
||||
data=message.encode("utf-8"),
|
||||
headers={"Title": title.encode("utf-8"), "Priority": "high"},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
urllib.request.urlopen(req, timeout=10).close()
|
||||
except Exception:
|
||||
pass # 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")
|
||||
if not topic:
|
||||
return
|
||||
if now.weekday() >= 5 or not (RAPPEL_DEBUT_H <= now.hour < RAPPEL_FIN_H):
|
||||
return
|
||||
|
||||
presence = load_presence(user_id)
|
||||
jour = _todays_pointage(user_id, now)
|
||||
|
||||
if presence.get("present") and not jour.get("matin_entree"):
|
||||
await asyncio.to_thread(send_ntfy, 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"):
|
||||
await asyncio.to_thread(send_ntfy, topic, "Tu as quitté les locaux sans dépointer la sortie.", "⏰ Pense à dépointer")
|
||||
|
||||
|
||||
async def reminder_loop():
|
||||
while True:
|
||||
now = datetime.now()
|
||||
for user_id in list_users():
|
||||
await check_user(user_id, now)
|
||||
await asyncio.sleep(CHECK_INTERVAL_SECONDS)
|
||||
Reference in New Issue
Block a user