Implémente système multi-utilisateurs avec classement
- Auth cookie simple (SSO-ready) : login/logout, redirection si non authentifié
- Données isolées par utilisateur dans /data/users/{user_id}/
- Migration automatique des données legacy au premier login
- Page classement avec podium et tableau comparatif (score, précision zéro, conformité)
- Nav mise à jour : Classement, affichage utilisateur courant, déconnexion
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
136
app/models.py
136
app/models.py
@ -1,25 +1,75 @@
|
||||
"""Flat-file storage: JSON files per week for pointages, one file for congés."""
|
||||
import json
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
DATA_DIR = Path("/data")
|
||||
POINTAGES_DIR = DATA_DIR / "pointages"
|
||||
CONGES_FILE = DATA_DIR / "conges.json"
|
||||
CONFIG_FILE = DATA_DIR / "config.json"
|
||||
|
||||
DEFAULT_CONFIG = {
|
||||
"heures_jour": 7.8, # 7h48 = 39h/semaine
|
||||
}
|
||||
|
||||
# ── User path helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
def _ensure_dirs():
|
||||
POINTAGES_DIR.mkdir(parents=True, exist_ok=True)
|
||||
def _user_root(user_id: str) -> Path:
|
||||
"""Return data directory for a user.
|
||||
|
||||
Legacy fallback: if no users/ directory exists yet but /data/pointages/ does,
|
||||
the first login migrates the data automatically (see migrate_legacy_to_user).
|
||||
After migration, each user gets /data/users/{user_id}/.
|
||||
"""
|
||||
return DATA_DIR / "users" / user_id
|
||||
|
||||
|
||||
# ---------- config ----------
|
||||
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)
|
||||
|
||||
|
||||
# ── Legacy migration (called once on first login) ─────────────────────────────
|
||||
|
||||
def needs_migration() -> bool:
|
||||
"""True when legacy flat data exists and no users/ directory has been created yet."""
|
||||
return (DATA_DIR / "pointages").exists() and not (DATA_DIR / "users").exists()
|
||||
|
||||
|
||||
def migrate_legacy_to_user(user_id: str):
|
||||
"""Move /data/pointages/ and /data/conges.json into /data/users/{user_id}/."""
|
||||
user_root = _user_root(user_id)
|
||||
user_root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
legacy_pt = DATA_DIR / "pointages"
|
||||
dest_pt = user_root / "pointages"
|
||||
if legacy_pt.exists() and not dest_pt.exists():
|
||||
shutil.copytree(legacy_pt, dest_pt)
|
||||
|
||||
legacy_cg = DATA_DIR / "conges.json"
|
||||
dest_cg = user_root / "conges.json"
|
||||
if legacy_cg.exists() and not dest_cg.exists():
|
||||
shutil.copy2(legacy_cg, dest_cg)
|
||||
|
||||
|
||||
# ── 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:
|
||||
_ensure_dirs()
|
||||
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
if CONFIG_FILE.exists():
|
||||
return json.loads(CONFIG_FILE.read_text())
|
||||
CONFIG_FILE.write_text(json.dumps(DEFAULT_CONFIG, indent=2))
|
||||
@ -27,78 +77,62 @@ def load_config() -> dict:
|
||||
|
||||
|
||||
def save_config(cfg: dict):
|
||||
_ensure_dirs()
|
||||
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
CONFIG_FILE.write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||||
|
||||
|
||||
# ---------- pointages ----------
|
||||
# ── Pointages ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _week_file(year: int, week: int) -> Path:
|
||||
return POINTAGES_DIR / f"{year}-W{week:02d}.json"
|
||||
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) -> list[dict]:
|
||||
f = _week_file(year, week)
|
||||
if f.exists():
|
||||
return json.loads(f.read_text())
|
||||
return []
|
||||
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]):
|
||||
_ensure_dirs()
|
||||
_week_file(year, week).write_text(json.dumps(pointages, indent=2, ensure_ascii=False))
|
||||
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 upsert_pointages(records: list[dict]):
|
||||
"""Insert or update pointage records, grouped by ISO week."""
|
||||
from datetime import datetime
|
||||
by_week: dict[tuple, dict[str, dict]] = {}
|
||||
for r in records:
|
||||
d = datetime.strptime(r["date"], "%Y-%m-%d").date()
|
||||
iso = d.isocalendar()
|
||||
key = (iso.year, iso.week)
|
||||
by_week.setdefault(key, {})
|
||||
by_week[key][r["date"]] = r
|
||||
|
||||
for (year, week), new_by_date in by_week.items():
|
||||
existing = {p["date"]: p for p in load_week_pointages(year, week)}
|
||||
existing.update(new_by_date)
|
||||
save_week_pointages(year, week, sorted(existing.values(), key=lambda p: p["date"]))
|
||||
|
||||
|
||||
def list_all_weeks() -> list[tuple[int, int]]:
|
||||
"""Return sorted list of (year, week) for which we have data."""
|
||||
_ensure_dirs()
|
||||
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 POINTAGES_DIR.glob("????-W??.json"):
|
||||
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 ----------
|
||||
# ── Congés ────────────────────────────────────────────────────────────────────
|
||||
|
||||
def load_conges() -> list[dict]:
|
||||
if CONGES_FILE.exists():
|
||||
return json.loads(CONGES_FILE.read_text())
|
||||
return []
|
||||
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]):
|
||||
_ensure_dirs()
|
||||
CONGES_FILE.write_text(json.dumps(conges, indent=2, ensure_ascii=False))
|
||||
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)
|
||||
)
|
||||
|
||||
|
||||
def toggle_conge(date: str, type_conge: str):
|
||||
def toggle_conge(date: str, type_conge: str, user_id: str = ""):
|
||||
assert type_conge in ("jour", "matin", "aprem")
|
||||
conges = load_conges()
|
||||
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:
|
||||
# Adding: remove conflicting entries first
|
||||
if type_conge == "jour":
|
||||
conges = [c for c in conges if c["date"] != date]
|
||||
else:
|
||||
@ -106,4 +140,4 @@ def toggle_conge(date: str, type_conge: str):
|
||||
conges.append({"date": date, "type": type_conge})
|
||||
|
||||
conges.sort(key=lambda c: (c["date"], c["type"]))
|
||||
save_conges(conges)
|
||||
save_conges(conges, user_id)
|
||||
|
||||
Reference in New Issue
Block a user