qihuaiflow/backend/python/app/api/admin/storage_admin.py

"""
OSS存储配置管理 API
Phase 3.1: 支持 MinIO / 阿里云OSS / AWS S3 等对象存储配置
"""
from fastapi import APIRouter, Depends, HTTPException, Query, Body
from sqlalchemy import or_, func, desc, asc, text
from sqlalchemy.orm import Session
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from datetime import datetime
import uuid
import json
import logging
import httpx

from ...database import get_db
from ...models import (
    gen_uuid,
)
from ...auth import get_current_admin
from ...middleware.log_service import LogService

router = APIRouter()
logger = logging.getLogger("aiflow")


# ============================================================
# Pydantic Models
# ============================================================

class OSSConfigCreate(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    provider: str = Field(min_length=1, max_length=20)
    endpoint: str = Field(min_length=1, max_length=500)
    access_key: str = Field(min_length=1, max_length=255)
    secret_key: str = Field(min_length=1, max_length=255)
    bucket: str = Field(min_length=1, max_length=200)
    region: Optional[str] = ""
    base_path: Optional[str] = ""
    is_default: bool = False
    is_enabled: bool = True
    config: Optional[Dict[str, Any]] = None


class OSSConfigUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=200)
    provider: Optional[str] = Field(None, min_length=1, max_length=20)
    endpoint: Optional[str] = Field(None, min_length=1, max_length=500)
    access_key: Optional[str] = Field(None, min_length=1, max_length=255)
    secret_key: Optional[str] = Field(None, min_length=1, max_length=255)
    bucket: Optional[str] = Field(None, min_length=1, max_length=200)
    region: Optional[str] = None
    base_path: Optional[str] = None
    is_default: Optional[bool] = None
    is_enabled: Optional[bool] = None
    config: Optional[Dict[str, Any]] = None


class OSSConnectionTest(BaseModel):
    """测试连接请求 - 可以测试已保存的配置或临时配置"""
    config_id: Optional[str] = None  # 已保存的配置ID
    # 以下为临时测试参数
    provider: Optional[str] = None
    endpoint: Optional[str] = None
    access_key: Optional[str] = None
    secret_key: Optional[str] = None
    bucket: Optional[str] = None
    region: Optional[str] = None


# ============================================================
# Helper Functions
# ============================================================

def config_to_dict(config) -> dict:
    """将OSSStorageConfig模型转换为字典"""
    result = {
        "id": config.id,
        "name": config.name,
        "provider": config.provider,
        "endpoint": config.endpoint,
        "access_key": config.access_key[:8] + "****" if config.access_key and len(config.access_key) > 8 else "****",
        "secret_key": "****",  # 永远不返回完整密钥
        "bucket": config.bucket,
        "region": config.region or "",
        "base_path": config.base_path or "",
        "is_default": getattr(config, "is_default", False),
        "is_enabled": config.is_enabled,
        "status": config.status or "disconnected",
        "created_at": config.created_at.isoformat() if config.created_at else None,
        "updated_at": config.updated_at.isoformat() if config.updated_at else None,
    }

    # 安全处理 JSON 字段
    config_raw = getattr(config, "config", None)
    if config_raw:
        try:
            result["config"] = json.loads(config_raw) if isinstance(config_raw, str) else config_raw
        except Exception:
            result["config"] = {}
    else:
        result["config"] = {}

    return result


