dc80e81db2
- Bouton F : marque un jour férié (dû 0, sans consommer de congé) ; il
qualifie les règles « ferie » et étend la fenêtre d'astreinte du jour.
- Samedi, dimanche et fériés : l'astreinte compte de 0h00 à 0h00 ; un jour
AS déclaré garde sa borne du lendemain 18h pour ne pas tronquer les
interventions nocturnes.
- Ratios d'astreinte éditables à la main dans Réglages (nom, jours, fenêtre,
ratio), avec ajout/suppression de règles et validation.
- Un quart d'heure entamé par deux interventions qui se suivent n'est compté
qu'une fois : les lignes d'événements somment au total du jour.
- Décompte mieux expliqué dans l'interface : quarts d'horloge (:00/:15/:30/:45),
fenêtre couverte par événement, exemple chiffré.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
429 lines
15 KiB
Python
429 lines
15 KiB
Python
"""Flat-file storage: JSON files per week for pointages, one file for congés."""
|
||
import json
|
||
import secrets
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from calcul import DEFAULT_PLAGES, DEFAULT_FACTEURS_ASTREINTE
|
||
|
||
DATA_DIR = Path("/data")
|
||
CONFIG_FILE = DATA_DIR / "config.json"
|
||
|
||
DEFAULT_CONFIG = {
|
||
"heures_semaine": 39.0, # heures hebdomadaires visées
|
||
"jours_par_semaine": 5, # jours travaillés dans la semaine — heures_jour en est dérivé
|
||
"log_level": "DEBUG", # DEBUG, INFO, WARNING ou ERROR
|
||
"allowed_email_domain": "", # ex: "sdv.fr" — seules ces adresses peuvent se connecter
|
||
"base_url": "", # ex: "https://pointeuse.sdv.fr" — utilisée pour les liens emails (anti open-redirect via Host header)
|
||
"mail_from": "", # ex: "no-reply@sdv.fr" — expéditeur des emails de connexion
|
||
"smtp": {
|
||
"host": "", # ex: "smtp.sdv.fr"
|
||
"port": 587,
|
||
"user": "",
|
||
"password": "",
|
||
"use_tls": True,
|
||
},
|
||
"plages": dict(DEFAULT_PLAGES),
|
||
"astreinte_facteurs": list(DEFAULT_FACTEURS_ASTREINTE),
|
||
}
|
||
|
||
# ── User path helpers ──────────────────────────────────────────────────────────
|
||
|
||
def _user_root(user_id: str) -> Path:
|
||
return DATA_DIR / "users" / user_id
|
||
|
||
|
||
def _pointages_dir(user_id: str) -> Path:
|
||
return _user_root(user_id) / "pointages"
|
||
|
||
|
||
def _conges_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "conges.json"
|
||
|
||
|
||
def _ensure_user_dirs(user_id: str):
|
||
_pointages_dir(user_id).mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
# ── User list ─────────────────────────────────────────────────────────────────
|
||
|
||
def list_users() -> list[str]:
|
||
users_dir = DATA_DIR / "users"
|
||
if not users_dir.exists():
|
||
return []
|
||
return sorted(p.name for p in users_dir.iterdir() if p.is_dir())
|
||
|
||
|
||
# ── Config (global) ───────────────────────────────────────────────────────────
|
||
|
||
def load_config() -> dict:
|
||
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
if CONFIG_FILE.exists():
|
||
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()
|
||
|
||
|
||
def save_config(cfg: dict):
|
||
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
CONFIG_FILE.write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def load_or_create_secret() -> bytes:
|
||
"""Charge ou crée la clé secrète utilisée pour signer les cookies de session.
|
||
|
||
Persistée hors config pour ne pas être réinitialisée à chaque modification des
|
||
réglages et rester stable entre les redémarrages.
|
||
"""
|
||
secret_file = DATA_DIR / "secret.key"
|
||
secret_file.parent.mkdir(parents=True, exist_ok=True)
|
||
if secret_file.exists():
|
||
try:
|
||
return bytes.fromhex(secret_file.read_text().strip())
|
||
except ValueError:
|
||
pass
|
||
secret = secrets.token_bytes(32)
|
||
secret_file.write_text(secret.hex())
|
||
return secret
|
||
|
||
|
||
def heures_jour_min() -> int:
|
||
"""Minutes attendues par jour, dérivées d'`heures_semaine` et `jours_par_semaine`.
|
||
|
||
Rétrocompat : si `heures_semaine` est absent mais `heures_jour` présent,
|
||
on utilise l'ancienne clé.
|
||
"""
|
||
cfg = load_config()
|
||
jours = int(cfg.get("jours_par_semaine", 5)) or 5
|
||
if "heures_semaine" in cfg:
|
||
return int(float(cfg["heures_semaine"]) * 60 / jours)
|
||
return int(float(cfg.get("heures_jour", 7.8)) * 60)
|
||
|
||
|
||
# ── Pointages ─────────────────────────────────────────────────────────────────
|
||
|
||
def _week_file(user_id: str, year: int, week: int) -> Path:
|
||
return _pointages_dir(user_id) / f"{year}-W{week:02d}.json"
|
||
|
||
|
||
def load_week_pointages(year: int, week: int, user_id: str = "") -> list[dict]:
|
||
f = _week_file(user_id, year, week)
|
||
return json.loads(f.read_text()) if f.exists() else []
|
||
|
||
|
||
def save_week_pointages(year: int, week: int, pointages: list[dict], user_id: str = ""):
|
||
_ensure_user_dirs(user_id)
|
||
_week_file(user_id, year, week).write_text(
|
||
json.dumps(pointages, indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
def list_all_weeks(user_id: str = "") -> list[tuple[int, int]]:
|
||
pt_dir = _pointages_dir(user_id)
|
||
if not pt_dir.exists():
|
||
return []
|
||
weeks = []
|
||
for f in pt_dir.glob("????-W??.json"):
|
||
parts = f.stem.split("-W")
|
||
if len(parts) == 2:
|
||
weeks.append((int(parts[0]), int(parts[1])))
|
||
return sorted(weeks)
|
||
|
||
|
||
# ── Congés ────────────────────────────────────────────────────────────────────
|
||
|
||
def load_conges(user_id: str = "") -> list[dict]:
|
||
f = _conges_file(user_id)
|
||
return json.loads(f.read_text()) if f.exists() else []
|
||
|
||
|
||
def save_conges(conges: list[dict], user_id: str = ""):
|
||
_ensure_user_dirs(user_id)
|
||
_conges_file(user_id).write_text(
|
||
json.dumps(conges, indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
# ── Astreintes ────────────────────────────────────────────────────────────────
|
||
|
||
def _astreintes_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "astreintes.json"
|
||
|
||
|
||
def load_astreintes(user_id: str = "") -> list[str]:
|
||
"""Dates marquées comme jours d'astreinte (triées, dédupliquées)."""
|
||
f = _astreintes_file(user_id)
|
||
if not f.exists():
|
||
return []
|
||
try:
|
||
dates = json.loads(f.read_text())
|
||
except json.JSONDecodeError:
|
||
return []
|
||
return sorted(set(d for d in dates if isinstance(d, str)))
|
||
|
||
|
||
def save_astreintes(dates: list[str], user_id: str = ""):
|
||
_ensure_user_dirs(user_id)
|
||
_astreintes_file(user_id).write_text(
|
||
json.dumps(sorted(set(dates)), indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
def toggle_astreinte(date: str, user_id: str = ""):
|
||
dates = load_astreintes(user_id)
|
||
if date in dates:
|
||
dates.remove(date)
|
||
else:
|
||
dates.append(date)
|
||
save_astreintes(dates, user_id)
|
||
|
||
|
||
# ── Jours fériés (marqués à la main) ──────────────────────────────────────────
|
||
|
||
def _feries_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "feries.json"
|
||
|
||
|
||
def load_feries(user_id: str = "") -> list[str]:
|
||
"""Dates marquées comme jours fériés (triées, dédupliquées)."""
|
||
f = _feries_file(user_id)
|
||
if not f.exists():
|
||
return []
|
||
try:
|
||
dates = json.loads(f.read_text())
|
||
except json.JSONDecodeError:
|
||
return []
|
||
return sorted(set(d for d in dates if isinstance(d, str)))
|
||
|
||
|
||
def save_feries(dates: list[str], user_id: str = ""):
|
||
_ensure_user_dirs(user_id)
|
||
_feries_file(user_id).write_text(
|
||
json.dumps(sorted(set(dates)), indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
def toggle_ferie(date: str, user_id: str = ""):
|
||
dates = load_feries(user_id)
|
||
if date in dates:
|
||
dates.remove(date)
|
||
else:
|
||
dates.append(date)
|
||
save_feries(dates, user_id)
|
||
|
||
|
||
def load_facteurs_astreinte() -> list[dict]:
|
||
"""Règles globales de comptage des heures d'astreinte (type de jour × fenêtre
|
||
horaire × facteur) ; défauts cohérents si absentes ou invalides dans la config."""
|
||
facteurs = load_config().get("astreinte_facteurs")
|
||
if facteurs and isinstance(facteurs, list) and all(isinstance(f, dict) for f in facteurs):
|
||
return facteurs
|
||
return DEFAULT_FACTEURS_ASTREINTE
|
||
|
||
|
||
def save_facteurs_astreinte(facteurs: list[dict]):
|
||
"""Règles globales de comptage des heures d'astreinte (config partagée)."""
|
||
cfg = load_config()
|
||
cfg["astreinte_facteurs"] = facteurs
|
||
save_config(cfg)
|
||
|
||
|
||
# ── Auth (email + password) ────────────────────────────────────────────────────
|
||
|
||
def _auth_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "auth.json"
|
||
|
||
|
||
def load_auth(user_id: str) -> dict:
|
||
f = _auth_file(user_id)
|
||
if f.exists():
|
||
return json.loads(f.read_text())
|
||
return {"email": None, "password_hash": None, "token": None, "token_expires": None}
|
||
|
||
|
||
def save_auth(user_id: str, **fields):
|
||
auth = load_auth(user_id)
|
||
auth.update(fields)
|
||
_ensure_user_dirs(user_id)
|
||
_auth_file(user_id).write_text(json.dumps(auth, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def find_user_by_reset_token(token: str) -> str | None:
|
||
for user_id in list_users():
|
||
if load_auth(user_id).get("token") == token:
|
||
return user_id
|
||
return None
|
||
|
||
|
||
# ── Notifications (ntfy topic + presence tracking) ────────────────────────────
|
||
|
||
def _notif_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "notif.json"
|
||
|
||
|
||
DEFAULT_NTFY_SERVER = "https://ntfy.sh"
|
||
DEFAULT_RAPPEL_DEBUT_H = 7
|
||
DEFAULT_RAPPEL_FIN_H = 20
|
||
|
||
|
||
def load_notif_config(user_id: str) -> dict:
|
||
"""Return {ntfy_topic, ntfy_server, ntfy_token, rappel_debut_h, rappel_fin_h, token}."""
|
||
f = _notif_file(user_id)
|
||
if f.exists():
|
||
cfg = json.loads(f.read_text())
|
||
cfg.setdefault("ntfy_server", DEFAULT_NTFY_SERVER)
|
||
cfg.setdefault("ntfy_token", "")
|
||
cfg.setdefault("rappel_debut_h", DEFAULT_RAPPEL_DEBUT_H)
|
||
cfg.setdefault("rappel_fin_h", DEFAULT_RAPPEL_FIN_H)
|
||
if cfg.get("token"):
|
||
return cfg
|
||
else:
|
||
cfg = {
|
||
"ntfy_topic": None, "ntfy_server": DEFAULT_NTFY_SERVER,
|
||
"ntfy_token": "",
|
||
"rappel_debut_h": DEFAULT_RAPPEL_DEBUT_H, "rappel_fin_h": DEFAULT_RAPPEL_FIN_H,
|
||
}
|
||
cfg["token"] = secrets.token_hex(16)
|
||
_ensure_user_dirs(user_id)
|
||
f.write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||
return cfg
|
||
|
||
|
||
def save_notif_config(
|
||
user_id: str, ntfy_topic: str, ntfy_server: str = "",
|
||
rappel_debut_h: int = DEFAULT_RAPPEL_DEBUT_H, rappel_fin_h: int = DEFAULT_RAPPEL_FIN_H,
|
||
ntfy_token: str = "",
|
||
):
|
||
cfg = load_notif_config(user_id)
|
||
cfg["ntfy_topic"] = ntfy_topic.strip() or None
|
||
cfg["ntfy_server"] = ntfy_server.strip().rstrip("/") or DEFAULT_NTFY_SERVER
|
||
cfg["ntfy_token"] = ntfy_token.strip()
|
||
cfg["rappel_debut_h"] = rappel_debut_h
|
||
cfg["rappel_fin_h"] = rappel_fin_h
|
||
_ensure_user_dirs(user_id)
|
||
_notif_file(user_id).write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def find_user_by_token(token: str) -> str | None:
|
||
for user_id in list_users():
|
||
f = _notif_file(user_id)
|
||
if f.exists() and json.loads(f.read_text()).get("token") == token:
|
||
return user_id
|
||
return None
|
||
|
||
|
||
def regenerate_presence_token(user_id: str) -> str:
|
||
"""Génère un nouveau token de présence (invalide les URLs Tasker précédentes)."""
|
||
cfg = load_notif_config(user_id)
|
||
cfg["token"] = secrets.token_hex(16)
|
||
_ensure_user_dirs(user_id)
|
||
_notif_file(user_id).write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||
return cfg["token"]
|
||
|
||
|
||
def _presence_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "presence.json"
|
||
|
||
|
||
def load_presence(user_id: str) -> dict:
|
||
f = _presence_file(user_id)
|
||
if f.exists():
|
||
return json.loads(f.read_text())
|
||
return {"present": False, "since": None}
|
||
|
||
|
||
def save_presence(user_id: str, present: bool, since: str | None):
|
||
_ensure_user_dirs(user_id)
|
||
_presence_file(user_id).write_text(
|
||
json.dumps({"present": present, "since": since}, indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
# ── Plages horaires (par utilisateur) ─────────────────────────────────────────
|
||
|
||
def _plages_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "plages.json"
|
||
|
||
|
||
def load_plages(user_id: str) -> dict:
|
||
"""Return this user's plage config, falling back to the (legacy) global config
|
||
default, then to DEFAULT_PLAGES, for any key not yet customized."""
|
||
base = {**DEFAULT_PLAGES, **load_config().get("plages", {})}
|
||
f = _plages_file(user_id)
|
||
if f.exists():
|
||
try:
|
||
return {**base, **json.loads(f.read_text())}
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return base
|
||
|
||
|
||
def save_plages(user_id: str, plages: dict):
|
||
_ensure_user_dirs(user_id)
|
||
_plages_file(user_id).write_text(json.dumps(plages, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def toggle_conge(date: str, type_conge: str, user_id: str = ""):
|
||
assert type_conge in ("jour", "matin", "aprem")
|
||
conges = load_conges(user_id)
|
||
already = any(c["date"] == date and c["type"] == type_conge for c in conges)
|
||
|
||
if already:
|
||
conges = [c for c in conges if not (c["date"] == date and c["type"] == type_conge)]
|
||
else:
|
||
if type_conge == "jour":
|
||
conges = [c for c in conges if c["date"] != date]
|
||
else:
|
||
conges = [c for c in conges if not (c["date"] == date and c["type"] == "jour")]
|
||
conges.append({"date": date, "type": type_conge})
|
||
|
||
conges.sort(key=lambda c: (c["date"], c["type"]))
|
||
save_conges(conges, user_id)
|
||
|
||
|
||
# ── Journaux (notifications + présence) ───────────────────────────────────────
|
||
|
||
LOG_TYPES = ("notif", "presence", "settings")
|
||
MAX_LOGS = 500
|
||
|
||
|
||
def _logs_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "logs.json"
|
||
|
||
|
||
def append_log(user_id: str, entry_type: str, event: str, message: str = "", ts: str | None = None):
|
||
"""Ajoute une entrée au journal de l'utilisateur (rotation sur MAX_LOGS)."""
|
||
assert entry_type in LOG_TYPES, f"type de journal invalide: {entry_type!r}"
|
||
f = _logs_file(user_id)
|
||
logs = []
|
||
if f.exists():
|
||
try:
|
||
logs = json.loads(f.read_text())
|
||
except json.JSONDecodeError:
|
||
logs = []
|
||
logs.append({
|
||
"ts": ts or datetime.now().isoformat(timespec="seconds"),
|
||
"type": entry_type,
|
||
"event": event,
|
||
"message": message,
|
||
})
|
||
if len(logs) > MAX_LOGS:
|
||
logs = logs[-MAX_LOGS:]
|
||
_ensure_user_dirs(user_id)
|
||
f.write_text(json.dumps(logs, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def load_logs(user_id: str) -> list[dict]:
|
||
"""Retourne les entrées du journal, des plus récentes aux plus anciennes."""
|
||
f = _logs_file(user_id)
|
||
if not f.exists():
|
||
return []
|
||
try:
|
||
logs = json.loads(f.read_text())
|
||
except json.JSONDecodeError:
|
||
return []
|
||
return list(reversed(logs))
|