from jose import JWTError, jwt from datetime import datetime, timedelta from fastapi.security import OAuth2PasswordBearer from fastapi import Depends, HTTPException SECRET_KEY = "your-secret-key-change-in-production" # 生产环境用环境变量 ALGORITHM = "HS256" oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token") def create_access_token(data: dict, expires_delta: timedelta): to_encode = data.copy() expire = datetime.utcnow() + expires_delta to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) def get_current_user(token: str = Depends(oauth2_scheme)): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") role: str = payload.get("role") if username is None: raise HTTPException(401, "无效令牌") return {"username": username, "role": role} except JWTError: raise HTTPException(401, "令牌验证失败")