rag/api/auth_routes.py

"""
认证与系统状态 API

路由:
- POST /auth/login - 模拟登录（仅开发环境）
- GET  /stats      - 系统统计 (管理员)
- GET  /health     - 健康检查
- GET  /auth/me    - 当前用户信息
"""

from flask import Blueprint, request, jsonify
from auth.gateway import require_gateway_auth, require_role, get_user_permissions, MOCK_USERS
import os
import time
from pathlib import Path
from dotenv import load_dotenv

# 加载 .env 文件（从项目根目录）
env_path = Path(__file__).parent.parent / '.env'
load_dotenv(env_path)

auth_bp = Blueprint('auth', __name__)


# ==================== 登录速率限制 ====================
# 简单的内存级速率限制：每个 IP 在时间窗口内最多允许 N 次登录尝试
_login_attempts = {}  # {ip: [(timestamp, ...), ...]}
_RATE_LIMIT_WINDOW = 300   # 5 分钟窗口
_RATE_LIMIT_MAX = 10       # 窗口内最多 10 次尝试


def _check_rate_limit(client_ip: str) -> bool:
    """检查是否超出登录速率限制，返回 True 表示允许"""
    now = time.time()
    if client_ip not in _login_attempts:
        _login_attempts[client_ip] = []

    # 清理过期记录
    _login_attempts[client_ip] = [
        t for t in _login_attempts[client_ip]
        if now - t < _RATE_LIMIT_WINDOW
    ]

    if len(_login_attempts[client_ip]) >= _RATE_LIMIT_MAX:
        return False

    _login_attempts[client_ip].append(now)
    return True


def _is_dev_mode() -> bool:
    """统一的开发模式判断"""
    return os.environ.get('DEV_MODE', 'true').lower() != 'false'


@auth_bp.route('/auth/login', methods=['POST'])
def mock_login():
    """
    模拟登录（仅开发环境）

    请求体:
    {
        "username": "admin",
        "password": "admin123"
    }

    返回:
    {
        "token": "mock-token-admin",
        "user": {
            "user_id": "admin001",
            "username": "admin",
            "role": "admin",
            "department": "管理部"
        }
    }

    测试账号:
    - admin / admin123 (管理员，管理部)
    - admin2 / admin456 (管理员，技术部)
    - admin3 / admin789 (管理员，运营部)
    - manager / manager123 (经理，财务部)
    - user / test123 (普通用户，技术部)
    """
    if not _is_dev_mode():
        return jsonify({"error": "仅开发环境可用，请设置 DEV_MODE=true"}), 403

    # 速率限制检查
    client_ip = request.remote_addr or 'unknown'
    if not _check_rate_limit(client_ip):
        return jsonify({"error": f"登录尝试过于频繁，请 {_RATE_LIMIT_WINDOW // 60} 分钟后再试"}), 429

    data = request.json or {}
    username = data.get('username')
    password = data.get('password')

    user = MOCK_USERS.get(username)
    if not user or user['password'] != password:
        return jsonify({"error": "用户名或密码错误"}), 401

    return jsonify({
        "token": f"mock-token-{username}",
        "user": {
            "user_id": user['user_id'],
            "username": username,
            "role": user['role'],
            "department": user['department']
        }
    })


@auth_bp.route('/stats', methods=['GET'])
@require_gateway_auth
@require_role('admin')
def get_stats():
    """获取系统统计信息（仅管理员）"""
    from flask import current_app
    session_manager = current_app.config.get('SESSION_MANAGER')
    if not session_manager:
        return jsonify({"error": "会话管理器未启用"}), 503
    return jsonify(session_manager.get_stats())


@auth_bp.route('/health', methods=['GET'])
def health():
    """健康检查"""
    return jsonify({
        "status": "ok",
        "knowledge_base": "多向量库模式 (按集合提供服务)",
        "bm25_index": "动态按需加载",
        "mode": "Agentic RAG"
    })


@auth_bp.route('/auth/me', methods=['GET'])
@require_gateway_auth
def get_current_user():
    """
    获取当前用户信息

    开发模式下支持模拟用户，生产模式下用户信息由后端控制。
    """
    user = request.current_user
    return jsonify({
        "user_id": user["user_id"],
        "username": user["username"],
        "role": user["role"],
        "department": user["department"],
        "permissions": get_user_permissions(user["role"])
    })


@auth_bp.route('/auth/users', methods=['GET'])
@require_gateway_auth
def get_users():
    """
    获取用户列表（仅开发环境）

    返回:
    {
        "users": [
            {
                "user_id": "admin001",
                "username": "admin",
                "role": "admin",
                "department": "管理部",
                "is_active": true
            }
        ]
    }
    """
    if not _is_dev_mode():
        return jsonify({"error": "仅开发环境可用"}), 403

    users = []
    for username, info in MOCK_USERS.items():
        users.append({
            "user_id": info["user_id"],
            "username": username,
            "role": info["role"],
            "department": info["department"],
            "is_active": True  # 模拟用户默认都是活跃状态
        })

    return jsonify({"users": users})


@auth_bp.route('/auth/users/<user_id>', methods=['PUT'])
@require_gateway_auth
def update_user(user_id):
    """
    更新用户信息（仅开发环境，模拟操作）

    请求体:
    {
        "is_active": false
    }
    """
    if not _is_dev_mode():
        return jsonify({"error": "仅开发环境可用"}), 403

    # 验证目标用户是否存在
    target_user = None
    for username, info in MOCK_USERS.items():
        if info['user_id'] == user_id:
            target_user = info
            break

    if not target_user:
        return jsonify({"error": f"用户 {user_id} 不存在"}), 404

    data = request.json or {}
    # 模拟操作：记录请求但不实际执行（mock 用户数据是静态的）
    return jsonify({
        "message": "操作成功（模拟）",
        "user_id": user_id,
        "applied_changes": data
    })


@auth_bp.route('/auth/change-password', methods=['POST'])
@require_gateway_auth
def change_password():
    """
    修改密码（仅开发环境，模拟操作）

    请求体:
    {
        "old_password": "xxx",
        "new_password": "xxx"
    }
    """
    if not _is_dev_mode():
        return jsonify({"error": "仅开发环境可用"}), 403

    data = request.json or {}
    old_password = data.get('old_password')
    new_password = data.get('new_password')

    if not old_password or not new_password:
        return jsonify({"error": "请提供旧密码和新密码"}), 400

    if len(new_password) < 6:
        return jsonify({"error": "新密码至少6位"}), 400

    # 验证当前用户的旧密码
    user = request.current_user
    username = user.get('username', '')
    mock_user = MOCK_USERS.get(username)
    if mock_user and mock_user['password'] != old_password:
        return jsonify({"error": "旧密码错误"}), 401

    # 模拟环境返回成功（不实际修改密码，mock 数据是静态的）
    return jsonify({"message": "密码修改成功（模拟）"})