taijibaga/taiji-AI-PAD
9
0
Fork 0
forked from xiaohei/taiji-AI-PAD
Code Pull Requests Activity

赌博前的备份

Browse Source
This commit is contained in:
zhanggangyong
2026-01-12 10:27:19 +00:00
parent b75f9671ff
commit e3d2cee85a
4 changed files with 862 additions and 1503 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/mcp-server/app/routes/frontend_integration.py
+19 -573
View File
@@ -897,226 +897,26 @@ async def admin_login(payload: Dict[str, str], db: AsyncSession = Depends(get_db
return {"token": token, "tokenType": "bearer", "email": email, "expiresIn": 60 * 60}
# 注意: /admin/dashboard/stats 接口已移至 admin.py，避免重复定义
# 该接口从 AgentBillingRecord 和 ModelBillingRecord 统计收入和调用次数
@router.get("/admin/channels")
async def admin_channels(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
import logging
logger = logging.getLogger(__name__)
channels = (await db.execute(select(Channel))).scalars().all()
# 获取每个渠道的租户数量
channel_tenant_counts = {}
for ch in channels:
tenant_count = (await db.execute(
select(func.count(User.id)).where(
User.channel_id == ch.id,
User.role == "user"
)
)).scalar() or 0
channel_tenant_counts[ch.id] = tenant_count
items = []
for ch in channels:
# 优先使用新字段 custom_agent_cpu_quota，如果为空则回退到旧字段 custom_agent_cpu
cpu_quota = ch.custom_agent_cpu_quota if ch.custom_agent_cpu_quota else ch.custom_agent_cpu
memory_quota = ch.custom_agent_memory_quota if ch.custom_agent_memory_quota else ch.custom_agent_memory
logger.debug(f"Channel {ch.name}: cpu_quota={ch.custom_agent_cpu_quota}, memory_quota={ch.custom_agent_memory_quota}, "
f"cpu={ch.custom_agent_cpu}, memory={ch.custom_agent_memory}")
items.append({
"id": str(ch.id),
"name": ch.name,
"email": ch.email,
"commissionRate": ch.commission_rate,
"channelCredit": float(ch.channel_credit) if ch.channel_credit else 0.0,
"customAgentCpu": float(cpu_quota) if cpu_quota else 0.0,
"customAgentMemory": float(memory_quota) if memory_quota else 0.0,
"status": ch.status or "active",
"createdAt": ch.created_at.isoformat() if ch.created_at else None,
"tenantCount": channel_tenant_counts.get(ch.id, 0),
"totalAllocatedCpu": 0.0, # TODO: 计算已分配给租户的CPU
"totalAllocatedMemory": 0.0, # TODO: 计算已分配给租户的内存
})
return {"success": True, "data": {"channels": items}, "message": None}
@router.post("/admin/channels/create")
async def admin_create_channel(payload: Dict[str, Any], db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
required = {"name", "email"}
missing = [k for k in required if not payload.get(k)]
if missing:
raise HTTPException(status_code=400, detail=f"Missing fields: {', '.join(missing)}")
channel = Channel(
name=payload["name"],
email=payload["email"],
commission_rate=payload.get("commissionRate", 0.0),
monthly_quota=payload.get("monthlyQuota", 0),
monthly_budget=payload.get("monthlyBudget", 0),
)
db.add(channel)
await db.commit()
await db.refresh(channel)
return {
"id": str(channel.id),
"name": channel.name,
"email": channel.email,
"commissionRate": channel.commission_rate,
"monthlyQuota": channel.monthly_quota,
"monthlyBudget": channel.monthly_budget,
}
@router.put("/admin/channels/{channel_id}/commission")
async def admin_update_channel_commission(channel_id: str, payload: Dict[str, Any], db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
channel = await db.get(Channel, uuid.UUID(channel_id)) if channel_id else None
if not channel:
raise HTTPException(status_code=404, detail="channel not found")
channel.commission_rate = payload.get("commissionRate", channel.commission_rate)
db.add(channel)
await db.commit()
await db.refresh(channel)
return {
"id": str(channel.id),
"name": channel.name,
"commissionRate": channel.commission_rate,
}
@router.get("/admin/channels/{channel_id}/resources")
async def admin_get_channel_resources(channel_id: str, db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
"""获取渠道的资源分配信息"""
channel = await db.get(Channel, uuid.UUID(channel_id)) if channel_id else None
if not channel:
raise HTTPException(status_code=404, detail="channel not found")
# 获取渠道的Agent配额
quotas = (await db.execute(
select(ChannelAgentQuota).where(ChannelAgentQuota.channel_id == channel.id)
)).scalars().all()
agents = []
for q in quotas:
agent = await db.get(Agent, q.agent_id)
if agent:
agents.append({
"agentId": str(q.agent_id),
"agentName": agent.name,
"quantity": q.quantity,
})
# 获取所有可用的模型供应商
models = (await db.execute(select(ProviderModel).where(ProviderModel.status == "active"))).scalars().all()
model_ids = [str(m.id) for m in models]
# 优先使用新字段，回退到旧字段
cpu_quota = channel.custom_agent_cpu_quota if channel.custom_agent_cpu_quota else channel.custom_agent_cpu
memory_quota = channel.custom_agent_memory_quota if channel.custom_agent_memory_quota else channel.custom_agent_memory
return {
"id": str(channel.id),
"channelName": channel.name,
"models": model_ids,
"agents": agents,
"customAgentResources": {
"cpu": float(cpu_quota) if cpu_quota else 0.0,
"memory": float(memory_quota) if memory_quota else 0.0,
},
"channelCredit": float(channel.channel_credit) if channel.channel_credit else 0.0,
}
@router.put("/admin/channels/{channel_id}/resources")
async def admin_update_channel_resources(channel_id: str, payload: Dict[str, Any], db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
import logging
logger = logging.getLogger(__name__)
channel = await db.get(Channel, uuid.UUID(channel_id)) if channel_id else None
if not channel:
raise HTTPException(status_code=404, detail="channel not found")
logger.info(f"更新渠道 {channel.name} 资源，payload: {payload}")
# 处理自定义Agent配额
custom_agent_quota = payload.get("customAgentQuota")
if custom_agent_quota:
if "cpuQuota" in custom_agent_quota:
cpu_value = custom_agent_quota["cpuQuota"]
channel.custom_agent_cpu_quota = cpu_value
channel.custom_agent_cpu = cpu_value # 同时更新旧字段
logger.info(f"设置 CPU 配额: {cpu_value}")
if "memoryQuota" in custom_agent_quota:
memory_value = custom_agent_quota["memoryQuota"]
channel.custom_agent_memory_quota = memory_value
channel.custom_agent_memory = memory_value # 同时更新旧字段
logger.info(f"设置内存配额: {memory_value}")
# 处理渠道信用额度
if "channelCredit" in payload:
channel.channel_credit = payload["channelCredit"]
# 简化为保存配额到 ChannelAgentQuota
if payload.get("agents"):
for agent in payload["agents"]:
agent_id = uuid.UUID(agent.get("agentId"))
quota = ChannelAgentQuota(channel_id=channel.id, agent_id=agent_id, quantity=agent.get("quantity", 0))
db.add(quota)
db.add(channel)
await db.commit()
await db.refresh(channel)
logger.info(f"更新后渠道 {channel.name}: cpu_quota={channel.custom_agent_cpu_quota}, memory_quota={channel.custom_agent_memory_quota}")
return {
"success": True,
"data": {
"channelId": str(channel.id),
"channelName": channel.name,
"customAgentCpu": float(channel.custom_agent_cpu_quota) if channel.custom_agent_cpu_quota else 0.0,
"customAgentMemory": float(channel.custom_agent_memory_quota) if channel.custom_agent_memory_quota else 0.0,
},
"message": "渠道资源分配成功"
}
@router.get("/admin/channels/applications")
async def admin_channel_applications() -> Dict[str, Any]:
return {"items": list(store.resource_applications.values())}
@router.put("/admin/channels/applications/{request_id}/approve")
async def admin_channel_applications_approve(request_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
if request_id not in store.resource_applications:
raise HTTPException(status_code=404, detail="application not found")
store.resource_applications[request_id]["status"] = "approved" if payload.get("approved") else "rejected"
store.resource_applications[request_id]["reason"] = payload.get("reason", "")
return store.resource_applications[request_id]
@router.get("/admin/resources/models")
async def admin_resources_models(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
models = (await db.execute(select(ProviderModel).order_by(ProviderModel.created_at.desc()))).scalars().all()
items = [
{
"id": str(m.id),
"name": m.name,
"apiUrl": m.api_url,
"supportedModels": m.supported_models,
"rpm": m.rpm,
"tpm": m.tpm,
"isActive": m.is_active,
}
for m in models
]
return {"items": items}
# =====================================================
# 注意: 以下 /admin/* 接口已移至 admin.py，避免重复定义
# 已删除的接口包括:
# - /admin/dashboard/stats (从 AgentBillingRecord 和 ModelBillingRecord 统计)
# - /admin/channels (使用 admin.py 中的完整实现)
# - /admin/channels/create (使用 admin.py 中带 LiteLLM 集成的实现)
# - /admin/channels/{channel_id}/commission (使用 admin.py 中的实现)
# - /admin/channels/{channel_id}/resources GET/PUT (使用 admin.py 中的实现)
# - /admin/channels/applications (使用 admin.py 中的数据库实现)
# - /admin/resources/models (使用 admin.py 中的 ModelProvider 实现)
# - /admin/resources/agents (使用 admin.py 中的完整实现)
# - /admin/monitoring/agents (使用 admin.py 中的 K8s+DB 实现)
# - /admin/billing/overview (使用 admin.py 中的正确计费表实现)
# - /admin/roles (使用 admin.py 中的实现)
# - /admin/channels/{channel_id}/admins (使用 admin.py 中的实现)
# - /admin/admins/create (使用 admin.py 中的实现)
# =====================================================
# 保留: /admin/resources/models/add - 独立功能，admin.py 中没有对应实现
@router.post("/admin/resources/models/add")
async def admin_resources_models_add(payload: Dict[str, Any], db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
required = {"name", "apiUrl", "apiKey", "supportedModels"}
@@ -1145,356 +945,7 @@ async def admin_resources_models_add(payload: Dict[str, Any], db: AsyncSession =
}
@router.get("/admin/resources/agents")
async def admin_resources_agents(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
"""获取所有 Agent 资源列表（平台端 + 自定义）
平台端 Agent 从 agent-manager (K8s) 获取
自定义 Agent 从本地数据库获取
"""
import logging
import traceback
logger = logging.getLogger(__name__)
items = []
k8s_agents = []
error_message = None
# 1. 从 agent-manager 获取 K8s 中运行的平台端 Agent
try:
print("=== 开始获取 agent-manager 数据 ===")
from app.agent_manager_client import get_agent_manager_client
client = get_agent_manager_client()
print(f"Agent Manager URL: {client.base_url}")
k8s_agents_result = await client.list_agents()
k8s_agents = k8s_agents_result.agents # 从 AgentListResult 中获取 agents 列表
print(f"从 agent-manager 获取到 {len(k8s_agents)} 个 Agent: {k8s_agents}")
logger.info(f"从 agent-manager 获取到 {len(k8s_agents)} 个 Agent")
for agent in k8s_agents:
# 解析资源配置
cpu_limit = "0"
memory_limit = "0"
try:
metrics = await client.get_agent_metrics(agent.get("name", ""))
cpu_limit = metrics.limits.get("cpu", "0") if hasattr(metrics, 'limits') else "0"
memory_limit = metrics.limits.get("memory", "0") if hasattr(metrics, 'limits') else "0"
except Exception as e:
logger.warning(f"获取 Agent {agent.get('name')} 资源指标失败: {e}")
# 解析 CPU (如 "500m" -> 0.5 核)
cpu_value = 0.0
if isinstance(cpu_limit, str):
if cpu_limit.endswith("m"):
cpu_value = float(cpu_limit[:-1]) / 1000
elif cpu_limit:
try:
cpu_value = float(cpu_limit)
except ValueError:
pass
# 解析内存 (如 "512Mi" -> 0.5 GB)
memory_value = 0.0
if isinstance(memory_limit, str):
if memory_limit.endswith("Mi"):
memory_value = float(memory_limit[:-2]) / 1024
elif memory_limit.endswith("Gi"):
memory_value = float(memory_limit[:-2])
elif memory_limit.endswith("Ki"):
memory_value = float(memory_limit[:-2]) / (1024 * 1024)
elif memory_limit:
try:
memory_value = float(memory_limit)
except ValueError:
pass
items.append({
"id": agent.get("name", ""),
"name": agent.get("name", ""),
"type": "platform",
"description": agent.get("template", "K8s Agent"),
"status": agent.get("status", "unknown"),
"healthStatus": "healthy" if agent.get("status") == "Running" else "unknown",
"cpu": round(cpu_value, 2),
"memory": round(memory_value, 2),
"maxInstances": 1,
"cpuRequest": agent.get("cpu_request", "100m"),
"cpuLimit": cpu_limit,
"memoryRequest": agent.get("memory_request", "128Mi"),
"memoryLimit": memory_limit,
"podName": agent.get("pod_name", ""),
"podIp": agent.get("pod_ip", ""),
"namespace": agent.get("namespace", "ai-agents"),
"template": agent.get("template", ""),
"createdAt": agent.get("created_at"),
"source": "k8s",
})
except Exception as e:
logger.warning(f"连接 agent-manager 失败: {e}")
# 2. 从本地数据库获取自定义 Agent
db_agents = (await db.execute(select(Agent).where(Agent.status != "inactive"))).scalars().all()
for a in db_agents:
# 检查是否已经从 K8s 获取过（避免重复）
if any(item.get("name") == a.name for item in items):
continue
items.append({
"id": str(a.id),
"name": a.name,
"type": a.type or "custom",
"description": a.description,
"status": a.status,
"healthStatus": a.health_status or "unknown",
"cpu": float(a.cpu or 0),
"memory": float(a.memory or 0),
"maxInstances": a.max_instances or 100,
"cpuRequest": a.cpu_request,
"cpuLimit": a.cpu_limit,
"memoryRequest": a.memory_request,
"memoryLimit": a.memory_limit,
"totalExecutions": a.total_executions or 0,
"successRate": float(a.success_rate or 0),
"createdAt": a.created_at.isoformat() if a.created_at else None,
"source": "database",
})
# 统计信息
platform_count = sum(1 for item in items if item.get("type") == "platform")
custom_count = sum(1 for item in items if item.get("type") == "custom")
total_cpu = sum(float(item.get("cpu", 0)) for item in items)
total_memory = sum(float(item.get("memory", 0)) for item in items)
return {
"success": True,
"data": {
"agents": items,
"summary": {
"total": len(items),
"platformAgents": platform_count,
"customAgents": custom_count,
"totalCpu": round(total_cpu, 2),
"totalMemory": round(total_memory, 2),
"k8sAgentsCount": len(k8s_agents),
"dbAgentsCount": len(db_agents),
}
},
"message": None
}
@router.put("/admin/resources/agents/{agent_id}")
async def admin_resources_agents_update(agent_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
return {"id": agent_id, **payload, "updatedAt": _now()}
@router.get("/admin/monitoring/agents")
async def admin_monitoring_agents(
agent_type: str = None,
health_status: str = None,
db: AsyncSession = Depends(get_db)
) -> Dict[str, Any]:
"""
获取 Agent 健康监控信息
参数:
- agent_type: 可选，筛选 Agent 类型 (platform/custom)
- health_status: 可选，筛选健康状态 (healthy/warning/critical/unknown)
"""
# 构建查询
query = select(Agent).where(Agent.status != "inactive")
if agent_type:
query = query.where(Agent.type == agent_type)
if health_status:
query = query.where(Agent.health_status == health_status)
agents = (await db.execute(query)).scalars().all()
# 构建响应
agent_list = []
summary = {
"total": 0,
"byType": {"platform": 0, "custom": 0},
"byHealthStatus": {"healthy": 0, "warning": 0, "critical": 0, "unknown": 0},
"byStatus": {"active": 0, "error": 0, "other": 0},
}
for agent in agents:
agent_data = {
"id": str(agent.id),
"name": agent.name,
"type": agent.type or "platform",
"status": agent.status,
"healthStatus": agent.health_status or "unknown",
"lastHealthCheck": agent.last_health_check.isoformat() if agent.last_health_check else None,
"healthMessage": agent.health_message,
"cpu": float(agent.cpu or 0),
"memory": float(agent.memory or 0),
"totalExecutions": agent.total_executions or 0,
"successRate": float(agent.success_rate or 0),
}
agent_list.append(agent_data)
# 更新统计
summary["total"] += 1
agent_type_val = agent.type or "platform"
if agent_type_val in summary["byType"]:
summary["byType"][agent_type_val] += 1
health = agent.health_status or "unknown"
if health in summary["byHealthStatus"]:
summary["byHealthStatus"][health] += 1
if agent.status == "active":
summary["byStatus"]["active"] += 1
elif agent.status == "error":
summary["byStatus"]["error"] += 1
else:
summary["byStatus"]["other"] += 1
return {
"success": True,
"data": {
"agents": agent_list,
"summary": summary,
},
"message": None,
}
@router.get("/admin/billing/overview")
async def admin_billing_overview(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
billing_rows = (
await db.execute(
select(Billing.cost, Billing.eu_consumed, Billing.created_at, Billing.channel_id, Billing.tenant_id)
.order_by(Billing.created_at.desc())
.limit(200)
)
).all()
total_eu = sum(float(row.eu_consumed or 0) for row in billing_rows)
total_cost = sum(float(row.cost or 0) for row in billing_rows)
channels = [
{
"channelName": str(row.channel_id) if row.channel_id else "",
"calls": 1,
"totalEU": float(row.eu_consumed or 0),
"totalCost": float(row.cost or 0),
}
for row in billing_rows
]
tenants = [
{
"tenantName": str(row.tenant_id) if row.tenant_id else "",
"channelName": str(row.channel_id) if row.channel_id else "",
"calls": 1,
"totalEU": float(row.eu_consumed or 0),
"totalCost": float(row.cost or 0),
}
for row in billing_rows
]
call_records = [
{
"timestamp": row.created_at.isoformat() if row.created_at else _now(),
"channelName": str(row.channel_id) if row.channel_id else "",
"tenantName": str(row.tenant_id) if row.tenant_id else "",
"agentName": "",
"duration": 0,
"eu": float(row.eu_consumed or 0),
"cost": float(row.cost or 0),
}
for row in billing_rows
]
return {"channels": channels, "tenants": tenants, "callRecords": call_records, "totalEU": total_eu, "totalCost": total_cost}
@router.get("/admin/roles")
async def admin_roles() -> Dict[str, Any]:
"""获取可用角色列表"""
return {
"items": [
{
"id": "super_admin",
"name": "超级管理员",
"description": "拥有系统所有权限",
"permissions": ["*"]
},
{
"id": "billing_admin",
"name": "计费管理员",
"description": "完整写入权限，可创建渠道、管理租户、计费操作",
"permissions": ["read:*", "write:channels", "write:tenants", "write:billing"]
},
{
"id": "operations_admin",
"name": "运维管理员",
"description": "只读权限，仅查看和监控",
"permissions": ["read:*"]
},
{
"id": "channel_admin",
"name": "渠道管理员",
"description": "渠道内部管理权限",
"permissions": ["read:channel", "write:tenants", "read:billing"]
},
{
"id": "user",
"name": "普通用户",
"description": "标准用户权限",
"permissions": ["read:self", "use:agents"]
}
]
}
@router.get("/admin/channels/{channel_id}/admins")
async def admin_get_channel_admins(channel_id: str, db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
"""获取渠道管理员列表"""
from models import User
channel = await db.get(Channel, uuid.UUID(channel_id)) if channel_id else None
if not channel:
raise HTTPException(status_code=404, detail="channel not found")
# 查询渠道下的管理员
admins = (await db.execute(
select(User).where(
User.channel_id == channel.id,
User.role == "channel_admin",
User.status == "active"
)
)).scalars().all()
items = [
{
"id": str(admin.id),
"name": admin.name or admin.full_name,
"email": admin.email,
"role": admin.role,
"status": admin.status,
"createdAt": admin.created_at.isoformat() if admin.created_at else None,
}
for admin in admins
]
return {
"channelId": str(channel.id),
"channelName": channel.name,
"admins": items
}
@router.post("/admin/admins/create")
async def admin_admins_create(payload: Dict[str, Any]) -> Dict[str, Any]:
admin_id = str(uuid.uuid4())
admin = {"id": admin_id, **payload, "createdAt": _now()}
store.channel_admins[admin_id] = admin
return admin
# 保留: /admin/providers/stats - 独立功能，admin.py 中没有对应实现
@router.get("/admin/providers/stats")
async def admin_providers_stats(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
models = (await db.execute(select(ProviderModel))).scalars().all()
@@ -1512,11 +963,6 @@ async def admin_providers_stats(db: AsyncSession = Depends(get_db)) -> Dict[str,
}
@router.get("/admin/channels/backend/stats")
async def admin_channels_backend_stats() -> Dict[str, Any]:
return {"channels": len(store.channels), "applications": len(store.resource_applications)}
# ----- Provider Management -----