Spaces:
Running
Running
File size: 3,691 Bytes
9b53f70 5edb87f 9b53f70 5edb87f 9b53f70 5edb87f 9b53f70 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | 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"])
@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)
# 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()}
@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", ""))
code = get_or_assign_code(username)
return {"ok": True, "code": code}
@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}
|