async def test_oss_connection(provider: str, endpoint: str, access_key: str, secret_key: str,
                                bucket: str, region: str = "") -> Dict[str, Any]:
    """测试OSS连接 - 使用httpx发送HTTP请求到对象存储端点"""
    import time
    start_time = time.time()

    try:
        if provider in ("minio", "aws_s3"):
            # 使用 AWS S3 兼容 API 测试
            # ListBuckets 请求
            import hmac
            import hashlib
            from urllib.parse import urlparse

            parsed = urlparse(endpoint)
            host = parsed.netloc or parsed.path
            if not host:
                return {"success": False, "error": "无法解析端点地址", "latency_ms": 0}

            # 先尝试简单的 bucket HEAD 请求
            bucket_url = f"{endpoint.rstrip('/')}/{bucket}"
            async with httpx.AsyncClient(timeout=10.0, verify=False) as client:
                resp = await client.head(bucket_url)
                latency = int((time.time() - start_time) * 1000)

                if resp.status_code in (200, 403):
                    # 200: bucket存在且有权限; 403: bucket存在但无权限(也是连通的表现)
                    return {
                        "success": True,
                        "message": f"连接成功！Bucket '{bucket}' 可访问",
                        "latency_ms": latency,
                        "status_code": resp.status_code
                    }
                elif resp.status_code == 404:
                    return {
                        "success": False,
                        "error": f"Bucket '{bucket}' 不存在",
                        "latency_ms": latency
                    }
                else:
                    return {
                        "success": False,
                        "error": f"连接失败: HTTP {resp.status_code}",
                        "latency_ms": latency
                    }

        elif provider == "aliyun_oss":
            # 阿里云OSS: 简单HEAD请求
            bucket_url = f"{endpoint.rstrip('/')}/{bucket}" if not endpoint.endswith(bucket) else endpoint
            async with httpx.AsyncClient(timeout=10.0, verify=False) as client:
                resp = await client.head(bucket_url)
                latency = int((time.time() - start_time) * 1000)

                if resp.status_code in (200, 403):
                    return {
                        "success": True,
                        "message": f"连接成功！Bucket '{bucket}' 可访问",
                        "latency_ms": latency,
                        "status_code": resp.status_code
                    }
                elif resp.status_code == 404:
                    return {
                        "success": False,
                        "error": f"Bucket '{bucket}' 不存在",
                        "latency_ms": latency
                    }
                else:
                    return {
                        "success": False,
                        "error": f"连接失败: HTTP {resp.status_code}",
                        "latency_ms": latency
                    }

        else:
            return {"success": False, "error": f"不支持的存储类型: {provider}", "latency_ms": 0}

    except httpx.ConnectError:
        latency = int((time.time() - start_time) * 1000)
        return {"success": False, "error": f"无法连接到 {endpoint}", "latency_ms": latency}
    except httpx.TimeoutException:
        latency = int((time.time() - start_time) * 1000)
        return {"success": False, "error": "连接超时(10秒)", "latency_ms": latency}
    except Exception as e:
        latency = int((time.time() - start_time) * 1000)
        return {"success": False, "error": f"测试失败: {str(e)}", "latency_ms": latency}


# ============================================================
# API Endpoints
# ============================================================

