feat: panel admin - gestion utilisateurs (#16)

- Route /api/admin avec middleware require_admin - Liste utilisateurs avec statut, role, dates - Actions: activer/desactiver, promouvoir/rétrograder admin, supprimer - Dashboard stats (utilisateurs, téléchargements) - Template admin_panel.html avec table responsive - Champ is_admin ajoute au modele User - Migration automatique colonne is_admin - Protection: impossible de modifier son propre compte Closes #16
2026-04-02 22:44:33 +00:00
parent c921aafadd
commit 2da2a5bb27
7 changed files with 289 additions and 1 deletions
@@ -0,0 +1,165 @@
+"""
+Admin panel routes for Ohm Stream Downloader API.
+"""
+
+from datetime import datetime
+from typing import Optional, List
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from fastapi.templating import Jinja2Templates
+from sqlmodel import Session, select
+
+from app.database import get_session, engine
+from app.models.auth import User, UserTable
+from app.routers.router_auth import get_current_user_from_token
+
+router = APIRouter(prefix="/api/admin", tags=["admin"])
+templates = Jinja2Templates(directory="templates")
+
+
+async def require_admin(current_user: User = Depends(get_current_user_from_token)) -> User:
+    """Dependency that requires the current user to be an admin."""
+    if not current_user.is_admin:
+        raise HTTPException(status_code=403, detail="Admin access required")
+    return current_user
+
+
+@router.get("/users")
+async def list_users(
+    current_user: User = Depends(require_admin),
+):
+    """List all users (admin only)"""
+    with Session(engine) as session:
+        statement = select(UserTable)
+        users = session.exec(statement).all()
+        return {
+            "users": [
+                {
+                    "id": u.id,
+                    "username": u.username,
+                    "email": u.email,
+                    "full_name": u.full_name,
+                    "is_active": u.is_active,
+                    "is_admin": u.is_admin,
+                    "created_at": u.created_at.isoformat() if u.created_at else None,
+                    "last_login": u.last_login.isoformat() if u.last_login else None,
+                }
+                for u in users
+            ],
+            "total": len(users),
+        }
+
+
+@router.get("/stats")
+async def get_admin_stats(
+    current_user: User = Depends(require_admin),
+):
+    """Get admin dashboard statistics"""
+    from app.download_manager import DownloadManager
+    from main import download_manager
+
+    with Session(engine) as session:
+        total_users = len(session.exec(select(UserTable)).all())
+        active_users = len(session.exec(select(UserTable).where(UserTable.is_active == True)).all())
+        admin_users = len(session.exec(select(UserTable).where(UserTable.is_admin == True)).all())
+
+    tasks = download_manager.get_all_tasks()
+    total_downloads = len(tasks)
+    completed_downloads = len([t for t in tasks if t.status == "completed"])
+    active_downloads = len([t for t in tasks if t.status in ("downloading", "pending")])
+
+    return {
+        "users": {
+            "total": total_users,
+            "active": active_users,
+            "admins": admin_users,
+        },
+        "downloads": {
+            "total": total_downloads,
+            "completed": completed_downloads,
+            "active": active_downloads,
+        },
+    }
+
+
+@router.put("/users/{user_id}/toggle-active")
+async def toggle_user_active(
+    user_id: str,
+    response: Response,
+    current_user: User = Depends(require_admin),
+):
+    """Activate or deactivate a user"""
+    with Session(engine) as session:
+        user = session.exec(select(UserTable).where(UserTable.id == user_id)).first()
+        if not user:
+            raise HTTPException(status_code=404, detail="User not found")
+        if user.id == current_user.id:
+            raise HTTPException(status_code=400, detail="Cannot modify your own account")
+        user.is_active = not user.is_active
+        session.add(user)
+        session.commit()
+        status = "active" if user.is_active else "inactive"
+        response.headers["HX-Trigger"] = f'{{"show-toast": {{"message": "User {user.username} is now {status}", "type": "success"}}}}'
+        return {"id": user_id, "is_active": user.is_active}
+
+
+@router.put("/users/{user_id}/toggle-admin")
+async def toggle_user_admin(
+    user_id: str,
+    response: Response,
+    current_user: User = Depends(require_admin),
+):
+    """Promote or demote a user to/from admin"""
+    with Session(engine) as session:
+        user = session.exec(select(UserTable).where(UserTable.id == user_id)).first()
+        if not user:
+            raise HTTPException(status_code=404, detail="User not found")
+        if user.id == current_user.id:
+            raise HTTPException(status_code=400, detail="Cannot modify your own admin status")
+        user.is_admin = not user.is_admin
+        session.add(user)
+        session.commit()
+        role = "admin" if user.is_admin else "user"
+        response.headers["HX-Trigger"] = f'{{"show-toast": {{"message": "{user.username} is now {role}", "type": "success"}}}}'
+        return {"id": user_id, "is_admin": user.is_admin}
+
+
+@router.delete("/users/{user_id}")
+async def delete_user(
+    user_id: str,
+    response: Response,
+    current_user: User = Depends(require_admin),
+):
+    """Delete a user"""
+    with Session(engine) as session:
+        user = session.exec(select(UserTable).where(UserTable.id == user_id)).first()
+        if not user:
+            raise HTTPException(status_code=404, detail="User not found")
+        if user.id == current_user.id:
+            raise HTTPException(status_code=400, detail="Cannot delete your own account")
+        username = user.username
+        session.delete(user)
+        session.commit()
+        response.headers["HX-Trigger"] = f'{{"show-toast": {{"message": "User {username} deleted", "type": "info"}}}}'
+        return {"deleted": user_id}
+
+
+@router.get("/ui")
+async def get_admin_ui(
+    request: Request,
+    current_user: Optional[User] = Depends(get_current_user_from_token),
+):
+    """Get admin panel UI"""
+    if current_user is None or not current_user.is_admin:
+        from app.routers.router_auth import get_optional_user
+        return templates.TemplateResponse(
+            "components/login_prompt.html", {"request": request}
+        )
+    
+    with Session(engine) as session:
+        users = session.exec(select(UserTable)).all()
+    
+    return templates.TemplateResponse(
+        "components/admin_panel.html",
+        {"request": request, "users": users, "current_user": current_user},
+    )