""" Meeconic Portal Backend — module admin / écoles (multi-école). Authentification (JWT signé) : POST /api/admin/login {password, school?} → rôle operator (mot de passe ADMIN_PASSWORD) ou director (mot de passe de l'école + code école) Écoles (operator) : GET /api/admin/schools → liste + chiffres clés POST /api/admin/schools → créer une école PATCH /api/admin/schools/{id} → renommer / (dés)activer / mot de passe directeur GET /api/admin/system → état du système (alertes, backup, organizer…) Toutes les routes ci-dessous sont limitées à UNE école : le directeur est enfermé dans la sienne (jeton), l'opérateur choisit via l'en-tête X-School. GET /api/admin/settings, PUT /api/admin/settings GET /api/admin/stats | /orders | /families GET|POST /api/admin/classes, PATCH|DELETE /api/admin/classes/{id} → classes de l'école (saisies ici) POST /api/admin/families, PATCH /api/admin/families/{code} → rattacher un code famille à ses classes PATCH /api/admin/orders/{id} GET /api/admin/export/printer.csv | labels | delivery-slip GET /api/admin/audit | /audit/export.csv GET /api/director/dashboard GET /api/qr.svg → QR code SVG (public) Seules les commandes PAYÉES comptent dans les chiffres, exports et lots d'impression. """ import asyncio import csv import io import json import logging import os import re import shutil import unicodedata import zipfile from dataclasses import dataclass from datetime import date, timedelta from typing import Optional import jwt from fastapi import APIRouter, File, Header, HTTPException, Query, Request, UploadFile from fastapi.responses import FileResponse, PlainTextResponse, Response from pydantic import BaseModel, Field from core import config from core.db import get_db, ensure_column, TABLE_OPTIONS from core.rate_limit import enforce as rate_limit_enforce from core.security import client_ip, constant_time_eq, jwt_secret, verify_password from core.timeutil import utcnow, utcnow_iso from services import ( ai_health, audit_service, card_service, media_service, photo_service, photographer_service, protection_service, school_service, ) from utils.code_generator import generate_unique_code log = logging.getLogger("meeconic.admin") JWT_ALGORITHM = "HS256" JWT_TTL_HOURS = 8 PAID_SQL = "payment_status='paid' AND status != 'cancelled'" router = APIRouter() # ── Schéma ──────────────────────────────────────────────────────────────────── def init_admin_db() -> None: school_service.init_school_tables() with get_db() as conn: conn.execute(f""" CREATE TABLE IF NOT EXISTS families ( code VARCHAR(32) NOT NULL PRIMARY KEY, family_name VARCHAR(200) NOT NULL DEFAULT '', -- legacy: schools may not send names, never filled again email VARCHAR(200), created_at VARCHAR(40), consent_at VARCHAR(40), consent_ip VARCHAR(64), school_id VARCHAR(64), KEY idx_families_school (school_id) ) {TABLE_OPTIONS}""") conn.execute(f""" CREATE TABLE IF NOT EXISTS students ( id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, family_code VARCHAR(32) NOT NULL, first_name VARCHAR(100) NOT NULL DEFAULT '', -- legacy, same reason last_name VARCHAR(100) NOT NULL DEFAULT '', class_name VARCHAR(100) NOT NULL, KEY idx_students_family (family_code), KEY idx_students_class (class_name) ) {TABLE_OPTIONS}""") conn.execute(f""" CREATE TABLE IF NOT EXISTS school_classes ( id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, school_id VARCHAR(64) NOT NULL, name VARCHAR(40) NOT NULL, student_count INT, UNIQUE KEY uq_school_class (school_id, name) ) {TABLE_OPTIONS}""") conn.execute(f""" CREATE TABLE IF NOT EXISTS school_products ( id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, school_id VARCHAR(64) NOT NULL, name_fr VARCHAR(80) NOT NULL, name_en VARCHAR(80) NOT NULL DEFAULT '', name_es VARCHAR(80) NOT NULL DEFAULT '', price DOUBLE NOT NULL, physical TINYINT NOT NULL DEFAULT 1, active TINYINT NOT NULL DEFAULT 1, has_image TINYINT NOT NULL DEFAULT 0, sort_order INT NOT NULL DEFAULT 0, created_at VARCHAR(40) NOT NULL, KEY idx_products_school (school_id) ) {TABLE_OPTIONS}""") ensure_column(conn, "orders", "class_name", "VARCHAR(100)") # Installations that came with classes only on the attachment rows get them as configured classes. conn.execute( "INSERT IGNORE INTO school_classes (school_id, name) " "SELECT DISTINCT f.school_id, s.class_name FROM students s JOIN families f ON f.code = s.family_code " "WHERE f.school_id IS NOT NULL AND f.school_id != ''") conn.commit() audit_service.init_audit_db() log.info("Admin tables ready") # ── Auth (JWT signés, sans état) ────────────────────────────────────────────── class LoginRequest(BaseModel): password: str = Field(max_length=200) school: Optional[str] = Field(default=None, max_length=60) def _issue(role: str, school: Optional[str]) -> str: now = utcnow() payload = {"role": role, "iat": now, "exp": now + timedelta(hours=JWT_TTL_HOURS)} if school: payload["school"] = school return jwt.encode(payload, jwt_secret(), algorithm=JWT_ALGORITHM) @router.post("/api/admin/login") def admin_login(body: LoginRequest, request: Request): ip = client_ip(request) # 5 tentatives / 15 min par IP — freine le brute-force sur les mots de passe. rate_limit_enforce(f"admin-login:{ip}", max_attempts=5, window_seconds=900, audit_action="auth.login_bruteforce_blocked", ip_address=ip) pw = body.password.strip() slug = (body.school or "").strip().lower() if config.ADMIN_PASSWORD and constant_time_eq(pw, config.ADMIN_PASSWORD): audit_service.log_event("auth.login", user_id="operator", ip_address=ip, severity="info") return {"token": _issue("operator", None), "role": "operator", "school": None} if slug: sch = school_service.get_school_with_hash(slug) if sch and sch["active"] and verify_password(pw, sch["director_password_hash"]): audit_service.log_event("auth.login", user_id="director", ip_address=ip, severity="info", school_id=slug) return {"token": _issue("director", slug), "role": "director", "school": slug} audit_service.log_event("auth.login_failed", ip_address=ip, severity="warning", details={"school": slug or None}) raise HTTPException(401, detail="Mot de passe incorrect.") @dataclass class Ctx: role: str school_id: Optional[str] def get_ctx(authorization: Optional[str], x_school: Optional[str], *roles: str, need_school: bool = True) -> Ctx: if not authorization or not authorization.startswith("Bearer "): raise HTTPException(401, detail="Token manquant.") try: payload = jwt.decode(authorization[7:], jwt_secret(), algorithms=[JWT_ALGORITHM]) except jwt.ExpiredSignatureError: raise HTTPException(401, detail="Session expired — please sign in again.") except jwt.InvalidTokenError: raise HTTPException(401, detail="Session invalide — reconnectez-vous.") role = payload.get("role") if roles and role not in roles: raise HTTPException(403, detail="Access denied for this role.") school_id: Optional[str] = None if role == "director": school_id = payload.get("school") sch = school_service.get_school(school_id) if school_id else None if not sch or not sch["active"]: raise HTTPException(401, detail="School disabled or not found.") elif x_school: school_id = x_school.strip().lower() if not school_service.get_school(school_id): raise HTTPException(404, detail="School not found.") if need_school and not school_id: raise HTTPException(400, detail="Select a school.") return Ctx(role=role, school_id=school_id) # ── Code famille ────────────────────────────────────────────────────────────── def _strip_accents(s: str) -> str: return "".join(c for c in unicodedata.normalize("NFD", s) if unicodedata.category(c) != "Mn") def new_family_code(conn, school_id: str) -> str: """Código único con el prefijo del colegio: MAR-4JC-IBD (887 millones de combinaciones por prefijo).""" return generate_unique_code(conn, school_service.code_prefix_for(school_id)) # ── Tri des classes (ordre scolaire français : CP, CE1, CE2, CM1, CM2…) ────── _LEVEL_ORDER = ["PS", "MS", "GS", "CP", "CE1", "CE2", "CM1", "CM2", "6E", "5E", "4E", "3E"] def class_sort_key(name: str): upper = name.upper() for i, lvl in enumerate(_LEVEL_ORDER): if upper.startswith(lvl) and (len(upper) == len(lvl) or not upper[len(lvl)].isdigit()): return (0, i, upper) return (1, 0, upper) # ── Helpers commandes ───────────────────────────────────────────────────────── def family_students_map(conn, sid: str) -> dict[str, list[dict]]: """family code → one entry per attached child, which carries ONLY the class (the schools may not give us names).""" out: dict[str, list[dict]] = {} for r in conn.execute( "SELECT s.family_code, s.class_name FROM students s " "JOIN families f ON f.code = s.family_code WHERE f.school_id=? ORDER BY s.id", (sid,)): out.setdefault(r["family_code"], []).append({"class_name": r["class_name"]}) return out def order_classes(row, fam_students: dict) -> list[str]: """The class the parent chose at checkout; otherwise the classes the family code is attached to.""" if row["class_name"]: return [row["class_name"]] return sorted({s["class_name"] for s in fam_students.get(row["family_code"], [])}, key=class_sort_key) def school_class_names(conn, sid: str) -> list[str]: return sorted((r["name"] for r in conn.execute("SELECT name FROM school_classes WHERE school_id=?", (sid,))), key=class_sort_key) def paid_orders(conn, sid: str, order_by: str = "") -> list: return conn.execute( f"SELECT * FROM orders WHERE school_id=? AND {PAID_SQL} {order_by}", (sid,)).fetchall() def item_buckets(items: list[dict]) -> dict: """Répartit le montant d'une commande : papier / magique IA / numérique. Compte les tirages.""" paper = magic = digital = 0.0 prints = 0 ai = 0 for it in items: amount = it["unit_price"] * it["quantity"] fid = it["format_id"] if it.get("is_magic"): magic += amount ai += it["quantity"] if fid in ("magic-print", "magic-pack"): prints += it["quantity"] elif fid == "digital": digital += amount else: paper += amount if fid in ("print-10x15", "print-20x30", "pack"): prints += it["quantity"] return {"paper": paper, "magic": magic, "digital": digital, "prints": prints, "ai": ai} def order_row_to_admin(row, fam_students: dict) -> dict: items = json.loads(row["items_json"]) classes = order_classes(row, fam_students) b = item_buckets(items) n_prints = b["prints"] n_digital = sum(i["quantity"] for i in items if "digital" in i["format_id"] or i["format_id"] == "pack") summary_parts = [] if n_prints: ai_part = f" (dont {b['ai']} IA)" if b["ai"] else "" summary_parts.append(f"{n_prints} tirage{'s' if n_prints > 1 else ''}{ai_part}") if n_digital: summary_parts.append("numérique") return { "id": row["id"], "short_id": row["id"][:8].upper(), "family_code": row["family_code"], "classes": classes, "items": items, "items_summary": " + ".join(summary_parts) or f"{len(items)} article(s)", "total": row["total"], "status": row["status"], "payment_status": row["payment_status"], "batch_status": row["batch_status"], "customer_email": row["customer_email"], "created_at": row["created_at"], "prints": n_prints, "ai": b["ai"], } def _count_photos(school_id: str, codes: list[str]) -> int: return sum(photo_service.count_family_photos(school_id, c) for c in codes) # ── Écoles (operator) ───────────────────────────────────────────────────────── class SchoolCreate(BaseModel): name: str = Field(min_length=1, max_length=120) slug: Optional[str] = Field(default=None, max_length=40) director_password: Optional[str] = Field(default=None, max_length=200) class SchoolPatch(BaseModel): name: Optional[str] = Field(default=None, max_length=120) active: Optional[bool] = None director_password: Optional[str] = Field(default=None, max_length=200) @router.get("/api/admin/schools") def list_schools(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): get_ctx(authorization, x_school, "operator", need_school=False) out = [] with get_db() as conn: for s in school_service.list_schools(): sid = s["id"] fam = conn.execute("SELECT COUNT(*) AS n FROM families WHERE school_id=?", (sid,)).fetchone()["n"] stu = conn.execute("SELECT COUNT(*) AS n FROM students s JOIN families f ON f.code=s.family_code " "WHERE f.school_id=?", (sid,)).fetchone()["n"] o = conn.execute(f"SELECT COUNT(*) AS n, COALESCE(SUM(total),0) AS rev FROM orders " f"WHERE school_id=? AND {PAID_SQL}", (sid,)).fetchone() out.append({**s, "families": fam, "students": stu, "orders": o["n"], "revenue": round(o["rev"])}) return {"schools": out} @router.post("/api/admin/schools") def create_school(body: SchoolCreate, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): get_ctx(authorization, x_school, "operator", need_school=False) try: school = school_service.create_school(body.name, body.slug, body.director_password or None) except school_service.SettingsError as e: raise HTTPException(400, detail=str(e)) audit_service.log_event("school.create", user_id="operator", ip_address=client_ip(request), resource_type="school", resource_id=school["id"], school_id=school["id"], details={"name": school["name"]}) return {"school": school} @router.patch("/api/admin/schools/{school_id}") def patch_school(school_id: str, body: SchoolPatch, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): get_ctx(authorization, x_school, "operator", need_school=False) try: school = school_service.update_school(school_id, name=body.name, active=body.active, director_password=body.director_password) except school_service.SettingsError as e: raise HTTPException(400, detail=str(e)) audit_service.log_event( "school.update", user_id="operator", ip_address=client_ip(request), resource_type="school", resource_id=school_id, school_id=school_id, severity="warning", details={"name": body.name, "active": body.active, "director_password_changed": body.director_password is not None}) return {"school": school} # ── Réglages de l'école ─────────────────────────────────────────────────────── class SettingsUpdate(BaseModel): settings: dict[str, str | int | float | bool | None] @router.get("/api/admin/settings") def admin_settings(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator", "director") fields = [{"key": k, "type": v["type"], "min": v.get("min"), "max": v.get("max")} for k, v in school_service.SETTINGS_SPEC.items()] return {"settings": school_service.get_settings(ctx.school_id), "school": school_service.get_school(ctx.school_id), "fields": fields} @router.put("/api/admin/settings") def admin_update_settings(body: SettingsUpdate, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") before = school_service.get_settings(ctx.school_id) try: after = school_service.update_settings(ctx.school_id, body.settings) except school_service.SettingsError as e: raise HTTPException(400, detail=str(e)) changed = {k: {"from": before.get(k), "to": after.get(k)} for k in body.settings if before.get(k) != after.get(k)} if changed: audit_service.log_event("settings.update", user_id=ctx.role, ip_address=client_ip(request), resource_type="school", resource_id=ctx.school_id, school_id=ctx.school_id, details=changed, severity="warning") return {"settings": after} # ── KPIs / stats opérateur ──────────────────────────────────────────────────── @router.get("/api/admin/stats") def admin_stats(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id with get_db() as conn: settings = school_service.get_settings(sid, conn) rows = paid_orders(conn, sid) codes = [r["code"] for r in conn.execute("SELECT code FROM families WHERE school_id=?", (sid,))] n_families = len(codes) n_students = conn.execute("SELECT COUNT(*) AS n FROM students s JOIN families f ON f.code=s.family_code " "WHERE f.school_id=?", (sid,)).fetchone()["n"] unpaid = conn.execute("SELECT COUNT(*) AS n FROM orders WHERE school_id=? AND payment_status IN ('unpaid','failed') " "AND status='pending'", (sid,)).fetchone()["n"] revenue = sum(r["total"] for r in rows) prints = ai = 0 buyers = set() new_hour = 0 hour_ago = (utcnow() - timedelta(hours=1)).isoformat() for r in rows: b = item_buckets(json.loads(r["items_json"])) prints += b["prints"] ai += b["ai"] buyers.add(r["family_code"]) if r["created_at"] >= hour_ago: new_hour += 1 closure = settings.get("closure_date", "") days_left = None try: days_left = (date.fromisoformat(closure) - utcnow().date()).days except Exception: pass share_pct = float(settings.get("revenue_share_pct") or 0) return { "school_id": sid, "school_name": settings.get("school_name"), "orders_count": len(rows), "pending_unpaid": unpaid, "new_last_hour": new_hour, "revenue": revenue, "buyer_rate": round(100 * len(buyers) / n_families) if n_families else 0, "prints_total": prints, "ai_total": ai, "closure_date": closure, "days_to_closure": days_left, "families_total": n_families, "students_total": n_students, "photos_received": _count_photos(sid, codes), "revenue_share_pct": share_pct, "revenue_share": round(revenue * share_pct / 100), "settings": settings, } # ── Commandes admin ─────────────────────────────────────────────────────────── @router.get("/api/admin/orders") def admin_orders( class_name: Optional[str] = Query(None), payment: str = Query("paid", pattern="^(paid|all)$"), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None), ): ctx = get_ctx(authorization, x_school, "operator") with get_db() as conn: fam_students = family_students_map(conn, ctx.school_id) if payment == "paid": rows = conn.execute(f"SELECT * FROM orders WHERE school_id=? AND {PAID_SQL} ORDER BY created_at DESC", (ctx.school_id,)).fetchall() else: rows = conn.execute("SELECT * FROM orders WHERE school_id=? ORDER BY created_at DESC", (ctx.school_id,)).fetchall() orders = [order_row_to_admin(r, fam_students) for r in rows] if class_name: orders = [o for o in orders if class_name in o["classes"]] return {"orders": orders} class BatchStatusRequest(BaseModel): batch_status: str @router.patch("/api/admin/orders/{order_id}") def admin_update_order(order_id: str, body: BatchStatusRequest, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") allowed = {"to_print", "printed", "labeled", "delivered", "cancelled"} if body.batch_status not in allowed: raise HTTPException(400, detail=f"Statut invalide. Valeurs : {sorted(allowed)}") with get_db() as conn: row = conn.execute("SELECT id, payment_status FROM orders WHERE id=? AND school_id=?", (order_id, ctx.school_id)).fetchone() if not row: raise HTTPException(404, detail="Commande introuvable.") conn.execute("UPDATE orders SET batch_status=? WHERE id=?", (body.batch_status, order_id)) if body.batch_status == "delivered": conn.execute("UPDATE orders SET status='delivered' WHERE id=?", (order_id,)) elif body.batch_status == "cancelled": conn.execute("UPDATE orders SET status='cancelled' WHERE id=?", (order_id,)) conn.commit() audit_service.log_event("order.batch_status", user_id=ctx.role, ip_address=client_ip(request), resource_type="order", resource_id=order_id, school_id=ctx.school_id, details={"batch_status": body.batch_status}, severity="warning" if body.batch_status == "cancelled" else "info") return {"success": True, "batch_status": body.batch_status} # ── Classes ─────────────────────────────────────────────────────────────────── def classes_summary(conn, sid: str) -> list[dict]: fam_students = family_students_map(conn, sid) rows = paid_orders(conn, sid) cls: dict[str, dict] = {} def entry(name: str, class_id=None, expected=None) -> dict: return cls.setdefault(name, { "id": class_id, "class_name": name, "student_count": expected, "students": 0, "families": 0, "orders": 0, "sales": 0.0, "buyer_families": set(), "prints": 0, }) for r in conn.execute("SELECT id, name, student_count FROM school_classes WHERE school_id=?", (sid,)): entry(r["name"], r["id"], r["student_count"]) for r in conn.execute( "SELECT s.class_name, COUNT(*) AS n, COUNT(DISTINCT s.family_code) AS f FROM students s " "JOIN families fa ON fa.code = s.family_code WHERE fa.school_id=? GROUP BY s.class_name", (sid,)): c = entry(r["class_name"]) c["students"], c["families"] = r["n"], r["f"] for r in rows: b = item_buckets(json.loads(r["items_json"])) # The class the parent picked at checkout is exact; a family attached to several classes (siblings) # without one splits the sale between them. classes = order_classes(r, fam_students) if not classes: continue share = r["total"] / len(classes) for name in classes: c = cls.get(name) if not c: continue c["sales"] += share c["orders"] += 1 c["buyer_families"].add(r["family_code"]) first = cls.get(classes[0]) if first: first["prints"] += b["prints"] out = [] for c in sorted(cls.values(), key=lambda x: class_sort_key(x["class_name"])): n_buyers = len(c.pop("buyer_families")) c["participation"] = round(100 * n_buyers / c["families"]) if c["families"] else 0 c["sales"] = round(c["sales"]) out.append(c) return out @router.get("/api/admin/classes") def admin_classes(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator", "director") with get_db() as conn: return {"classes": classes_summary(conn, ctx.school_id)} class ClassIn(BaseModel): name: str = Field(min_length=1, max_length=40) student_count: Optional[int] = Field(default=None, ge=0, le=5000) class ClassPatch(BaseModel): name: Optional[str] = Field(default=None, min_length=1, max_length=40) student_count: Optional[int] = Field(default=None, ge=0, le=5000) def _clean_class(name: str) -> str: return re.sub(r"\s+", " ", name).strip() @router.post("/api/admin/classes") def create_class(body: ClassIn, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") name = _clean_class(body.name) if not name: raise HTTPException(400, detail="Class name required.") with get_db() as conn: if conn.execute("SELECT 1 FROM school_classes WHERE school_id=? AND name=?", (ctx.school_id, name)).fetchone(): raise HTTPException(409, detail="This class already exists.") cur = conn.execute("INSERT INTO school_classes (school_id, name, student_count) VALUES (?,?,?)", (ctx.school_id, name, body.student_count)) conn.commit() new_id = cur.lastrowid audit_service.log_event("class.create", user_id=ctx.role, ip_address=client_ip(request), school_id=ctx.school_id, details={"class_name": name, "student_count": body.student_count}) return {"class": {"id": new_id, "class_name": name, "student_count": body.student_count}} @router.patch("/api/admin/classes/{class_id}") def update_class(class_id: int, body: ClassPatch, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id with get_db() as conn: cur = conn.execute("SELECT * FROM school_classes WHERE id=? AND school_id=?", (class_id, sid)).fetchone() if not cur: raise HTTPException(404, detail="Class not found.") name = cur["name"] if body.name is not None: new_name = _clean_class(body.name) if not new_name: raise HTTPException(400, detail="Class name required.") if new_name != name: if conn.execute("SELECT 1 FROM school_classes WHERE school_id=? AND name=?", (sid, new_name)).fetchone(): raise HTTPException(409, detail="This class already exists.") conn.execute("UPDATE school_classes SET name=? WHERE id=?", (new_name, class_id)) # The name is what families and orders point at: they move with it. codes = [r["code"] for r in conn.execute("SELECT code FROM families WHERE school_id=?", (sid,))] for code in codes: conn.execute("UPDATE students SET class_name=? WHERE family_code=? AND class_name=?", (new_name, code, name)) conn.execute("UPDATE orders SET class_name=? WHERE class_name=? AND school_id=?", (new_name, name, sid)) name = new_name if "student_count" in body.model_fields_set: conn.execute("UPDATE school_classes SET student_count=? WHERE id=?", (body.student_count, class_id)) conn.commit() audit_service.log_event("class.update", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, details={"id": class_id, "class_name": name}) return {"success": True, "class_name": name} @router.delete("/api/admin/classes/{class_id}") def delete_class(class_id: int, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id with get_db() as conn: cur = conn.execute("SELECT * FROM school_classes WHERE id=? AND school_id=?", (class_id, sid)).fetchone() if not cur: raise HTTPException(404, detail="Class not found.") attached = conn.execute( "SELECT COUNT(*) AS n FROM students s JOIN families f ON f.code=s.family_code " "WHERE f.school_id=? AND s.class_name=?", (sid, cur["name"])).fetchone()["n"] if attached: raise HTTPException(409, detail=f"{attached} family code(s) are attached to this class — move them first.") conn.execute("DELETE FROM school_classes WHERE id=?", (class_id,)) conn.commit() audit_service.log_event("class.delete", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, details={"class_name": cur["name"]}, severity="warning") return {"success": True} @router.delete("/api/admin/families/{code}") def delete_family(code: str, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Remove a family code (a wrong one, a test, or what blocks deleting a class). A code with PAID orders is never removed: it is the proof of a sale. The photos go to the restorable trash, not to /dev/null.""" ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id code = code.strip().upper() with get_db() as conn: if not conn.execute("SELECT 1 FROM families WHERE code=? AND school_id=?", (code, sid)).fetchone(): raise HTTPException(404, detail="Family not found.") paid = conn.execute(f"SELECT COUNT(*) AS n FROM orders WHERE family_code=? AND {PAID_SQL}", (code,)).fetchone()["n"] if paid: raise HTTPException(409, detail=f"This code has {paid} paid order(s) and cannot be deleted.") dropped = conn.execute("DELETE FROM orders WHERE family_code=? AND school_id=?", (code, sid)).rowcount conn.execute("DELETE FROM students WHERE family_code=?", (code,)) conn.execute("DELETE FROM families WHERE code=?", (code,)) conn.commit() photos = photo_service.delete_family_photos(sid, code) photo_service.invalidate_families_cache(sid) audit_service.log_event("family.delete", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, resource_type="family", resource_id=code, details={"unpaid_orders_removed": dropped, "files_to_trash": photos}, severity="critical") return {"success": True, "photos_moved_to_trash": photos} # ── Photographer app (/shoot): private link and class QR payloads ───────────────────────────────── @router.get("/api/admin/shoot") def admin_shoot(request: Request, base: Optional[str] = Query(None, max_length=200), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id from services import shoot_service token = shoot_service.get_token(sid) portal = _portal_base(school_service.get_settings(sid), request, base) with get_db() as conn: classes = school_class_names(conn, sid) return { "enabled": school_service.get_settings(sid).get("shoot_mode") == "true", "link": f"{portal}/shoot?t={token}" if token else None, "classes": [{"class_name": c, "qr": shoot_service.class_qr_payload(sid, c)} for c in classes], } @router.post("/api/admin/shoot/token") def admin_shoot_token(request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """(Re)generate the phone link. The previous one stops working at once: use it when a phone is lost.""" ctx = get_ctx(authorization, x_school, "operator") from services import shoot_service shoot_service.rotate_token(ctx.school_id) audit_service.log_event("shoot.token_rotated", user_id=ctx.role, ip_address=client_ip(request), school_id=ctx.school_id, severity="warning") return {"success": True} # ── Catalogue produits (porte-clés, mugs…) avec visuel d'exemple ───────────────── MAX_PRODUCT_IMAGE_BYTES = 6 * 1024 * 1024 PRODUCT_IMAGE_PX = 900 def product_image_path(school_id: str, product_id: int): return config.PRODUCTS_DIR / school_id / f"{int(product_id)}.webp" def product_to_dict(row, school_id: str) -> dict: return { "id": row["id"], "name_fr": row["name_fr"], "name_en": row["name_en"], "name_es": row["name_es"], "price": row["price"], "physical": bool(row["physical"]), "active": bool(row["active"]), "image_url": f"/api/portal/product-image/{school_id}/{row['id']}.webp?v={row['has_image']}" if row["has_image"] else None, } class ProductIn(BaseModel): name_fr: str = Field(min_length=1, max_length=80) name_en: str = Field(default="", max_length=80) name_es: str = Field(default="", max_length=80) price: float = Field(ge=0, le=10000) physical: bool = True class ProductPatch(BaseModel): name_fr: Optional[str] = Field(default=None, min_length=1, max_length=80) name_en: Optional[str] = Field(default=None, max_length=80) name_es: Optional[str] = Field(default=None, max_length=80) price: Optional[float] = Field(default=None, ge=0, le=10000) physical: Optional[bool] = None active: Optional[bool] = None @router.get("/api/admin/products") def list_products(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") with get_db() as conn: rows = conn.execute("SELECT * FROM school_products WHERE school_id=? ORDER BY sort_order, id", (ctx.school_id,)).fetchall() return {"products": [product_to_dict(r, ctx.school_id) for r in rows]} @router.post("/api/admin/products") def create_product(body: ProductIn, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id with get_db() as conn: cur = conn.execute( "INSERT INTO school_products (school_id, name_fr, name_en, name_es, price, physical, created_at) " "VALUES (?,?,?,?,?,?,?)", (sid, body.name_fr.strip(), body.name_en.strip(), body.name_es.strip(), round(body.price, 2), int(body.physical), utcnow_iso())) conn.commit() row = conn.execute("SELECT * FROM school_products WHERE id=?", (cur.lastrowid,)).fetchone() audit_service.log_event("product.create", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, details={"id": row["id"], "name": row["name_fr"], "price": row["price"]}) return {"product": product_to_dict(row, sid)} @router.patch("/api/admin/products/{product_id}") def update_product(product_id: int, body: ProductPatch, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id fields = {k: v for k, v in body.model_dump(exclude_unset=True).items() if v is not None} if "price" in fields: fields["price"] = round(fields["price"], 2) for flag in ("physical", "active"): if flag in fields: fields[flag] = int(fields[flag]) for k in ("name_fr", "name_en", "name_es"): if k in fields: fields[k] = fields[k].strip() if fields.get("name_fr") == "": raise HTTPException(400, detail="Product name required.") with get_db() as conn: if not conn.execute("SELECT 1 FROM school_products WHERE id=? AND school_id=?", (product_id, sid)).fetchone(): raise HTTPException(404, detail="Product not found.") if fields: sets = ", ".join(f"{k}=?" for k in fields) # keys come from the model above, never from the client conn.execute(f"UPDATE school_products SET {sets} WHERE id=?", (*fields.values(), product_id)) conn.commit() row = conn.execute("SELECT * FROM school_products WHERE id=?", (product_id,)).fetchone() audit_service.log_event("product.update", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, details={"id": product_id, **{k: v for k, v in fields.items()}}) return {"product": product_to_dict(row, sid)} @router.delete("/api/admin/products/{product_id}") def delete_product(product_id: int, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id with get_db() as conn: row = conn.execute("SELECT * FROM school_products WHERE id=? AND school_id=?", (product_id, sid)).fetchone() if not row: raise HTTPException(404, detail="Product not found.") conn.execute("DELETE FROM school_products WHERE id=?", (product_id,)) conn.commit() try: product_image_path(sid, product_id).unlink(missing_ok=True) except OSError: pass audit_service.log_event("product.delete", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, details={"id": product_id, "name": row["name_fr"]}, severity="warning") return {"success": True} @router.post("/api/admin/products/{product_id}/image") async def upload_product_image(product_id: int, request: Request, file: UploadFile = File(...), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Example visual of the product. Re-encoded to WebP: whatever the client sends, only an image we produced is ever served (no SVG/HTML smuggled through, no EXIF).""" # This handler stays `async` only because the upload is read with `await`; every blocking step # (auth lookups, MySQL, disk) goes through a worker thread. ctx = await asyncio.to_thread(get_ctx, authorization, x_school, "operator") sid = ctx.school_id def _exists() -> bool: with get_db() as conn: return bool(conn.execute("SELECT 1 FROM school_products WHERE id=? AND school_id=?", (product_id, sid)).fetchone()) if not await asyncio.to_thread(_exists): raise HTTPException(404, detail="Product not found.") raw = await file.read(MAX_PRODUCT_IMAGE_BYTES + 1) if len(raw) > MAX_PRODUCT_IMAGE_BYTES: raise HTTPException(413, detail="Image too large (max 6 MB).") def convert() -> bytes: from PIL import Image, ImageOps with Image.open(io.BytesIO(raw)) as im: im = ImageOps.exif_transpose(im) im.thumbnail((PRODUCT_IMAGE_PX, PRODUCT_IMAGE_PX)) out = io.BytesIO() im.convert("RGBA" if "A" in im.getbands() else "RGB").save(out, "WEBP", quality=84) return out.getvalue() try: data = await asyncio.to_thread(convert) except Exception: raise HTTPException(400, detail="Not a valid image.") ip = client_ip(request) def _save() -> dict: path = product_image_path(sid, product_id) path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) with get_db() as conn: conn.execute("UPDATE school_products SET has_image=has_image+1 WHERE id=?", (product_id,)) # cache-buster conn.commit() row = conn.execute("SELECT * FROM school_products WHERE id=?", (product_id,)).fetchone() audit_service.log_event("product.image", user_id=ctx.role, ip_address=ip, school_id=sid, details={"id": product_id, "bytes": len(data)}) return product_to_dict(row, sid) return {"product": await asyncio.to_thread(_save)} @router.get("/api/portal/product-image/{school_id}/{product_id}.webp") def product_image(school_id: str, product_id: int): """Public: a catalogue picture is marketing material, not a family's photo.""" with get_db() as conn: row = conn.execute("SELECT 1 FROM school_products WHERE id=? AND school_id=? AND active=1 AND has_image>0", (product_id, school_id)).fetchone() path = product_image_path(school_id, product_id) if not row or not path.is_file(): raise HTTPException(404, detail="Image not found.") return FileResponse(path, media_type="image/webp", headers={"Cache-Control": "public, max-age=86400"}) # ── Familles / cartes ───────────────────────────────────────────────────────── @router.get("/api/admin/families") def admin_families(q: Optional[str] = Query(None), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id with get_db() as conn: fam_students = family_students_map(conn, sid) buyers = {r["family_code"] for r in conn.execute( f"SELECT DISTINCT family_code FROM orders WHERE school_id=? AND {PAID_SQL}", (sid,))} rows = conn.execute("SELECT * FROM families WHERE school_id=? ORDER BY code", (sid,)).fetchall() out = [] for r in rows: classes = [s["class_name"] for s in fam_students.get(r["code"], [])] fam = { "code": r["code"], "email": r["email"], "classes": classes, "is_sibling": len(classes) > 1, "has_orders": r["code"] in buyers, "card_url": media_service.card_url(sid, r["code"]), "created_at": r["created_at"] or "", } if q and q.lower() not in f"{fam['code']} {fam['email'] or ''} {' '.join(classes)}".lower(): continue out.append(fam) # Class, then the order the children were photographed in: the teacher hands the sheets out in that order. out.sort(key=lambda f: (class_sort_key(f["classes"][0] if f["classes"] else "~"), f["created_at"], f["code"])) return {"families": out, "total": len(out)} # The schools may not send names, so there is no student import: the operator types the classes (see # /api/admin/classes) and attaches each family code to its class(es) here. class FamilyCreate(BaseModel): classes: list[str] = Field(min_length=1, max_length=12) email: Optional[str] = Field(default=None, max_length=200) code: Optional[str] = Field(default=None, max_length=24) class FamilyPatch(BaseModel): classes: Optional[list[str]] = Field(default=None, min_length=1, max_length=12) email: Optional[str] = Field(default=None, max_length=200) _EMAIL_OK = re.compile(r"[^@\s]+@[^@\s]+\.[^@\s]+") def _valid_classes(conn, sid: str, classes: list[str]) -> list[str]: known = set(school_class_names(conn, sid)) cleaned = [_clean_class(c) for c in classes] unknown = sorted({c for c in cleaned if c not in known}) if not cleaned or unknown: raise HTTPException(400, detail="Unknown class: " + ", ".join(unknown or ["(empty)"]) + ". Create it in the classes list first.") return cleaned def _family_payload(conn, sid: str, code: str) -> dict: row = conn.execute("SELECT * FROM families WHERE code=? AND school_id=?", (code, sid)).fetchone() classes = [r["class_name"] for r in conn.execute( "SELECT class_name FROM students WHERE family_code=? ORDER BY id", (code,))] has_orders = bool(conn.execute( f"SELECT 1 FROM orders WHERE family_code=? AND {PAID_SQL} LIMIT 1", (code,)).fetchone()) return {"code": code, "email": row["email"], "classes": classes, "is_sibling": len(classes) > 1, "has_orders": has_orders, "card_url": media_service.card_url(sid, code)} @router.post("/api/admin/families") def create_family(body: FamilyCreate, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Attach a new family code to its class(es). One code per family, whatever the number of children: the individual and the sibling photos all end up under it.""" ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id email = (body.email or "").strip() or None if email and not _EMAIL_OK.fullmatch(email): raise HTTPException(400, detail="Invalid e-mail address.") with get_db() as conn: classes = _valid_classes(conn, sid, body.classes) if body.code: code = body.code.strip().upper() if not re.fullmatch(r"[A-Z0-9][A-Z0-9-]{2,22}", code): raise HTTPException(400, detail="Invalid code.") if conn.execute("SELECT 1 FROM families WHERE code=?", (code,)).fetchone(): raise HTTPException(409, detail="This code already exists.") else: code = new_family_code(conn, sid) conn.execute("INSERT INTO families (code, family_name, email, created_at, school_id) VALUES (?,?,?,?,?)", (code, "", email, utcnow_iso(), sid)) for cls in classes: conn.execute("INSERT INTO students (family_code, class_name) VALUES (?,?)", (code, cls)) conn.commit() family = _family_payload(conn, sid, code) photo_service.invalidate_families_cache(sid) audit_service.log_event("family.create", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, details={"code": code, "classes": classes}) return {"family": family} @router.patch("/api/admin/families/{code}") def update_family(code: str, body: FamilyPatch, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Re-attach a code to other classes (the sibling case, or a mistake at the shooting).""" ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id code = code.strip().upper() with get_db() as conn: if not conn.execute("SELECT 1 FROM families WHERE code=? AND school_id=?", (code, sid)).fetchone(): raise HTTPException(404, detail="Family not found.") if body.classes is not None: classes = _valid_classes(conn, sid, body.classes) conn.execute("DELETE FROM students WHERE family_code=?", (code,)) for cls in classes: conn.execute("INSERT INTO students (family_code, class_name) VALUES (?,?)", (code, cls)) if "email" in body.model_fields_set: email = (body.email or "").strip() or None if email and not _EMAIL_OK.fullmatch(email): raise HTTPException(400, detail="Invalid e-mail address.") conn.execute("UPDATE families SET email=? WHERE code=?", (email, code)) conn.commit() family = _family_payload(conn, sid, code) photo_service.invalidate_families_cache(sid) audit_service.log_event("family.update", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, details={"code": code, "classes": family["classes"]}) return {"family": family} # ── Family cards (PNG, see services/card_service.py) ─────────────────────────── CARD_LANGS = ("fr", "en", "es") def _portal_base(settings: dict, request: Request, explicit: Optional[str]) -> str: """Portal origin printed in the QR: the school's setting wins, then what the operator's browser reports.""" url = (settings.get("portal_url") or explicit or "").strip() if url and not url.startswith("http"): url = "https://" + url if not re.match(r"^https?://[^\s/]+(:\d+)?(/[^\s]*)?$", url): url = f"{request.url.scheme}://{request.headers.get('host', request.url.netloc)}" return url.rstrip("/") def _render_family_card(conn, sid: str, code: str, lang: str, base: str) -> Optional[bytes]: fam = conn.execute("SELECT code FROM families WHERE code=? AND school_id=?", (code, sid)).fetchone() if not fam: return None classes = [r["class_name"] for r in conn.execute( "SELECT class_name FROM students WHERE family_code=? ORDER BY id", (code,))] st = school_service.get_settings(sid) # The sheet shows the child's photos so the parents recognise it: originals only, never the AI ones. photos = [f for f in photo_service.list_family_files(sid, fam["code"]) if photo_service.photo_kind(f.name) == "original"] return card_service.render_card( code=fam["code"], classes=classes, school_name=st.get("school_name") or "", portal_url=base, rgpd_days=st.get("rgpd_days") or "60", lang=lang, brand_color=st.get("brand_color") or "", photos=photos) @router.get("/api/media/card/{school_id}/{code}.png") def family_card_png(school_id: str, code: str, request: Request, exp: int = Query(...), sig: str = Query(...), lang: str = Query("fr"), base: Optional[str] = Query(None, max_length=200)): if not media_service.check_card_signature(school_id, code, exp, sig): raise HTTPException(403, detail="Link expired or invalid.") lang = lang if lang in CARD_LANGS else "fr" def build() -> Optional[bytes]: with get_db() as conn: return _render_family_card(conn, school_id, code, lang, _portal_base(school_service.get_settings(school_id), request, base)) png = build() if png is None: raise HTTPException(404, detail="Family not found.") return Response(content=png, media_type="image/png", headers={"Cache-Control": "private, max-age=300"}) @router.get("/api/admin/cards.zip") def family_cards_zip(request: Request, lang: str = Query("fr"), class_name: Optional[str] = Query(None), base: Optional[str] = Query(None, max_length=200), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Every card of the school (or one class) as PNG files, grouped in one folder per class.""" ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id lang = lang if lang in CARD_LANGS else "fr" def build() -> tuple[bytes, int]: st = school_service.get_settings(sid) portal = _portal_base(st, request, base) buf = io.BytesIO() n = 0 with get_db() as conn: fam_students = family_students_map(conn, sid) rows = conn.execute("SELECT code, created_at FROM families WHERE school_id=? ORDER BY created_at, code", (sid,)).fetchall() ordered = sorted( ((r, fam_students.get(r["code"], [])) for r in rows), key=lambda rs: (class_sort_key(rs[1][0]["class_name"] if rs[1] else "~"), rs[0]["created_at"] or "", rs[0]["code"])) seq: dict[str, int] = {} with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: # PNGs are already compressed for r, studs in ordered: cls = studs[0]["class_name"] if studs else "no-class" if class_name and cls != class_name: continue png = _render_family_card(conn, sid, r["code"], lang, portal) if png: folder = _strip_accents(re.sub(r"[^\w.-]+", "_", cls)).strip("_") or "no-class" seq[folder] = seq.get(folder, 0) + 1 zf.writestr(f"{folder}/{seq[folder]:03d}_{r['code']}.png", png) # shooting order n += 1 return buf.getvalue(), n data, n = build() if n == 0: raise HTTPException(404, detail="No cards to export.") audit_service.log_event("cards.export", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, details={"cards": n, "class": class_name, "lang": lang}) return Response(content=data, media_type="application/zip", headers={"Content-Disposition": f'attachment; filename="meeconic_cards_{sid}.zip"'}) # ── Audit log ───────────────────────────────────────────────────────────────── def _audit_scope(ctx: Ctx) -> Optional[str]: """Directeur : toujours son école. Opérateur : l'école sélectionnée, sinon tout.""" return ctx.school_id @router.get("/api/admin/audit") def get_audit( action: Optional[str] = Query(None), severity: Optional[str] = Query(None), date_from: Optional[str] = Query(None), date_to: Optional[str] = Query(None), limit: int = Query(100, le=500), offset: int = Query(0, ge=0), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None), ): ctx = get_ctx(authorization, x_school, "operator", "director", need_school=False) sid = _audit_scope(ctx) events = audit_service.list_events(action=action, severity=severity, date_from=date_from, date_to=date_to, school_id=sid, limit=limit, offset=offset) total = audit_service.count_events(action=action, severity=severity, date_from=date_from, date_to=date_to, school_id=sid) return {"events": events, "total": total} @router.get("/api/admin/audit/export.csv") def export_audit_csv( action: Optional[str] = Query(None), severity: Optional[str] = Query(None), date_from: Optional[str] = Query(None), date_to: Optional[str] = Query(None), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None), ): ctx = get_ctx(authorization, x_school, "operator", "director", need_school=False) events = audit_service.list_events(action=action, severity=severity, date_from=date_from, date_to=date_to, school_id=_audit_scope(ctx), limit=5000) buf = io.StringIO() w = csv.writer(buf, delimiter=";") w.writerow(["date_utc", "utilisateur", "action", "type_ressource", "id_ressource", "ip", "details", "niveau"]) for e in events: w.writerow([ e["created_at"], e["user_id"] or "", e["action"], e["resource_type"] or "", e["resource_id"] or "", e["ip_address"] or "", json.dumps(e["details"], ensure_ascii=False) if e["details"] else "", e["severity"], ]) return PlainTextResponse( content="" + buf.getvalue(), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="meeconic_audit.csv"'}, ) # ── Exports ─────────────────────────────────────────────────────────────────── def _csv_safe(v) -> str: """Neutralise l'injection de formules Excel (=, +, -, @) dans les exports.""" s = "" if v is None else str(v) return "'" + s if s[:1] in ("=", "+", "-", "@") else s @router.get("/api/admin/export/printer.csv") def export_printer_csv(class_name: Optional[str] = Query(None), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") with get_db() as conn: fam_students = family_students_map(conn, ctx.school_id) rows = paid_orders(conn, ctx.school_id, "ORDER BY created_at") buf = io.StringIO() w = csv.writer(buf, delimiter=";") w.writerow(["commande", "code_famille", "classes", "fichier_photo", "format", "quantite", "magique_ia", "metier_ia", "statut_lot"]) for r in rows: o = order_row_to_admin(r, fam_students) if class_name and class_name not in o["classes"]: continue for it in o["items"]: w.writerow([ o["short_id"], o["family_code"], _csv_safe(" / ".join(o["classes"])), it["photo_filename"], _csv_safe(it["format_label"]), it["quantity"], "oui" if it.get("is_magic") else "non", _csv_safe(it.get("job_label") or ""), o["batch_status"], ]) return PlainTextResponse( content="" + buf.getvalue(), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="meeconic_imprimerie.csv"'}, ) @router.get("/api/admin/export/labels") def export_labels(class_name: Optional[str] = Query(None), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") with get_db() as conn: fam_students = family_students_map(conn, ctx.school_id) rows = paid_orders(conn, ctx.school_id, "ORDER BY created_at") by_class: dict[str, dict] = {} for r in rows: o = order_row_to_admin(r, fam_students) for c in o["classes"]: if class_name and c != class_name: continue g = by_class.setdefault(c, {"class_name": c, "orders": set(), "prints": 0, "labels": []}) g["orders"].add(o["id"]) g["labels"].append({ "class_name": c, "order_id": o["short_id"], "family_code": o["family_code"], "summary": o["items_summary"], "prints": o["prints"], "ai": o["ai"], }) out = [] for g in sorted(by_class.values(), key=lambda x: class_sort_key(x["class_name"])): g["prints"] = sum(l["prints"] for l in g["labels"]) g["orders"] = len(g["orders"]) out.append(g) return {"classes": out} @router.get("/api/admin/export/delivery-slip") def export_delivery_slip(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Bordereau de livraison à l'école : liste de vérification par classe + ligne de signature.""" ctx = get_ctx(authorization, x_school, "operator") with get_db() as conn: settings = school_service.get_settings(ctx.school_id, conn) fam_students = family_students_map(conn, ctx.school_id) rows = paid_orders(conn, ctx.school_id, "ORDER BY created_at") by_class: dict[str, dict] = {} for r in rows: o = order_row_to_admin(r, fam_students) for c in o["classes"]: g = by_class.setdefault(c, {"class_name": c, "order_ids": set(), "prints": 0, "items": []}) g["order_ids"].add(o["id"]) g["items"].append({ "family_code": o["family_code"], "order_id": o["short_id"], "summary": o["items_summary"], "prints": o["prints"], "ai": o["ai"], }) classes = [] total_orders = total_prints = 0 for g in sorted(by_class.values(), key=lambda x: class_sort_key(x["class_name"])): prints = sum(i["prints"] for i in g["items"]) classes.append({"class_name": g["class_name"], "orders": len(g["order_ids"]), "prints": prints, "items": g["items"]}) total_orders += len(g["order_ids"]) total_prints += prints return { "school_name": settings.get("school_name"), "delivery_week": settings.get("delivery_week"), "generated_at": utcnow_iso(), "total_classes": len(classes), "total_orders": total_orders, "total_prints": total_prints, "classes": classes, } # ── État de l'ingestion (FTP / organizer / IA) ───────────────────────────────── @router.get("/api/admin/ingestion") def admin_ingestion(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") from services.organizer_service import get_organizer_status, incoming_dir_for organizer = {"available": True, **get_organizer_status(ctx.school_id)} ftp_url = school_service.get_settings(ctx.school_id).get("ftp_url", "") return { "organizer": organizer, "incoming_pending": organizer["pending_count"], "incoming_dir": str(incoming_dir_for(ctx.school_id).name) or "incoming", "ftp_host": config.FTP_HOST, "ftp_port": config.FTP_PORT, "ftp_url": ftp_url, "ai_configured": bool(config.STORYFACE_API_KEY), } @router.get("/api/admin/organizer/status") def organizer_status(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Estado del organizer DE ESTE COLEGIO con su diagnóstico de salud (lo sondea el banner del admin). `stalled` = el bucle murió, o hay fotos esperando y nada se ha movido en ORGANIZER_STALL_SECS. Va también `ai`: la salud de los servicios de IA / protección de caras. Es el mismo tipo de información (qué falla en segundo plano) y este sondeo ya existe, así que no hace falta otro endpoint.""" ctx = get_ctx(authorization, x_school, "operator") from services.organizer_service import get_organizer_status status = get_organizer_status(ctx.school_id) # La salud de la IA es un extra de este sondeo, no su motivo: si falla, el banner sigue diciendo si el # organizer está vivo (sin esto, un ImportError se llevaba la respuesta entera justo cuando más falta hacía). try: status["ai"] = ai_health.get_status(ctx.school_id) except Exception as exc: log.warning("Salud de IA no disponible: %s", exc) status["ai"] = None return status @router.get("/api/admin/setup") def admin_setup_checklist(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """First-run checklist for the selected school: what is configured, what is still missing, in the order to do it. `section` tells the UI which Settings tab / tab to open (settings sub-tabs: school, sales, photos, access).""" from services import email_service ctx = get_ctx(authorization, x_school, "operator") sid = ctx.school_id st = school_service.get_settings(sid) from services import shoot_service with get_db() as conn: families = conn.execute("SELECT COUNT(*) AS n FROM families WHERE school_id=?", (sid,)).fetchone()["n"] classes = conn.execute("SELECT COUNT(*) AS n FROM school_classes WHERE school_id=?", (sid,)).fetchone()["n"] photographers = len(photographer_service.list_photographers(sid)) shoot_on = st.get("shoot_mode") == "true" # There is no import: families are never known in advance. With the photographer app they are created by the # shooting itself (so the only things to prepare are the classes and the phone link); without it the operator # attaches codes to classes by hand. family_steps = ( [{"id": "shoot_link", "done": bool(shoot_service.get_token(sid)), "tab": "settings", "section": "photos"}] if shoot_on else [{"id": "families", "done": families > 0, "tab": "cards", "section": None}]) steps = [ {"id": "school_name", "done": bool(st.get("school_name")), "tab": "settings", "section": "school"}, {"id": "portal_url", "done": bool(st.get("portal_url")), "tab": "settings", "section": "school"}, {"id": "classes", "done": classes > 0, "tab": "cards", "section": None}, *family_steps, {"id": "photographers", "done": photographers > 0, "tab": "settings", "section": "photos"}, {"id": "shoot_date", "done": bool(st.get("shoot_date")), "tab": "settings", "section": "school"}, {"id": "stripe", "done": bool(config.STRIPE_SECRET_KEY and config.STRIPE_WEBHOOK_SECRET), "tab": "settings", "section": "access"}, {"id": "email", "done": email_service.is_configured(), "tab": "settings", "section": "access"}, ] return {"steps": steps, "done": sum(s["done"] for s in steps), "total": len(steps)} @router.get("/api/admin/system") def admin_system(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Vue d'ensemble technique pour l'opérateur (diagnostic sans accès serveur).""" get_ctx(authorization, x_school, "operator", need_school=False) from core.leader import is_leader from services import backup_service, email_service db_ok = True try: with get_db() as conn: conn.execute("SELECT 1").fetchone() except Exception: db_ok = False last_backup = None try: files = sorted(backup_service.BACKUPS_DIR.glob("meeconic_*.sql.gz"), key=lambda f: f.stat().st_mtime) if files: last_backup = files[-1].name except Exception: pass try: free_mb = shutil.disk_usage(config.PHOTOS_DIR).free // (1024 * 1024) except Exception: free_mb = None with get_db() as conn: crit = conn.execute("SELECT COUNT(*) AS n FROM audit_logs WHERE severity='critical' AND created_at >= ?", ((utcnow() - timedelta(days=1)).isoformat(),)).fetchone()["n"] return { "env": config.APP_ENV, "db_ok": db_ok, "stripe_configured": bool(config.STRIPE_SECRET_KEY), "stripe_webhook_configured": bool(config.STRIPE_WEBHOOK_SECRET), "smtp_configured": email_service.is_configured(), "email_enabled": config._bool("EMAIL_ENABLED"), "alerts_email": config.ALERT_EMAIL or None, "sentry_enabled": bool(config.SENTRY_DSN), "ai_configured": bool(config.STORYFACE_API_KEY), "background_tasks": config.BACKGROUND_TASKS_ENABLED, "this_worker_is_leader": {n: is_leader(n) for n in ("cleanup", "backup", "monitor", "organizer")}, "last_backup": last_backup, "disk_free_mb": free_mb, "critical_events_24h": crit, } # ── Dashboard directeur (lecture seule) ─────────────────────────────────────── @router.get("/api/director/dashboard") def director_dashboard(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator", "director") sid = ctx.school_id with get_db() as conn: settings = school_service.get_settings(sid, conn) rows = paid_orders(conn, sid) n_families = conn.execute("SELECT COUNT(*) AS n FROM families WHERE school_id=?", (sid,)).fetchone()["n"] n_students = conn.execute("SELECT COUNT(*) AS n FROM students s JOIN families f ON f.code=s.family_code " "WHERE f.school_id=?", (sid,)).fetchone()["n"] classes = classes_summary(conn, sid) revenue = paper = magic = digital = 0.0 buyers = set() today = utcnow().date().isoformat() today_count = 0 last_order_at = None for r in rows: b = item_buckets(json.loads(r["items_json"])) revenue += r["total"] paper += b["paper"] magic += b["magic"] digital += b["digital"] buyers.add(r["family_code"]) if r["created_at"][:10] == today: today_count += 1 if last_order_at is None or r["created_at"] > last_order_at: last_order_at = r["created_at"] # Date de suppression des photos : jour de prise de vue + durée de conservation. rgpd_delete = "" try: rgpd_delete = (date.fromisoformat(settings["shoot_date"]) + timedelta(days=int(settings["rgpd_days"]))).isoformat() except Exception: pass share_pct = float(settings.get("revenue_share_pct") or 0) return { "school_name": settings.get("school_name"), "orders_count": len(rows), "orders_today": today_count, "revenue": round(revenue), "revenue_paper": round(paper), "revenue_magic": round(magic), "revenue_digital": round(digital), "revenue_share_pct": share_pct, "revenue_share": round(revenue * share_pct / 100), "partner_name": settings.get("partner_name"), "participation": round(100 * len(buyers) / n_families) if n_families else 0, "buyer_families": len(buyers), "families_total": n_families, "students_total": n_students, "last_order_at": last_order_at, "classes": classes, "calendar": { "shoot_date": settings.get("shoot_date"), "closure_date": settings.get("closure_date"), "delivery_week": settings.get("delivery_week"), "rgpd_delete_date": rgpd_delete, }, } # ── QR code SVG (public) ────────────────────────────────────────────────────── @router.get("/api/qr.svg") def qr_svg(data: str = Query(..., min_length=1, max_length=300), scale: int = Query(4, ge=1, le=20)): import segno # micro=False : segno choisit un Micro QR par défaut pour les chaînes courtes, # or cv2.QRCodeDetector (organizer) ne sait pas les détecter. qr = segno.make(data, error="m", micro=False) buf = io.BytesIO() # border=4 (quiet zone standard) : indispensable à la détection par l'organizer. qr.save(buf, kind="svg", scale=scale, dark="#0A1F44", border=4) return Response(content=buf.getvalue(), media_type="image/svg+xml", headers={"Cache-Control": "public, max-age=86400"}) # ── Fotógrafos (prefijo del archivo de la cámara → carpeta) ─────────────────── class PhotographersBody(BaseModel): photographers: dict[str, str] @router.get("/api/admin/photographers") def get_photographers(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") return {"photographers": photographer_service.list_photographers(ctx.school_id), "implicit": not photographer_service.get_photographer_map(ctx.school_id)} @router.put("/api/admin/photographers") def set_photographers(body: PhotographersBody, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") try: updated = photographer_service.update_photographer_map(ctx.school_id, body.photographers) except ValueError as e: raise HTTPException(400, detail=str(e)) audit_service.log_event("photographer.update", user_id=ctx.role, ip_address=client_ip(request), resource_type="photographer_map", school_id=ctx.school_id, details={"map": updated}, severity="warning") return {"photographers": photographer_service.list_photographers(ctx.school_id), "map": updated} # ── Protección de fotos del portal (caras en servidor + overlay en el navegador) ── @router.get("/api/admin/protection") def get_protection(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator", "director") return {"protection": protection_service.get_protection(ctx.school_id), "heads_available": bool(config.STORYFACE_API_KEY)} @router.put("/api/admin/protection") def put_protection(body: dict, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") try: cfg = protection_service.set_protection(ctx.school_id, body.get("protection", body)) except ValueError as e: raise HTTPException(400, detail=str(e)) audit_service.log_event("protection.update", user_id=ctx.role, ip_address=client_ip(request), resource_type="school", resource_id=ctx.school_id, school_id=ctx.school_id, details={"heads": cfg["protection_heads_enabled"], "overlay": cfg["protection_overlay"]}, severity="warning") return {"protection": cfg} # ── Gestión de fotos (corregir un QR mal leído, papelera) ───────────────────── # Mismas herramientas que hec-photo-app. Borrar NO destruye: mueve a la papelera (recuperable # TRASH_RETENTION_DAYS días, nunca más que la retención RGPD del colegio). DESTRUCTIVE_CONFIRM_WORD = "BORRAR" def _require_confirm_word(confirm: Optional[str]) -> None: """Pestillo contra el borrado masivo accidental (un enlace, un script a medias, un doble clic).""" if (confirm or "").strip().upper() != DESTRUCTIVE_CONFIRM_WORD: raise HTTPException(400, detail=f"Confirmation required: send confirm={DESTRUCTIVE_CONFIRM_WORD}") @router.get("/api/admin/photos/families") def photos_families(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") fams = photo_service.get_all_families(ctx.school_id) return {"families": [f.model_dump() for f in fams]} @router.get("/api/admin/photos/by-date") def photos_by_date( date_from: Optional[str] = Query(None, pattern=r"^\d{4}-\d{2}-\d{2}$"), date_to: Optional[str] = Query(None, pattern=r"^\d{4}-\d{2}-\d{2}$"), time_from: Optional[str] = Query(None, pattern=r"^\d{2}:\d{2}$"), time_to: Optional[str] = Query(None, pattern=r"^\d{2}:\d{2}$"), photographer: Optional[str] = None, limit: int = Query(300, ge=1, le=1000), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None), ): ctx = get_ctx(authorization, x_school, "operator") photos, total = photo_service.list_photos_by_date(ctx.school_id, date_from, date_to, time_from, time_to, photographer, limit) for p in photos: p["thumb_url"] = media_service.admin_thumb_url(ctx.school_id, p["photographer"], p["qr_code"], p["filename"]) return {"photos": photos, "total": len(photos), "total_matching": total} class PhotoMoveBody(BaseModel): photographer: str qr_code: str filename: str target_qr_code: str = Field(min_length=2, max_length=24) target_photographer: Optional[str] = None @router.post("/api/admin/photos/move") def photos_move(body: PhotoMoveBody, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") try: resolved = photo_service.move_photo(ctx.school_id, body.photographer, body.qr_code, body.filename, body.target_qr_code, body.target_photographer) except FileNotFoundError as e: raise HTTPException(404, detail=str(e)) except ValueError as e: raise HTTPException(400, detail=str(e)) audit_service.log_event( "photo.move", user_id=ctx.role, ip_address=client_ip(request), resource_type="photo", resource_id=f"{body.qr_code}/{body.filename}", school_id=ctx.school_id, severity="warning", details={"from": {"photographer": body.photographer, "qr_code": body.qr_code}, "to": {"photographer": body.target_photographer or body.photographer, "qr_code": resolved}}) return {"success": True, "qr_code": resolved} @router.delete("/api/admin/photos/family/{code}") def photos_delete_family(code: str, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") try: n = photo_service.delete_family_photos(ctx.school_id, code) except ValueError as e: raise HTTPException(400, detail=str(e)) audit_service.log_event("photo.delete_family", user_id=ctx.role, ip_address=client_ip(request), resource_type="family", resource_id=code, school_id=ctx.school_id, details={"files": n}, severity="warning") return {"success": True, "deleted": n} @router.delete("/api/admin/photos/all") def photos_delete_all(request: Request, confirm: str = Query(""), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") _require_confirm_word(confirm) n = photo_service.delete_all_photos(ctx.school_id) audit_service.log_event("photo.delete_all", user_id=ctx.role, ip_address=client_ip(request), resource_type="school", resource_id=ctx.school_id, school_id=ctx.school_id, details={"files": n}, severity="critical") return {"success": True, "deleted": n} def _sales_data_counts(conn, sid: str) -> dict: one = lambda sql: conn.execute(sql, (sid,)).fetchone()["n"] # noqa: E731 return { "orders": one("SELECT COUNT(*) AS n FROM orders WHERE school_id=?"), "paid_orders": one(f"SELECT COUNT(*) AS n FROM orders WHERE school_id=? AND {PAID_SQL}"), "families": one("SELECT COUNT(*) AS n FROM families WHERE school_id=?"), } @router.get("/api/admin/reset-data") def reset_data_preview(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """What a reset would remove, so the confirmation dialog can show real numbers.""" ctx = get_ctx(authorization, x_school, "operator") with get_db() as conn: return _sales_data_counts(conn, ctx.school_id) @router.post("/api/admin/reset-data") def reset_data(request: Request, confirm: str = Query(""), families: bool = Query(False), authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): """Wipe the TEST data of ONE school before going live: every order (paid ones included — unlike `delete_family`, which protects the proof of a sale) and, with `families=true`, the family codes, their class links and the photos (moved to the restorable trash). Classes, products, prices, settings and the audit journal are kept: the journal is the only trace that this happened. Payments already taken in Stripe are NOT touched — refund or ignore them there.""" ctx = get_ctx(authorization, x_school, "operator") _require_confirm_word(confirm) sid = ctx.school_id with get_db() as conn: before = _sales_data_counts(conn, sid) conn.execute("DELETE FROM orders WHERE school_id=?", (sid,)) if families: conn.execute("DELETE FROM students WHERE family_code IN (SELECT code FROM families WHERE school_id=?)", (sid,)) conn.execute("DELETE FROM families WHERE school_id=?", (sid,)) conn.commit() photos = 0 if families: photos = photo_service.delete_all_photos(sid) photo_service.invalidate_families_cache(sid) audit_service.log_event("school.reset_data", user_id=ctx.role, ip_address=client_ip(request), school_id=sid, resource_type="school", resource_id=sid, details={**before, "families_removed": families, "files_to_trash": photos}, severity="critical") return {"success": True, **before, "files_to_trash": photos} @router.delete("/api/admin/photos/{photographer}/{code}/{filename}") def photos_delete_one(photographer: str, code: str, filename: str, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") try: ok = photo_service.delete_photo(ctx.school_id, code, filename) except ValueError as e: raise HTTPException(400, detail=str(e)) if not ok: raise HTTPException(404, detail="Photo introuvable.") audit_service.log_event("photo.delete", user_id=ctx.role, ip_address=client_ip(request), resource_type="photo", resource_id=f"{code}/{filename}", school_id=ctx.school_id, severity="warning") return {"success": True} @router.get("/api/admin/photos/trash") def photos_trash(authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") entries = photo_service.list_trash(ctx.school_id) return {"entries": entries, "retention_days": config.TRASH_RETENTION_DAYS} @router.post("/api/admin/photos/trash/{entry_id}/restore") def photos_restore(entry_id: str, request: Request, authorization: Optional[str] = Header(None), x_school: Optional[str] = Header(None)): ctx = get_ctx(authorization, x_school, "operator") try: restored = photo_service.restore_trash_entry(ctx.school_id, entry_id) except ValueError as e: raise HTTPException(400, detail=str(e)) audit_service.log_event("photo.trash_restore", user_id=ctx.role, ip_address=client_ip(request), resource_type="trash_batch", resource_id=entry_id, school_id=ctx.school_id, details={"restored": restored}, severity="warning") return {"success": True, "restored": restored}