chat-ai-api/app/dependencies/auth.py

from fastapi import HTTPException, Query, Depends
from fastapi.security import OAuth2PasswordBearer
from typing import Optional, Annotated
import jwt

from ..db.redis_client import get_app_user
from ..config.config import Config

config = Config()
oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="/users/token", auto_error=False)


async def resolve_username(
    jwt_token: Annotated[Optional[str], Depends(oauth2_scheme_optional)] = None,
    source: Optional[str] = Query(default=None),
    token: Optional[str] = Query(default=None),
) -> str:
    if source == "app" and token:
        app_user = get_app_user(token)
        if not app_user:
            raise HTTPException(status_code=401, detail="无效的 App token")
        return f"app_{app_user['userId']}"
    if not jwt_token:
        raise HTTPException(status_code=401, detail="未提供认证令牌")
    try:
        payload = jwt.decode(jwt_token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
        sub = payload.get("sub")
        if not sub:
            raise HTTPException(status_code=401, detail="无效的令牌")
        return sub
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="无效的令牌")