import traceback from fastapi import APIRouter, HTTPException from ..config import settings from ..services.auth.storage import ( list_users, add_user, remove_user, is_authorized, list_pending, request_access, approve_pending, reject_pending, ) router = APIRouter(prefix="/api/auth", tags=["auth"]) @router.post("/login") def login(body: dict): username = (body.get("username") or "").strip() if not username: return {"ok": False, "authorized": False, "error": "Missing username"} try: authorized = is_authorized(username) return {"ok": True, "authorized": authorized, "username": username} except Exception as ex: return {"ok": False, "error": f"{type(ex).__name__}: {ex}", "trace": traceback.format_exc()} @router.get("/check") def check_authorized(username: str = ""): if not username.strip(): return {"ok": False, "authorized": False} return {"ok": True, "authorized": is_authorized(username)} @router.post("/request-access") def request_access_endpoint(body: dict): username = (body.get("username") or "").strip() if not username: return {"ok": False, "error": "Missing username"} request_access(username) return {"ok": True} @router.get("/admin/users") def admin_list_users(token: str = ""): if token != settings.VIDEO_ADMIN_TOKEN: raise HTTPException(403, "Not authorized") return {"ok": True, "users": list_users()} @router.post("/admin/users") def admin_add_user(body: dict): if body.get("token") != settings.VIDEO_ADMIN_TOKEN: raise HTTPException(403, "Not authorized") username = (body.get("username") or "").strip() if not username: raise HTTPException(400, "Missing username") add_user(username, body.get("note", "")) return {"ok": True} @router.delete("/admin/users/{username}") def admin_remove_user(username: str, token: str = ""): if token != settings.VIDEO_ADMIN_TOKEN: raise HTTPException(403, "Not authorized") removed = remove_user(username) return {"ok": removed} @router.get("/admin/pending") def admin_list_pending(token: str = ""): if token != settings.VIDEO_ADMIN_TOKEN: raise HTTPException(403, "Not authorized") return {"ok": True, "pending": list_pending()} @router.post("/admin/pending/approve") def admin_approve_pending(body: dict): if body.get("token") != settings.VIDEO_ADMIN_TOKEN: raise HTTPException(403, "Not authorized") username = (body.get("username") or "").strip() if not username: raise HTTPException(400, "Missing username") ok = approve_pending(username) if not ok: raise HTTPException(404, "Username not in pending list") return {"ok": True} @router.delete("/admin/pending/{username}") def admin_reject_pending(username: str, token: str = ""): if token != settings.VIDEO_ADMIN_TOKEN: raise HTTPException(403, "Not authorized") ok = reject_pending(username) if not ok: raise HTTPException(404, "Username not in pending list") return {"ok": True} @router.get("/admin/export") def admin_export(token: str = ""): if token != settings.VIDEO_ADMIN_TOKEN: raise HTTPException(403, "Not authorized") users = list_users() csv = ", ".join(u["username"] for u in users) return {"ok": True, "csv": csv, "users": users}