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 服务实例
    如果服务未初始化，抛出异常
    使用动态导入避免循环导入问题
    """
    # 动态获取 vn_service，避免循环导入时的静态绑定问题
    import sys
    api_module = sys.modules.get('sanguo_web.api')
    if api_module is None:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="API module not loaded"
        )

    vn_service = getattr(api_module, 'vn_service', None)

    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]:
    """
    验证用户凭证
    生产环境应使用数据库和环境变量存储密码
    """
    import os

    # 从环境变量读取管理员密码
    admin_password = os.environ.get("ADMIN_PASSWORD", "admin123")

    # 验证管理员账户
    if username == "admin" and password == admin_password:
        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",
]