from fastapi import Depends, HTTPException, Header, Cookie from typing import Optional from sqlalchemy.orm import Session from app.models.database import get_db from app.models.user import UserDB, UserRole from app.core.security import verify_token from app.core.redis_client import redis_client async def get_current_user( authorization: Optional[str] = Header(None), access_token: Optional[str] = Cookie(None), db: Session = Depends(get_db) ) -> UserDB: # 优先使用Header中的token,其次使用Cookie中的token token = None if authorization and authorization.startswith("Bearer "): token = authorization.split(" ")[1] elif access_token: token = access_token if not token: raise HTTPException(status_code=401, detail="未提供有效的认证信息") sub = verify_token(token) if not sub: raise HTTPException(status_code=401, detail="Token已过期或无效") # 读取 redis 缓存数据 redis_key = f"user:{sub}" user_data = redis_client.get_client().get(redis_key) if user_data: user = UserDB.model_validate_json(user_data) else: user = db.query(UserDB).filter(UserDB.phone == sub).first() if user: redis_client.get_client().set(redis_key, user.model_dump_json()) if not user: raise HTTPException(status_code=401, detail="用户未登录") return user async def get_admin_user( current_user: UserDB = Depends(get_current_user) ) -> UserDB: if UserRole.ADMIN not in current_user.roles: raise HTTPException(status_code=403, detail="需要管理员权限") return current_user async def get_deliveryman_user( current_user: UserDB = Depends(get_current_user) ) -> UserDB: """验证配送员权限""" if UserRole.DELIVERYMAN not in current_user.roles: raise HTTPException(status_code=403, detail="需要配送员权限") return current_user async def get_merchant_user( current_user: UserDB = Depends(get_current_user) ) -> UserDB: """验证商家权限""" if UserRole.MERCHANT not in current_user.roles: raise HTTPException(status_code=403, detail="需要商家权限") return current_user