#!/usr/bin/env python3 """Suivi Production — Gestion de projets avec rôles utilisateurs""" import sqlite3, os, secrets, re, logging, itertools, threading from datetime import date, datetime, timedelta, timezone from pathlib import Path from functools import wraps from logging.handlers import RotatingFileHandler from flask import Flask, g, render_template, request, redirect, url_for, jsonify, session, send_file from werkzeug.security import generate_password_hash, check_password_hash from werkzeug.middleware.proxy_fix import ProxyFix BASE_DIR = Path(__file__).resolve().parent DB_PATH = BASE_DIR / "suivi.db" COLLABORATEURS = ["NIZAR", "YOUSRA", "IMANE", "SANAA", "HAYAT"] ROLES = ["super_admin", "admin", "membre"] STATUTS = {"NOUVEAU": "Nouveau", "EN COURS": "En cours", "LIVRER": "Livré"} CONGE_STATUTS = {"EN ATTENTE": "En attente", "VALIDÉ": "Validé", "REFUSÉ": "Refusé", "ANNULÉ": "Annulé"} OSM_ETATS = ["Nouveau", "Transféré pour étude", "Faisabilité OK", "Qualification IAV", "Rejetée", "Abandonnée par IAV"] EQUIPE_OSM = "OSM" EQUIPE_STANDARD = "Standard" app = Flask(__name__) # ── Security: SECRET_KEY ────────────────────────────────────────────────── _SECRET = os.environ.get("SECRET_KEY") if not _SECRET: _SECRET = secrets.token_hex(32) import sys print("⚠️ SECRET_KEY not set in environment — using random key.", file=sys.stderr) print(" Sessions will be invalidated on every restart.", file=sys.stderr) app.config["SECRET_KEY"] = _SECRET app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(hours=8) app.config["SESSION_COOKIE_HTTPONLY"] = True app.config["SESSION_COOKIE_SAMESITE"] = "Lax" # ── ProxyFix: trust X-Forwarded-For from 1 proxy (cPanel/Apache) ────────── app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1) # ── Logging ──────────────────────────────────────────────────────────────── LOG_PATH = BASE_DIR / "suivi.log" def _setup_logging(): logger = logging.getLogger("suivi") logger.setLevel(logging.INFO) if not logger.handlers: h = RotatingFileHandler(str(LOG_PATH), maxBytes=5*1024*1024, backupCount=3) h.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) logger.addHandler(h) return logger log = _setup_logging() # Route → action mapping for audit log ACTION_MAP = { "login": "LOGIN_SUCCESS", "logout": "LOGOUT", "api_new_projet": "PROJET_CREATE", "api_update_projet": "PROJET_UPDATE", "api_delete_projet": "PROJET_DELETE", "api_add_traitement": "TRAITEMENT_ADD", "api_delete_traitement": "TRAITEMENT_DELETE", "api_user_new": "USER_CREATE", "api_user_edit": "USER_UPDATE", "api_user_delete": "USER_DELETE", "api_invitation_new": "INVITATION_CREATE", "api_invitation_delete": "INVITATION_DELETE", } # ── Helpers ──────────────────────────────────────────────────────────────── def validate_password(password): """Retourne (ok, message). Exige ≥8 caractères + ≥1 majuscule + ≥1 chiffre + ≥1 caractère spécial.""" if len(password) < 8: return False, "Le mot de passe doit contenir au moins 8 caractères." if not re.search(r"[A-Z]", password): return False, "Le mot de passe doit contenir au moins une lettre majuscule." if not re.search(r"\d", password): return False, "Le mot de passe doit contenir au moins un chiffre." if not re.search(r"[!@#$%^&*(),.?\":{}|<>_\-+=\[\]\\;'/`~]", password): return False, "Le mot de passe doit contenir au moins un caractère spécial." return True, "" def is_admin(): """True si l'utilisateur est admin, super_admin, ou NIZAR""" return role in ("super_admin", "admin") or full_name.upper() == "NIZAR" def is_super_admin(): """True si l'utilisateur est super_admin.""" return session.get("role") == "super_admin" def login_required(f): @wraps(f) def dec(*a, **kw): if "user_id" not in session: return redirect(url_for("login")) return f(*a, **kw) return dec def admin_required(f): @wraps(f) def dec(*a, **kw): if "user_id" not in session: return redirect(url_for("login")) if not is_admin(): return redirect(url_for("dashboard")) return f(*a, **kw) return dec def super_admin_required(f): @wraps(f) def dec(*a, **kw): if "user_id" not in session: return redirect(url_for("login")) if session.get("role") != "super_admin": return redirect(url_for("dashboard")) return f(*a, **kw) return dec # ── CSRF Protection ─────────────────────────────────────────────────────── def _generate_csrf(): """Generate or return existing CSRF token in session.""" if "_csrf_token" not in session: session["_csrf_token"] = secrets.token_hex(32) return session["_csrf_token"] # Exempt routes from CSRF check (no token needed) _CSRF_EXEMPT = {"login", "static", "invite_register"} @app.before_request def csrf_check(): """Validate CSRF token on all state-changing requests.""" if request.method in ("GET", "HEAD", "OPTIONS"): return None if request.endpoint in _CSRF_EXEMPT: return None token = (request.headers.get("X-CSRF-Token") or request.form.get("_csrf_token") or (request.get_json(silent=True) or {}).get("_csrf_token")) if not token or not secrets.compare_digest(str(token), session.get("_csrf_token", "")): log.warning("CSRF_FAIL ip=%s endpoint=%s", request.remote_addr, request.endpoint) if request.content_type and "application/json" in request.content_type: return jsonify({"ok": False, "error": "Jeton CSRF invalide ou manquant"}), 403 return "Jeton CSRF invalide — veuillez rafraîchir la page.", 403 return None @app.context_processor def inject_csrf(): """Inject CSRF token into all templates.""" return {"csrf_token": _generate_csrf()} # ── Session timeout & rate-limit cleanup ────────────────────────────────── # Deterministic request counter for periodic cleanup (replaces hash(request.path)) _cleanup_counter = itertools.cycle(range(50)) _cleanup_lock = threading.Lock() @app.before_request def check_session_timeout(): if "user_id" in session: last = session.get("_last_active") now_ts = datetime.now().timestamp() if last is not None and (now_ts - last) > 8 * 3600: session.clear() return redirect(url_for("login")) session["_last_active"] = now_ts session.permanent = True # Clean old login attempts every ~50 requests if request.endpoint != "static": with _cleanup_lock: if next(_cleanup_counter) == 0: try: db = get_db() db.execute("DELETE FROM login_attempts WHERE attempted_at < datetime('now', '-1 day')") db.commit() except Exception: pass # ── Force password change ───────────────────────────────────────────────── PASSWORD_CHANGE_EXEMPT = {"change_password", "logout", "static", "login", "invite_register"} @app.before_request def enforce_password_change(): """Redirect users who must change their password.""" if "user_id" not in session: return None if request.endpoint in PASSWORD_CHANGE_EXEMPT: return None if session.get("must_change_password"): return redirect(url_for("change_password")) return None def get_db(): if "db" not in g: g.db = sqlite3.connect(str(DB_PATH)) g.db.row_factory = sqlite3.Row g.db.execute("PRAGMA journal_mode=WAL") g.db.execute("PRAGMA foreign_keys=ON") return g.db @app.teardown_appcontext def close_db(exc): db = g.pop("db", None) if db is not None: db.close() def get_current_user(): if "user_id" not in session: return None db = get_db() return db.execute("SELECT * FROM users WHERE id=?", (session["user_id"],)).fetchone() # ── Init & migrate ───────────────────────────────────────────────────────── def init_db(): """Create DB schema. Does NOT seed users — use seed.py for that.""" db = sqlite3.connect(str(DB_PATH)) db.executescript(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, full_name TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'membre', must_change_password INTEGER NOT NULL DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS projets ( id INTEGER PRIMARY KEY AUTOINCREMENT, nom TEXT NOT NULL UNIQUE, numero_requete TEXT DEFAULT '', total_sites INTEGER NOT NULL DEFAULT 0, statut TEXT NOT NULL DEFAULT 'NOUVEAU', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS affectations ( id INTEGER PRIMARY KEY AUTOINCREMENT, projet_id INTEGER NOT NULL REFERENCES projets(id) ON DELETE CASCADE, collaborateur TEXT NOT NULL, prevu_sites INTEGER NOT NULL DEFAULT 0, helico_cr INTEGER NOT NULL DEFAULT 0, UNIQUE(projet_id, collaborateur) ); CREATE TABLE IF NOT EXISTS traitements ( id INTEGER PRIMARY KEY AUTOINCREMENT, projet_id INTEGER NOT NULL REFERENCES projets(id) ON DELETE CASCADE, collaborateur TEXT NOT NULL, date_traitement DATE NOT NULL, nb_sites INTEGER NOT NULL DEFAULT 0, helico INTEGER NOT NULL DEFAULT 0, helico_cr INTEGER NOT NULL DEFAULT 0, cr INTEGER NOT NULL DEFAULT 0, UNIQUE(projet_id, collaborateur, date_traitement) ); """) db.executescript(""" CREATE TABLE IF NOT EXISTS clients_bal ( id INTEGER PRIMARY KEY AUTOINCREMENT, nom TEXT NOT NULL UNIQUE ); CREATE TABLE IF NOT EXISTS demandes_bal ( id INTEGER PRIMARY KEY AUTOINCREMENT, date_reception DATE NOT NULL, client TEXT NOT NULL, nb_sites INTEGER NOT NULL DEFAULT 1, collaborateur TEXT NOT NULL, date_traitement DATE, delai_min INTEGER, statut TEXT NOT NULL DEFAULT 'EN ATTENTE', date_cloture DATE, commentaire TEXT DEFAULT '', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) db.executescript(""" CREATE TABLE IF NOT EXISTS types_conge ( id INTEGER PRIMARY KEY AUTOINCREMENT, nom TEXT NOT NULL UNIQUE, couleur TEXT NOT NULL DEFAULT '#3b82f6', actif INTEGER NOT NULL DEFAULT 1 ); CREATE TABLE IF NOT EXISTS conges ( id INTEGER PRIMARY KEY AUTOINCREMENT, demandeur_id INTEGER NOT NULL REFERENCES users(id), personne_id INTEGER NOT NULL REFERENCES users(id), type_id INTEGER NOT NULL REFERENCES types_conge(id), date_debut DATE NOT NULL, date_fin DATE NOT NULL, nb_jours REAL NOT NULL DEFAULT 1, statut TEXT NOT NULL DEFAULT 'EN ATTENTE', commentaire TEXT DEFAULT '', valide_par INTEGER REFERENCES users(id), date_validation TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS jours_feries ( id INTEGER PRIMARY KEY AUTOINCREMENT, date_ferie DATE NOT NULL UNIQUE, nom TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS suivi_osm ( id INTEGER PRIMARY KEY AUTOINCREMENT, ref_osm TEXT NOT NULL, date_arrivee DATE NOT NULL, date_envoi DATE, priorisation INTEGER NOT NULL DEFAULT 0, etat TEXT NOT NULL DEFAULT 'Nouveau', collaborateur_id INTEGER REFERENCES users(id), commentaire TEXT DEFAULT '', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) db.executescript(""" CREATE TABLE IF NOT EXISTS invitations ( id INTEGER PRIMARY KEY AUTOINCREMENT, token TEXT NOT NULL UNIQUE, username TEXT NOT NULL, full_name TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'membre', created_by INTEGER NOT NULL REFERENCES users(id), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP NOT NULL, used_at TIMESTAMP ); CREATE TABLE IF NOT EXISTS login_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT NOT NULL, username TEXT NOT NULL, attempted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # Seed default clients for client in ["Bouygues Construction","Engie","Saint Gobain","SUEZ","SUEZ TECHNIQUE"]: db.execute("INSERT OR IGNORE INTO clients_bal (nom) VALUES (?)", (client,)) db.commit() db.close() log.info("DB initialized: %s", DB_PATH) def migrate_db(): """Add missing columns and tables without data loss.""" db = sqlite3.connect(str(DB_PATH)) for sql in [ "ALTER TABLE projets ADD COLUMN numero_requete TEXT DEFAULT ''", "ALTER TABLE projets ADD COLUMN statut TEXT NOT NULL DEFAULT 'NOUVEAU'", "ALTER TABLE affectations ADD COLUMN helico_cr INTEGER NOT NULL DEFAULT 0", "ALTER TABLE traitements ADD COLUMN helico INTEGER NOT NULL DEFAULT 0", "ALTER TABLE traitements ADD COLUMN helico_cr INTEGER NOT NULL DEFAULT 0", "ALTER TABLE traitements ADD COLUMN cr INTEGER NOT NULL DEFAULT 0", "ALTER TABLE users ADD COLUMN must_change_password INTEGER NOT NULL DEFAULT 0", "ALTER TABLE users ADD COLUMN equipe TEXT DEFAULT 'Standard'", ]: try: db.execute(sql) except sqlite3.OperationalError: pass for sql in [ """CREATE TABLE IF NOT EXISTS invitations ( id INTEGER PRIMARY KEY AUTOINCREMENT, token TEXT NOT NULL UNIQUE, username TEXT NOT NULL, full_name TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'membre', created_by INTEGER NOT NULL REFERENCES users(id), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP NOT NULL, used_at TIMESTAMP )""", """CREATE TABLE IF NOT EXISTS login_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT NOT NULL, username TEXT NOT NULL, attempted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""", """CREATE TABLE IF NOT EXISTS types_conge ( id INTEGER PRIMARY KEY AUTOINCREMENT, nom TEXT NOT NULL UNIQUE, couleur TEXT NOT NULL DEFAULT '#3b82f6', actif INTEGER NOT NULL DEFAULT 1 )""", """CREATE TABLE IF NOT EXISTS conges ( id INTEGER PRIMARY KEY AUTOINCREMENT, demandeur_id INTEGER NOT NULL REFERENCES users(id), personne_id INTEGER NOT NULL REFERENCES users(id), type_id INTEGER NOT NULL REFERENCES types_conge(id), date_debut DATE NOT NULL, date_fin DATE NOT NULL, nb_jours REAL NOT NULL DEFAULT 1, statut TEXT NOT NULL DEFAULT 'EN ATTENTE', commentaire TEXT DEFAULT '', valide_par INTEGER REFERENCES users(id), date_validation TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""", """CREATE TABLE IF NOT EXISTS jours_feries ( id INTEGER PRIMARY KEY AUTOINCREMENT, date_ferie DATE NOT NULL UNIQUE, nom TEXT NOT NULL )""", """CREATE TABLE IF NOT EXISTS suivi_osm ( id INTEGER PRIMARY KEY AUTOINCREMENT, ref_osm TEXT NOT NULL, date_arrivee DATE NOT NULL, date_envoi DATE, priorisation INTEGER NOT NULL DEFAULT 0, etat TEXT NOT NULL DEFAULT 'Nouveau', collaborateur_id INTEGER REFERENCES users(id), commentaire TEXT DEFAULT '', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""", ]: try: db.execute(sql) except sqlite3.OperationalError: pass db.commit() db.close() log.info("DB migrated") # ── Auth routes ──────────────────────────────────────────────────────────── MAX_LOGIN_ATTEMPTS = 5 LOGIN_BLOCK_MINUTES = 15 @app.route("/login", methods=["GET", "POST"]) def login(): error = None if request.method == "POST": username = request.form.get("login", "").strip().lower() password = request.form.get("password", "") ip = request.remote_addr or "0.0.0.0" db = get_db() # Rate limit check cutoff = (datetime.now(timezone.utc) - timedelta(minutes=LOGIN_BLOCK_MINUTES)).strftime("%Y-%m-%d %H:%M:%S") recent = db.execute( "SELECT COUNT(*) FROM login_attempts WHERE ip=? AND attempted_at > ?", (ip, cutoff) ).fetchone()[0] if recent >= MAX_LOGIN_ATTEMPTS: error = f"Trop de tentatives. Réessayez dans {LOGIN_BLOCK_MINUTES} minutes." log.warning("LOGIN_BLOCKED ip=%s username=%s", ip, username) return render_template("login.html", error=error) user = db.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone() if user and check_password_hash(user["password_hash"], password): db.execute("DELETE FROM login_attempts WHERE ip=?", (ip,)) db.commit() session.clear() session["user_id"] = user["id"] session["username"] = user["username"] session["full_name"] = user["full_name"] session["role"] = user["role"] session["_last_active"] = datetime.now().timestamp() session["must_change_password"] = bool(user["must_change_password"]) session.permanent = True log.info("LOGIN_SUCCESS username=%s ip=%s", username, ip) if user["must_change_password"]: return redirect(url_for("change_password")) return redirect(url_for("modules")) db.execute("INSERT INTO login_attempts (ip, username) VALUES (?,?)", (ip, username)) db.commit() remaining = MAX_LOGIN_ATTEMPTS - recent - 1 log.warning("LOGIN_FAIL username=%s ip=%s remaining=%s", username, ip, remaining) if remaining > 0: error = f"Identifiant ou mot de passe incorrect. {remaining} tentative(s) restante(s)." else: error = f"Trop de tentatives. Réessayez dans {LOGIN_BLOCK_MINUTES} minutes." return render_template("login.html", error=error) @app.route("/change-password", methods=["GET", "POST"]) @login_required def change_password(): error = None if request.method == "POST": pw = request.form.get("password", "") pw2 = request.form.get("password_confirm", "") if pw != pw2: error = "Les mots de passe ne correspondent pas." else: ok, msg = validate_password(pw) if not ok: error = msg else: db = get_db() db.execute("UPDATE users SET password_hash=?, must_change_password=0 WHERE id=?", (generate_password_hash(pw), session["user_id"])) db.commit() session["must_change_password"] = False log.info("PASSWORD_CHANGED user_id=%s", session["user_id"]) return redirect(url_for("modules")) return render_template("change_password.html", error=error, user={"full_name": session.get("full_name", "")}) @app.route("/modules") @login_required def modules(): user = get_current_user() return render_template("modules.html", user=user, is_admin=is_admin()) @app.route("/logout") def logout(): uid = session.get("user_id", "?") session.clear() log.info("LOGOUT user_id=%s", uid) return redirect(url_for("login")) # ── Invitations ──────────────────────────────────────────────────────────── INVITE_EXPIRY_HOURS = 48 @app.route("/invite/", methods=["GET", "POST"]) def invite_register(token): db = get_db() inv = db.execute("SELECT * FROM invitations WHERE token=?", (token,)).fetchone() if not inv: return render_template("invite_register.html", invitation=None, error="Lien d'invitation invalide.", success=None) if inv["used_at"]: return render_template("invite_register.html", invitation=None, error="Cette invitation a déjà été utilisée.", success=None) if datetime.now() > datetime.fromisoformat(inv["expires_at"]): return render_template("invite_register.html", invitation=None, error="Cette invitation a expiré.", success=None) if request.method == "POST": password = request.form.get("password", "") password_confirm = request.form.get("password_confirm", "") if password != password_confirm: return render_template("invite_register.html", invitation=inv, error="Les mots de passe ne correspondent pas.", success=None) ok, msg = validate_password(password) if not ok: return render_template("invite_register.html", invitation=inv, error=msg, success=None) try: # New invited users must change password on first login db.execute("INSERT INTO users (username, password_hash, full_name, role, must_change_password) VALUES (?,?,?,?,1)", (inv["username"], generate_password_hash(password), inv["full_name"], inv["role"])) db.execute("UPDATE invitations SET used_at=? WHERE id=?", (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), inv["id"])) db.commit() log.info("INVITE_ACCEPTED username=%s", inv["username"]) return render_template("invite_register.html", invitation=None, error=None, success="Compte activé ! Vous pouvez maintenant vous connecter.") except sqlite3.IntegrityError: return render_template("invite_register.html", invitation=inv, error="Ce login existe déjà. Contactez un administrateur.", success=None) return render_template("invite_register.html", invitation=inv, error=None, success=None) @app.route("/api/admin/invitations/new", methods=["POST"]) @super_admin_required def api_invitation_new(): data = request.get_json() username = data.get("username", "").strip().lower() full_name = data.get("full_name", "").strip().upper() role = data.get("role", "membre") if not username or not full_name: return jsonify({"ok": False, "error": "Login et nom complet obligatoires"}), 400 if role not in ("membre", "admin"): return jsonify({"ok": False, "error": "Rôle invalide"}), 400 db = get_db() if db.execute("SELECT id FROM users WHERE username=?", (username,)).fetchone(): return jsonify({"ok": False, "error": "Ce login existe déjà"}), 409 existing = db.execute( "SELECT id FROM invitations WHERE username=? AND used_at IS NULL AND expires_at > ?", (username, datetime.now().strftime("%Y-%m-%d %H:%M:%S")) ).fetchone() if existing: return jsonify({"ok": False, "error": "Une invitation en attente existe déjà pour ce login"}), 409 token = secrets.token_urlsafe(32) expires_at = (datetime.now() + timedelta(hours=INVITE_EXPIRY_HOURS)).strftime("%Y-%m-%d %H:%M:%S") db.execute( "INSERT INTO invitations (token, username, full_name, role, created_by, expires_at) VALUES (?,?,?,?,?,?)", (token, username, full_name, role, session["user_id"], expires_at)) db.commit() log.info("INVITATION_CREATED by=%s for=%s role=%s", session["username"], username, role) return jsonify({"ok": True, "token": token}) @app.route("/api/admin/invitations/", methods=["DELETE"]) @super_admin_required def api_invitation_delete(iid): db = get_db() db.execute("DELETE FROM invitations WHERE id=? AND used_at IS NULL", (iid,)) db.commit() log.info("INVITATION_DELETED id=%s by=%s", iid, session["username"]) return jsonify({"ok": True}) # ── Module BAL ───────────────────────────────────────────────────────────── BAL_STATUTS = ["EN ATTENTE", "EN COURS", "CLÔTURÉ"] @app.route("/bal") @login_required def bal_list(): user = get_current_user() db = get_db() demandes = db.execute("SELECT * FROM demandes_bal ORDER BY date_reception DESC, id DESC").fetchall() clients = [r["nom"] for r in db.execute("SELECT nom FROM clients_bal ORDER BY nom")] return render_template("bal_list.html", user=user, demandes=demandes, clients=clients, collaborateurs=COLLABORATEURS, statuts=BAL_STATUTS, is_admin=is_admin()) @app.route("/bal/new", methods=["GET","POST"]) @login_required def bal_new(): user = get_current_user() db = get_db() if request.method == "POST": db.execute("""INSERT INTO demandes_bal (date_reception, client, nb_sites, collaborateur, date_traitement, delai_min, statut, date_cloture, commentaire) VALUES (?,?,?,?,?,?,?,?,?)""", (request.form["date_reception"], request.form["client"], int(request.form["nb_sites"]), request.form["collaborateur"], request.form.get("date_traitement") or None, int(request.form["delai_min"]) if request.form.get("delai_min") else None, request.form["statut"], request.form.get("date_cloture") or None, request.form.get("commentaire",""))) db.commit() log.info("BAL_DEMANDE_CREATED by=%s client=%s", session["username"], request.form["client"]) return redirect(url_for("bal_list")) clients = [r["nom"] for r in db.execute("SELECT nom FROM clients_bal ORDER BY nom")] return render_template("bal_form.html", user=user, demande=None, clients=clients, collaborateurs=COLLABORATEURS, statuts=BAL_STATUTS, is_admin=is_admin(), today=date.today().isoformat()) @app.route("/bal//edit", methods=["GET","POST"]) @login_required @admin_required def bal_edit(did): user = get_current_user() db = get_db() demande = db.execute("SELECT * FROM demandes_bal WHERE id=?", (did,)).fetchone() if not demande: return "Introuvable", 404 if request.method == "POST": db.execute("""UPDATE demandes_bal SET date_reception=?,client=?,nb_sites=?,collaborateur=?,date_traitement=?,delai_min=?,statut=?,date_cloture=?,commentaire=? WHERE id=?""", (request.form["date_reception"], request.form["client"], int(request.form["nb_sites"]), request.form["collaborateur"], request.form.get("date_traitement") or None, int(request.form["delai_min"]) if request.form.get("delai_min") else None, request.form["statut"], request.form.get("date_cloture") or None, request.form.get("commentaire",""), did)) db.commit() log.info("BAL_DEMANDE_UPDATED id=%s by=%s", did, session["username"]) return redirect(url_for("bal_list")) clients = [r["nom"] for r in db.execute("SELECT nom FROM clients_bal ORDER BY nom")] return render_template("bal_form.html", user=user, demande=demande, clients=clients, collaborateurs=COLLABORATEURS, statuts=BAL_STATUTS, is_admin=is_admin()) @app.route("/bal//delete", methods=["POST"]) @login_required @admin_required def bal_delete(did): db = get_db() db.execute("DELETE FROM demandes_bal WHERE id=?", (did,)) db.commit() log.info("BAL_DEMANDE_DELETED id=%s by=%s", did, session["username"]) return redirect(url_for("bal_list")) @app.route("/api/bal/clients", methods=["POST"]) @login_required @admin_required def api_bal_add_client(): db = get_db() nom = request.form.get("nom","").strip() if not nom: return jsonify({"ok":False}), 400 try: db.execute("INSERT INTO clients_bal (nom) VALUES (?)", (nom,)) db.commit() log.info("BAL_CLIENT_ADDED name=%s by=%s", nom, session["username"]) return jsonify({"ok":True,"nom":nom}) except sqlite3.IntegrityError: return jsonify({"ok":False,"error":"Existe déjà"}), 409 @app.route("/api/bal/clients/", methods=["DELETE"]) @login_required @admin_required def api_bal_delete_client(cid): db = get_db() db.execute("DELETE FROM clients_bal WHERE id=?", (cid,)) db.commit() log.info("BAL_CLIENT_DELETED id=%s by=%s", cid, session["username"]) return jsonify({"ok":True}) @app.route("/api/bal/kpi") @login_required @admin_required def api_bal_kpi(): db = get_db() total = db.execute("SELECT COUNT(*) FROM demandes_bal").fetchone()[0] clot = db.execute("SELECT COUNT(*) FROM demandes_bal WHERE statut='CLÔTURÉ'").fetchone()[0] encours = db.execute("SELECT COUNT(*) FROM demandes_bal WHERE statut='EN COURS'").fetchone()[0] attente = db.execute("SELECT COUNT(*) FROM demandes_bal WHERE statut='EN ATTENTE'").fetchone()[0] by_collab = db.execute(""" SELECT collaborateur, COUNT(*) AS nb, SUM(CASE WHEN statut='CLÔTURÉ' THEN 1 ELSE 0 END) AS cloturees FROM demandes_bal GROUP BY collaborateur ORDER BY nb DESC """).fetchall() by_client = db.execute(""" SELECT client, COUNT(*) AS nb, SUM(nb_sites) AS sites FROM demandes_bal GROUP BY client ORDER BY nb DESC LIMIT 10 """).fetchall() return jsonify({"total":total,"cloturees":clot,"en_cours":encours,"en_attente":attente, "by_collab":[dict(r) for r in by_collab], "by_client":[dict(r) for r in by_client]}) @app.route("/api/bal/kpi/export") @login_required @admin_required def api_bal_kpi_export(): """Export BAL KPI data as Excel workbook.""" from io import BytesIO from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils import get_column_letter db = get_db() total = db.execute("SELECT COUNT(*) FROM demandes_bal").fetchone()[0] clot = db.execute("SELECT COUNT(*) FROM demandes_bal WHERE statut='CLÔTURÉ'").fetchone()[0] encours = db.execute("SELECT COUNT(*) FROM demandes_bal WHERE statut='EN COURS'").fetchone()[0] attente = db.execute("SELECT COUNT(*) FROM demandes_bal WHERE statut='EN ATTENTE'").fetchone()[0] by_collab = db.execute(""" SELECT collaborateur, COUNT(*) AS nb, SUM(CASE WHEN statut='CLÔTURÉ' THEN 1 ELSE 0 END) AS cloturees FROM demandes_bal GROUP BY collaborateur ORDER BY nb DESC """).fetchall() by_client = db.execute(""" SELECT client, COUNT(*) AS nb, SUM(nb_sites) AS sites FROM demandes_bal GROUP BY client ORDER BY nb DESC """).fetchall() HEADER_FONT = Font(bold=True, color="FFFFFF", size=11) HEADER_FILL = PatternFill(start_color="1E293B", end_color="1E293B", fill_type="solid") HEADER_ALIGN = Alignment(horizontal="center", vertical="center") CELL_ALIGN = Alignment(vertical="center") NUM_ALIGN = Alignment(horizontal="right", vertical="center") TOTAL_FILL = PatternFill(start_color="F1F5F9", end_color="F1F5F9", fill_type="solid") TOTAL_FONT = Font(bold=True) GREEN_FILL = PatternFill(start_color="D1FAE5", end_color="D1FAE5", fill_type="solid") def style_header(ws, headers, row=1): for col, h in enumerate(headers, 1): cell = ws.cell(row=row, column=col, value=h) cell.font = HEADER_FONT cell.fill = HEADER_FILL cell.alignment = HEADER_ALIGN def style_cell(cell, is_number=False): cell.font = Font(size=10) cell.alignment = NUM_ALIGN if is_number else CELL_ALIGN def auto_width(ws, min_w=10, max_w=30): for col_cells in ws.columns: col_letter = get_column_letter(col_cells[0].column) max_len = max((len(str(c.value or "")) for c in col_cells), default=0) ws.column_dimensions[col_letter].width = min(max(max_len + 2, min_w), max_w) wb = Workbook() wb.remove(wb.active) # ── Sheet 1: Synthèse ── ws1 = wb.create_sheet("Synthèse BAL") style_header(ws1, ["Indicateur", "Valeur"]) rows1 = [ ("Total demandes", total), ("Clôturées", clot), ("En cours", encours), ("En attente", attente), ("Taux de clôture", f"{round(clot / total * 100) if total else 0}%"), ] for r, (label, val) in enumerate(rows1, 2): c1 = ws1.cell(row=r, column=1, value=label) c1.font = Font(bold=True, size=10) c1.alignment = CELL_ALIGN c2 = ws1.cell(row=r, column=2, value=val) style_cell(c2, isinstance(val, (int, float))) auto_width(ws1, min_w=14) # ── Sheet 2: Par Collaborateur ── ws2 = wb.create_sheet("Par Collaborateur") style_header(ws2, ["Collaborateur", "Nb Demandes", "Clôturées", "Taux Clôture"]) tN = tC = 0 for r_idx, c in enumerate(by_collab, 2): taux = round(c["cloturees"] / c["nb"] * 100) if c["nb"] else 0 tN += c["nb"]; tC += c["cloturees"] for col, v in enumerate([c["collaborateur"], c["nb"], c["cloturees"], f"{taux}%"], 1): cell = ws2.cell(row=r_idx, column=col, value=v) style_cell(cell, col >= 2) tr2 = len(by_collab) + 2 taux_t = round(tC / tN * 100) if tN else 0 for col, v in enumerate(["TOTAL", tN, tC, f"{taux_t}%"], 1): cell = ws2.cell(row=tr2, column=col, value=v) cell.font = TOTAL_FONT cell.fill = TOTAL_FILL style_cell(cell, col >= 2) auto_width(ws2) # ── Sheet 3: Par Client ── ws3 = wb.create_sheet("Par Client") style_header(ws3, ["Client", "Nb Demandes", "Total Sites"]) for r_idx, c in enumerate(by_client, 2): for col, v in enumerate([c["client"], c["nb"], c["sites"]], 1): cell = ws3.cell(row=r_idx, column=col, value=v) style_cell(cell, col >= 2) auto_width(ws3) # ── Save ── output = BytesIO() wb.save(output) output.seek(0) filename = f"KPI_BAL_{date.today().isoformat()}.xlsx" return send_file( output, mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", as_attachment=True, download_name=filename ) @app.route("/bal/kpi") @login_required @admin_required def bal_kpi(): user = get_current_user() return render_template("bal_kpi.html", user=user, is_admin=is_admin()) @app.route("/bal/clients") @login_required @admin_required def bal_clients(): user = get_current_user() db = get_db() clients = db.execute("SELECT * FROM clients_bal ORDER BY nom").fetchall() return render_template("bal_clients.html", user=user, clients=clients, is_admin=True) @app.route("/bal/collaborateurs") @login_required @admin_required def bal_collaborateurs(): user = get_current_user() return render_template("bal_collaborateurs.html", user=user, collaborateurs=COLLABORATEURS, is_admin=True) @app.route("/settings") @login_required @admin_required def settings(): user = get_current_user() db = get_db() users = db.execute("SELECT * FROM users ORDER BY created_at").fetchall() clients = db.execute("SELECT * FROM clients_bal ORDER BY nom").fetchall() invitations = db.execute( "SELECT * FROM invitations ORDER BY created_at DESC" ).fetchall() if session.get("role") == "super_admin" else [] return render_template("settings.html", user=user, users=users, clients=clients, collaborateurs=COLLABORATEURS, ROLES=ROLES, is_admin=True, invitations=invitations, now=datetime.now().strftime("%Y-%m-%d %H:%M:%S")) # ── Dashboard ────────────────────────────────────────────────────────────── @app.route("/") @login_required def dashboard(): user = get_current_user() db = get_db() admin = is_admin() if admin: projets = db.execute(""" SELECT p.*, (SELECT COUNT(*) FROM affectations WHERE projet_id=p.id) AS nb_collabs, COALESCE((SELECT SUM(nb_sites) FROM traitements WHERE projet_id=p.id),0) AS total_traites FROM projets p ORDER BY p.created_at DESC """).fetchall() else: projets = db.execute(""" SELECT DISTINCT p.*, (SELECT COUNT(*) FROM affectations WHERE projet_id=p.id) AS nb_collabs, COALESCE((SELECT SUM(nb_sites) FROM traitements WHERE projet_id=p.id),0) AS total_traites FROM projets p JOIN affectations a ON a.projet_id = p.id WHERE a.collaborateur = ? ORDER BY p.created_at DESC """, (user["full_name"],)).fetchall() result = [] for p in projets: collabs = db.execute(""" SELECT a.collaborateur, a.prevu_sites, a.helico_cr, COALESCE(SUM(t.nb_sites),0) AS traites FROM affectations a LEFT JOIN traitements t ON t.projet_id=a.projet_id AND t.collaborateur=a.collaborateur WHERE a.projet_id=? GROUP BY a.collaborateur ORDER BY a.collaborateur """, (p["id"],)).fetchall() result.append({"projet": p, "collaborateurs": [dict(c) for c in collabs]}) return render_template("dashboard.html", user=user, projets=result, collaborateurs=COLLABORATEURS, STATUTS=STATUTS, is_admin=admin) @app.route("/kpi") @login_required @admin_required def kpi(): user = get_current_user() return render_template("kpi.html", user=user, collaborateurs=COLLABORATEURS, is_admin=True) # ── Projet detail ────────────────────────────────────────────────────────── @app.route("/projet/") @login_required def projet_detail(pid): user = get_current_user() db = get_db() projet = db.execute("SELECT * FROM projets WHERE id=?", (pid,)).fetchone() if not projet: return "Projet introuvable", 404 collabs = db.execute(""" SELECT a.*, COALESCE(SUM(t.nb_sites),0) AS traites FROM affectations a LEFT JOIN traitements t ON t.projet_id=a.projet_id AND t.collaborateur=a.collaborateur WHERE a.projet_id=? GROUP BY a.collaborateur ORDER BY a.collaborateur """, (pid,)).fetchall() return render_template("projet.html", user=user, projet=projet, collabs=collabs, collaborateurs=COLLABORATEURS, STATUTS=STATUTS) # ── Calendar ─────────────────────────────────────────────────────────────── @app.route("/projet//") @login_required def calendrier(pid, collaborateur): user = get_current_user() if not is_admin() and collaborateur.upper() != user["full_name"].upper(): return redirect(url_for("dashboard")) db = get_db() projet = db.execute("SELECT * FROM projets WHERE id=?", (pid,)).fetchone() if not projet: return "Projet introuvable", 404 aff = db.execute( "SELECT prevu_sites FROM affectations WHERE projet_id=? AND collaborateur=?", (pid, collaborateur)).fetchone() today = date.today() year = int(request.args.get("year", today.year)) month = int(request.args.get("month", today.month)) start = date(year, month, 1) end = date(year, month+1, 1) - timedelta(days=1) if month < 12 else date(year+1, 1, 1) - timedelta(days=1) traitements = db.execute(""" SELECT date_traitement, nb_sites, helico, helico_cr, cr FROM traitements WHERE projet_id=? AND collaborateur=? AND date_traitement BETWEEN ? AND ? """, (pid, collaborateur, start.isoformat(), end.isoformat())).fetchall() tmap = {} for r in traitements: tmap[r["date_traitement"]] = { "nb": r["nb_sites"], "helico": bool(r["helico"]), "helico_cr": bool(r["helico_cr"]), "cr": bool(r["cr"]) } total_traite = db.execute( "SELECT COALESCE(SUM(nb_sites),0) FROM traitements WHERE projet_id=? AND collaborateur=?", (pid, collaborateur)).fetchone()[0] cal = [] for _ in range(start.weekday()): cal.append(None) for d in range(1, end.day+1): cal.append(d) pm, py = ((month-1) or 12, year if month > 1 else year-1) nm, ny = ((month % 12)+1, year if month < 12 else year+1) return render_template("calendrier.html", user=user, projet=projet, collaborateur=collaborateur, prevu=aff["prevu_sites"] if aff else 0, total_traite=total_traite, year=year, month=month, cal=cal, tmap=tmap, prev_month=pm, prev_year=py, next_month=nm, next_year=ny, today=today, is_admin=is_admin()) # ── Admin: Users ─────────────────────────────────────────────────────────── @app.route("/admin/users") @super_admin_required def admin_users(): user = get_current_user() db = get_db() users = db.execute("SELECT * FROM users ORDER BY created_at").fetchall() invitations = db.execute( "SELECT * FROM invitations ORDER BY created_at DESC" ).fetchall() return render_template("admin_users.html", user=user, users=users, ROLES=ROLES, invitations=invitations, now=datetime.now().strftime("%Y-%m-%d %H:%M:%S")) @app.route("/api/admin/users/new", methods=["POST"]) @super_admin_required def api_user_new(): data = request.get_json() username = data.get("username", "").strip().lower() password = data.get("password", "").strip() full_name = data.get("full_name", "").strip().upper() role = data.get("role", "membre") if not username or not password or not full_name: return jsonify({"ok": False, "error": "Tous les champs sont obligatoires"}), 400 if role not in ROLES: return jsonify({"ok": False, "error": "Rôle invalide"}), 400 ok, msg = validate_password(password) if not ok: return jsonify({"ok": False, "error": msg}), 400 db = get_db() try: db.execute("INSERT INTO users (username, password_hash, full_name, role, must_change_password) VALUES (?,?,?,?,1)", (username, generate_password_hash(password), full_name, role)) db.commit() log.info("USER_CREATED username=%s role=%s by=%s", username, role, session["username"]) return jsonify({"ok": True}) except sqlite3.IntegrityError: return jsonify({"ok": False, "error": "Ce login existe déjà"}), 409 @app.route("/api/admin/users/", methods=["PUT"]) @super_admin_required def api_user_edit(uid): data = request.get_json() db = get_db() target = db.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone() if not target: return jsonify({"ok": False, "error": "Utilisateur introuvable"}), 404 if target["role"] == "super_admin" and uid != session["user_id"]: return jsonify({"ok": False, "error": "Impossible de modifier un autre Super Admin"}), 403 username = data.get("username", target["username"]).strip().lower() full_name = data.get("full_name", target["full_name"]).strip().upper() role = data.get("role", target["role"]) if role not in ROLES: return jsonify({"ok": False, "error": "Rôle invalide"}), 400 if data.get("password"): ok, msg = validate_password(data["password"]) if not ok: return jsonify({"ok": False, "error": msg}), 400 db.execute("UPDATE users SET username=?, password_hash=?, full_name=?, role=? WHERE id=?", (username, generate_password_hash(data["password"]), full_name, role, uid)) else: db.execute("UPDATE users SET username=?, full_name=?, role=? WHERE id=?", (username, full_name, role, uid)) db.commit() log.info("USER_UPDATED uid=%s by=%s", uid, session["username"]) return jsonify({"ok": True}) @app.route("/api/admin/users/", methods=["DELETE"]) @super_admin_required def api_user_delete(uid): if uid == session["user_id"]: return jsonify({"ok": False, "error": "Vous ne pouvez pas supprimer votre propre compte"}), 400 db = get_db() target = db.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone() if target and target["role"] == "super_admin": return jsonify({"ok": False, "error": "Impossible de supprimer un Super Admin"}), 403 db.execute("DELETE FROM users WHERE id=?", (uid,)) db.commit() log.info("USER_DELETED uid=%s by=%s", uid, session["username"]) return jsonify({"ok": True}) # ── API Projets ──────────────────────────────────────────────────────────── @app.route("/api/projets") @login_required def api_projets(): db = get_db() projets = db.execute(""" SELECT p.*, (SELECT COUNT(*) FROM affectations WHERE projet_id=p.id) AS nb_collabs, COALESCE((SELECT SUM(nb_sites) FROM traitements WHERE projet_id=p.id),0) AS total_traites FROM projets p ORDER BY p.created_at DESC """).fetchall() result = [] for p in projets: collabs = db.execute(""" SELECT a.collaborateur, a.prevu_sites, a.helico_cr, COALESCE(SUM(t.nb_sites),0) AS traites FROM affectations a LEFT JOIN traitements t ON t.projet_id=a.projet_id AND t.collaborateur=a.collaborateur WHERE a.projet_id=? GROUP BY a.collaborateur ORDER BY a.collaborateur """, (p["id"],)).fetchall() result.append({"id": p["id"], "nom": p["nom"], "numero_requete": p["numero_requete"], "total_sites": p["total_sites"], "statut": p["statut"], "created_at": p["created_at"], "nb_collabs": p["nb_collabs"], "total_traites": p["total_traites"], "collaborateurs": [dict(c) for c in collabs]}) return jsonify(result) @app.route("/api/projet/new", methods=["POST"]) @admin_required def api_new_projet(): data = request.get_json() nom = data.get("nom", "").strip().upper() if not nom: return jsonify({"ok": False, "error": "Nom requis"}), 400 db = get_db() try: cur = db.execute("INSERT INTO projets (nom, numero_requete, total_sites) VALUES (?,?,?)", (nom, data.get("numero_requete", ""), int(data.get("total_sites", 0)))) pid = cur.lastrowid for c in data.get("collaborateurs", []): db.execute("INSERT INTO affectations (projet_id, collaborateur, prevu_sites, helico_cr) VALUES (?,?,?,?)", (pid, c["collaborateur"], int(c.get("prevu_sites", 0)), 1 if c.get("helico_cr") else 0)) db.commit() log.info("PROJET_CREATED id=%s nom=%s by=%s", pid, nom, session["username"]) return jsonify({"ok": True, "id": pid}) except sqlite3.IntegrityError: return jsonify({"ok": False, "error": "Ce projet existe déjà"}), 409 @app.route("/api/projet/", methods=["PUT"]) @admin_required def api_update_projet(pid): data = request.get_json() db = get_db() db.execute("UPDATE projets SET nom=?, numero_requete=?, total_sites=?, statut=? WHERE id=?", (data.get("nom", "").strip().upper(), data.get("numero_requete", ""), int(data.get("total_sites", 0)), data.get("statut", "NOUVEAU"), pid)) db.execute("DELETE FROM affectations WHERE projet_id=?", (pid,)) for c in data.get("collaborateurs", []): db.execute("INSERT INTO affectations (projet_id, collaborateur, prevu_sites, helico_cr) VALUES (?,?,?,?)", (pid, c["collaborateur"], int(c.get("prevu_sites", 0)), 1 if c.get("helico_cr") else 0)) db.commit() log.info("PROJET_UPDATED id=%s by=%s", pid, session["username"]) return jsonify({"ok": True}) @app.route("/api/projet/", methods=["DELETE"]) @admin_required def api_delete_projet(pid): get_db().execute("DELETE FROM projets WHERE id=?", (pid,)) get_db().commit() log.info("PROJET_DELETED id=%s by=%s", pid, session["username"]) return jsonify({"ok": True}) def _check_project_affectation(pid, collab): """Verify collaborator is assigned to the project. Returns True if assigned.""" db = get_db() aff = db.execute( "SELECT 1 FROM affectations WHERE projet_id=? AND collaborateur=?", (pid, collab) ).fetchone() return aff is not None @app.route("/api/projet//traitement", methods=["POST"]) @login_required def api_add_traitement(pid): data = request.get_json() collab = data.get("collaborateur", "") if not is_admin() and collab.upper() != session.get("full_name", "").upper(): return jsonify({"ok": False, "error": "Accès refusé"}), 403 # Verify collaborator is assigned to this project if not _check_project_affectation(pid, collab): return jsonify({"ok": False, "error": "Collaborateur non affecté à ce projet"}), 403 date_str = data.get("date", "") nb = int(data.get("nb_sites", 0)) helico = 1 if data.get("helico") else 0 helico_cr = 1 if data.get("helico_cr") else 0 cr = 1 if data.get("cr") else 0 if not collab or not date_str: return jsonify({"ok": False}), 400 db = get_db() db.execute("""INSERT INTO traitements (projet_id, collaborateur, date_traitement, nb_sites, helico, helico_cr, cr) VALUES (?,?,?,?,?,?,?) ON CONFLICT(projet_id, collaborateur, date_traitement) DO UPDATE SET nb_sites=?, helico=?, helico_cr=?, cr=?""", (pid, collab, date_str, nb, helico, helico_cr, cr, nb, helico, helico_cr, cr)) db.commit() log.info("TRAITEMENT_ADD pid=%s collab=%s date=%s nb=%s by=%s", pid, collab, date_str, nb, session["username"]) total = db.execute("SELECT COALESCE(SUM(nb_sites),0) FROM traitements WHERE projet_id=? AND collaborateur=?", (pid, collab)).fetchone()[0] return jsonify({"ok": True, "total": total}) @app.route("/api/projet//traitement/", methods=["DELETE"]) @login_required def api_delete_traitement(pid, date_str): collab = request.args.get("collaborateur", "") if not is_admin() and collab.upper() != session.get("full_name", "").upper(): return jsonify({"ok": False, "error": "Accès refusé"}), 403 # Verify collaborator is assigned to this project if not _check_project_affectation(pid, collab): return jsonify({"ok": False, "error": "Collaborateur non affecté à ce projet"}), 403 db = get_db() db.execute("DELETE FROM traitements WHERE projet_id=? AND collaborateur=? AND date_traitement=?", (pid, collab, date_str)) db.commit() log.info("TRAITEMENT_DELETE pid=%s collab=%s date=%s by=%s", pid, collab, date_str, session["username"]) total = db.execute("SELECT COALESCE(SUM(nb_sites),0) FROM traitements WHERE projet_id=? AND collaborateur=?", (pid, collab)).fetchone()[0] return jsonify({"ok": True, "total": total}) # ── API KPI ──────────────────────────────────────────────────────────────── def _build_taux_sheet(ws, data, key_field, capacite): """Build a pivot-style taux de charge sheet for Excel export.""" from openpyxl.styles import Font, PatternFill, Alignment, Border, Side COLLABS_TC = ["NIZAR", "YOUSRA", "IMANE", "SANAA", "HAYAT"] HEADER_FONT = Font(bold=True, color="FFFFFF", size=11) HEADER_FILL = PatternFill(start_color="1E293B", end_color="1E293B", fill_type="solid") HEADER_ALIGN = Alignment(horizontal="center", vertical="center") NUM_ALIGN = Alignment(horizontal="right", vertical="center") CELL_ALIGN = Alignment(vertical="center") THIN_BORDER = Border(bottom=Side(style="thin", color="E2E8F0")) headers = [key_field.capitalize()] + COLLABS_TC for col, h in enumerate(headers, 1): cell = ws.cell(row=1, column=col, value=h) cell.font = HEADER_FONT cell.fill = HEADER_FILL cell.alignment = HEADER_ALIGN periods = sorted(set(r[key_field] for r in data)) if not periods: ws.cell(row=2, column=1, value="Aucune donnée") return for r_idx, period in enumerate(periods, 2): ws.cell(row=r_idx, column=1, value=period).font = Font(bold=True, size=10) ws.cell(row=r_idx, column=1).border = THIN_BORDER for c_idx, collab in enumerate(COLLABS_TC, 2): entry = next((r for r in data if r[key_field] == period and r["collaborateur"] == collab), None) if entry and entry["jours"] > 0: taux = round(entry["sites"] / (entry["jours"] * capacite) * 100) val = f"{taux}%" else: val = "—" cell = ws.cell(row=r_idx, column=c_idx, value=val) cell.font = Font(size=10) cell.alignment = NUM_ALIGN cell.border = THIN_BORDER # Auto-width from openpyxl.utils import get_column_letter for col_cells in ws.columns: col_letter = get_column_letter(col_cells[0].column) max_len = max((len(str(c.value or "")) for c in col_cells), default=0) ws.column_dimensions[col_letter].width = min(max(max_len + 2, 10), 28) @app.route("/api/kpi") @login_required @admin_required def api_kpi(): db = get_db() by_collab = db.execute(""" SELECT a.collaborateur, COUNT(DISTINCT a.projet_id) AS nb_projets, COALESCE(SUM(a.prevu_sites),0) AS total_prevu, COALESCE((SELECT SUM(nb_sites) FROM traitements t WHERE t.collaborateur=a.collaborateur),0) AS total_traite FROM affectations a GROUP BY a.collaborateur ORDER BY total_traite DESC """).fetchall() collab_projets = {} for c in by_collab: projets = db.execute(""" SELECT p.nom, p.statut, p.total_sites, a.prevu_sites, COALESCE((SELECT SUM(nb_sites) FROM traitements t WHERE t.projet_id=p.id AND t.collaborateur=a.collaborateur),0) AS traites FROM affectations a JOIN projets p ON p.id=a.projet_id WHERE a.collaborateur=? ORDER BY traites DESC """, (c["collaborateur"],)).fetchall() collab_projets[c["collaborateur"]] = [dict(r) for r in projets] weekly = db.execute(""" SELECT strftime('%Y-%W', date_traitement) AS week, collaborateur, SUM(nb_sites) AS total FROM traitements WHERE date_traitement >= date('now', '-84 days') GROUP BY week, collaborateur ORDER BY week """).fetchall() statuts = db.execute("SELECT statut, COUNT(*) AS nb, COALESCE(SUM(total_sites),0) AS sites FROM projets GROUP BY statut").fetchall() CAPACITE = 70 taux_hebdo = db.execute(""" SELECT strftime('%Y-W%W', date_traitement) AS semaine, collaborateur, SUM(nb_sites) AS sites, COUNT(DISTINCT date_traitement) AS jours FROM traitements GROUP BY semaine, collaborateur ORDER BY semaine """).fetchall() taux_mensuel = db.execute(""" SELECT strftime('%Y-%m', date_traitement) AS mois, collaborateur, SUM(nb_sites) AS sites, COUNT(DISTINCT date_traitement) AS jours FROM traitements GROUP BY mois, collaborateur ORDER BY mois """).fetchall() return jsonify({ "by_collab": [dict(r) for r in by_collab], "collab_projets": collab_projets, "weekly": [dict(r) for r in weekly], "statuts": [dict(r) for r in statuts], "taux_hebdo": [dict(r) for r in taux_hebdo], "taux_mensuel": [dict(r) for r in taux_mensuel], "capacite": CAPACITE, }) @app.route("/api/kpi/export") @login_required @admin_required def api_kpi_export(): """Export all KPI data as an Excel workbook with multiple sheets.""" from io import BytesIO from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter db = get_db() # ── Collect data (same queries as api_kpi) ── by_collab = db.execute(""" SELECT a.collaborateur, COUNT(DISTINCT a.projet_id) AS nb_projets, COALESCE(SUM(a.prevu_sites),0) AS total_prevu, COALESCE((SELECT SUM(nb_sites) FROM traitements t WHERE t.collaborateur=a.collaborateur),0) AS total_traite FROM affectations a GROUP BY a.collaborateur ORDER BY total_traite DESC """).fetchall() collab_projets = {} for c in by_collab: projets = db.execute(""" SELECT p.nom, p.statut, p.total_sites, a.prevu_sites, COALESCE((SELECT SUM(nb_sites) FROM traitements t WHERE t.projet_id=p.id AND t.collaborateur=a.collaborateur),0) AS traites FROM affectations a JOIN projets p ON p.id=a.projet_id WHERE a.collaborateur=? ORDER BY traites DESC """, (c["collaborateur"],)).fetchall() collab_projets[c["collaborateur"]] = [dict(r) for r in projets] weekly = db.execute(""" SELECT strftime('%Y-%W', date_traitement) AS week, collaborateur, SUM(nb_sites) AS total FROM traitements WHERE date_traitement >= date('now', '-84 days') GROUP BY week, collaborateur ORDER BY week """).fetchall() statuts = db.execute("SELECT statut, COUNT(*) AS nb, COALESCE(SUM(total_sites),0) AS sites FROM projets GROUP BY statut").fetchall() CAPACITE = 70 taux_hebdo = db.execute(""" SELECT strftime('%Y-W%W', date_traitement) AS semaine, collaborateur, SUM(nb_sites) AS sites, COUNT(DISTINCT date_traitement) AS jours FROM traitements GROUP BY semaine, collaborateur ORDER BY semaine """).fetchall() taux_mensuel = db.execute(""" SELECT strftime('%Y-%m', date_traitement) AS mois, collaborateur, SUM(nb_sites) AS sites, COUNT(DISTINCT date_traitement) AS jours FROM traitements GROUP BY mois, collaborateur ORDER BY mois """).fetchall() # ── Styles ── HEADER_FONT = Font(bold=True, color="FFFFFF", size=11) HEADER_FILL = PatternFill(start_color="1E293B", end_color="1E293B", fill_type="solid") HEADER_ALIGN = Alignment(horizontal="center", vertical="center") CELL_ALIGN = Alignment(vertical="center") NUM_ALIGN = Alignment(horizontal="right", vertical="center") TOTAL_FILL = PatternFill(start_color="F1F5F9", end_color="F1F5F9", fill_type="solid") TOTAL_FONT = Font(bold=True) THIN_BORDER = Border( bottom=Side(style="thin", color="E2E8F0") ) STATUT_COLORS = { "LIVRER": PatternFill(start_color="D1FAE5", end_color="D1FAE5", fill_type="solid"), "EN COURS": PatternFill(start_color="FEF3C7", end_color="FEF3C7", fill_type="solid"), "NOUVEAU": PatternFill(start_color="DBEAFE", end_color="DBEAFE", fill_type="solid"), } def style_header(ws, headers, row=1): for col, h in enumerate(headers, 1): cell = ws.cell(row=row, column=col, value=h) cell.font = HEADER_FONT cell.fill = HEADER_FILL cell.alignment = HEADER_ALIGN def style_data_cell(cell, is_number=False): cell.font = Font(size=10) cell.alignment = NUM_ALIGN if is_number else CELL_ALIGN cell.border = THIN_BORDER def auto_width(ws, min_w=10, max_w=28): for col_cells in ws.columns: col_letter = get_column_letter(col_cells[0].column) max_len = max((len(str(c.value or "")) for c in col_cells), default=0) ws.column_dimensions[col_letter].width = min(max(max_len + 2, min_w), max_w) SL = {"NOUVEAU": "Nouveau", "EN COURS": "En cours", "LIVRER": "Livré"} # ═══ Create Workbook ═══ wb = Workbook() # Remove default sheet wb.remove(wb.active) # ── Sheet 1: Synthèse ── ws1 = wb.create_sheet("Synthèse") total_projets = sum(s["nb"] for s in statuts) total_sites = sum(s["sites"] for s in statuts) livres = next((s for s in statuts if s["statut"] == "LIVRER"), {"nb": 0, "sites": 0}) encours = next((s for s in statuts if s["statut"] == "EN COURS"), {"nb": 0, "sites": 0}) nouveaux = next((s for s in statuts if s["statut"] == "NOUVEAU"), {"nb": 0, "sites": 0}) taux_livraison = round(livres["nb"] / total_projets * 100) if total_projets else 0 taux_sites = round(livres["sites"] / total_sites * 100) if total_sites else 0 headers1 = ["Indicateur", "Valeur"] style_header(ws1, headers1) rows1 = [ ("Total Projets", total_projets), ("Total Sites", total_sites), ("Projets Livrés", livres["nb"]), ("Sites Livrés", livres["sites"]), ("Taux de Livraison (projets)", f"{taux_livraison}%"), ("Taux de Livraison (sites)", f"{taux_sites}%"), ("Projets En Cours", encours["nb"]), ("Projets Nouveaux", nouveaux["nb"]), ("Nb Collaborateurs", len(by_collab)), ] for r, (label, val) in enumerate(rows1, 2): c1 = ws1.cell(row=r, column=1, value=label) c1.font = Font(bold=True, size=10) c1.alignment = CELL_ALIGN c1.border = THIN_BORDER c2 = ws1.cell(row=r, column=2, value=val) style_data_cell(c2, isinstance(val, (int, float))) auto_width(ws1, min_w=14) # ── Sheet 2: Par Statut ── ws2 = wb.create_sheet("Par Statut") headers2 = ["Statut", "Nb Projets", "% Projets", "Nb Sites", "% Sites", "Sites Moy."] style_header(ws2, headers2) for r, s in enumerate(statuts, 2): pct_p = round(s["nb"] / total_projets * 100, 1) if total_projets else 0 pct_s = round(s["sites"] / total_sites * 100, 1) if total_sites else 0 moy = round(s["sites"] / s["nb"]) if s["nb"] else 0 vals = [SL.get(s["statut"], s["statut"]), s["nb"], f"{pct_p}%", s["sites"], f"{pct_s}%", moy] for col, v in enumerate(vals, 1): c = ws2.cell(row=r, column=col, value=v) style_data_cell(c, col >= 2) # Total row total_vals = ["TOTAL", total_projets, "100%", total_sites, "100%", round(total_sites / total_projets) if total_projets else 0] tr = len(statuts) + 2 for col, v in enumerate(total_vals, 1): c = ws2.cell(row=tr, column=col, value=v) c.font = TOTAL_FONT c.fill = TOTAL_FILL style_data_cell(c, col >= 2) auto_width(ws2) # ── Sheet 3: Par Collaborateur ── ws3 = wb.create_sheet("Par Collaborateur") headers3 = ["Collaborateur", "Projets", "Sites", "Livrés", "En Cours", "Nouveaux", "Taux Livr."] style_header(ws3, headers3) tP = tS = tL = tE = tN = 0 for r_idx, cdata in enumerate(by_collab, 2): projs = collab_projets.get(cdata["collaborateur"], []) nbP = len(projs) nbS = sum(p["total_sites"] for p in projs) nbL = sum(1 for p in projs if p["statut"] == "LIVRER") nbE = sum(1 for p in projs if p["statut"] == "EN COURS") nbN = sum(1 for p in projs if p["statut"] == "NOUVEAU") taux = round(nbL / nbP * 100) if nbP else 0 tP += nbP; tS += nbS; tL += nbL; tE += nbE; tN += nbN vals = [cdata["collaborateur"], nbP, nbS, nbL, nbE, nbN, f"{taux}%"] for col, v in enumerate(vals, 1): c = ws3.cell(row=r_idx, column=col, value=v) style_data_cell(c, col >= 2) # Total row taux_t = round(tL / tP * 100) if tP else 0 total3 = ["TOTAL", tP, tS, tL, tE, tN, f"{taux_t}%"] tr3 = len(by_collab) + 2 for col, v in enumerate(total3, 1): c = ws3.cell(row=tr3, column=col, value=v) c.font = TOTAL_FONT c.fill = TOTAL_FILL style_data_cell(c, col >= 2) auto_width(ws3) # ── Sheet 4: Top Projets ── ws4 = wb.create_sheet("Top Projets") headers4 = ["Projet", "Collaborateur", "Nb Sites", "Statut"] style_header(ws4, headers4) all_projs = [] for collab, projs in collab_projets.items(): for p in projs: all_projs.append((p["nom"], collab, p["total_sites"], p["statut"])) all_projs.sort(key=lambda x: -x[2]) for r_idx, (nom, collab, sites, statut) in enumerate(all_projs, 2): vals = [nom, collab, sites, SL.get(statut, statut)] for col, v in enumerate(vals, 1): c = ws4.cell(row=r_idx, column=col, value=v) style_data_cell(c, col == 3) if col == 4 and statut in STATUT_COLORS: c.fill = STATUT_COLORS[statut] auto_width(ws4) # ── Sheet 5: Taux de Charge Hebdo ── ws5 = wb.create_sheet("Taux Charge Hebdo") _build_taux_sheet(ws5, taux_hebdo, "semaine", CAPACITE) # ── Sheet 6: Taux de Charge Mensuel ── ws6 = wb.create_sheet("Taux Charge Mensuel") _build_taux_sheet(ws6, taux_mensuel, "mois", CAPACITE) # ── Save to BytesIO ── output = BytesIO() wb.save(output) output.seek(0) filename = f"KPI_Suivi_Production_{date.today().isoformat()}.xlsx" return send_file( output, mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", as_attachment=True, download_name=filename ) # ── Congés ───────────────────────────────────────────────────────────────── CONGE_STATUTS_ORDER = ["EN ATTENTE", "VALIDÉ", "REFUSÉ", "ANNULÉ"] CONGE_STATUTS_BADGE = { "EN ATTENTE": "badge-yellow", "VALIDÉ": "badge-green", "REFUSÉ": "badge-red", "ANNULÉ": "badge-orange" } @app.route("/conges") @login_required def conges_list(): user = get_current_user() db = get_db() admin = is_admin() if admin: conges = db.execute(""" SELECT c.*, tc.nom AS type_nom, tc.couleur, d.full_name AS demandeur_nom, p.full_name AS personne_nom, v.full_name AS valideur_nom, p.equipe FROM conges c JOIN users d ON d.id = c.demandeur_id JOIN users p ON p.id = c.personne_id JOIN types_conge tc ON tc.id = c.type_id LEFT JOIN users v ON v.id = c.valide_par ORDER BY c.date_debut DESC, c.id DESC """).fetchall() else: conges = db.execute(""" SELECT c.*, tc.nom AS type_nom, tc.couleur, d.full_name AS demandeur_nom, p.full_name AS personne_nom, v.full_name AS valideur_nom, p.equipe FROM conges c JOIN users d ON d.id = c.demandeur_id JOIN users p ON p.id = c.personne_id JOIN types_conge tc ON tc.id = c.type_id LEFT JOIN users v ON v.id = c.valide_par WHERE c.personne_id = ? ORDER BY c.date_debut DESC, c.id DESC """, (user["id"],)).fetchall() types = db.execute("SELECT * FROM types_conge WHERE actif=1 ORDER BY nom").fetchall() personnes = db.execute("SELECT id, full_name, equipe FROM users ORDER BY full_name").fetchall() return render_template("conges_list.html", user=user, conges=conges, types=types, personnes=personnes, statuts=CONGE_STATUTS_ORDER, is_admin=admin, statut_badges=CONGE_STATUTS_BADGE) @app.route("/conges/new", methods=["GET", "POST"]) @login_required def conges_new(): user = get_current_user() db = get_db() if request.method == "POST": personne_id = int(request.form.get("personne_id", user["id"])) # Non-admin can only create for themselves if not is_admin() and personne_id != user["id"]: return redirect(url_for("conges_list")) db.execute("""INSERT INTO conges (demandeur_id, personne_id, type_id, date_debut, date_fin, nb_jours, statut, commentaire) VALUES (?,?,?,?,?,?,?,?)""", (user["id"], personne_id, int(request.form["type_id"]), request.form["date_debut"], request.form["date_fin"], float(request.form.get("nb_jours", 1)), request.form.get("statut", "EN ATTENTE"), request.form.get("commentaire", ""))) db.commit() log.info("CONGE_CREATED by=%s for=%s", session["username"], personne_id) return redirect(url_for("conges_list")) types = db.execute("SELECT * FROM types_conge WHERE actif=1 ORDER BY nom").fetchall() personnes = db.execute("SELECT id, full_name FROM users ORDER BY full_name").fetchall() return render_template("conges_form.html", user=user, conge=None, types=types, personnes=personnes, is_admin=is_admin(), today=date.today().isoformat()) @app.route("/conges//edit", methods=["GET", "POST"]) @login_required def conges_edit(cid): user = get_current_user() db = get_db() conge = db.execute("SELECT * FROM conges WHERE id=?", (cid,)).fetchone() if not conge: return "Introuvable", 404 # Only admin or the original demandeur can edit (if still pending) if not is_admin() and (conge["demandeur_id"] != user["id"] or conge["statut"] != "EN ATTENTE"): return redirect(url_for("conges_list")) if request.method == "POST": personne_id = int(request.form.get("personne_id", conge["personne_id"])) if not is_admin(): personne_id = conge["personne_id"] db.execute("""UPDATE conges SET personne_id=?, type_id=?, date_debut=?, date_fin=?, nb_jours=?, commentaire=?, updated_at=CURRENT_TIMESTAMP WHERE id=?""", (personne_id, int(request.form["type_id"]), request.form["date_debut"], request.form["date_fin"], float(request.form.get("nb_jours", conge["nb_jours"])), request.form.get("commentaire", conge["commentaire"] or ""), cid)) db.commit() log.info("CONGE_UPDATED id=%s by=%s", cid, session["username"]) return redirect(url_for("conges_list")) types = db.execute("SELECT * FROM types_conge WHERE actif=1 ORDER BY nom").fetchall() personnes = db.execute("SELECT id, full_name FROM users ORDER BY full_name").fetchall() return render_template("conges_form.html", user=user, conge=conge, types=types, personnes=personnes, is_admin=is_admin()) @app.route("/conges//statut", methods=["POST"]) @login_required @admin_required def conges_statut(cid): db = get_db() statut = request.form.get("statut", "") if statut not in CONGE_STATUTS: return redirect(url_for("conges_list")) now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") db.execute("UPDATE conges SET statut=?, valide_par=?, date_validation=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (statut, session["user_id"], now, cid)) db.commit() log.info("CONGE_STATUT id=%s statut=%s by=%s", cid, statut, session["username"]) return redirect(url_for("conges_list")) @app.route("/conges//delete", methods=["POST"]) @login_required @admin_required def conges_delete(cid): db = get_db() db.execute("DELETE FROM conges WHERE id=?", (cid,)) db.commit() log.info("CONGE_DELETED id=%s by=%s", cid, session["username"]) return redirect(url_for("conges_list")) @app.route("/conges/calendrier") @login_required def conges_calendrier(): user = get_current_user() db = get_db() admin = is_admin() today = date.today() year = int(request.args.get("year", today.year)) month = int(request.args.get("month", today.month)) personne = request.args.get("personne", "") start = date(year, month, 1) if month == 12: end = date(year + 1, 1, 1) - timedelta(days=1) else: end = date(year, month + 1, 1) - timedelta(days=1) if admin and personne: conges_query = """SELECT c.*, tc.nom, tc.couleur, p.full_name, u.equipe FROM conges c JOIN users p ON p.id = c.personne_id JOIN types_conge tc ON tc.id = c.type_id JOIN users u ON u.id = c.personne_id WHERE c.statut IN ('EN ATTENTE','VALIDÉ') AND u.full_name = ? AND date_fin >= ? AND date_debut <= ?""" conges_rows = db.execute(conges_query, (personne.upper(), start.isoformat(), end.isoformat())).fetchall() elif admin and not personne: conges_query = """SELECT c.*, tc.nom, tc.couleur, p.full_name, u.equipe FROM conges c JOIN users p ON p.id = c.personne_id JOIN types_conge tc ON tc.id = c.type_id JOIN users u ON u.id = c.personne_id WHERE c.statut IN ('EN ATTENTE','VALIDÉ') AND date_fin >= ? AND date_debut <= ?""" conges_rows = db.execute(conges_query, (start.isoformat(), end.isoformat())).fetchall() else: conges_query = """SELECT c.*, tc.nom, tc.couleur, p.full_name, u.equipe FROM conges c JOIN users p ON p.id = c.personne_id JOIN types_conge tc ON tc.id = c.type_id JOIN users u ON u.id = c.personne_id WHERE c.personne_id = ? AND c.statut IN ('EN ATTENTE','VALIDÉ') AND date_fin >= ? AND date_debut <= ?""" conges_rows = db.execute(conges_query, (user["id"], start.isoformat(), end.isoformat())).fetchall() feries = db.execute("SELECT date_ferie, nom FROM jours_feries WHERE date_ferie BETWEEN ? AND ?", (start.isoformat(), end.isoformat())).fetchall() feries_map = {r["date_ferie"]: r["nom"] for r in feries} personnes = db.execute("SELECT full_name, equipe FROM users ORDER BY full_name").fetchall() cal = [] first_weekday = start.weekday() for _ in range(first_weekday): cal.append(None) for d in range(1, end.day + 1): ds = f"{year:04d}-{month:02d}-{d:02d}" entry = {"day": d, "date": ds, "conges": [], "ferie": feries_map.get(ds)} for c in conges_rows: if c["date_debut"] <= ds <= c["date_fin"]: entry["conges"].append({"nom": c["full_name"], "type": c["nom"], "couleur": c["couleur"], "equipe": c["equipe"]}) cal.append(entry) pm = month - 1 if month > 1 else 12 py = year if month > 1 else year - 1 nm = month + 1 if month < 12 else 1 ny = year if month < 12 else year + 1 return render_template("conges_calendrier.html", user=user, is_admin=admin, year=year, month=month, cal=cal, prev_month=pm, prev_year=py, next_month=nm, next_year=ny, today=today, personnes=personnes, filtre_personne=personne) # ── Suivi OSM ────────────────────────────────────────────────────────────── @app.route("/osm") @login_required def osm_list(): user = get_current_user() db = get_db() admin = is_admin() filtre_priorite = request.args.get("priorite", "") filtre_etat = request.args.get("etat", "") filtre_collab = request.args.get("collab", "") query = """SELECT s.*, u.full_name AS collaborateur_nom FROM suivi_osm s LEFT JOIN users u ON u.id = s.collaborateur_id WHERE 1=1""" params = [] if not admin: osm_user_ids = [r["id"] for r in db.execute( "SELECT id FROM users WHERE equipe=?", (EQUIPE_OSM,)).fetchall()] if osm_user_ids: placeholders = ",".join("?" * len(osm_user_ids)) query += f" AND s.collaborateur_id IN ({placeholders})" params.extend(osm_user_ids) else: query += " AND 0" if filtre_priorite: query += " AND s.priorisation=?" params.append(1 if filtre_priorite == "oui" else 0) if filtre_etat: query += " AND s.etat=?" params.append(filtre_etat) if filtre_collab: query += " AND u.full_name=?" params.append(filtre_collab.upper()) query += " ORDER BY s.priorisation DESC, s.created_at DESC" dossiers = db.execute(query, params).fetchall() nb_prioritaires = db.execute("SELECT COUNT(*) FROM suivi_osm WHERE priorisation=1").fetchone()[0] collaborateurs_osm = db.execute( "SELECT id, full_name FROM users WHERE equipe=? ORDER BY full_name", (EQUIPE_OSM,)).fetchall() return render_template("suivi_osm.html", user=user, dossiers=dossiers, etats=OSM_ETATS, is_admin=admin, nb_prioritaires=nb_prioritaires, collaborateurs=collaborateurs_osm, filtre_priorite=filtre_priorite, filtre_etat=filtre_etat, filtre_collab=filtre_collab) @app.route("/osm/new", methods=["GET", "POST"]) @login_required @admin_required def osm_new(): user = get_current_user() db = get_db() if request.method == "POST": ref = request.form["ref_osm"].strip() if not ref: return redirect(url_for("osm_list")) db.execute("""INSERT INTO suivi_osm (ref_osm, date_arrivee, date_envoi, priorisation, etat, collaborateur_id, commentaire) VALUES (?,?,?,?,?,?,?)""", (ref, request.form["date_arrivee"], request.form.get("date_envoi") or None, 1 if request.form.get("priorisation") == "1" else 0, request.form.get("etat", "Nouveau"), int(request.form["collaborateur_id"]) if request.form.get("collaborateur_id") else None, request.form.get("commentaire", ""))) db.commit() log.info("OSM_CREATED ref=%s by=%s", ref, session["username"]) return redirect(url_for("osm_list")) collaborateurs_osm = db.execute( "SELECT id, full_name FROM users WHERE equipe=? ORDER BY full_name", (EQUIPE_OSM,)).fetchall() return render_template("suivi_osm_form.html", user=user, dossier=None, etats=OSM_ETATS, collaborateurs=collaborateurs_osm, today=date.today().isoformat()) @app.route("/osm//edit", methods=["GET", "POST"]) @login_required @admin_required def osm_edit(oid): user = get_current_user() db = get_db() dossier = db.execute("SELECT * FROM suivi_osm WHERE id=?", (oid,)).fetchone() if not dossier: return "Introuvable", 404 if request.method == "POST": ref = request.form["ref_osm"].strip() if not ref: return redirect(url_for("osm_list")) db.execute("""UPDATE suivi_osm SET ref_osm=?, date_arrivee=?, date_envoi=?, priorisation=?, etat=?, collaborateur_id=?, commentaire=?, updated_at=CURRENT_TIMESTAMP WHERE id=?""", (ref, request.form["date_arrivee"], request.form.get("date_envoi") or None, 1 if request.form.get("priorisation") == "1" else 0, request.form.get("etat", "Nouveau"), int(request.form["collaborateur_id"]) if request.form.get("collaborateur_id") else None, request.form.get("commentaire", ""), oid)) db.commit() log.info("OSM_UPDATED ref=%s by=%s", ref, session["username"]) return redirect(url_for("osm_list")) collaborateurs_osm = db.execute( "SELECT id, full_name FROM users WHERE equipe=? ORDER BY full_name", (EQUIPE_OSM,)).fetchall() return render_template("suivi_osm_form.html", user=user, dossier=dossier, etats=OSM_ETATS, collaborateurs=collaborateurs_osm) @app.route("/osm//delete", methods=["POST"]) @login_required @admin_required def osm_delete(oid): db = get_db() db.execute("DELETE FROM suivi_osm WHERE id=?", (oid,)) db.commit() log.info("OSM_DELETED id=%s by=%s", oid, session["username"]) return redirect(url_for("osm_list")) # ── Main ─────────────────────────────────────────────────────────────────── if __name__ == "__main__": if not DB_PATH.exists(): init_db() print("⚠️ Base de données créée. Lancez seed.py pour créer les utilisateurs initiaux.") else: migrate_db() from waitress import serve print("Suivi Production → http://0.0.0.0:5050") serve(app, host="0.0.0.0", port=5050) # ── Auto-init pour Passenger / cPanel ───────────────────────────────────── _AUTO_DB = None try: _AUTO_DB = sqlite3.connect(str(DB_PATH)) _AUTO_DB.execute("SELECT 1 FROM users LIMIT 1") user_count = _AUTO_DB.execute("SELECT COUNT(*) FROM users").fetchone()[0] except sqlite3.OperationalError: if _AUTO_DB: _AUTO_DB.close() if not DB_PATH.exists() or DB_PATH.stat().st_size == 0: init_db() else: migrate_db() user_count = 0 if _AUTO_DB: _AUTO_DB.close() # Auto-seed si aucun utilisateur n'existe (premier déploiement) if not user_count: from werkzeug.security import generate_password_hash _db = sqlite3.connect(str(DB_PATH)) migrate_db() _db.row_factory = sqlite3.Row # Créer les utilisateurs par défaut _users = [ ("admin", "Administrateur", "super_admin", "Admin@2026", "Standard"), ("nizar", "NIZAR", "admin", "Nizar@2026", "OSM"), ("yousra", "YOUSRA", "membre", "Yousra@2026","Standard"), ("imane", "IMANE", "membre", "Imane@2026", "Standard"), ("sanaa", "SANAA", "membre", "Sanaa@2026", "Standard"), ("hayat", "HAYAT", "membre", "Hayat@2026", "Standard"), ("yousef", "YOUSEF", "membre", "Yousef@2026","OSM"), ] for username, full_name, role, pw, equipe in _users: _db.execute( "INSERT OR IGNORE INTO users (username, password_hash, full_name, role, must_change_password, equipe) VALUES (?,?,?,?,1,?)", (username, generate_password_hash(pw), full_name, role, equipe)) # Types de congé for nom, couleur in [("Congé annuel","#3b82f6"),("Congé maladie","#ef4444"), ("Congé exceptionnel","#f59e0b"),("Récupération","#10b981"), ("Maternité","#ec4899")]: _db.execute("INSERT OR IGNORE INTO types_conge (nom, couleur) VALUES (?,?)", (nom, couleur)) # Jours fériés Maroc for date_str, nom in [ ("2026-01-01","Nouvel An"),("2026-01-11","Manifeste de l'Indépendance"), ("2026-03-18","Aïd al-Fitr (est.)"),("2026-03-19","Aïd al-Fitr (est.)"), ("2026-05-01","Fête du Travail"),("2026-05-25","Aïd al-Adha (est.)"), ("2026-05-26","Aïd al-Adha (est.)"),("2026-06-15","Nouvel An Hijri (est.)"), ("2026-07-30","Fête du Trône"),("2026-08-14","Allégeance Oued Eddahab"), ("2026-08-20","Révolution du Roi et du Peuple"),("2026-08-21","Fête de la Jeunesse"), ("2026-08-24","Naissance du Prophète (est.)"),("2026-11-06","Marche Verte"), ("2026-11-18","Fête de l'Indépendance"), ]: _db.execute("INSERT OR IGNORE INTO jours_feries (date_ferie, nom) VALUES (?,?)", (date_str, nom)) # Clients BAL par défaut for client in ["Bouygues Construction","Engie","Saint Gobain","SUEZ","SUEZ TECHNIQUE"]: _db.execute("INSERT OR IGNORE INTO clients_bal (nom) VALUES (?)", (client,)) _db.commit() _db.close() log.warning("AUTO-SEED: 7 utilisateurs + types congés + jours fériés créés.") # Pour Passenger WSGI (cPanel) application = app