2ee317a46e
Permet de vérifier en un clic que le serveur et le topic ntfy
renseignés fonctionnent, avant d'attendre un vrai rappel. Le bouton
envoie une notification de test via la même chaîne que les rappels
réels (donc tracée dans /logs), et affiche le résultat inline.
La fonction send_ntfy retourne désormais un booléen de succès pour
permettre ce retour utilisateur.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
716 lines
29 KiB
Python
716 lines
29 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
import re
|
|
import secrets
|
|
from datetime import datetime, timedelta
|
|
from itertools import groupby
|
|
|
|
import bcrypt
|
|
from fastapi import FastAPI, Form, Request, Response
|
|
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from mail import send_mail
|
|
from models import (
|
|
load_config, load_week_pointages, save_week_pointages, load_conges,
|
|
toggle_conge, list_all_weeks,
|
|
load_notif_config, save_notif_config, find_user_by_token, save_presence,
|
|
load_auth, save_auth, find_user_by_reset_token,
|
|
load_plages, save_plages, heures_jour_min,
|
|
append_log, load_logs,
|
|
)
|
|
from calcul import compute_week, minutes_to_hhmm, hhmm_to_minutes, heures_dues
|
|
from stats import compute_all_stats
|
|
from export import export_semaine_ods, export_annee_ods
|
|
from notifications import reminder_loop, send_ntfy
|
|
|
|
logging.basicConfig(
|
|
level=getattr(logging, load_config().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())
|
|
templates = Jinja2Templates(directory="templates")
|
|
templates.env.filters["zip"] = zip
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
|
|
JOURS_FR = ["Lundi", "Mardi", "Mercredi", "Jeudi", "Vendredi"]
|
|
_USERNAME_RE = re.compile(r"^[a-zA-Z0-9_.\-]{1,64}$")
|
|
_EMAIL_RE = re.compile(r"^([a-zA-Z0-9_.\-]{1,64})@([a-zA-Z0-9.\-]{1,255})$")
|
|
_HHMM_RE = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$")
|
|
|
|
|
|
# ── Auth helpers ──────────────────────────────────────────────────────────────
|
|
|
|
def _get_user(request: Request) -> str | None:
|
|
"""Extract current user from cookie."""
|
|
uid = request.cookies.get("user_id", "").strip()
|
|
return uid if _USERNAME_RE.match(uid) else None
|
|
|
|
|
|
def allowed_email_domain() -> str:
|
|
return load_config().get("allowed_email_domain", "")
|
|
|
|
|
|
def mail_from_address() -> str:
|
|
return load_config().get("mail_from") or "no-reply@pointeuse.local"
|
|
|
|
|
|
def smtp_config() -> dict:
|
|
return load_config().get("smtp", {})
|
|
|
|
|
|
def _require_user(request: Request):
|
|
"""Return user_id or redirect to login."""
|
|
uid = _get_user(request)
|
|
if not uid:
|
|
return None
|
|
return uid
|
|
|
|
|
|
# ── Shared helpers ────────────────────────────────────────────────────────────
|
|
|
|
def week_dates(year: int, week: int) -> list[str]:
|
|
first = datetime.strptime(f"{year}-W{week:02d}-1", "%G-W%V-%u").date()
|
|
return [(first + timedelta(days=i)).strftime("%Y-%m-%d") for i in range(5)]
|
|
|
|
|
|
def _weeks_by_year(all_weeks: list) -> dict:
|
|
result = {}
|
|
for yr, wks in groupby(sorted(all_weeks, reverse=True), key=lambda x: x[0]):
|
|
result[yr] = sorted(wks, key=lambda x: x[1], reverse=True)
|
|
return result
|
|
|
|
|
|
def _build_soldes_ctx(year, week, conges, h_jour, user_id):
|
|
dates = week_dates(year, week)
|
|
raw = load_week_pointages(year, week, user_id)
|
|
p_by_date = {p["date"]: p for p in raw}
|
|
pointages = [
|
|
p_by_date.get(d, {"date": d, "matin_entree": None, "matin_sortie": None,
|
|
"aprem_entree": None, "aprem_sortie": None})
|
|
for d in dates
|
|
]
|
|
return {"result": compute_week(pointages, conges, h_jour, load_plages(user_id))}
|
|
|
|
|
|
def _build_jour_ctx(date, conges, h_jour, user_id):
|
|
from calcul import heures_travaillees
|
|
d = datetime.strptime(date, "%Y-%m-%d").date()
|
|
iso = d.isocalendar()
|
|
raw = load_week_pointages(iso.year, iso.week, user_id)
|
|
p_by_date = {p["date"]: p for p in raw}
|
|
p = p_by_date.get(date, {"date": date, "matin_entree": None, "matin_sortie": None,
|
|
"aprem_entree": None, "aprem_sortie": None})
|
|
trav = heures_travaillees(p)
|
|
du = heures_dues(date, conges, h_jour)
|
|
delta = trav - du
|
|
return {
|
|
**p,
|
|
"travaille_min": trav, "du_min": du, "delta_min": delta,
|
|
"travaille": minutes_to_hhmm(trav),
|
|
"du": minutes_to_hhmm(du),
|
|
"delta": minutes_to_hhmm(delta),
|
|
"conge": {c["type"] for c in conges if c["date"] == date},
|
|
}
|
|
|
|
|
|
def _oob_calc(date, jour):
|
|
trav_inner = f'<span class="calc-num">{"—" if not jour["travaille_min"] else jour["travaille"]}</span>'
|
|
if jour["travaille_min"]:
|
|
cls = "delta-pos" if jour["delta_min"] > 0 else ("delta-neg" if jour["delta_min"] < 0 else "delta-zero")
|
|
sign = "+" if jour["delta_min"] > 0 else ""
|
|
delta_inner = f'<span class="{cls}">{sign}{jour["delta"]}</span>'
|
|
else:
|
|
delta_inner = '<span class="delta-zero">—</span>'
|
|
return (
|
|
f'<span id="calc-{date}-trav" hx-swap-oob="true">{trav_inner}</span>'
|
|
f'<span id="calc-{date}-delta" hx-swap-oob="true">{delta_inner}</span>'
|
|
)
|
|
|
|
|
|
def _oob_week_extras(result):
|
|
html = ""
|
|
for jour in result["jours"]:
|
|
date = jour["date"]
|
|
cm = jour["delta_cumul_min"]
|
|
if jour.get("cumul_visible"):
|
|
if cm == 0:
|
|
cumul_inner = '<span class="delta-zero">0h00</span>'
|
|
else:
|
|
cls = "delta-pos" if cm > 0 else "delta-neg"
|
|
sign = "+" if cm > 0 else "-"
|
|
cumul_inner = f'<span class="{cls}">{sign}{jour["delta_cumul"]}</span>'
|
|
else:
|
|
cumul_inner = '<span class="delta-zero">—</span>'
|
|
html += f'<span id="calc-{date}-cumul" hx-swap-oob="true">{cumul_inner}</span>'
|
|
parts = []
|
|
if jour.get("cible_matin"):
|
|
parts.append(f'<span class="cible-num">M {jour["cible_matin_entree"]} → {jour["cible_matin"]}</span>')
|
|
if jour.get("cible_aprem"):
|
|
parts.append(f'<span class="cible-num">A {jour["cible_aprem_entree"]} → {jour["cible_aprem"]}</span>')
|
|
cible_inner = " ".join(parts) if parts else '<span class="delta-zero">—</span>'
|
|
html += f'<span id="calc-{date}-cible" hx-swap-oob="true">{cible_inner}</span>'
|
|
return html
|
|
|
|
|
|
def _oob_time_cells(date, jour, plages):
|
|
cg = jour["conge"]
|
|
matin_cg = "matin" in cg or "jour" in cg
|
|
aprem_cg = "aprem" in cg or "jour" in cg
|
|
matin_warn = not matin_cg and jour["du_min"] > 0 and (
|
|
(jour.get("matin_entree") and jour["matin_entree"] > plages["matin_debut"]) or
|
|
(jour.get("matin_sortie") and jour["matin_sortie"] < plages["matin_fin"])
|
|
)
|
|
aprem_warn = not aprem_cg and jour["du_min"] > 0 and (
|
|
(jour.get("aprem_entree") and jour["aprem_entree"] > plages["aprem_debut"]) or
|
|
(jour.get("aprem_sortie") and jour["aprem_sortie"] < plages["aprem_fin"])
|
|
)
|
|
matin_cls = "td-fill cg" if matin_cg else ("td-fill warn" if matin_warn else "td-fill")
|
|
aprem_cls = "td-fill cg" if aprem_cg else ("td-fill warn" if aprem_warn else "td-fill")
|
|
matin_html = (
|
|
f'<div id="cg-matin-{date}" hx-swap-oob="true" class="{matin_cls}">'
|
|
f'<div class="time-pair">'
|
|
f'<input class="time-input" type="time" name="matin_entree" data-date="{date}" value="{jour["matin_entree"] or ""}">'
|
|
f'<span class="time-sep">→</span>'
|
|
f'<input class="time-input" type="time" name="matin_sortie" data-date="{date}" value="{jour["matin_sortie"] or ""}">'
|
|
f'</div></div>'
|
|
)
|
|
aprem_html = (
|
|
f'<div id="cg-aprem-{date}" hx-swap-oob="true" class="{aprem_cls}">'
|
|
f'<div class="time-pair">'
|
|
f'<input class="time-input" type="time" name="aprem_entree" data-date="{date}" value="{jour["aprem_entree"] or ""}">'
|
|
f'<span class="time-sep">→</span>'
|
|
f'<input class="time-input" type="time" name="aprem_sortie" data-date="{date}" value="{jour["aprem_sortie"] or ""}">'
|
|
f'</div></div>'
|
|
)
|
|
return matin_html + aprem_html
|
|
|
|
|
|
def _htmx_conge_and_soldes(request, date, d, conges, h_jour, user_id):
|
|
iso = d.isocalendar()
|
|
plages = load_plages(user_id)
|
|
jour = _build_jour_ctx(date, conges, h_jour, user_id)
|
|
cg = jour["conge"]
|
|
btn_ma_on = " on" if "matin" in cg else ""
|
|
btn_am_on = " on" if "aprem" in cg else ""
|
|
btn_j_on = " jour-on" if "jour" in cg else ""
|
|
btns_html = (
|
|
f'<div id="cg-btn-{date}" hx-swap-oob="true">'
|
|
f'<div class="conge-btns">'
|
|
f'<button hx-post="/conge/{date}/matin" hx-swap="none" class="btn-cg{btn_ma_on}" title="Congé matin">MA</button>'
|
|
f'<button hx-post="/conge/{date}/aprem" hx-swap="none" class="btn-cg{btn_am_on}" title="Congé après-midi">AM</button>'
|
|
f'<button hx-post="/conge/{date}/jour" hx-swap="none" class="btn-cg btn-jour{btn_j_on}" title="Journée entière">J</button>'
|
|
f'</div></div>'
|
|
)
|
|
soldes_ctx = _build_soldes_ctx(iso.year, iso.week, conges, h_jour, user_id)
|
|
soldes_html = templates.get_template("_soldes.html").render({"request": request, **soldes_ctx})
|
|
return HTMLResponse(
|
|
_oob_time_cells(date, jour, plages)
|
|
+ btns_html
|
|
+ _oob_calc(date, jour)
|
|
+ _oob_week_extras(soldes_ctx["result"])
|
|
+ soldes_html
|
|
)
|
|
|
|
|
|
def _htmx_calc_and_soldes(request, date, d, conges, h_jour, user_id):
|
|
iso = d.isocalendar()
|
|
plages = load_plages(user_id)
|
|
jour = _build_jour_ctx(date, conges, h_jour, user_id)
|
|
soldes_ctx = _build_soldes_ctx(iso.year, iso.week, conges, h_jour, user_id)
|
|
soldes_html = templates.get_template("_soldes.html").render({"request": request, **soldes_ctx})
|
|
return HTMLResponse(
|
|
_oob_calc(date, jour)
|
|
+ _oob_time_cells(date, jour, plages)
|
|
+ _oob_week_extras(soldes_ctx["result"])
|
|
+ soldes_html
|
|
)
|
|
|
|
|
|
# ── Login / logout ────────────────────────────────────────────────────────────
|
|
|
|
RESET_TOKEN_VALIDITY_HOURS = 24
|
|
|
|
|
|
def _login_ctx(request, step, email="", error=None):
|
|
return {"request": request, "step": step, "email": email, "error": error}
|
|
|
|
|
|
@app.get("/login", response_class=HTMLResponse)
|
|
def login_page(request: Request):
|
|
if _get_user(request):
|
|
return RedirectResponse("/", status_code=302)
|
|
return templates.TemplateResponse("login.html", _login_ctx(request, "email"))
|
|
|
|
|
|
@app.post("/login", response_class=HTMLResponse)
|
|
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.",
|
|
))
|
|
|
|
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))
|
|
|
|
# No account yet: (re)send a link to set the password
|
|
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}"
|
|
try:
|
|
send_mail(
|
|
email,
|
|
"Définis ton mot de passe — Décompte Horaire",
|
|
f"Clique sur ce lien pour définir ton mot de passe (valable {RESET_TOKEN_VALIDITY_HOURS}h) :\n\n{link}\n",
|
|
mail_from_address(),
|
|
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.",
|
|
))
|
|
|
|
return templates.TemplateResponse("login.html", _login_ctx(request, "sent", email))
|
|
|
|
|
|
@app.post("/login/password", response_class=HTMLResponse)
|
|
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
|
|
|
|
|
|
def _valid_reset_token(token: str) -> str | None:
|
|
user_id = find_user_by_reset_token(token)
|
|
if not user_id:
|
|
return None
|
|
expires = load_auth(user_id).get("token_expires")
|
|
if not expires or datetime.fromisoformat(expires) < datetime.now():
|
|
return None
|
|
return user_id
|
|
|
|
|
|
@app.get("/set-password/{token}", response_class=HTMLResponse)
|
|
def set_password_page(request: Request, token: str):
|
|
valid = _valid_reset_token(token) is not None
|
|
return templates.TemplateResponse("set_password.html", {
|
|
"request": request, "token": token, "valid": valid, "error": None,
|
|
})
|
|
|
|
|
|
@app.post("/set-password/{token}", response_class=HTMLResponse)
|
|
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.",
|
|
})
|
|
|
|
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)
|
|
return resp
|
|
|
|
|
|
@app.get("/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
|
|
|
|
|
|
# ── Main routes ───────────────────────────────────────────────────────────────
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def index(request: Request):
|
|
if not _get_user(request):
|
|
return RedirectResponse("/login", status_code=302)
|
|
today = datetime.today()
|
|
iso = today.isocalendar()
|
|
return RedirectResponse(f"/semaine/{iso.year}/{iso.week}")
|
|
|
|
|
|
@app.get("/semaine/{year}/{week}", response_class=HTMLResponse)
|
|
def semaine(request: Request, year: int, week: int):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
h_jour = heures_jour_min()
|
|
plages = load_plages(user_id)
|
|
conges = load_conges(user_id)
|
|
soldes_ctx = _build_soldes_ctx(year, week, conges, h_jour, user_id)
|
|
|
|
dates = week_dates(year, week)
|
|
raw = load_week_pointages(year, week, user_id)
|
|
p_by_date = {p["date"]: p for p in raw}
|
|
pointages = [
|
|
p_by_date.get(d, {"date": d, "matin_entree": None, "matin_sortie": None,
|
|
"aprem_entree": None, "aprem_sortie": None})
|
|
for d in dates
|
|
]
|
|
result = compute_week(pointages, conges, h_jour, plages)
|
|
|
|
prev_d = datetime.strptime(f"{year}-W{week:02d}-1", "%G-W%V-%u").date() - timedelta(weeks=1)
|
|
next_d = datetime.strptime(f"{year}-W{week:02d}-1", "%G-W%V-%u").date() + timedelta(weeks=1)
|
|
all_weeks = list_all_weeks(user_id)
|
|
|
|
return templates.TemplateResponse("semaine.html", {
|
|
"request": request,
|
|
"year": year, "week": week,
|
|
"result": result,
|
|
**soldes_ctx,
|
|
"plages": plages,
|
|
"prev_year": prev_d.isocalendar().year, "prev_week": prev_d.isocalendar().week,
|
|
"next_year": next_d.isocalendar().year, "next_week": next_d.isocalendar().week,
|
|
"jours_fr": JOURS_FR,
|
|
"all_weeks": all_weeks,
|
|
"weeks_by_year": _weeks_by_year(all_weeks),
|
|
"current_user": user_id,
|
|
})
|
|
|
|
|
|
@app.post("/pointage/{date}", response_class=HTMLResponse)
|
|
def save_pointage(
|
|
request: Request, date: str,
|
|
matin_entree: str = Form(""), matin_sortie: str = Form(""),
|
|
aprem_entree: str = Form(""), aprem_sortie: str = Form(""),
|
|
):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
def clean(v):
|
|
v = v.strip(); return v if v else None
|
|
|
|
d = datetime.strptime(date, "%Y-%m-%d").date()
|
|
iso = d.isocalendar()
|
|
raw = load_week_pointages(iso.year, iso.week, user_id)
|
|
p_by_date = {p["date"]: p for p in raw}
|
|
p_by_date[date] = {
|
|
"date": date,
|
|
"matin_entree": clean(matin_entree), "matin_sortie": clean(matin_sortie),
|
|
"aprem_entree": clean(aprem_entree), "aprem_sortie": clean(aprem_sortie),
|
|
}
|
|
save_week_pointages(iso.year, iso.week, sorted(p_by_date.values(), key=lambda x: x["date"]), user_id)
|
|
conges = load_conges(user_id)
|
|
return _htmx_calc_and_soldes(request, date, d, conges, heures_jour_min(), user_id)
|
|
|
|
|
|
@app.get("/refresh/{date}", response_class=HTMLResponse)
|
|
def refresh_pointage(request: Request, date: str):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
d = datetime.strptime(date, "%Y-%m-%d").date()
|
|
conges = load_conges(user_id)
|
|
return _htmx_calc_and_soldes(request, date, d, conges, heures_jour_min(), user_id)
|
|
|
|
|
|
@app.post("/conge/{date}/{type_conge}", response_class=HTMLResponse)
|
|
def toggle_conge_route(date: str, type_conge: str, request: Request):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
toggle_conge(date, type_conge, user_id)
|
|
d = datetime.strptime(date, "%Y-%m-%d").date()
|
|
if request.headers.get("HX-Request"):
|
|
conges = load_conges(user_id)
|
|
return _htmx_conge_and_soldes(request, date, d, conges, heures_jour_min(), user_id)
|
|
iso = d.isocalendar()
|
|
return RedirectResponse(f"/semaine/{iso.year}/{iso.week}", status_code=303)
|
|
|
|
|
|
@app.get("/stats", response_class=HTMLResponse)
|
|
def stats_page(request: Request):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
stats = compute_all_stats(user_id)
|
|
data_json = json.dumps({
|
|
"time_stats": {
|
|
slot: {
|
|
"mean": s["mean"], "sigma": s["sigma"],
|
|
"mean_hhmm": s["mean_hhmm"], "sigma_min": s["sigma_min"],
|
|
"n": s["n"], "dated_values": s["dated_values"],
|
|
} if s else None
|
|
for slot, s in stats["time_stats"].items()
|
|
},
|
|
"compliance": stats["compliance"],
|
|
"weekly_balances": stats["weekly_balances"],
|
|
"today": stats["today"],
|
|
})
|
|
return templates.TemplateResponse("stats.html", {
|
|
"request": request,
|
|
"stats": stats,
|
|
"data_json": data_json,
|
|
"current_user": user_id,
|
|
})
|
|
|
|
|
|
ODS_MEDIA_TYPE = "application/vnd.oasis.opendocument.spreadsheet"
|
|
|
|
|
|
@app.get("/export/semaine/{year}/{week}")
|
|
def export_semaine(request: Request, year: int, week: int):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
content = export_semaine_ods(year, week, user_id)
|
|
filename = f"pointages-{user_id}-{year}-S{week:02d}.ods"
|
|
logger.info("export ODS semaine: user_id=%r year=%d week=%d", user_id, year, week)
|
|
return Response(
|
|
content=content, media_type=ODS_MEDIA_TYPE,
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
@app.get("/export/annee/{year}")
|
|
def export_annee(request: Request, year: int):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
content = export_annee_ods(year, user_id)
|
|
filename = f"pointages-{user_id}-{year}.ods"
|
|
logger.info("export ODS année: user_id=%r year=%d", user_id, year)
|
|
return Response(
|
|
content=content, media_type=ODS_MEDIA_TYPE,
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
@app.get("/settings", response_class=HTMLResponse)
|
|
def settings_page(request: Request):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
notif = load_notif_config(user_id)
|
|
plages = load_plages(user_id)
|
|
base_url = str(request.base_url).rstrip("/")
|
|
return templates.TemplateResponse("settings.html", {
|
|
"request": request,
|
|
"current_user": user_id,
|
|
"ntfy_topic": notif.get("ntfy_topic") or "",
|
|
"ntfy_server": notif.get("ntfy_server") or "https://ntfy.sh",
|
|
"rappel_debut_h": notif.get("rappel_debut_h", 7),
|
|
"rappel_fin_h": notif.get("rappel_fin_h", 20),
|
|
"arrivee_url": f"{base_url}/presence/{notif['token']}/arrivee",
|
|
"depart_url": f"{base_url}/presence/{notif['token']}/depart",
|
|
"plages": plages,
|
|
"saved": request.query_params.get("saved") == "1",
|
|
"plages_saved": request.query_params.get("plages_saved") == "1",
|
|
"plages_error": request.query_params.get("plages_error") == "1",
|
|
})
|
|
|
|
|
|
@app.post("/settings", response_class=HTMLResponse)
|
|
def settings_save(
|
|
request: Request, ntfy_topic: str = Form(""), ntfy_server: str = Form(""),
|
|
rappel_debut_h: int = Form(7), rappel_fin_h: int = Form(20),
|
|
):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
rappel_debut_h = max(0, min(23, rappel_debut_h))
|
|
rappel_fin_h = max(rappel_debut_h + 1, min(24, rappel_fin_h))
|
|
|
|
logger.info(
|
|
"réglages notif mis à jour: user_id=%r ntfy_topic=%r ntfy_server=%r plage=%sh-%sh",
|
|
user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h,
|
|
)
|
|
save_notif_config(user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h)
|
|
return RedirectResponse("/settings?saved=1", status_code=303)
|
|
|
|
|
|
@app.post("/settings/test-ntfy", response_class=HTMLResponse)
|
|
def settings_test_ntfy(request: Request):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
notif = load_notif_config(user_id)
|
|
topic = notif.get("ntfy_topic")
|
|
server = notif.get("ntfy_server", "https://ntfy.sh")
|
|
|
|
if not topic:
|
|
logger.warning("test ntfy: pas de topic configuré pour %r", user_id)
|
|
return HTMLResponse(
|
|
'<div id="ntfy-test-result" class="settings-error">'
|
|
'Renseigne et enregistre un topic ntfy avant de tester.'
|
|
'</div>'
|
|
)
|
|
|
|
logger.info("test ntfy: envoi de test pour %r sur %r/%r", user_id, server, topic)
|
|
ok = send_ntfy(
|
|
server, topic,
|
|
"Ceci est une notification de test — ta configuration ntfy fonctionne.",
|
|
"Test pointeuse", user_id=user_id,
|
|
)
|
|
cls = "settings-saved" if ok else "settings-error"
|
|
msg = ("Notification envoyée. Vérifie ton téléphone, elle doit arriver en quelques secondes."
|
|
if ok else
|
|
"Échec de l'envoi. Vérifie le serveur et le topic, et l'optimisation batterie de l'app ntfy.")
|
|
return HTMLResponse(
|
|
f'<div id="ntfy-test-result" class="{cls}">{msg}</div>'
|
|
)
|
|
|
|
|
|
@app.post("/settings/plages", response_class=HTMLResponse)
|
|
def settings_plages_save(
|
|
request: Request,
|
|
matin_debut: str = Form(""), matin_fin: str = Form(""),
|
|
aprem_debut: str = Form(""), aprem_fin: str = Form(""),
|
|
pause_dejeuner_fin: str = Form(""),
|
|
arrivee_visee: str = Form(""), depart_vise: str = Form(""),
|
|
):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
fields = {
|
|
"matin_debut": matin_debut, "matin_fin": matin_fin,
|
|
"aprem_debut": aprem_debut, "aprem_fin": aprem_fin,
|
|
"pause_dejeuner_fin": pause_dejeuner_fin,
|
|
# Heures visées, optionnelles (vide = comportement par défaut)
|
|
"arrivee_visee": arrivee_visee, "depart_vise": depart_vise,
|
|
}
|
|
optionnels = {"arrivee_visee", "depart_vise"}
|
|
if not all(_HHMM_RE.match(v) for k, v in fields.items() if v or k not in optionnels):
|
|
logger.warning("réglages plages rejetés (format invalide): user_id=%r valeurs=%r", user_id, fields)
|
|
return RedirectResponse("/settings?plages_error=1", status_code=303)
|
|
if not (fields["matin_debut"] < fields["matin_fin"] < fields["aprem_debut"] < fields["aprem_fin"]):
|
|
logger.warning("réglages plages rejetés (ordre invalide): user_id=%r valeurs=%r", user_id, fields)
|
|
return RedirectResponse("/settings?plages_error=1", status_code=303)
|
|
if (fields["arrivee_visee"] and fields["arrivee_visee"] >= fields["matin_fin"]) or \
|
|
(fields["depart_vise"] and fields["depart_vise"] < fields["aprem_fin"]):
|
|
logger.warning("réglages plages rejetés (heures visées incohérentes): user_id=%r valeurs=%r", user_id, fields)
|
|
return RedirectResponse("/settings?plages_error=1", status_code=303)
|
|
|
|
logger.info("réglages plages mis à jour: user_id=%r plages=%r", user_id, fields)
|
|
save_plages(user_id, fields)
|
|
return RedirectResponse("/settings?plages_saved=1", status_code=303)
|
|
|
|
|
|
@app.get("/presence/{token}/arrivee")
|
|
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())
|
|
append_log(user_id, "presence", "arrivée", "Présence détectée par Tasker")
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/presence/{token}/depart")
|
|
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())
|
|
append_log(user_id, "presence", "départ", "Départ détecté par Tasker")
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/logs", response_class=HTMLResponse)
|
|
def logs_page(request: Request):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
|
|
logs = load_logs(user_id)
|
|
return templates.TemplateResponse("logs.html", {
|
|
"request": request,
|
|
"current_user": user_id,
|
|
"logs": logs,
|
|
"filter": request.query_params.get("filter", "all"),
|
|
})
|
|
|
|
|
|
@app.get("/aide", response_class=HTMLResponse)
|
|
def aide_page(request: Request):
|
|
user_id = _get_user(request)
|
|
if not user_id:
|
|
return RedirectResponse("/login", status_code=302)
|
|
return templates.TemplateResponse("aide.html", {
|
|
"request": request,
|
|
"current_user": user_id,
|
|
})
|
|
|
|
|