from fastapi import APIRouter, Request, Depends, Form from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from sqlalchemy import text from ..dependencies import get_current_user from ..database import SessionLocal, SessionLocalDemo from ..auth import verify_password, create_access_token, hash_password from ..services.audit_service import log_login, log_logout, log_login_failed from ..config import APP_NAME, APP_VERSION router = APIRouter() templates = Jinja2Templates(directory="app/templates") @router.get("/login", response_class=HTMLResponse) async def login_page(request: Request): return templates.TemplateResponse("login.html", { "request": request, "app_name": APP_NAME, "version": APP_VERSION, "error": None }) @router.post("/login") async def login(request: Request, username: str = Form(...), password: str = Form(...), mode: str = Form("reel")): # Select DB based on mode factory = SessionLocalDemo if mode == "demo" else SessionLocal db = factory() try: row = db.execute(text("SELECT id, username, password_hash, role, is_active FROM users WHERE LOWER(username) = LOWER(:u)"), {"u": username}).fetchone() if not row: log_login_failed(db, request, username) db.commit() return templates.TemplateResponse("login.html", { "request": request, "app_name": APP_NAME, "version": APP_VERSION, "error": "Utilisateur inconnu" }) if not row.is_active: log_login_failed(db, request, username) db.commit() return templates.TemplateResponse("login.html", { "request": request, "app_name": APP_NAME, "version": APP_VERSION, "error": "Compte desactive" }) try: ok = verify_password(password, row.password_hash) except Exception: ok = False if not ok: log_login_failed(db, request, username) db.commit() return templates.TemplateResponse("login.html", { "request": request, "app_name": APP_NAME, "version": APP_VERSION, "error": "Mot de passe incorrect" }) # Include mode in JWT token token = create_access_token({"sub": row.username, "role": row.role, "uid": row.id, "mode": mode}) user = {"sub": row.username, "role": row.role, "uid": row.id, "mode": mode} log_login(db, request, user) db.commit() # Redirect perms = db.execute(text("SELECT module FROM user_permissions WHERE user_id = :uid"), {"uid": row.id}).fetchall() modules = {r.module for r in perms} redirect_url = "/quickwin" if modules == {"quickwin"} else "/dashboard" response = RedirectResponse(url=redirect_url, status_code=303) response.set_cookie(key="access_token", value=token, httponly=True, samesite="lax", max_age=3600) return response finally: db.close() @router.get("/logout") async def logout(request: Request): user = get_current_user(request) if user: mode = user.get("mode", "reel") factory = SessionLocalDemo if mode == "demo" else SessionLocal db = factory() try: log_logout(db, request, user) db.commit() finally: db.close() response = RedirectResponse(url="/login", status_code=302) response.delete_cookie("access_token") return response