@router.get("/stats/overview", summary="存储配置统计概览")
async def get_storage_stats(
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取存储配置的统计信息"""
    from ...models import OSSStorageConfig

    total = db.query(func.count(OSSStorageConfig.id)).scalar() or 0
    enabled_count = db.query(func.count(OSSStorageConfig.id)).filter(
        OSSStorageConfig.is_enabled == True
    ).scalar() or 0
    default_count = db.query(func.count(OSSStorageConfig.id)).filter(
        OSSStorageConfig.is_default == True
    ).scalar() or 0

    # 按类型统计
    by_provider = {}
    providers = db.query(OSSStorageConfig.provider, func.count(OSSStorageConfig.id)).group_by(
        OSSStorageConfig.provider
    ).all()
    for provider, count in providers:
        by_provider[provider] = count

    # 按状态统计
    by_status = {}
    statuses = db.query(OSSStorageConfig.status, func.count(OSSStorageConfig.id)).group_by(
        OSSStorageConfig.status
    ).all()
    for status, count in statuses:
        by_status[status] = count

    return {
        "code": 0,
        "data": {
            "total": total,
            "enabled_count": enabled_count,
            "default_count": default_count,
            "by_provider": by_provider,
            "by_status": by_status,
        },
        "message": "ok"
    }


@router.get("/", summary="获取OSS存储配置列表")
async def list_storage_configs(
    provider: str = Query(None, description="按提供商筛选"),
    is_enabled: bool = Query(None, description="按启用状态筛选"),
    keyword: str = Query(None, description="搜索关键词"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取OSS存储配置列表，支持分页和筛选"""
    from ...models import OSSStorageConfig

    query = db.query(OSSStorageConfig)

    if provider:
        query = query.filter(OSSStorageConfig.provider == provider)
    if is_enabled is not None:
        query = query.filter(OSSStorageConfig.is_enabled == is_enabled)
    if keyword:
        keyword_filter = or_(
            OSSStorageConfig.name.ilike(f"%{keyword}%"),
            OSSStorageConfig.endpoint.ilike(f"%{keyword}%"),
            OSSStorageConfig.bucket.ilike(f"%{keyword}%"),
        )
        query = query.filter(keyword_filter)

    total = query.count()
    items = query.order_by(desc(OSSStorageConfig.is_default), desc(OSSStorageConfig.updated_at)) \
        .offset((page - 1) * page_size).limit(page_size).all()

    return {
        "code": 0,
        "data": {
            "items": [config_to_dict(item) for item in items],
            "total": total,
            "page": page,
            "page_size": page_size,
        },
        "message": "ok"
    }


@router.get("/{config_id}", summary="获取OSS存储配置详情")
async def get_storage_config(
    config_id: str,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取单个OSS存储配置的详细信息"""
    from ...models import OSSStorageConfig

    config = db.query(OSSStorageConfig).filter(OSSStorageConfig.id == config_id).first()
    if not config:
        raise HTTPException(status_code=404, detail="存储配置不存在")

    return {
        "code": 0,
        "data": config_to_dict(config),
        "message": "ok"
    }


@router.post("/", summary="创建OSS存储配置")
async def create_storage_config(
    req: OSSConfigCreate,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """创建新的OSS存储配置"""
    from ...models import OSSStorageConfig

    # 检查名称重复
    existing = db.query(OSSStorageConfig).filter(OSSStorageConfig.name == req.name).first()
    if existing:
        raise HTTPException(status_code=400, detail="配置名称已存在")

    # 如果设为默认，先取消其他默认
    if req.is_default:
        db.query(OSSStorageConfig).filter(OSSStorageConfig.is_default == True).update(
            {"is_default": False}
        )

    config = OSSStorageConfig(
        id=gen_uuid(),
        name=req.name,
        provider=req.provider,
        endpoint=req.endpoint,
        access_key=req.access_key,
        secret_key=req.secret_key,
        bucket=req.bucket,
        region=req.region or "",
        base_path=req.base_path or "",
        is_default=req.is_default,
        is_enabled=req.is_enabled,
        status="disconnected",
        config=req.config,
    )

    db.add(config)
    db.commit()
    db.refresh(config)

    # 审计日志
    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="create_oss_config",
        target_type="oss_storage_config",
        target_id=config.id,
        detail=f"创建了OSS存储配置: {config.name} ({config.provider})"
    )

    return {
        "code": 0,
        "data": config_to_dict(config),
        "message": "创建成功"
    }


@router.put("/{config_id}", summary="更新OSS存储配置")
async def update_storage_config(
    config_id: str,
    req: OSSConfigUpdate,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """更新OSS存储配置"""
    from ...models import OSSStorageConfig

    config = db.query(OSSStorageConfig).filter(OSSStorageConfig.id == config_id).first()
    if not config:
        raise HTTPException(status_code=404, detail="存储配置不存在")

    # 检查名称重复
    if req.name is not None and req.name != config.name:
        existing = db.query(OSSStorageConfig).filter(
            OSSStorageConfig.name == req.name,
            OSSStorageConfig.id != config_id
        ).first()
        if existing:
            raise HTTPException(status_code=400, detail="配置名称已存在")

    # 如果设为默认，先取消其他默认
    if req.is_default and not config.is_default:
        db.query(OSSStorageConfig).filter(OSSStorageConfig.is_default == True).update(
            {"is_default": False}
        )

    # 更新字段
    update_fields = ["name", "provider", "endpoint", "access_key", "secret_key",
                     "bucket", "region", "base_path", "is_default", "is_enabled"]
    for field in update_fields:
        value = getattr(req, field, None)
        if value is not None:
            setattr(config, field, value)
            # 如果修改了连接参数，重置状态
            if field in ("endpoint", "access_key", "secret_key", "bucket", "provider"):
                config.status = "disconnected"

    if req.config is not None:
        config.config = req.config

    config.updated_at = datetime.now()

    db.commit()
    db.refresh(config)

    # 审计日志
    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="update_oss_config",
        target_type="oss_storage_config",
        target_id=config.id,
        detail=f"更新了OSS存储配置: {config.name}"
    )

    return {
        "code": 0,
        "data": config_to_dict(config),
        "message": "更新成功"
    }


@router.delete("/{config_id}", summary="删除OSS存储配置")
async def delete_storage_config(
    config_id: str,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """删除OSS存储配置"""
    from ...models import OSSStorageConfig

    config = db.query(OSSStorageConfig).filter(OSSStorageConfig.id == config_id).first()
    if not config:
        raise HTTPException(status_code=404, detail="存储配置不存在")

    name = config.name
    db.delete(config)
    db.commit()

    # 审计日志
    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="delete_oss_config",
        target_type="oss_storage_config",
        target_id=config_id,
        detail=f"删除了OSS存储配置: {name}"
    )

    return {
        "code": 0,
        "data": None,
        "message": f"已删除配置: {name}"
    }


@router.put("/{config_id}/toggle", summary="启用/禁用存储配置")
async def toggle_storage_config(
    config_id: str,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """切换存储配置的启用/禁用状态"""
    from ...models import OSSStorageConfig

    config = db.query(OSSStorageConfig).filter(OSSStorageConfig.id == config_id).first()
    if not config:
        raise HTTPException(status_code=404, detail="存储配置不存在")

    config.is_enabled = not config.is_enabled
    config.updated_at = datetime.now()
    db.commit()
    db.refresh(config)

    action = "启用" if config.is_enabled else "禁用"
    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="toggle_oss_config",
        target_type="oss_storage_config",
        target_id=config.id,
        detail=f"{action}了OSS存储配置: {config.name}"
    )

    return {
        "code": 0,
        "data": config_to_dict(config),
        "message": f"已{action}"
    }


@router.put("/{config_id}/set-default", summary="设为默认存储配置")
async def set_default_storage_config(
    config_id: str,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """将指定配置设为默认存储"""
    from ...models import OSSStorageConfig

    config = db.query(OSSStorageConfig).filter(OSSStorageConfig.id == config_id).first()
    if not config:
        raise HTTPException(status_code=404, detail="存储配置不存在")

    # 取消所有默认
    db.query(OSSStorageConfig).filter(OSSStorageConfig.is_default == True).update(
        {"is_default": False}
    )

    # 设置新默认
    config.is_default = True
    config.updated_at = datetime.now()
    db.commit()
    db.refresh(config)

    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="set_default_oss_config",
        target_type="oss_storage_config",
        target_id=config.id,
        detail=f"将 {config.name} 设为默认存储配置"
    )

    return {
        "code": 0,
        "data": config_to_dict(config),
        "message": f"已将 {config.name} 设为默认存储"
    }


@router.post("/test", summary="测试OSS连接")
async def test_storage_connection(
    req: OSSConnectionTest,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """测试OSS存储连接"""
    from ...models import OSSStorageConfig

    provider = req.provider
    endpoint = req.endpoint
    access_key = req.access_key
    secret_key = req.secret_key
    bucket = req.bucket
    region = req.region or ""

    # 如果提供了config_id，从数据库获取配置
    if req.config_id:
        config = db.query(OSSStorageConfig).filter(OSSStorageConfig.id == req.config_id).first()
        if not config:
            raise HTTPException(status_code=404, detail="存储配置不存在")
        provider = config.provider
        endpoint = config.endpoint
        access_key = config.access_key
        secret_key = config.secret_key
        bucket = config.bucket
        region = config.region or ""

    if not all([provider, endpoint, access_key, secret_key, bucket]):
        raise HTTPException(status_code=400, detail="缺少必要的连接参数")

    result = await test_oss_connection(provider, endpoint, access_key, secret_key, bucket, region)

    # 如果是已保存的配置，更新连接状态
    if req.config_id:
        config = db.query(OSSStorageConfig).filter(OSSStorageConfig.id == req.config_id).first()
        if config:
            config.status = "connected" if result["success"] else "disconnected"
            config.updated_at = datetime.now()
            db.commit()

    return {
        "code": 0,
        "data": result,
        "message": "连接测试完成"
    }