LWX/ai_teaching_system/backend/services/auth.py

# -*- coding: utf-8 -*-
"""JWT 实现学生与教师角色细粒度权限控制"""
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials, APIKeyHeader

from backend.config import get_settings

settings = get_settings()
security = HTTPBearer(auto_error=False)


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)


def decode_token(token: str) -> Optional[dict]:
    try:
        return jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
    except JWTError:
        return None


async def get_current_user_id(credentials: HTTPAuthorizationCredentials = Depends(security)) -> int:
    if credentials is None:
        raise HTTPException(status_code=401, detail="需要登录")
    payload = decode_token(credentials.credentials)
    if payload is None:
        raise HTTPException(status_code=401, detail="token 无效")
    uid = payload.get("sub")
    if uid is None:
        raise HTTPException(status_code=401, detail="token 无效")
    return int(uid)


async def get_current_user_role(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str:
    if credentials is None:
        return "student"
    payload = decode_token(credentials.credentials)
    if payload is None:
        return "student"
    return payload.get("role", "student")


def require_teacher(role: str = Depends(get_current_user_role)):
    if role != "teacher":
        raise HTTPException(status_code=403, detail="需要教师权限")
    return role