55efac9cdd
Toute modification des paramètres (topic/serveur/jeton ntfy, plages
horaires, régénération du token de présence Tasker) est désormais
enregistrée dans /logs sous un nouveau type "Réglages", avec un onglet
dédié. Chaque entrée résume ce qui a changé (valeurs des plages, plage
des rappels, topic, jeton défini/vide, URLs invalidées), pour aider à
diagnostiquer a posteriori un problème de configuration.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
315 lines
11 KiB
Python
315 lines
11 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
|
|
|
|
DATA_DIR = Path("/data")
|
|
CONFIG_FILE = DATA_DIR / "config.json"
|
|
|
|
DEFAULT_CONFIG = {
|
|
"heures_jour": 7.8, # 7h48 = 39h/semaine
|
|
"log_level": "DEBUG", # DEBUG, INFO, WARNING ou ERROR
|
|
"allowed_email_domain": "", # ex: "sdv.fr" — seules ces adresses peuvent se connecter
|
|
"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),
|
|
}
|
|
|
|
# ── 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 heures_jour_min() -> int:
|
|
return int(float(load_config().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)
|
|
)
|
|
|
|
|
|
# ── 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))
|