feat: VoiceScript STT+OCR 자막기능 수정
This commit is contained in:
50
app/auth.py
50
app/auth.py
@@ -1,22 +1,11 @@
|
||||
"""
|
||||
인증 모듈 — 다중 사용자 JSON 파일 기반
|
||||
사용자 구조:
|
||||
{
|
||||
"password": "...",
|
||||
"role": "admin" | "user",
|
||||
"permissions": {
|
||||
"stt": true | false,
|
||||
"ocr": true | false,
|
||||
"allowed_stt_models": ["medium", "large-v3", ...], # 빈 배열 = 모두 허용
|
||||
"allowed_ocr_models": ["granite3.2-vision", ...] # 빈 배열 = 모두 허용
|
||||
}
|
||||
}
|
||||
권한: stt | ocr | subtitle
|
||||
"""
|
||||
import os, json, threading
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from fastapi import Depends, HTTPException
|
||||
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
||||
from jose import JWTError, jwt
|
||||
|
||||
@@ -32,20 +21,15 @@ USERS_FILE = DATA_DIR / "users.json"
|
||||
_lock = threading.Lock()
|
||||
bearer = HTTPBearer(auto_error=False)
|
||||
|
||||
|
||||
# ── 파일 I/O ──────────────────────────────────────────────────
|
||||
def _load() -> dict:
|
||||
if not USERS_FILE.exists(): return {}
|
||||
with open(USERS_FILE, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
with open(USERS_FILE, "r", encoding="utf-8") as f: return json.load(f)
|
||||
|
||||
def _save(users: dict):
|
||||
USERS_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(USERS_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(users, f, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
# ── 초기화 ────────────────────────────────────────────────────
|
||||
def init_users():
|
||||
with _lock:
|
||||
users = _load()
|
||||
@@ -53,15 +37,12 @@ def init_users():
|
||||
"password": ADMIN_PASSWORD,
|
||||
"role": "admin",
|
||||
"permissions": {
|
||||
"stt": True, "ocr": True,
|
||||
"allowed_stt_models": [], # 빈 배열 = 제한 없음
|
||||
"allowed_ocr_models": [],
|
||||
"stt": True, "ocr": True, "subtitle": True,
|
||||
"allowed_stt_models": [], "allowed_ocr_models": [],
|
||||
},
|
||||
}
|
||||
_save(users)
|
||||
|
||||
|
||||
# ── CRUD ──────────────────────────────────────────────────────
|
||||
def authenticate(username: str, password: str):
|
||||
with _lock: users = _load()
|
||||
u = users.get(username)
|
||||
@@ -80,9 +61,9 @@ def create_user(username: str, password: str, permissions: dict) -> tuple:
|
||||
with _lock:
|
||||
users = _load()
|
||||
if username in users: return False, "이미 존재하는 사용자입니다"
|
||||
# 기본값 보완
|
||||
permissions.setdefault("allowed_stt_models", [])
|
||||
permissions.setdefault("allowed_ocr_models", [])
|
||||
permissions.setdefault("subtitle", False)
|
||||
users[username] = {"password": password, "role": "user", "permissions": permissions}
|
||||
_save(users)
|
||||
return True, "사용자가 생성되었습니다"
|
||||
@@ -94,6 +75,7 @@ def update_user(username: str, permissions: dict, password: str = None) -> tuple
|
||||
if username not in users: return False, "사용자를 찾을 수 없습니다"
|
||||
permissions.setdefault("allowed_stt_models", [])
|
||||
permissions.setdefault("allowed_ocr_models", [])
|
||||
permissions.setdefault("subtitle", False)
|
||||
users[username]["permissions"] = permissions
|
||||
if password: users[username]["password"] = password
|
||||
_save(users)
|
||||
@@ -107,14 +89,10 @@ def delete_user(username: str) -> tuple:
|
||||
del users[username]; _save(users)
|
||||
return True, "삭제되었습니다"
|
||||
|
||||
|
||||
# ── JWT ───────────────────────────────────────────────────────
|
||||
def create_access_token(username: str) -> str:
|
||||
exp = datetime.utcnow() + timedelta(hours=EXPIRE_HOURS)
|
||||
return jwt.encode({"sub": username, "exp": exp}, SECRET_KEY, algorithm=ALGORITHM)
|
||||
|
||||
|
||||
# ── FastAPI 의존성 ─────────────────────────────────────────────
|
||||
def require_auth(credentials: HTTPAuthorizationCredentials = Depends(bearer)) -> dict:
|
||||
if credentials is None:
|
||||
raise HTTPException(401, "인증이 필요합니다", headers={"WWW-Authenticate": "Bearer"})
|
||||
@@ -126,16 +104,24 @@ def require_auth(credentials: HTTPAuthorizationCredentials = Depends(bearer)) ->
|
||||
if not u: raise JWTError()
|
||||
return {"username": username, **u}
|
||||
except JWTError:
|
||||
raise HTTPException(401, "토큰이 유효하지 않거나 만료되었습니다", headers={"WWW-Authenticate": "Bearer"})
|
||||
raise HTTPException(401, "토큰이 유효하지 않거나 만료되었습니다",
|
||||
headers={"WWW-Authenticate": "Bearer"})
|
||||
|
||||
def require_admin(user: dict = Depends(require_auth)) -> dict:
|
||||
if user.get("role") != "admin": raise HTTPException(403, "관리자 권한이 필요합니다")
|
||||
return user
|
||||
|
||||
def require_stt(user: dict = Depends(require_auth)) -> dict:
|
||||
if not user.get("permissions", {}).get("stt", False): raise HTTPException(403, "STT 사용 권한이 없습니다")
|
||||
if not user.get("permissions", {}).get("stt", False):
|
||||
raise HTTPException(403, "STT 사용 권한이 없습니다")
|
||||
return user
|
||||
|
||||
def require_ocr(user: dict = Depends(require_auth)) -> dict:
|
||||
if not user.get("permissions", {}).get("ocr", False): raise HTTPException(403, "OCR 사용 권한이 없습니다")
|
||||
if not user.get("permissions", {}).get("ocr", False):
|
||||
raise HTTPException(403, "OCR 사용 권한이 없습니다")
|
||||
return user
|
||||
|
||||
def require_subtitle(user: dict = Depends(require_auth)) -> dict:
|
||||
if not user.get("permissions", {}).get("subtitle", False):
|
||||
raise HTTPException(403, "자막 사용 권한이 없습니다")
|
||||
return user
|
||||
|
||||
Reference in New Issue
Block a user