taiji-AI-PAD/services/mcp-server/app/routes/admin.py

"""
超级管理员API路由
"""

from datetime import datetime, timedelta
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import select, func, and_, desc, or_
from sqlalchemy.ext.asyncio import AsyncSession
import uuid

from database import get_db
from models import (
    User, Channel, Agent, ResourceAllocation,
    BillingRecord, Application, ModelProvider
)
from app.auth import require_auth, get_password_hash
from app.schemas import (
    SuccessResponse,
    CreateChannelRequest,
    UpdateChannelRequest,
    UpdateAgentConfigRequest,
    ChannelInfo,
    ChannelResourceAllocation,
    ApplicationInfo,
    ReviewApplicationRequest,
    AdminBillingResponse,
)

router = APIRouter(prefix="/api/admin", tags=["超级管理员"])


def _verify_admin_permission(principal: dict):
    """验证超级管理员权限"""
    role = principal.get("claims", {}).get("role")
    if role not in ["super_admin", "admin"]:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="需要超级管理员权限"
        )


# ============= 概览 =============

@router.get("/dashboard/stats", response_model=SuccessResponse)
async def get_admin_dashboard_stats(
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    获取平台全局统计
    """
    _verify_admin_permission(principal)
    
    # 渠道总数
    channels_count = await db.execute(select(func.count(Channel.id)))
    total_channels = channels_count.scalar() or 0
    
    # 租户总数
    tenants_count = await db.execute(
        select(func.count(User.id)).where(User.role == "user")
    )
    total_tenants = tenants_count.scalar() or 0
    
    # Agent总数
    agents_count = await db.execute(select(func.count(Agent.id)))
    total_agents = agents_count.scalar() or 0
    
    # 总调用次数
    calls_count = await db.execute(select(func.count(BillingRecord.id)))
    total_calls = calls_count.scalar() or 0
    
    # 总收入
    revenue = await db.execute(select(func.sum(BillingRecord.cost)))
    total_revenue = float(revenue.scalar() or 0)
    
    return SuccessResponse(
        data={
            "totalChannels": total_channels,
            "totalTenants": total_tenants,
            "totalAgents": total_agents,
            "totalCalls": total_calls,
            "totalRevenue": total_revenue,
        }
    )


# ============= 渠道管理 =============

@router.get("/channels", response_model=SuccessResponse)
async def list_channels(
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    获取所有渠道列表
    """
    _verify_admin_permission(principal)
    
    result = await db.execute(select(Channel))
    channels = result.scalars().all()
    
    data = [
        {
            "id": str(channel.id),
            "name": channel.name,
            "email": channel.email,
            "commissionRate": float(channel.commission_rate),
            "channelCredit": float(channel.channel_credit),
            "customAgentCpu": float(channel.custom_agent_cpu),
            "customAgentMemory": float(channel.custom_agent_memory),
            "status": channel.status,
            "createdAt": channel.created_at.isoformat(),
        }
        for channel in channels
    ]
    
    return SuccessResponse(data={"channels": data})


@router.post("/channels/create", response_model=SuccessResponse)
async def create_channel(
    req: CreateChannelRequest,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    创建渠道
    """
    _verify_admin_permission(principal)
    
    # 检查邮箱是否已存在
    result = await db.execute(
        select(Channel).where(Channel.email == req.email)
    )
    if result.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="邮箱已被使用"
        )
    
    # 创建渠道
    password_hash = get_password_hash(req.password)
    channel = Channel(
        name=req.name,
        email=req.email,
        password_hash=password_hash,
        commission_rate=req.commissionRate,
        channel_credit=0,
        custom_agent_cpu=2,
        custom_agent_memory=4,
        status="active",
    )
    
    db.add(channel)
    await db.commit()
    await db.refresh(channel)
    
    return SuccessResponse(
        data={
            "id": str(channel.id),
            "name": channel.name,
            "email": channel.email,
        },
        message="渠道创建成功"
    )


@router.put("/channels/{channel_id}", response_model=SuccessResponse)
async def update_channel(
    channel_id: str,
    req: UpdateChannelRequest,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    更新渠道信息
    """
    _verify_admin_permission(principal)
    
    # 验证渠道存在
    result = await db.execute(
        select(Channel).where(Channel.id == channel_id)
    )
    channel = result.scalar_one_or_none()
    
    if not channel:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="渠道不存在"
        )
    
    # 如果更新邮箱，检查邮箱是否已被其他渠道使用
    if req.email and req.email != channel.email:
        email_check = await db.execute(
            select(Channel).where(
                and_(
                    Channel.email == req.email,
                    Channel.id != channel_id
                )
            )
        )
        if email_check.scalar_one_or_none():
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="邮箱已被其他渠道使用"
            )
        channel.email = req.email
    
    # 更新其他字段
    if req.name is not None:
        channel.name = req.name
    if req.commissionRate is not None:
        channel.commission_rate = req.commissionRate
    if req.status is not None:
        channel.status = req.status
    
    await db.commit()
    await db.refresh(channel)
    
    return SuccessResponse(
        data={
            "id": str(channel.id),
            "name": channel.name,
            "email": channel.email,
            "commissionRate": float(channel.commission_rate),
            "status": channel.status,
        },
        message="渠道信息更新成功"
    )


