Unified_Authentication_Platform/backend/app/api/v1/endpoints/login_logs.py

from typing import Any, Optional
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session

from app.api.v1 import deps
from app.models.user import User, UserRole
from app.models.login_log import LoginMethod, AuthType
from app.schemas.login_log import LoginLogListResponse
from app.services.login_log_service import LoginLogService

router = APIRouter()

@router.get("/", response_model=LoginLogListResponse, summary="获取登录日志列表")
def get_login_logs(
    db: Session = Depends(deps.get_db),
    current_user: User = Depends(deps.get_current_active_user),
    skip: int = 0,
    limit: int = 20,
    mobile: Optional[str] = None,
    ip_address: Optional[str] = None,
    status: Optional[int] = Query(None, description="1: Success, 0: Failed"),
    login_method: Optional[LoginMethod] = None,
    auth_type: Optional[AuthType] = None,
    start_date: Optional[datetime] = None,
    end_date: Optional[datetime] = None
) -> Any:
    """
    获取统一认证登录日志。
    只有超级管理员可以查看所有日志。
    普通用户只能查看自己的日志（待定，目前假设只有管理员看运维日志）。
    """
    # 鉴权：只有 SUPER_ADMIN 和 DEVELOPER 可以查看运维日志？
    # 根据需求描述“运维管理里面有一个统一登录日志”，通常隐含管理员权限。
    if current_user.role not in [UserRole.SUPER_ADMIN, UserRole.DEVELOPER]:
        raise HTTPException(status_code=403, detail="权限不足")
    
    total, items = LoginLogService.get_logs(
        db=db,
        skip=skip,
        limit=limit,
        mobile=mobile,
        ip_address=ip_address,
        status=status,
        login_method=login_method,
        auth_type=auth_type,
        start_date=start_date,
        end_date=end_date
    )
    
    return {
        "total": total,
        "items": items
    }