sanguo_vnpy_v2/sanguo_web/api/deps.py

"""
依赖注入模块
提供认证、数据库等依赖注入函数
"""
from typing import Generator, Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)


# JWT 配置
SECRET_KEY = "sanguo_secret_key_change_in_production"  # 生产环境应从配置读取
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


security = HTTPBearer()


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """创建 JWT Token"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)

    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


def verify_token(token: str) -> dict:
    """验证 JWT Token"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError as e:
        logger.warning(f"Token verification failed: {e}")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    """
    获取当前用户依赖
    从 Authorization header 中解析 JWT Token
    """
    token = credentials.credentials

    try:
        payload = verify_token(token)

        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication credentials",
                headers={"WWW-Authenticate": "Bearer"},
            )

        # 返回用户信息（实际应从数据库获取）
        return {
            "username": username,
            "is_active": payload.get("is_active", True),
            "exp": payload.get("exp")
        }

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error getting current user: {e}")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )


async def get_optional_user(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(HTTPBearer(auto_error=False))
) -> Optional[dict]:
    """
    可选的用户认证依赖
    允许未登录用户访问，但如果提供了 Token 则会验证
    """
    if credentials is None:
        return None

    try:
        return await get_current_user(credentials)
    except HTTPException:
        return None


# ============================================
# VeighNa 服务依赖
# ============================================

async def get_vn_service():
    """
    获取 VeighNa 服务实例
    如果服务未初始化，抛出异常
    使用延迟导入避免循环导入问题
    """
    from . import vn_service

    if vn_service is None:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="VeighNa service is not initialized"
        )

    return vn_service


# ============================================
# 简单认证（开发环境）
# ============================================

# 硬编码的用户数据库（生产环境应使用真实数据库）
FAKE_USERS_DB = {
    "admin": {
        "username": "admin",
        "full_name": "Administrator",
        "email": "admin@sanguo.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36Wh0/mqGKnKM0lQ5lEqxKe",  # "secret"
        "is_active": True,
    }
}


def authenticate_user(username: str, password: str) -> Optional[dict]:
    """
    验证用户凭证
    开发环境使用硬编码用户，生产环境应使用数据库
    """
    # 开发环境简单验证
    if username == "admin" and password == "admin123":
        return {
            "username": "admin",
            "is_active": True
        }

    # 生产环境应使用数据库和 passlib
    # user = FAKE_USERS_DB.get(username)
    # if not user:
    #     return None
    # if not user["is_active"]:
    #     return None
    # from passlib.context import CryptContext
    # pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
    # if not pwd_context.verify(password, user["hashed_password"]):
    #     return None
    # return user

    return None


__all__ = [
    "create_access_token",
    "verify_token",
    "get_current_user",
    "get_optional_user",
    "get_vn_service",
    "authenticate_user",
]