@router.delete("/channels/{channel_id}", response_model=SuccessResponse)
async def delete_channel(
    channel_id: str,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    删除渠道（软删除，标记为不活跃）
    """
    _verify_admin_permission(principal)
    
    # 验证渠道存在
    result = await db.execute(
        select(Channel).where(Channel.id == channel_id)
    )
    channel = result.scalar_one_or_none()
    
    if not channel:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="渠道不存在"
        )
    
    # 检查是否有关联的租户
    tenants_result = await db.execute(
        select(func.count(User.id)).where(
            and_(
                User.channel_id == channel_id,
                User.role == "user",
                User.status == "active"
            )
        )
    )
    tenant_count = tenants_result.scalar() or 0
    
    if tenant_count > 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"渠道下有 {tenant_count} 个活跃租户，无法删除。请先移除或停用所有租户。"
        )
    
    # 软删除：标记为不活跃
    channel.status = "inactive"
    await db.commit()
    
    return SuccessResponse(
        data={"id": str(channel.id)},
        message="渠道已删除"
    )


@router.put("/channels/{channel_id}/resources", response_model=SuccessResponse)
async def allocate_channel_resources(
    channel_id: str,
    req: ChannelResourceAllocation,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    统一管理渠道资源
    """
    _verify_admin_permission(principal)
    
    # 验证渠道
    result = await db.execute(
        select(Channel).where(Channel.id == channel_id)
    )
    channel = result.scalar_one_or_none()
    
    if not channel:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="渠道不存在"
        )
    
    # 删除现有资源分配
    await db.execute(
        select(ResourceAllocation).where(
            and_(
                ResourceAllocation.target_id == channel_id,
                ResourceAllocation.target_type == "channel"
            )
        )
    )
    
    # 分配模型供应商资源
    for model_id in req.models:
        allocation = ResourceAllocation(
            target_id=channel_id,
            target_type="channel",
            resource_type="model",
            resource_id=model_id,
        )
        db.add(allocation)
    
    # 分配Agent资源
    for agent_alloc in req.agents:
        allocation = ResourceAllocation(
            target_id=channel_id,
            target_type="channel",
            resource_type="agent",
            resource_id=agent_alloc.agentId,
            quantity=agent_alloc.quantity,
        )
        db.add(allocation)
    
    # 更新自定义Agent资源和授信额度
    if req.customAgentResources:
        channel.custom_agent_cpu = req.customAgentResources.cpu
        channel.custom_agent_memory = req.customAgentResources.memory
    
    channel.channel_credit = req.channelCredit
    
    await db.commit()
    
    return SuccessResponse(message="渠道资源分配成功")


# ============= 申请审批 =============

