app.py.save

"""
Flask ML Web Application - Iris Flower Classifier
Database schema:
  - users      (id, username, email, password, created)
  - sessions   (id, user_id, login_at, logout_at)
  - predictions(id, username, sepal_length, sepal_width, petal_length, petal_width, prediction, confidence, timestamp)
"""

import sqlite3
import hashlib
import os
from functools import wraps
from datetime import datetime

import joblib
import numpy as np
from flask import (Flask, render_template, request, redirect,
                   url_for, session, flash, g)

# ── App setup ─────────────────────────────────────────────────────────────────
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY", "iris-ml-secret-2024")

BASE_DIR  = os.path.dirname(os.path.abspath(__file__))
DB_PATH   = os.path.join(BASE_DIR, "users.db")
MODEL_DIR = os.path.join(BASE_DIR, "model")

# ── Load ML artefacts ─────────────────────────────────────────────────────────
model       = joblib.load(os.path.join(MODEL_DIR, "iris_model.pkl"))
scaler      = joblib.load(os.path.join(MODEL_DIR, "scaler.pkl"))
class_names = joblib.load(os.path.join(MODEL_DIR, "class_names.pkl"))

FLOWER_INFO = {
    "setosa":     {"emoji": "🌸", "description": "Iris setosa is a hardy species found in arctic and subarctic regions.", "color": "#f9a8d4"},
    "versicolor": {"emoji": "💜", "description": "Iris versicolor (Blue Flag Iris) is native to North America.",         "color": "#c4b5fd"},
    "virginica":  {"emoji": "🌺", "description": "Iris virginica (Virginia Iris) thrives in moist, coastal habitats.",   "color": "#6ee7b7"},
}

# ── Database helpers ───────────────────────────────────────────────────────────
def get_db():
    db = getattr(g, "_database", None)
    if db is None:
        db = g._database = sqlite3.connect(DB_PATH)
        db.row_factory = sqlite3.Row
    return db

@app.teardown_appcontext
def close_db(_):
    db = getattr(g, "_database", None)
    if db is not None:
        db.close()

