code_public/pointeuse-optimisator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ajoute une journalisation détaillée (login, mail, présence, rappels)

Browse Source 
Niveau configurable via LOG_LEVEL (défaut DEBUG). Les erreurs de config
JSON invalide remontent maintenant un message clair (fichier + position)
au lieu d'un JSONDecodeError brut.
This commit is contained in:
Antoine
2026-07-17 22:03:10 +02:00
parent e6a523f76f
commit 2d4179169b
4 changed files with 84 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
app/mail.py
View File

@ -1,11 +1,15 @@
"""Mail delivery via a configured SMTP relay (host/port/credentials in config.json)."""
import logging
import smtplib
from email.mime.text import MIMEText
logger = logging.getLogger("pointeuse.mail")
def send_mail(to_addr: str, subject: str, body: str, from_addr: str, smtp_cfg: dict):
host = smtp_cfg.get("host")
if not host:
logger.error("envoi annulé: aucun serveur SMTP configuré (smtp.host manquant dans config.json)")
raise RuntimeError("Aucun serveur SMTP configuré (smtp.host manquant dans config.json).")
port = int(smtp_cfg.get("port", 587))
user = smtp_cfg.get("user")
@ -17,9 +21,21 @@ def send_mail(to_addr: str, subject: str, body: str, from_addr: str, smtp_cfg: d
msg["From"] = from_addr
msg["To"] = to_addr
with smtplib.SMTP(host, port, timeout=10) as smtp:
if use_tls:
smtp.starttls()
if user:
smtp.login(user, password)
smtp.send_message(msg)
logger.info(
"envoi mail: to=%r from=%r subject=%r via smtp=%s:%s user=%r tls=%r",
to_addr, from_addr, subject, host, port, user, use_tls,
)
try:
with smtplib.SMTP(host, port, timeout=10) as smtp:
smtp.set_debuglevel(1)
if use_tls:
logger.debug("STARTTLS vers %s:%s", host, port)
smtp.starttls()
if user:
logger.debug("login SMTP en tant que %r", user)
smtp.login(user, password)
smtp.send_message(msg)
logger.info("mail envoyé avec succès à %r", to_addr)
except Exception:
logger.exception("échec de l'envoi du mail à %r via %s:%s", to_addr, host, port)
raise

41
app/main.py
View File

@ -1,5 +1,7 @@
import asyncio
import json
import logging
import os
import re
import secrets
from datetime import datetime, timedelta
@ -23,9 +25,28 @@ from stats import compute_all_stats
from classement import compute_classement
from notifications import reminder_loop
logging.basicConfig(
level=getattr(logging, os.environ.get("LOG_LEVEL", "DEBUG").upper(), logging.DEBUG),
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger("pointeuse")
app = FastAPI(title="Décompte Horaire")
@app.on_event("startup")
async def _log_config_at_startup():
cfg = load_config()
smtp = cfg.get("smtp", {})
logger.info(
"config chargée: heures_jour=%r allowed_email_domain=%r mail_from=%r "
"smtp.host=%r smtp.port=%r smtp.user=%r smtp.password=%s smtp.use_tls=%r",
cfg.get("heures_jour"), cfg.get("allowed_email_domain"), cfg.get("mail_from"),
smtp.get("host"), smtp.get("port"), smtp.get("user"),
"***" if smtp.get("password") else "(vide)", smtp.get("use_tls"),
)
@app.on_event("startup")
async def _start_reminder_loop():
asyncio.create_task(reminder_loop())
@ -251,7 +272,9 @@ def login_submit(request: Request, email: str = Form("")):
email = email.strip().lower()
m = _EMAIL_RE.match(email)
domain = allowed_email_domain()
logger.info("POST /login: email=%r domaine_attendu=%r", email, domain)
if not m or not domain or m.group(2) != domain:
logger.warning("login refusé (domaine invalide): email=%r", email)
return templates.TemplateResponse("login.html", _login_ctx(
request, "email", email,
f"Seules les adresses @{domain or '?'} sont autorisées.",
@ -259,6 +282,7 @@ def login_submit(request: Request, email: str = Form("")):
user_id = m.group(1)
auth = load_auth(user_id)
logger.debug("login: user_id=%r compte_existant=%s", user_id, bool(auth.get("password_hash")))
if auth.get("password_hash"):
return templates.TemplateResponse("login.html", _login_ctx(request, "password", email))
@ -267,6 +291,7 @@ def login_submit(request: Request, email: str = Form("")):
token = secrets.token_urlsafe(32)
expires = (datetime.now() + timedelta(hours=RESET_TOKEN_VALIDITY_HOURS)).isoformat()
save_auth(user_id, email=email, token=token, token_expires=expires)
logger.info("nouveau token de création de compte pour %r (expire %s)", user_id, expires)
base_url = str(request.base_url).rstrip("/")
link = f"{base_url}/set-password/{token}"
@ -279,6 +304,7 @@ def login_submit(request: Request, email: str = Form("")):
smtp_config(),
)
except Exception:
logger.exception("échec envoi email d'invitation à %r", email)
return templates.TemplateResponse("login.html", _login_ctx(
request, "email", email, "Impossible d'envoyer l'email pour le moment, réessaie plus tard.",
))
@ -291,16 +317,20 @@ def login_password(request: Request, email: str = Form(""), password: str = Form
email = email.strip().lower()
m = _EMAIL_RE.match(email)
domain = allowed_email_domain()
logger.info("POST /login/password: email=%r", email)
if not m or not domain or m.group(2) != domain:
logger.warning("login/password refusé (domaine invalide): email=%r", email)
return RedirectResponse("/login", status_code=302)
user_id = m.group(1)
ph = load_auth(user_id).get("password_hash")
if not ph or not bcrypt.checkpw(password.encode(), ph.encode()):
logger.warning("mot de passe incorrect pour %r", user_id)
return templates.TemplateResponse("login.html", _login_ctx(
request, "password", email, "Mot de passe incorrect.",
))
logger.info("connexion réussie: user_id=%r", user_id)
resp = RedirectResponse("/", status_code=302)
resp.set_cookie("user_id", user_id, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30)
return resp
@ -328,10 +358,12 @@ def set_password_page(request: Request, token: str):
def set_password_submit(request: Request, token: str, password: str = Form(""), password2: str = Form("")):
user_id = _valid_reset_token(token)
if not user_id:
logger.warning("POST /set-password: token invalide ou expiré")
return templates.TemplateResponse("set_password.html", {
"request": request, "token": token, "valid": False, "error": None,
})
if len(password) < 8 or password != password2:
logger.info("POST /set-password: mot de passe rejeté (règles) pour %r", user_id)
return templates.TemplateResponse("set_password.html", {
"request": request, "token": token, "valid": True,
"error": "Les mots de passe ne correspondent pas ou font moins de 8 caractères.",
@ -339,6 +371,7 @@ def set_password_submit(request: Request, token: str, password: str = Form(""),
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
save_auth(user_id, password_hash=password_hash, token=None, token_expires=None)
logger.info("mot de passe défini pour %r", user_id)
resp = RedirectResponse("/", status_code=302)
resp.set_cookie("user_id", user_id, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30)
@ -346,7 +379,8 @@ def set_password_submit(request: Request, token: str, password: str = Form(""),
@app.get("/logout")
def logout():
def logout(request: Request):
logger.info("déconnexion: user_id=%r", _get_user(request))
resp = RedirectResponse("/login", status_code=302)
resp.delete_cookie("user_id")
return resp
@ -506,6 +540,7 @@ def settings_save(request: Request, ntfy_topic: str = Form("")):
if not user_id:
return RedirectResponse("/login", status_code=302)
logger.info("réglages notif mis à jour: user_id=%r ntfy_topic=%r", user_id, ntfy_topic)
save_notif_config(user_id, ntfy_topic)
return RedirectResponse("/settings?saved=1", status_code=303)
@ -514,7 +549,9 @@ def settings_save(request: Request, ntfy_topic: str = Form("")):
def presence_arrivee(token: str):
user_id = find_user_by_token(token)
if not user_id:
logger.warning("GET /presence/.../arrivee: token inconnu")
return JSONResponse({"error": "invalid token"}, status_code=404)
logger.info("présence: arrivée détectée pour %r", user_id)
save_presence(user_id, present=True, since=datetime.now().isoformat())
return {"ok": True}
@ -523,7 +560,9 @@ def presence_arrivee(token: str):
def presence_depart(token: str):
user_id = find_user_by_token(token)
if not user_id:
logger.warning("GET /presence/.../depart: token inconnu")
return JSONResponse({"error": "invalid token"}, status_code=404)
logger.info("présence: départ détecté pour %r", user_id)
save_presence(user_id, present=False, since=datetime.now().isoformat())
return {"ok": True}

5
app/models.py
View File

@ -51,7 +51,10 @@ def list_users() -> list[str]:
def load_config() -> dict:
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
if CONFIG_FILE.exists():
return json.loads(CONFIG_FILE.read_text())
try:
return json.loads(CONFIG_FILE.read_text())
except json.JSONDecodeError as e:
raise ValueError(f"{CONFIG_FILE} invalide : {e}") from e
CONFIG_FILE.write_text(json.dumps(DEFAULT_CONFIG, indent=2))
return DEFAULT_CONFIG.copy()

20
app/notifications.py
View File

@ -1,10 +1,13 @@
"""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
logger = logging.getLogger("pointeuse.notifications")
NTFY_BASE = "https://ntfy.sh"
CHECK_INTERVAL_SECONDS = 300
RAPPEL_DEBUT_H = 7
@ -20,8 +23,10 @@ def send_ntfy(topic: str, message: str, title: str):
)
try:
urllib.request.urlopen(req, timeout=10).close()
logger.info("notif ntfy envoyée: topic=%r title=%r", topic, title)
except Exception:
pass # best-effort: a missed reminder isn't worth crashing the loop
logger.exception("échec de l'envoi ntfy: topic=%r title=%r", topic, title)
# best-effort: a missed reminder isn't worth crashing the loop
def _todays_pointage(user_id: str, today: datetime) -> dict:
@ -35,22 +40,33 @@ async def check_user(user_id: str, now: datetime):
notif = load_notif_config(user_id)
topic = notif.get("ntfy_topic")
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)
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, 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")
async def reminder_loop():
logger.info("boucle de rappels démarrée (intervalle=%ss)", CHECK_INTERVAL_SECONDS)
while True:
now = datetime.now()
for user_id in list_users():
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)