@router.get("/channels/applications", response_model=SuccessResponse)
async def list_applications(
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    获取所有渠道申请
    """
    _verify_admin_permission(principal)
    
    result = await db.execute(
        select(Application, Channel)
        .join(Channel, Application.channel_id == Channel.id)
        .order_by(desc(Application.created_at))
    )
    
    data = []
    for app, channel in result.all():
        details = {}
        if app.type == "model":
            details = {
                "modelName": app.model_name,
                "rpm": app.rpm,
                "tpm": app.tpm,
            }
        else:
            details = {
                "agentType": app.agent_type,
                "quantity": app.quantity,
            }
        
        data.append({
            "id": str(app.id),
            "channelId": str(app.channel_id),
            "channelName": channel.name,
            "type": app.type,
            "details": details,
            "reason": app.reason or "",
            "status": app.status,
            "createdAt": app.created_at.isoformat(),
        })
    
    return SuccessResponse(data={"data": data})


@router.put("/channels/applications/{application_id}/review", response_model=SuccessResponse)
async def review_application(
    application_id: str,
    req: ReviewApplicationRequest,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    审批申请
    """
    _verify_admin_permission(principal)
    user_id = principal.get("user_id")
    
    # 查询申请
    result = await db.execute(
        select(Application).where(Application.id == application_id)
    )
    application = result.scalar_one_or_none()
    
    if not application:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="申请不存在"
        )
    
    if application.status != "pending":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="该申请已被处理"
        )
    
    # 更新申请状态
    application.status = "approved" if req.approved else "rejected"
    application.reviewed_by = user_id
    application.review_reason = req.reason
    application.reviewed_at = datetime.utcnow()
    
    # 如果批准，自动分配资源
    if req.approved:
        if application.type == "model":
            # 查找模型供应商
            result = await db.execute(
                select(ModelProvider).where(ModelProvider.name == application.model_name)
            )
            model_provider = result.scalar_one_or_none()
            
            if model_provider:
                allocation = ResourceAllocation(
                    target_id=str(application.channel_id),
                    target_type="channel",
                    resource_type="model",
                    resource_id=str(model_provider.id),
                    rpm=application.rpm,
                    tpm=application.tpm,
                )
                db.add(allocation)
        
        elif application.type == "agent":
            # 查找Agent（简化处理，实际应该根据agent_type查找）
            result = await db.execute(
                select(Agent).where(Agent.name == application.agent_type).limit(1)
            )
            agent = result.scalar_one_or_none()
            
            if agent:
                allocation = ResourceAllocation(
                    target_id=str(application.channel_id),
                    target_type="channel",
                    resource_type="agent",
                    resource_id=str(agent.id),
                    quantity=application.quantity,
                )
                db.add(allocation)
    
    await db.commit()
    
    return SuccessResponse(
        message=f"申请已{'批准' if req.approved else '拒绝'}"
    )


# ============= 资源管理 =============