def init_db():
    with app.app_context():
        db = get_db()

        # Table users : id, username, email, password, created
        db.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id       INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email    TEXT UNIQUE NOT NULL,
                password TEXT NOT NULL,
                created  TEXT NOT NULL
            )
        """)

        # Table sessions : trace chaque connexion / déconnexion
        db.execute("""
            CREATE TABLE IF NOT EXISTS sessions (
                id        INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id   INTEGER NOT NULL,
                login_at  TEXT NOT NULL,
                logout_at TEXT,
                FOREIGN KEY (user_id) REFERENCES users(id)
            )
        """)

        # Table predictions
        db.execute("""
            CREATE TABLE IF NOT EXISTS predictions (
                id           INTEGER PRIMARY KEY AUTOINCREMENT,
                username     TEXT NOT NULL,
                sepal_length REAL,
                sepal_width  REAL,
                petal_length REAL,
                petal_width  REAL,
                prediction   TEXT,
                confidence   REAL,
                timestamp    TEXT NOT NULL
            )
        """)

        # Migration : ajouter email si la colonne manque dans une ancienne DB
        try:
            db.execute("ALTER TABLE users ADD COLUMN email TEXT NOT NULL DEFAULT ''")
        except sqlite3.OperationalError:
            pass  # colonne déjà présente

        # Compte admin par défaut
        pw_hash = hashlib.sha256("password123".encode()).hexdigest()
        try:
            db.execute(
                "INSERT INTO users (username, email, password, created) VALUES (?, ?, ?, ?)",
                ("admin", "admin@irisai.local", pw_hash, datetime.now().isoformat())
            )
        except sqlite3.IntegrityError:
            pass
        db.commit()

def hash_pw(pw: str) -> str:
    return hashlib.sha256(pw.encode()).hexdigest()

# ── Auth decorator ─────────────────────────────────────────────────────────────
def login_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        if "username" not in session:
            flash("Veuillez vous connecter pour accéder à cette page.", "warning")
            return redirect(url_for("login"))
        return f(*args, **kwargs)
    return decorated

# ── Routes ─────────────────────────────────────────────────────────────────────
@app.route("/")
def index():
    if "username" in session:
        return redirect(url_for("predict"))
    return redirect(url_for("login"))

@app.route("/login", methods=["GET", "POST"])
def login():
    if "username" in session:
        return redirect(url_for("predict"))
    if request.method == "POST":
        username = request.form.get("username", "").strip()
        password = request.form.get("password", "")
        db   = get_db()
        user = db.execute(
            "SELECT * FROM users WHERE username = ? AND password = ?",
            (username, hash_pw(password))
        ).fetchone()
        if user:
            session["username"]      = username
            session["user_id"]       = user["id"]

            # Enregistrer la connexion dans la table sessions
            db.execute(
                "INSERT INTO sessions (user_id, login_at) VALUES (?, ?)",
                (user["id"], datetime.now().isoformat())
            )
            cursor = db.execute("SELECT last_insert_rowid()")
            session["session_db_id"] = cursor.fetchone()[0]
            db.commit()

            flash(f"Bienvenue, {username} ! 👋", "success")
            return redirect(url_for("predict"))
        flash("Nom d'utilisateur ou mot de passe incorrect.", "danger")
    return render_template("login.html")

@app.route("/register", methods=["GET", "POST"])
def register():
    if "username" in session:
        return redirect(url_for("predict"))
    if request.method == "POST":
        username = request.form.get("username", "").strip()
        email    = request.form.get("email", "").strip().lower()
        password = request.form.get("password", "")
        confirm  = request.form.get("confirm", "")
        if not username or not email or not password:
            flash("Tous les champs sont obligatoires.", "danger")
        elif "@" not in email or "." not in email:
            flash("Adresse email invalide.", "danger")
        elif password != confirm:
            flash("Les mots de passe ne correspondent pas.", "danger")
        elif len(password) < 6:
            flash("Le mot de passe doit comporter au moins 6 caractères.", "danger")
        else:
            db = get_db()
            try:
                db.execute(
                    "INSERT INTO users (username, email, password, created) VALUES (?, ?, ?, ?)",
                    (username, email, hash_pw(password), datetime.now().isoformat())
                )
                db.commit()
                flash("Compte créé ! Vous pouvez vous connecter.", "success")
                return redirect(url_for("login"))
            except sqlite3.IntegrityError as e:
                if "username" in str(e):
                    flash("Ce nom d'utilisateur est déjà pris.", "danger")
                else:
                    flash("Cette adresse email est déjà utilisée.", "danger")
    return render_template("register.html")

@app.route("/logout")
def logout():
    # Enregistrer l'heure de déconnexion dans sessions
    session_db_id = session.get("session_db_id")
    if session_db_id:
        db = get_db()
        db.execute(
            "UPDATE sessions SET logout_at = ? WHERE id = ?",
            (datetime.now().isoformat(), session_db_id)
        )
        db.commit()

    session.clear()
    flash("Vous avez été déconnecté.", "info")
    return redirect(url_for("login"))

@app.route("/predict", methods=["GET", "POST"])
@login_required
def predict():
    result    = None
    form_data = {}

    if request.method == "POST":
        try:
            sepal_length = float(request.form["sepal_length"])
            sepal_width  = float(request.form["sepal_width"])
            petal_length = float(request.form["petal_length"])
            petal_width  = float(request.form["petal_width"])

            if not (0 < sepal_length < 20 and 0 < sepal_width < 20 and
                    0 < petal_length < 20 and 0 < petal_width < 20):
                flash("Entrez des mesures réalistes (0–20 cm).", "warning")
                return redirect(url_for("predict"))

            form_data = {
                "sepal_length": sepal_length, "sepal_width": sepal_width,
                "petal_length": petal_length, "petal_width": petal_width,
            }

            X         = np.array([[sepal_length, sepal_width, petal_length, petal_width]])
            X_scaled  = scaler.transform(X)
            pred_idx  = model.predict(X_scaled)[0]
            proba     = model.predict_proba(X_scaled)[0]
            pred_name = class_names[pred_idx]
            confidence = float(proba[pred_idx]) * 100

            db = get_db()
            db.execute(
                """INSERT INTO predictions
                   (username, sepal_length, sepal_width, petal_length, petal_width,
                    prediction, confidence, timestamp)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                (session["username"], sepal_length, sepal_width,
                 petal_length, petal_width, pred_name, confidence,
                 datetime.now().isoformat())
            )
            db.commit()

            all_probs = {class_names[i]: round(float(p) * 100, 1) for i, p in enumerate(proba)}
            result = {
                "name":       pred_name,
                "confidence": round(confidence, 1),
                "info":       FLOWER_INFO.get(pred_name, {}),
                "all_probs":  all_probs,
            }

        except (ValueError, KeyError):
            flash("Entrée invalide — veuillez saisir des valeurs numériques.", "danger")

    db      = get_db()
    history = db.execute(
        "SELECT * FROM predictions WHERE username = ? ORDER BY id DESC LIMIT 5",
        (session["username"],)
    ).fetchall()

    return render_template("predict.html",
                           result=result,
                           form_data=form_data,
                           history=history,
                           username=session["username"])

# ── Entry point ────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    init_db()
    port = int(os.environ.get("PORT", 5000))
    app.run(host="0.0.0.0", port=port, debug=False)


# ── Admin dashboard (temps réel, refresh auto) ────────────────────────────────
@app.route("/admin")
@login_required
def admin():
    # Seulement l'admin peut accéder
    if session.get("username") != "admin":
        flash("Accès réservé à l'administrateur.", "danger")
        return redirect(url_for("predict"))
    db = get_db()

    users = db.execute(
        "SELECT id, username, email, created FROM users ORDER BY id DESC"
    ).fetchall()

    sessions_log = db.execute("""
        SELECT s.id, u.username, u.email, s.login_at, s.logout_at
        FROM sessions s
        JOIN users u ON s.user_id = u.id
        ORDER BY s.id DESC
        LIMIT 50
    """).fetchall()

    stats = {
        "total_users":    db.execute("SELECT COUNT(*) FROM users").fetchone()[0],
        "total_sessions": db.execute("SELECT COUNT(*) FROM sessions").fetchone()[0],
        "active_now":     db.execute("SELECT CO