Spaces:
Running
Running
| 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, | |
| get_or_assign_code, list_codes, | |
| ) | |
| router = APIRouter(prefix="/api/auth", tags=["auth"]) | |
| 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) | |
| # Mã U00N để log dùng làm user_id ngắn gọn. Lỗi cấp mã -> None, FE tự | |
| # lùi về username; KHÔNG chặn đăng nhập. | |
| code = get_or_assign_code(username) if authorized else None | |
| return {"ok": True, "authorized": authorized, "username": username, "code": code} | |
| except Exception as ex: | |
| return {"ok": False, "error": f"{type(ex).__name__}: {ex}", "trace": traceback.format_exc()} | |
| def check_authorized(username: str = ""): | |
| if not username.strip(): | |
| return {"ok": False, "authorized": False} | |
| return {"ok": True, "authorized": is_authorized(username)} | |
| 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} | |
| def admin_list_users(token: str = ""): | |
| if token != settings.VIDEO_ADMIN_TOKEN: | |
| raise HTTPException(403, "Not authorized") | |
| return {"ok": True, "users": list_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", "")) | |
| code = get_or_assign_code(username) | |
| return {"ok": True, "code": code} | |
| 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} | |
| def admin_list_pending(token: str = ""): | |
| if token != settings.VIDEO_ADMIN_TOKEN: | |
| raise HTTPException(403, "Not authorized") | |
| return {"ok": True, "pending": list_pending()} | |
| 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} | |
| 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} | |
| 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} | |