@router.get("/resources/models", response_model=SuccessResponse)
async def list_model_providers(
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    获取所有模型供应商
    """
    _verify_admin_permission(principal)
    
    result = await db.execute(select(ModelProvider))
    providers = result.scalars().all()
    
    data = [
        {
            "id": str(provider.id),
            "name": provider.name,
            "provider": provider.provider,
            "apiUrl": provider.api_url,
            "supportedModels": provider.supported_models,
            "rpm": provider.rpm,
            "tpm": provider.tpm,
            "status": provider.status,
            "isActive": provider.is_active,
        }
        for provider in providers
    ]
    
    return SuccessResponse(data={"providers": data})


@router.get("/resources/agents", response_model=SuccessResponse)
async def list_all_agents(
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    获取所有Agent资源
    """
    _verify_admin_permission(principal)
    
    result = await db.execute(select(Agent))
    agents = result.scalars().all()
    
    data = [
        {
            "id": str(agent.id),
            "name": agent.name,
            "type": agent.type,
            "category": agent.category,
            "cpu": float(agent.cpu),
            "memory": float(agent.memory),
            "status": agent.status,
        }
        for agent in agents
    ]
    
    return SuccessResponse(data={"agents": data})


@router.delete("/resources/agents/{agent_id}", response_model=SuccessResponse)
async def delete_agent_resource(
    agent_id: str,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    删除Agent资源（软删除，标记为不活跃）
    """
    _verify_admin_permission(principal)
    
    # 验证Agent存在
    result = await db.execute(
        select(Agent).where(Agent.id == agent_id)
    )
    agent = result.scalar_one_or_none()
    
    if not agent:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Agent资源不存在"
        )
    
    # 软删除：标记为不活跃
    agent.status = "inactive"
    await db.commit()
    
    return SuccessResponse(
        data={"id": str(agent.id), "name": agent.name},
        message="Agent资源已删除"
    )


@router.put("/resources/agents/{agent_id}/config", response_model=SuccessResponse)
async def update_agent_config(
    agent_id: str,
    req: UpdateAgentConfigRequest,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    更新Agent资源配置（CPU、内存、最大实例数）
    """
    _verify_admin_permission(principal)
    
    # 验证Agent存在
    result = await db.execute(
        select(Agent).where(Agent.id == agent_id)
    )
    agent = result.scalar_one_or_none()
    
    if not agent:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Agent资源不存在"
        )
    
    # 更新配置
    if req.cpu is not None:
        agent.cpu = req.cpu
    if req.memory is not None:
        agent.memory = req.memory
    if req.maxInstances is not None:
        agent.max_instances = req.maxInstances
    
    await db.commit()
    await db.refresh(agent)
    
    return SuccessResponse(
        data={
            "id": str(agent.id),
            "name": agent.name,
            "cpu": float(agent.cpu),
            "memory": float(agent.memory),
            "maxInstances": agent.max_instances,
        },
        message="Agent资源配置更新成功"
    )


# ============= 监控 =============

@router.get("/monitoring/agents", response_model=SuccessResponse)
async def monitor_agents(
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    监控Agent健康状态和性能指标
    """
    _verify_admin_permission(principal)
    
    result = await db.execute(
        select(Agent).where(Agent.type == "platform")
    )
    agents = result.scalars().all()
    
    data = [
        {
            "id": str(agent.id),
            "name": agent.name,
            "status": agent.status,
            "totalExecutions": agent.total_executions,
            "successRate": float(agent.success_rate),
            "avgExecutionTime": float(agent.avg_execution_time),
            "cpu": float(agent.cpu),
            "memory": float(agent.memory),
        }
        for agent in agents
    ]
    
    return SuccessResponse(data={"agents": data})


# ============= 计费（三维度） =============

@router.get("/billing/overview", response_model=SuccessResponse)
async def get_billing_overview(
    startTime: str = Query(...),
    endTime: str = Query(...),
    channelName: Optional[str] = Query(None),
    tenantName: Optional[str] = Query(None),
    minCalls: Optional[int] = Query(None),
    maxCalls: Optional[int] = Query(None),
    export: Optional[str] = Query(None, pattern="^(excel|csv|pdf)$"),
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    获取三维度计费统计（渠道/租户/调用）
    """
    _verify_admin_permission(principal)
    
    # 解析时间
    start_dt = datetime.fromisoformat(startTime.replace("Z", "+00:00"))
    end_dt = datetime.fromisoformat(endTime.replace("Z", "+00:00"))
    
    # 渠道统计
    channel_stats_result = await db.execute(
        select(
            Channel.id,
            Channel.name,
            func.count(BillingRecord.id).label("calls"),
            func.sum(BillingRecord.eu).label("total_eu"),
            func.sum(BillingRecord.cost).label("total_cost"),
        )
        .select_from(BillingRecord)
        .join(Channel, BillingRecord.channel_id == Channel.id)
        .where(
            and_(
                BillingRecord.timestamp >= start_dt,
                BillingRecord.timestamp <= end_dt,
            )
        )
        .group_by(Channel.id, Channel.name)
    )
    
    channel_stats = [
        {
            "channelId": str(row.id),
            "channelName": row.name,
            "calls": row.calls,
            "totalEU": float(row.total_eu or 0),
            "totalCost": float(row.total_cost or 0),
        }
        for row in channel_stats_result.all()
    ]
    
    # 租户统计
    tenant_stats_result = await db.execute(
        select(
            User.id,
            User.name,
            Channel.name.label("channel_name"),
            func.count(BillingRecord.id).label("calls"),
            func.sum(BillingRecord.eu).label("total_eu"),
            func.sum(BillingRecord.cost).label("total_cost"),
        )
        .select_from(BillingRecord)
        .join(User, BillingRecord.tenant_id == User.id)
        .join(Channel, User.channel_id == Channel.id)
        .where(
            and_(
                BillingRecord.timestamp >= start_dt,
                BillingRecord.timestamp <= end_dt,
            )
        )
        .group_by(User.id, User.name, Channel.name)
    )
    
    tenant_stats = [
        {
            "tenantId": str(row.id),
            "tenantName": row.name,
            "channelName": row.channel_name,
            "calls": row.calls,
            "totalEU": float(row.total_eu or 0),
            "totalCost": float(row.total_cost or 0),
        }
        for row in tenant_stats_result.all()
    ]
    
    # 调用记录
    records_result = await db.execute(
        select(BillingRecord, Channel.name, User.name)
        .join(Channel, BillingRecord.channel_id == Channel.id)
        .join(User, BillingRecord.tenant_id == User.id)
        .where(
            and_(
                BillingRecord.timestamp >= start_dt,
                BillingRecord.timestamp <= end_dt,
            )
        )
        .order_by(desc(BillingRecord.timestamp))
        .limit(100)
    )
    
    call_records = [
        {
            "id": str(record.id),
            "timestamp": record.timestamp.isoformat(),
            "channelName": channel_name,
            "tenantName": tenant_name,
            "agentName": record.agent_name,
            "duration": record.duration,
            "eu": record.eu,
            "cost": float(record.cost),
        }
        for record, channel_name, tenant_name in records_result.all()
    ]
    
    # 如果是导出请求
    if export:
        file_url = f"https://exports.taiji-ai.com/admin/{export}/billing_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}.{export}"
        expires_at = (datetime.utcnow() + timedelta(hours=24)).isoformat()
        
        return SuccessResponse(
            data={
                "fileUrl": file_url,
                "format": export,
                "expiresAt": expires_at,
            }
        )
    
    return SuccessResponse(
        data={
            "channelStats": channel_stats,
            "tenantStats": tenant_stats,
            "callRecords": call_records,
        }
    )