更新太极测试渠道

This commit is contained in:
Ubuntu
2025-12-28 13:28:16 +00:00
parent 78a2394426
commit ee368e8812
25 changed files with 3804 additions and 264 deletions
+227
View File
@@ -0,0 +1,227 @@
name: MCP Server CI/CD
on:
push:
branches:
- main
- develop
paths:
- 'services/mcp-server/**'
- '.github/workflows/mcp-server-deploy.yml'
pull_request:
branches:
- main
- develop
paths:
- 'services/mcp-server/**'
workflow_dispatch:
inputs:
environment:
description: 'Deployment environment'
required: true
default: 'staging'
type: choice
options:
- staging
- production
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}/mcp-server
AZURE_REGISTRY: taiji.azurecr.io
AZURE_IMAGE_NAME: taiji-mcp-server
jobs:
build-and-push:
name: Build and Push Docker Image
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
id-token: write
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to GitHub Container Registry
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Log in to Azure Container Registry
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
uses: docker/login-action@v3
with:
registry: ${{ env.AZURE_REGISTRY }}
username: ${{ secrets.AZURE_CLIENT_ID }}
password: ${{ secrets.AZURE_CLIENT_SECRET }}
- name: Extract metadata
id: meta
uses: docker/metadata-action@v5
with:
images: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
${{ env.AZURE_REGISTRY }}/${{ env.AZURE_IMAGE_NAME }}
tags: |
type=ref,event=branch
type=ref,event=pr
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}
type=sha,prefix={{branch}}-
type=raw,value=latest,enable={{is_default_branch}}
- name: Build and push Docker image
uses: docker/build-push-action@v5
with:
context: ./services/mcp-server
file: ./services/mcp-server/Dockerfile
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
cache-from: type=gha
cache-to: type=gha,mode=max
platforms: linux/amd64,linux/arm64
- name: Generate build summary
run: |
echo "### 🚀 Build Summary" >> $GITHUB_STEP_SUMMARY
echo "" >> $GITHUB_STEP_SUMMARY
echo "**Image:** \`${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}\`" >> $GITHUB_STEP_SUMMARY
echo "**Tags:**" >> $GITHUB_STEP_SUMMARY
echo "\`\`\`" >> $GITHUB_STEP_SUMMARY
echo "${{ steps.meta.outputs.tags }}" >> $GITHUB_STEP_SUMMARY
echo "\`\`\`" >> $GITHUB_STEP_SUMMARY
deploy-staging:
name: Deploy to Staging
needs: build-and-push
if: github.event_name == 'push' && github.ref == 'refs/heads/develop'
runs-on: ubuntu-latest
environment:
name: staging
url: https://staging-mcp.taiji-ai.com
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up kubectl
uses: azure/setup-kubectl@v3
with:
version: 'v1.28.0'
- name: Configure kubectl
run: |
mkdir -p $HOME/.kube
echo "${{ secrets.KUBE_CONFIG_STAGING }}" | base64 -d > $HOME/.kube/config
- name: Update deployment image
run: |
kubectl set image deployment/mcp-server \
mcp-server=${{ env.AZURE_REGISTRY }}/${{ env.AZURE_IMAGE_NAME }}:develop \
-n taiji-ai
- name: Wait for rollout
run: |
kubectl rollout status deployment/mcp-server -n taiji-ai --timeout=5m
- name: Verify deployment
run: |
kubectl get pods -n taiji-ai -l app=mcp-server
kubectl get svc -n taiji-ai -l app=mcp-server
deploy-production:
name: Deploy to Production
needs: build-and-push
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
runs-on: ubuntu-latest
environment:
name: production
url: https://mcp.taiji-ai.com
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up kubectl
uses: azure/setup-kubectl@v3
with:
version: 'v1.28.0'
- name: Configure kubectl
run: |
mkdir -p $HOME/.kube
echo "${{ secrets.KUBE_CONFIG_PRODUCTION }}" | base64 -d > $HOME/.kube/config
- name: Update deployment image
run: |
kubectl set image deployment/mcp-server \
mcp-server=${{ env.AZURE_REGISTRY }}/${{ env.AZURE_IMAGE_NAME }}:latest \
-n taiji-ai
- name: Wait for rollout
run: |
kubectl rollout status deployment/mcp-server -n taiji-ai --timeout=5m
- name: Verify deployment
run: |
kubectl get pods -n taiji-ai -l app=mcp-server
kubectl get svc -n taiji-ai -l app=mcp-server
- name: Send deployment notification
if: always()
uses: 8398a7/action-slack@v3
with:
status: ${{ job.status }}
text: 'MCP Server deployment to production: ${{ job.status }}'
webhook_url: ${{ secrets.SLACK_WEBHOOK }}
env:
SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK }}
health-check:
name: Post-Deployment Health Check
needs: [deploy-staging, deploy-production]
if: always() && (needs.deploy-staging.result == 'success' || needs.deploy-production.result == 'success')
runs-on: ubuntu-latest
steps:
- name: Determine environment
id: env
run: |
if [[ "${{ github.ref }}" == "refs/heads/main" ]]; then
echo "url=https://mcp.taiji-ai.com" >> $GITHUB_OUTPUT
echo "env=production" >> $GITHUB_OUTPUT
else
echo "url=https://staging-mcp.taiji-ai.com" >> $GITHUB_OUTPUT
echo "env=staging" >> $GITHUB_OUTPUT
fi
- name: Health check
run: |
echo "Checking health endpoint: ${{ steps.env.outputs.url }}/health"
for i in {1..10}; do
if curl -f -s "${{ steps.env.outputs.url }}/health" > /dev/null; then
echo "✅ Health check passed!"
exit 0
fi
echo "Attempt $i failed, waiting 10s..."
sleep 10
done
echo "❌ Health check failed after 10 attempts"
exit 1
- name: API smoke test
run: |
echo "Running API smoke tests..."
# Test agents endpoint
curl -f -s "${{ steps.env.outputs.url }}/api/v1/agents" || exit 1
# Test health endpoint
curl -f -s "${{ steps.env.outputs.url }}/health" || exit 1
echo "✅ Smoke tests passed!"
+122
View File
@@ -0,0 +1,122 @@
#!/usr/bin/env python3
"""
创建超级管理员账号
直接通过数据库创建,不需要认证
"""
import sys
import os
import asyncio
# 添加services/mcp-server到路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'services', 'mcp-server'))
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select
from models import User
from config import settings
import bcrypt
import uuid
def get_password_hash(password: str) -> str:
"""加密密码(使用bcrypt)"""
password_bytes = password.encode('utf-8')
salt = bcrypt.gensalt()
hashed = bcrypt.hashpw(password_bytes, salt)
return hashed.decode('utf-8')
async def create_super_admin():
"""创建超级管理员"""
try:
# 准备数据库URL
database_url = settings.database_url
if database_url.startswith("postgresql://") and "+asyncpg" not in database_url:
database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
engine = create_async_engine(database_url, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async with AsyncSessionLocal() as session:
# 检查用户是否已存在
result = await session.execute(
select(User).where(User.email == "superadmin@taiji-ai.com")
)
existing_user = result.scalar_one_or_none()
if existing_user:
print(f" ⚠ 用户已存在: superadmin@taiji-ai.com")
# 更新密码和角色
password_hash = get_password_hash("Admin@123456")
existing_user.password_hash = password_hash
existing_user.hashed_password = password_hash
existing_user.role = "super_admin"
existing_user.is_admin = True
existing_user.is_active = True
existing_user.name = "超级管理员"
existing_user.username = "superadmin"
existing_user.full_name = "超级管理员"
await session.commit()
print(f" ✓ 更新成功: superadmin@taiji-ai.com")
return True
# 创建新用户
password_hash = get_password_hash("Admin@123456")
user = User(
name="超级管理员",
email="superadmin@taiji-ai.com",
password_hash=password_hash,
hashed_password=password_hash,
username="superadmin",
full_name="超级管理员",
role="super_admin",
is_active=True,
is_admin=True,
status="active",
balance=0,
credit_limit=0,
)
session.add(user)
await session.commit()
await session.refresh(user)
print(f" ✓ 创建成功: superadmin@taiji-ai.com (角色: super_admin)")
return True
except Exception as e:
print(f" ✗ 创建失败: {e}")
import traceback
traceback.print_exc()
return False
finally:
if engine:
await engine.dispose()
async def main():
"""主函数"""
print("="*80)
print("创建超级管理员账号")
print("="*80)
print()
print("创建超级管理员...")
success = await create_super_admin()
if success:
print("\n" + "="*80)
print("✓ 超级管理员创建成功!")
print("="*80)
print("\n账号信息:")
print(" 邮箱: superadmin@taiji-ai.com")
print(" 密码: Admin@123456")
print(" 角色: super_admin")
sys.exit(0)
else:
print("\n" + "="*80)
print("✗ 超级管理员创建失败!")
print("="*80)
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
+136
View File
@@ -0,0 +1,136 @@
# AKS 部署检查清单
## 部署前准备
- [ ] Azure CLI 已安装并登录 (`az login`)
- [ ] kubectl 已安装
- [ ] Docker 已安装
- [ ] AKS 集群已创建
- [ ] Azure Container Registry (ACR) 已创建
- [ ] 已获取 ACR 名称、AKS 资源组名称、AKS 集群名称
## 配置更新
- [ ] 更新 `secrets.yaml` 中的数据库连接字符串(已改为 `taiji.postgres.database.azure.com`)
- [ ] 更新 `secrets.yaml` 中的 Redis 连接字符串
- [ ] 更新 `secrets.yaml` 中的所有 API 密钥(OpenRouter, RapidAPI 等)
- [ ] 确认数据库用户名和密码正确
## 部署步骤
### 方式 1: 使用自动化脚本(推荐)
```bash
cd k8s
./deploy.sh <ACR_NAME> <AKS_RESOURCE_GROUP> <AKS_CLUSTER_NAME>
```
### 方式 2: 手动部署
1. **获取 AKS 凭据**
```bash
az aks get-credentials --resource-group <RESOURCE_GROUP> --name <CLUSTER_NAME>
```
2. **创建命名空间**
```bash
kubectl apply -f namespace.yaml
```
3. **创建 Secret**
```bash
# 方式 A: 使用 kubectl 命令
kubectl create secret generic taiji-secrets \
--from-literal=DATABASE_URL="postgresql+asyncpg://taiji:PASSWORD@taiji.postgres.database.azure.com:5432/postgres?sslmode=require" \
--from-literal=ASYNC_DATABASE_URL="postgresql+asyncpg://taiji:PASSWORD@taiji.postgres.database.azure.com:5432/postgres?sslmode=require" \
--from-literal=REDIS_URL="rediss://:REDIS_PASSWORD@taiji.southeastasia.redis.azure.net:10000/0?ssl_cert_reqs=none" \
--from-literal=SECRET_KEY="zsbgnw" \
--from-literal=ENCRYPTION_KEY="zsbgnw" \
--from-literal=LITELLM_MASTER_KEY="sk-taiji-master-key" \
--from-literal=OPENROUTER_API_KEY="YOUR_KEY" \
--from-literal=OPENROUTER_BASE_URL="https://openrouter.ai/api/v1" \
--from-literal=RAPIDAPI_KEY="YOUR_KEY" \
--from-literal=RAPIDAPI_HOST="YOUR_HOST" \
-n taiji-ai
```
4. **构建并推送 Docker 镜像**
```bash
ACR_NAME="your-acr-name"
az acr login --name $ACR_NAME
docker build -t $ACR_NAME.azurecr.io/taiji-mcp-server:latest ../services/mcp-server
docker push $ACR_NAME.azurecr.io/taiji-mcp-server:latest
docker build -t $ACR_NAME.azurecr.io/taiji-data-ingestion:latest ../services/data-ingestion
docker push $ACR_NAME.azurecr.io/taiji-data-ingestion:latest
docker build -t $ACR_NAME.azurecr.io/taiji-litellm-gateway:latest ../services/model-gateway
docker push $ACR_NAME.azurecr.io/taiji-litellm-gateway:latest
```
5. **更新 Deployment 文件中的 ACR 名称**
```bash
sed -i "s/YOUR_ACR_NAME/$ACR_NAME/g" litellm-gateway.yaml
sed -i "s/YOUR_ACR_NAME/$ACR_NAME/g" data-ingestion.yaml
sed -i "s/YOUR_ACR_NAME/$ACR_NAME/g" mcp-server.yaml
```
6. **创建 ACR 拉取 Secret**
```bash
kubectl create secret docker-registry acr-secret \
--docker-server=$ACR_NAME.azurecr.io \
--docker-username=$(az acr credential show --name $ACR_NAME --query username -o tsv) \
--docker-password=$(az acr credential show --name $ACR_NAME --query passwords[0].value -o tsv) \
--namespace=taiji-ai
```
7. **应用 ConfigMap**
```bash
kubectl apply -f configmap.yaml
```
8. **部署服务(按依赖顺序)**
```bash
kubectl apply -f nats.yaml
kubectl wait --for=condition=available --timeout=300s deployment/nats -n taiji-ai
kubectl apply -f litellm-gateway.yaml
kubectl apply -f data-ingestion.yaml
kubectl apply -f mcp-server.yaml
kubectl apply -f api-gateway.yaml
kubectl apply -f monitoring.yaml # 可选
```
## 部署后验证
- [ ] 检查所有 Pod 状态:`kubectl get pods -n taiji-ai`
- [ ] 检查服务状态:`kubectl get services -n taiji-ai`
- [ ] 获取 API Gateway 外部 IP:`kubectl get service api-gateway-service -n taiji-ai`
- [ ] 测试健康检查:`curl http://<EXTERNAL-IP>/health`
- [ ] 查看日志确认无错误:`kubectl logs -f deployment/mcp-server -n taiji-ai`
## 重要提示
1. **数据库已更新为 taiji**:确保所有配置指向 `taiji.postgres.database.azure.com`
2. **不影响本地 Docker**:此部署完全独立,不会影响本地运行的 docker-compose 环境
3. **存储类**:确保 AKS 集群支持 `managed-premium` 存储类
4. **资源限制**:根据实际需求调整各服务的资源请求和限制
## 故障排查
如果遇到问题:
1. 查看 Pod 详情:`kubectl describe pod <pod-name> -n taiji-ai`
2. 查看事件:`kubectl get events -n taiji-ai --sort-by='.lastTimestamp'`
3. 查看日志:`kubectl logs <pod-name> -n taiji-ai`
4. 检查 Secret:`kubectl get secret taiji-secrets -n taiji-ai -o yaml`
## 下一步
部署成功后:
- 配置域名和 SSL 证书(如需要)
- 设置监控告警
- 配置自动扩缩容(HPA)
- 设置备份策略
+148
View File
@@ -0,0 +1,148 @@
# AKS 部署状态报告
## 部署时间
2025-12-28
## 部署信息
- **AKS 集群**: taiji-ai-pda
- **资源组**: taiji-ai-pda
- **ACR**: taiji.azurecr.io
- **命名空间**: taiji-ai
- **数据库**: taiji.postgres.database.azure.com
## 服务状态
### ✅ 已成功部署
1. **API Gateway (Nginx)**
- 状态: Running (2/2 replicas)
- 外部 IP: 135.171.159.241
- 健康检查: ✅ 正常 (http://135.171.159.241/health)
- 访问地址: http://135.171.159.241
2. **ConfigMap 和 Secrets**
- ✅ taiji-config (ConfigMap)
- ✅ taiji-secrets (Secret)
- ✅ acr-secret (ACR 拉取密钥)
### ⚠️ 需要修复
1. **MCP Server**
- 状态: CrashLoopBackOff
- 问题: `exec format error` - Python 解释器架构不匹配
- 影响: MCP Server 无法启动
- 建议: 需要检查 Dockerfile 和构建环境
2. **Data Ingestion**
- 状态: CrashLoopBackOff
- 问题: 与 MCP Server 相同
- 影响: 数据接入服务无法启动
3. **LiteLLM Gateway**
- 状态: CrashLoopBackOff
- 问题: 与 MCP Server 相同
- 影响: LLM 网关服务无法启动
4. **NATS**
- 状态: CrashLoopBackOff / ContainerCreating
- 问题: Command 配置问题(已修复,但可能需要重新部署)
- 影响: 消息队列服务无法启动
## 已完成的步骤
1. ✅ 创建命名空间
2. ✅ 创建 ConfigMap
3. ✅ 创建 Secrets(数据库已更新为 taiji)
4. ✅ 创建 ACR 拉取密钥
5. ✅ 构建并推送 Docker 镜像
6. ✅ 部署所有服务
7. ✅ API Gateway 正常运行
## 待解决问题
### 1. Python 服务架构问题
**问题**: `exec format error` 表示 Python 解释器无法执行,可能是架构不匹配。
**可能原因**:
- 构建环境与运行环境架构不一致
- Docker 镜像构建时未正确指定平台
- Python 解释器路径问题
**解决方案**:
1. 检查构建环境架构: `uname -m` 和 `docker version`
2. 使用 `docker buildx` 构建多架构镜像
3. 在 Dockerfile 中明确指定 Python 路径
4. 考虑使用 `python3` 而不是 `python`
### 2. NATS 服务问题
**问题**: NATS 容器启动失败
**已修复**: 更新了 command 配置,但可能需要删除旧 Pod 重新创建
**解决方案**:
```bash
kubectl delete deployment nats -n taiji-ai
kubectl apply -f k8s/nats.yaml
```
## 下一步操作
1. **修复 Python 服务架构问题**
```bash
# 检查当前构建环境
uname -m
docker version
# 使用 buildx 构建
docker buildx build --platform linux/amd64 -t taiji.azurecr.io/taiji-mcp-server:latest ./services/mcp-server --push
```
2. **重新部署服务**
```bash
kubectl rollout restart deployment/mcp-server -n taiji-ai
kubectl rollout restart deployment/data-ingestion -n taiji-ai
kubectl rollout restart deployment/litellm-gateway -n taiji-ai
```
3. **验证部署**
```bash
kubectl get pods -n taiji-ai
kubectl logs -f deployment/mcp-server -n taiji-ai
```
## 访问信息
- **API Gateway**: http://135.171.159.241
- **健康检查**: http://135.171.159.241/health
- **API 文档**: http://135.171.159.241/docs (需要 MCP Server 运行)
## 监控命令
```bash
# 查看所有 Pod 状态
kubectl get pods -n taiji-ai
# 查看服务状态
kubectl get services -n taiji-ai
# 查看部署状态
kubectl get deployments -n taiji-ai
# 查看日志
kubectl logs -f deployment/mcp-server -n taiji-ai
kubectl logs -f deployment/data-ingestion -n taiji-ai
kubectl logs -f deployment/litellm-gateway -n taiji-ai
# 查看事件
kubectl get events -n taiji-ai --sort-by='.lastTimestamp'
```
## 注意事项
1. **数据库配置**: 已更新为 `taiji.postgres.database.azure.com`
2. **不影响本地 Docker**: 此部署完全独立,不影响本地 docker-compose 环境
3. **资源限制**: 根据实际需求调整各服务的 CPU 和内存限制
4. **存储**: PVC 使用 `managed-premium` 存储类
+316
View File
@@ -0,0 +1,316 @@
# AKS 部署指南
本文档说明如何将 taiji-AI-PAD 部署到 Azure Kubernetes Service (AKS)。
## 前置要求
1. **Azure CLI** 已安装并登录 (`az login`)
2. **kubectl** 已安装
3. **Docker** 已安装
4. **AKS 集群** 已创建
5. **Azure Container Registry (ACR)** 已创建(用于存储 Docker 镜像)
## 重要变更
### 数据库配置
- **PostgreSQL 数据库已更新为 `taiji.postgres.database.azure.com`**
- 请确保在 `secrets.yaml` 中更新数据库连接字符串
## 部署步骤
### 1. 准备环境变量
设置以下环境变量(或直接在脚本中替换):
```bash
export ACR_NAME="your-acr-name"
export AKS_RESOURCE_GROUP="your-resource-group"
export AKS_CLUSTER_NAME="your-aks-cluster-name"
```
### 2. 更新 Secret 配置
**重要:** 在部署前,必须更新 `secrets.yaml` 文件中的敏感信息:
```yaml
# 数据库连接字符串 - 已更新为 taiji 数据库
DATABASE_URL: "postgresql+asyncpg://taiji:YOUR_PASSWORD@taiji.postgres.database.azure.com:5432/postgres?sslmode=require"
# Redis 连接字符串
REDIS_URL: "rediss://:YOUR_REDIS_PASSWORD@taiji.southeastasia.redis.azure.net:10000/0?ssl_cert_reqs=none"
# 其他 API 密钥
OPENROUTER_API_KEY: "YOUR_OPENROUTER_API_KEY"
RAPIDAPI_KEY: "YOUR_RAPIDAPI_KEY"
# ... 等等
```
### 3. 创建 Secret
有两种方式创建 Secret:
#### 方式 1: 使用 kubectl 命令(推荐用于测试)
```bash
kubectl create secret generic taiji-secrets \
--from-literal=DATABASE_URL="postgresql+asyncpg://taiji:PASSWORD@taiji.postgres.database.azure.com:5432/postgres?sslmode=require" \
--from-literal=ASYNC_DATABASE_URL="postgresql+asyncpg://taiji:PASSWORD@taiji.postgres.database.azure.com:5432/postgres?sslmode=require" \
--from-literal=REDIS_URL="rediss://:REDIS_PASSWORD@taiji.southeastasia.redis.azure.net:10000/0?ssl_cert_reqs=none" \
--from-literal=SECRET_KEY="zsbgnw" \
--from-literal=ENCRYPTION_KEY="zsbgnw" \
--from-literal=LITELLM_MASTER_KEY="sk-taiji-master-key" \
--from-literal=OPENROUTER_API_KEY="YOUR_KEY" \
--from-literal=OPENROUTER_BASE_URL="https://openrouter.ai/api/v1" \
--from-literal=RAPIDAPI_KEY="YOUR_KEY" \
--from-literal=RAPIDAPI_HOST="YOUR_HOST" \
-n taiji-ai
```
#### 方式 2: 使用 Azure Key Vault(推荐用于生产环境)
```bash
# 安装 Key Vault CSI 驱动
az aks enable-addons --addons azure-keyvault-secrets-provider --name $AKS_CLUSTER_NAME --resource-group $AKS_RESOURCE_GROUP
# 创建 SecretProviderClass(参考 Azure 文档)
```
### 4. 运行部署脚本
```bash
cd k8s
chmod +x deploy.sh
./deploy.sh $ACR_NAME $AKS_RESOURCE_GROUP $AKS_CLUSTER_NAME
```
或者手动执行步骤:
```bash
# 1. 获取 AKS 凭据
az aks get-credentials --resource-group $AKS_RESOURCE_GROUP --name $AKS_CLUSTER_NAME
# 2. 创建命名空间
kubectl apply -f namespace.yaml
# 3. 构建并推送镜像
az acr login --name $ACR_NAME
docker build -t $ACR_NAME.azurecr.io/taiji-mcp-server:latest ../services/mcp-server
docker push $ACR_NAME.azurecr.io/taiji-mcp-server:latest
# ... 其他服务类似
# 4. 更新 Deployment 文件中的 ACR 名称
sed -i "s/YOUR_ACR_NAME/$ACR_NAME/g" *.yaml
# 5. 创建 ACR 拉取 Secret
kubectl create secret docker-registry acr-secret \
--docker-server=$ACR_NAME.azurecr.io \
--docker-username=$(az acr credential show --name $ACR_NAME --query username -o tsv) \
--docker-password=$(az acr credential show --name $ACR_NAME --query passwords[0].value -o tsv) \
--namespace=taiji-ai
# 6. 应用配置
kubectl apply -f configmap.yaml
kubectl apply -f secrets.yaml # 或使用上面创建的 secret
# 7. 部署服务(按依赖顺序)
kubectl apply -f nats.yaml
kubectl apply -f litellm-gateway.yaml
kubectl apply -f data-ingestion.yaml
kubectl apply -f mcp-server.yaml
kubectl apply -f api-gateway.yaml
kubectl apply -f monitoring.yaml # 可选
```
## 验证部署
### 检查 Pod 状态
```bash
kubectl get pods -n taiji-ai
```
所有 Pod 应该处于 `Running` 状态。
### 检查服务状态
```bash
kubectl get services -n taiji-ai
```
### 获取 API Gateway 外部 IP
```bash
kubectl get service api-gateway-service -n taiji-ai
```
访问 `http://<EXTERNAL-IP>/health` 验证服务是否正常。
### 查看日志
```bash
# 查看 MCP Server 日志
kubectl logs -f deployment/mcp-server -n taiji-ai
# 查看 Data Ingestion 日志
kubectl logs -f deployment/data-ingestion -n taiji-ai
# 查看所有 Pod 日志
kubectl logs -f -l app=mcp-server -n taiji-ai
```
## 服务架构
```
Internet
|
v
[LoadBalancer] API Gateway (Nginx)
|
+---> MCP Server (3 replicas)
+---> Data Ingestion (2 replicas)
+---> LiteLLM Gateway (2 replicas)
|
v
[NATS] (消息队列)
|
v
[Azure PostgreSQL] (taiji.postgres.database.azure.com)
[Azure Redis] (taiji.southeastasia.redis.azure.net)
```
## 扩缩容
### 手动扩缩容
```bash
# 扩展 MCP Server 到 5 个副本
kubectl scale deployment mcp-server --replicas=5 -n taiji-ai
# 扩展 Data Ingestion 到 3 个副本
kubectl scale deployment data-ingestion --replicas=3 -n taiji-ai
```
### 自动扩缩容(HPA)
```bash
# 为 MCP Server 创建 HPA
kubectl autoscale deployment mcp-server \
--cpu-percent=70 \
--min=3 \
--max=10 \
-n taiji-ai
```
## 更新部署
### 更新镜像
```bash
# 1. 构建新镜像
docker build -t $ACR_NAME.azurecr.io/taiji-mcp-server:v1.1.0 ../services/mcp-server
docker push $ACR_NAME.azurecr.io/taiji-mcp-server:v1.1.0
# 2. 更新 Deployment
kubectl set image deployment/mcp-server \
mcp-server=$ACR_NAME.azurecr.io/taiji-mcp-server:v1.1.0 \
-n taiji-ai
# 3. 查看滚动更新状态
kubectl rollout status deployment/mcp-server -n taiji-ai
```
### 回滚
```bash
# 查看历史版本
kubectl rollout history deployment/mcp-server -n taiji-ai
# 回滚到上一个版本
kubectl rollout undo deployment/mcp-server -n taiji-ai
# 回滚到指定版本
kubectl rollout undo deployment/mcp-server --to-revision=2 -n taiji-ai
```
## 故障排查
### Pod 无法启动
```bash
# 查看 Pod 详情
kubectl describe pod <pod-name> -n taiji-ai
# 查看事件
kubectl get events -n taiji-ai --sort-by='.lastTimestamp'
```
### 镜像拉取失败
```bash
# 检查 ACR Secret
kubectl get secret acr-secret -n taiji-ai
# 验证 ACR 登录
az acr login --name $ACR_NAME
```
### 数据库连接问题
```bash
# 检查 Secret 中的数据库 URL
kubectl get secret taiji-secrets -n taiji-ai -o yaml
# 测试数据库连接(在 Pod 内)
kubectl exec -it <pod-name> -n taiji-ai -- bash
# 然后测试数据库连接
```
## 监控
### Prometheus
访问 Prometheus(需要端口转发):
```bash
kubectl port-forward service/prometheus-service 9090:9090 -n taiji-ai
```
然后访问 `http://localhost:9090`
### Grafana
访问 Grafana(需要端口转发):
```bash
kubectl port-forward service/grafana-service 3000:3000 -n taiji-ai
```
然后访问 `http://localhost:3000`(用户名/密码: admin/admin)
## 清理
```bash
# 删除所有资源
kubectl delete namespace taiji-ai
# 或删除特定服务
kubectl delete -f mcp-server.yaml
kubectl delete -f data-ingestion.yaml
# ... 等等
```
## 注意事项
1. **数据库已更新为 taiji**:确保 `secrets.yaml` 中的数据库连接字符串指向 `taiji.postgres.database.azure.com`
2. **不影响本地 Docker**:此部署完全独立于本地 docker-compose 环境
3. **存储类**:PVC 使用 `managed-premium` 存储类,确保 AKS 集群支持
4. **资源限制**:根据实际需求调整各服务的资源请求和限制
5. **安全**:生产环境建议使用 Azure Key Vault 管理密钥
## 支持
如有问题,请查看:
- Kubernetes 日志:`kubectl logs -n taiji-ai`
- Pod 事件:`kubectl describe pod -n taiji-ai`
- 服务状态:`kubectl get all -n taiji-ai`
+79
View File
@@ -0,0 +1,79 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: api-gateway
namespace: taiji-ai
labels:
app: api-gateway
component: gateway
spec:
replicas: 2
selector:
matchLabels:
app: api-gateway
template:
metadata:
labels:
app: api-gateway
component: gateway
spec:
containers:
- name: nginx
image: nginx:alpine
ports:
- containerPort: 80
name: http
- containerPort: 443
name: https
resources:
requests:
memory: "128Mi"
cpu: "100m"
limits:
memory: "256Mi"
cpu: "500m"
volumeMounts:
- name: nginx-config
mountPath: /etc/nginx/nginx.conf
subPath: nginx.conf
livenessProbe:
httpGet:
path: /health
port: 80
initialDelaySeconds: 10
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 80
initialDelaySeconds: 5
periodSeconds: 10
volumes:
- name: nginx-config
configMap:
name: taiji-config
items:
- key: nginx.conf
path: nginx.conf
---
apiVersion: v1
kind: Service
metadata:
name: api-gateway-service
namespace: taiji-ai
labels:
app: api-gateway
spec:
type: LoadBalancer # 使用 LoadBalancer 暴露到公网
ports:
- port: 80
targetPort: 80
protocol: TCP
name: http
- port: 443
targetPort: 443
protocol: TCP
name: https
selector:
app: api-gateway
+160
View File
@@ -0,0 +1,160 @@
apiVersion: v1
kind: ConfigMap
metadata:
name: taiji-config
namespace: taiji-ai
data:
# NATS配置
NATS_URL: "nats://nats-service:4222"
# LiteLLM配置
LITELLM_URL: "http://litellm-gateway-service:4000"
LITELLM_API_KEY: "sk-taiji-master-key"
# 应用配置
LOG_LEVEL: "INFO"
LOG_FORMAT: "json"
# MCP Server配置
MCP_TIMEOUT: "30"
MCP_MAX_RETRIES: "3"
# Agent配置
MAX_AGENTS_PER_USER: "100"
AGENT_EXECUTION_TIMEOUT: "300"
# 监控配置
ENABLE_METRICS: "true"
METRICS_PORT: "8001"
# Nginx配置(通过ConfigMap挂载)
nginx.conf: |
user nginx;
worker_processes auto;
error_log /var/log/nginx/error.log notice;
pid /var/run/nginx.pid;
events {
worker_connections 1024;
use epoll;
multi_accept on;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent" "$http_x_forwarded_for" '
'rt=$request_time ut="$upstream_response_time"';
access_log /var/log/nginx/access.log main;
sendfile on;
tcp_nopush on;
tcp_nodelay on;
keepalive_timeout 65;
types_hash_max_size 2048;
client_max_body_size 50M;
gzip on;
gzip_vary on;
gzip_min_length 1024;
gzip_proxied any;
gzip_comp_level 6;
gzip_types text/plain text/css text/xml text/javascript application/json application/javascript application/xml+rss application/atom+xml image/svg+xml;
upstream mcp-server {
least_conn;
server mcp-server-service:8000 max_fails=3 fail_timeout=30s;
keepalive 32;
}
upstream data-ingestion {
least_conn;
server data-ingestion-service:8000 max_fails=3 fail_timeout=30s;
keepalive 32;
}
upstream litellm-gateway {
least_conn;
server litellm-gateway-service:4000 max_fails=3 fail_timeout=30s;
keepalive 32;
}
limit_req_zone $binary_remote_addr zone=api:10m rate=100r/m;
limit_req_zone $binary_remote_addr zone=auth:10m rate=20r/m;
server {
listen 80;
server_name _;
add_header X-Frame-Options DENY;
add_header X-Content-Type-Options nosniff;
add_header X-XSS-Protection "1; mode=block";
add_header Referrer-Policy "strict-origin-when-cross-origin";
location /health {
access_log off;
return 200 "OK\n";
add_header Content-Type text/plain;
}
location /api/mcp/ {
limit_req zone=api burst=50 nodelay;
proxy_pass http://mcp-server/;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_connect_timeout 30s;
proxy_send_timeout 30s;
proxy_read_timeout 30s;
proxy_buffering on;
proxy_buffer_size 4k;
proxy_buffers 8 4k;
}
location /api/data/ {
limit_req zone=api burst=30 nodelay;
proxy_pass http://data-ingestion/;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_connect_timeout 60s;
proxy_send_timeout 60s;
proxy_read_timeout 300s;
}
location /api/llm/ {
limit_req zone=api burst=20 nodelay;
proxy_pass http://litellm-gateway/;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_connect_timeout 60s;
proxy_send_timeout 60s;
proxy_read_timeout 300s;
}
location /docs {
proxy_pass http://mcp-server/docs;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
location / {
return 404 '{"error": "Not Found", "message": "请使用正确的API端点"}';
add_header Content-Type application/json;
}
}
}
+116
View File
@@ -0,0 +1,116 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: data-ingestion
namespace: taiji-ai
labels:
app: data-ingestion
component: ingestion
spec:
replicas: 2
selector:
matchLabels:
app: data-ingestion
template:
metadata:
labels:
app: data-ingestion
component: ingestion
spec:
imagePullSecrets:
- name: acr-secret
containers:
- name: data-ingestion
image: taiji.azurecr.io/taiji-data-ingestion:latest
ports:
- containerPort: 8000
name: http
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: DATABASE_URL
- name: ASYNC_DATABASE_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: ASYNC_DATABASE_URL
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: REDIS_URL
- name: NATS_URL
valueFrom:
configMapKeyRef:
name: taiji-config
key: NATS_URL
- name: RAPIDAPI_KEY
valueFrom:
secretKeyRef:
name: taiji-secrets
key: RAPIDAPI_KEY
- name: RAPIDAPI_HOST
valueFrom:
secretKeyRef:
name: taiji-secrets
key: RAPIDAPI_HOST
- name: OPENROUTER_API_KEY
valueFrom:
secretKeyRef:
name: taiji-secrets
key: OPENROUTER_API_KEY
- name: OPENROUTER_BASE_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: OPENROUTER_BASE_URL
- name: LOG_LEVEL
valueFrom:
configMapKeyRef:
name: taiji-config
key: LOG_LEVEL
resources:
requests:
memory: "512Mi"
cpu: "200m"
limits:
memory: "1Gi"
cpu: "1000m"
volumeMounts:
- name: logs
mountPath: /app/logs
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 10
volumes:
- name: logs
emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
name: data-ingestion-service
namespace: taiji-ai
labels:
app: data-ingestion
spec:
type: ClusterIP
ports:
- port: 8000
targetPort: 8000
protocol: TCP
name: http
selector:
app: data-ingestion
Executable
+160
View File
@@ -0,0 +1,160 @@
#!/bin/bash
# AKS 部署脚本
# 使用方法: ./deploy.sh <ACR_NAME> <AKS_RESOURCE_GROUP> <AKS_CLUSTER_NAME>
set -e
# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# 检查参数
if [ $# -lt 3 ]; then
echo -e "${RED}错误: 缺少参数${NC}"
echo "使用方法: $0 <ACR_NAME> <AKS_RESOURCE_GROUP> <AKS_CLUSTER_NAME>"
echo "示例: $0 myregistry myresourcegroup myakscluster"
exit 1
fi
ACR_NAME=$1
AKS_RESOURCE_GROUP=$2
AKS_CLUSTER_NAME=$3
echo -e "${GREEN}开始部署到 AKS...${NC}"
# 1. 获取 AKS 凭据
echo -e "${YELLOW}步骤 1: 获取 AKS 凭据...${NC}"
az aks get-credentials --resource-group $AKS_RESOURCE_GROUP --name $AKS_CLUSTER_NAME --overwrite-existing
# 2. 创建命名空间
echo -e "${YELLOW}步骤 2: 创建命名空间...${NC}"
kubectl apply -f namespace.yaml
# 3. 构建并推送 Docker 镜像到 ACR
echo -e "${YELLOW}步骤 3: 构建并推送 Docker 镜像到 ACR...${NC}"
# 登录到 ACR
az acr login --name $ACR_NAME
# 构建并推送各个服务的镜像
echo "构建 mcp-server 镜像..."
docker build -t $ACR_NAME.azurecr.io/taiji-mcp-server:latest ./services/mcp-server
docker push $ACR_NAME.azurecr.io/taiji-mcp-server:latest
echo "构建 data-ingestion 镜像..."
docker build -t $ACR_NAME.azurecr.io/taiji-data-ingestion:latest ./services/data-ingestion
docker push $ACR_NAME.azurecr.io/taiji-data-ingestion:latest
echo "构建 litellm-gateway 镜像..."
docker build -t $ACR_NAME.azurecr.io/taiji-litellm-gateway:latest ./services/model-gateway
docker push $ACR_NAME.azurecr.io/taiji-litellm-gateway:latest
# 4. 更新 Deployment 文件中的 ACR 名称
echo -e "${YELLOW}步骤 4: 更新 Deployment 文件中的 ACR 名称...${NC}"
sed -i "s/YOUR_ACR_NAME/$ACR_NAME/g" litellm-gateway.yaml
sed -i "s/YOUR_ACR_NAME/$ACR_NAME/g" data-ingestion.yaml
sed -i "s/YOUR_ACR_NAME/$ACR_NAME/g" mcp-server.yaml
# 5. 创建 Secret(需要用户输入)
echo -e "${YELLOW}步骤 5: 创建 Secret...${NC}"
echo -e "${RED}请确保已更新 secrets.yaml 文件中的敏感信息!${NC}"
read -p "是否已更新 secrets.yaml 文件?(y/n) " -n 1 -r
echo
if [[ ! $REPLY =~ ^[Yy]$ ]]; then
echo -e "${RED}请先更新 secrets.yaml 文件,然后重新运行此脚本${NC}"
exit 1
fi
# 创建 Secret(使用 kubectl create secret 命令,而不是直接 apply secrets.yaml)
echo "请手动创建 Secret,使用以下命令:"
echo "kubectl create secret generic taiji-secrets --from-file=secrets.yaml -n taiji-ai"
echo "或者使用 Azure Key Vault"
read -p "是否已创建 Secret?(y/n) " -n 1 -r
echo
if [[ ! $REPLY =~ ^[Yy]$ ]]; then
echo -e "${YELLOW}跳过 Secret 创建,请稍后手动创建${NC}"
fi
# 6. 创建 ACR 拉取 Secret(如果需要)
echo -e "${YELLOW}步骤 6: 创建 ACR 拉取 Secret...${NC}"
# 获取 ACR 登录服务器
ACR_LOGIN_SERVER=$(az acr show --name $ACR_NAME --query loginServer -o tsv)
# 获取 Service Principal 信息(如果使用)
# 这里假设使用 Azure CLI 的凭据
kubectl create secret docker-registry acr-secret \
--docker-server=$ACR_LOGIN_SERVER \
--docker-username=$(az acr credential show --name $ACR_NAME --query username -o tsv) \
--docker-password=$(az acr credential show --name $ACR_NAME --query passwords[0].value -o tsv) \
--namespace=taiji-ai \
--dry-run=client -o yaml | kubectl apply -f -
# 7. 应用 ConfigMap
echo -e "${YELLOW}步骤 7: 应用 ConfigMap...${NC}"
kubectl apply -f configmap.yaml
# 8. 部署服务(按依赖顺序)
echo -e "${YELLOW}步骤 8: 部署服务...${NC}"
# 8.1 部署 NATS(消息队列)
echo "部署 NATS..."
kubectl apply -f nats.yaml
# 等待 NATS 就绪
echo "等待 NATS 就绪..."
kubectl wait --for=condition=available --timeout=300s deployment/nats -n taiji-ai
# 8.2 部署 LiteLLM Gateway
echo "部署 LiteLLM Gateway..."
kubectl apply -f litellm-gateway.yaml
# 8.3 部署 Data Ingestion
echo "部署 Data Ingestion..."
kubectl apply -f data-ingestion.yaml
# 8.4 部署 MCP Server
echo "部署 MCP Server..."
kubectl apply -f mcp-server.yaml
# 8.5 部署 API Gateway
echo "部署 API Gateway..."
kubectl apply -f api-gateway.yaml
# 8.6 部署监控服务(可选)
read -p "是否部署监控服务 (Prometheus/Grafana)?(y/n) " -n 1 -r
echo
if [[ $REPLY =~ ^[Yy]$ ]]; then
echo "部署监控服务..."
kubectl apply -f monitoring.yaml
fi
# 9. 等待所有服务就绪
echo -e "${YELLOW}步骤 9: 等待所有服务就绪...${NC}"
kubectl wait --for=condition=available --timeout=600s deployment/litellm-gateway -n taiji-ai || true
kubectl wait --for=condition=available --timeout=600s deployment/data-ingestion -n taiji-ai || true
kubectl wait --for=condition=available --timeout=600s deployment/mcp-server -n taiji-ai || true
kubectl wait --for=condition=available --timeout=600s deployment/api-gateway -n taiji-ai || true
# 10. 显示部署状态
echo -e "${GREEN}部署完成!${NC}"
echo ""
echo "部署状态:"
kubectl get deployments -n taiji-ai
echo ""
echo "服务状态:"
kubectl get services -n taiji-ai
echo ""
echo "Pod 状态:"
kubectl get pods -n taiji-ai
echo ""
# 11. 获取 API Gateway 的外部 IP
echo -e "${GREEN}获取 API Gateway 外部 IP...${NC}"
API_GATEWAY_IP=$(kubectl get service api-gateway-service -n taiji-ai -o jsonpath='{.status.loadBalancer.ingress[0].ip}' 2>/dev/null || echo "正在分配中...")
echo "API Gateway 外部 IP: $API_GATEWAY_IP"
echo ""
echo -e "${GREEN}部署完成!${NC}"
echo "访问地址: http://$API_GATEWAY_IP"
+104
View File
@@ -0,0 +1,104 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: litellm-gateway
namespace: taiji-ai
labels:
app: litellm-gateway
component: gateway
spec:
replicas: 2
selector:
matchLabels:
app: litellm-gateway
template:
metadata:
labels:
app: litellm-gateway
component: gateway
spec:
imagePullSecrets:
- name: acr-secret
containers:
- name: litellm-gateway
image: taiji.azurecr.io/taiji-litellm-gateway:latest
ports:
- containerPort: 4000
name: http
env:
- name: LITELLM_MASTER_KEY
valueFrom:
secretKeyRef:
name: taiji-secrets
key: LITELLM_MASTER_KEY
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: DATABASE_URL
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: REDIS_URL
- name: OPENROUTER_API_KEY
valueFrom:
secretKeyRef:
name: taiji-secrets
key: OPENROUTER_API_KEY
- name: OPENROUTER_BASE_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: OPENROUTER_BASE_URL
resources:
requests:
memory: "512Mi"
cpu: "200m"
limits:
memory: "1Gi"
cpu: "1000m"
volumeMounts:
- name: logs
mountPath: /app/logs
livenessProbe:
httpGet:
path: /health
port: 4000
httpHeaders:
- name: Authorization
value: "Bearer sk-taiji-master-key"
initialDelaySeconds: 60
periodSeconds: 30
timeoutSeconds: 25
readinessProbe:
httpGet:
path: /health
port: 4000
httpHeaders:
- name: Authorization
value: "Bearer sk-taiji-master-key"
initialDelaySeconds: 30
periodSeconds: 10
timeoutSeconds: 25
volumes:
- name: logs
emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
name: litellm-gateway-service
namespace: taiji-ai
labels:
app: litellm-gateway
spec:
type: ClusterIP
ports:
- port: 4000
targetPort: 4000
protocol: TCP
name: http
selector:
app: litellm-gateway
+118
View File
@@ -0,0 +1,118 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: mcp-server
namespace: taiji-ai
labels:
app: mcp-server
component: server
spec:
replicas: 3
selector:
matchLabels:
app: mcp-server
template:
metadata:
labels:
app: mcp-server
component: server
spec:
imagePullSecrets:
- name: acr-secret
containers:
- name: mcp-server
image: taiji.azurecr.io/taiji-mcp-server:latest
ports:
- containerPort: 8000
name: http
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: DATABASE_URL
- name: ASYNC_DATABASE_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: ASYNC_DATABASE_URL
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: taiji-secrets
key: REDIS_URL
- name: NATS_URL
valueFrom:
configMapKeyRef:
name: taiji-config
key: NATS_URL
- name: LITELLM_URL
valueFrom:
configMapKeyRef:
name: taiji-config
key: LITELLM_URL
- name: LITELLM_API_KEY
valueFrom:
secretKeyRef:
name: taiji-secrets
key: LITELLM_API_KEY
- name: SECRET_KEY
valueFrom:
secretKeyRef:
name: taiji-secrets
key: SECRET_KEY
- name: ENCRYPTION_KEY
valueFrom:
secretKeyRef:
name: taiji-secrets
key: ENCRYPTION_KEY
- name: LOG_LEVEL
valueFrom:
configMapKeyRef:
name: taiji-config
key: LOG_LEVEL
- name: WORKERS
value: "4"
resources:
requests:
memory: "1Gi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "2000m"
volumeMounts:
- name: logs
mountPath: /app/logs
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 40
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 10
volumes:
- name: logs
emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
name: mcp-server-service
namespace: taiji-ai
labels:
app: mcp-server
spec:
type: ClusterIP
ports:
- port: 8000
targetPort: 8000
protocol: TCP
name: http
selector:
app: mcp-server
+212
View File
@@ -0,0 +1,212 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: prometheus
namespace: taiji-ai
labels:
app: prometheus
component: monitoring
spec:
replicas: 1
selector:
matchLabels:
app: prometheus
template:
metadata:
labels:
app: prometheus
component: monitoring
spec:
containers:
- name: prometheus
image: prom/prometheus:latest
ports:
- containerPort: 9090
name: http
command:
- '--config.file=/etc/prometheus/prometheus.yml'
- '--storage.tsdb.path=/prometheus'
- '--web.console.libraries=/etc/prometheus/console_libraries'
- '--web.console.templates=/etc/prometheus/consoles'
- '--web.enable-lifecycle'
resources:
requests:
memory: "512Mi"
cpu: "200m"
limits:
memory: "2Gi"
cpu: "1000m"
volumeMounts:
- name: prometheus-config
mountPath: /etc/prometheus/prometheus.yml
subPath: prometheus.yml
- name: prometheus-data
mountPath: /prometheus
volumes:
- name: prometheus-config
configMap:
name: prometheus-config
- name: prometheus-data
persistentVolumeClaim:
claimName: prometheus-pvc
---
apiVersion: v1
kind: Service
metadata:
name: prometheus-service
namespace: taiji-ai
labels:
app: prometheus
spec:
type: ClusterIP
ports:
- port: 9090
targetPort: 9090
protocol: TCP
name: http
selector:
app: prometheus
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: grafana
namespace: taiji-ai
labels:
app: grafana
component: monitoring
spec:
replicas: 1
selector:
matchLabels:
app: grafana
template:
metadata:
labels:
app: grafana
component: monitoring
spec:
containers:
- name: grafana
image: grafana/grafana:latest
ports:
- containerPort: 3000
name: http
env:
- name: GF_SECURITY_ADMIN_PASSWORD
value: "admin"
resources:
requests:
memory: "256Mi"
cpu: "100m"
limits:
memory: "512Mi"
cpu: "500m"
volumeMounts:
- name: grafana-data
mountPath: /var/lib/grafana
volumes:
- name: grafana-data
persistentVolumeClaim:
claimName: grafana-pvc
---
apiVersion: v1
kind: Service
metadata:
name: grafana-service
namespace: taiji-ai
labels:
app: grafana
spec:
type: ClusterIP
ports:
- port: 3000
targetPort: 3000
protocol: TCP
name: http
selector:
app: grafana
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: prometheus-pvc
namespace: taiji-ai
spec:
accessModes:
- ReadWriteOnce
resources:
requests:
storage: 50Gi
storageClassName: managed-premium
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: grafana-pvc
namespace: taiji-ai
spec:
accessModes:
- ReadWriteOnce
resources:
requests:
storage: 10Gi
storageClassName: managed-premium
---
apiVersion: v1
kind: ConfigMap
metadata:
name: prometheus-config
namespace: taiji-ai
data:
prometheus.yml: |
global:
scrape_interval: 15s
evaluation_interval: 15s
scrape_configs:
- job_name: 'mcp-server'
kubernetes_sd_configs:
- role: pod
namespaces:
names:
- taiji-ai
relabel_configs:
- source_labels: [__meta_kubernetes_pod_label_app]
action: keep
regex: mcp-server
- source_labels: [__meta_kubernetes_pod_ip]
action: replace
target_label: __address__
replacement: $1:8000
- job_name: 'data-ingestion'
kubernetes_sd_configs:
- role: pod
namespaces:
names:
- taiji-ai
relabel_configs:
- source_labels: [__meta_kubernetes_pod_label_app]
action: keep
regex: data-ingestion
- source_labels: [__meta_kubernetes_pod_ip]
action: replace
target_label: __address__
replacement: $1:8000
- job_name: 'litellm-gateway'
kubernetes_sd_configs:
- role: pod
namespaces:
names:
- taiji-ai
relabel_configs:
- source_labels: [__meta_kubernetes_pod_label_app]
action: keep
regex: litellm-gateway
- source_labels: [__meta_kubernetes_pod_ip]
action: replace
target_label: __address__
replacement: $1:4000
+8
View File
@@ -0,0 +1,8 @@
apiVersion: v1
kind: Namespace
metadata:
name: taiji-ai
labels:
name: taiji-ai
environment: production
+94
View File
@@ -0,0 +1,94 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: nats
namespace: taiji-ai
labels:
app: nats
component: messaging
spec:
replicas: 1
selector:
matchLabels:
app: nats
template:
metadata:
labels:
app: nats
component: messaging
spec:
containers:
- name: nats
image: nats:2.10-alpine
ports:
- containerPort: 4222
name: client
- containerPort: 6222
name: routing
- containerPort: 8222
name: monitoring
command: ["/nats-server"]
args: ["-js", "-m", "8222"]
resources:
requests:
memory: "256Mi"
cpu: "100m"
limits:
memory: "512Mi"
cpu: "500m"
volumeMounts:
- name: nats-data
mountPath: /data
livenessProbe:
tcpSocket:
port: 4222
initialDelaySeconds: 10
periodSeconds: 30
readinessProbe:
tcpSocket:
port: 4222
initialDelaySeconds: 5
periodSeconds: 10
volumes:
- name: nats-data
persistentVolumeClaim:
claimName: nats-pvc
---
apiVersion: v1
kind: Service
metadata:
name: nats-service
namespace: taiji-ai
labels:
app: nats
spec:
type: ClusterIP
ports:
- port: 4222
targetPort: 4222
protocol: TCP
name: client
- port: 6222
targetPort: 6222
protocol: TCP
name: routing
- port: 8222
targetPort: 8222
protocol: TCP
name: monitoring
selector:
app: nats
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: nats-pvc
namespace: taiji-ai
spec:
accessModes:
- ReadWriteOnce
resources:
requests:
storage: 10Gi
storageClassName: managed-premium # Azure 存储类
+60
View File
@@ -0,0 +1,60 @@
# Kubernetes Secrets配置
# 注意:实际部署时请使用 kubectl create secret 命令或 Azure Key Vault
# 此文件仅作为模板参考
apiVersion: v1
kind: Secret
metadata:
name: taiji-secrets
namespace: taiji-ai
type: Opaque
stringData:
# 数据库连接字符串 - 已更新为 taiji 数据库
# 格式: postgresql+asyncpg://用户名:密码@taiji.postgres.database.azure.com:5432/数据库名?sslmode=require
# 请替换为实际的用户名和密码
DATABASE_URL: "postgresql+asyncpg://taiji:YOUR_PASSWORD@taiji.postgres.database.azure.com:5432/postgres?sslmode=require"
ASYNC_DATABASE_URL: "postgresql+asyncpg://taiji:YOUR_PASSWORD@taiji.postgres.database.azure.com:5432/postgres?sslmode=require"
# Redis连接字符串
# 请替换为实际的Redis密码
REDIS_URL: "rediss://:YOUR_REDIS_PASSWORD@taiji.southeastasia.redis.azure.net:10000/0?ssl_cert_reqs=none"
# JWT密钥
SECRET_KEY: "zsbgnw"
JWT_SECRET: "zsbgnw"
# 加密密钥
ENCRYPTION_KEY: "zsbgnw"
# LiteLLM API密钥
LITELLM_MASTER_KEY: "sk-taiji-master-key"
LITELLM_API_KEY: "sk-taiji-master-key"
# OpenRouter配置
OPENROUTER_API_KEY: "YOUR_OPENROUTER_API_KEY"
OPENROUTER_BASE_URL: "https://openrouter.ai/api/v1"
# RapidAPI配置
RAPIDAPI_KEY: "YOUR_RAPIDAPI_KEY"
RAPIDAPI_HOST: "YOUR_RAPIDAPI_HOST"
# Azure存储连接字符串(可选)
AZURE_STORAGE_CONNECTION_STRING: "YOUR_AZURE_STORAGE_CONNECTION_STRING"
---
# ACR拉取密钥(使用Service Principal)
# 生成方式:
# kubectl create secret docker-registry acr-secret \
# --docker-server=${ACR_NAME}.azurecr.io \
# --docker-username=${SP_APP_ID} \
# --docker-password=${SP_PASSWORD} \
# --namespace=taiji-ai
apiVersion: v1
kind: Secret
metadata:
name: acr-secret
namespace: taiji-ai
type: kubernetes.io/dockerconfigjson
data:
.dockerconfigjson: <BASE64_ENCODED_DOCKER_CONFIG>
-162
View File
@@ -1,162 +0,0 @@
#!/usr/bin/env python3
"""
生成初始用户的SQL插入语句
使用bcrypt加密密码
"""
import uuid
from datetime import datetime
import bcrypt
def hash_password(password: str) -> str:
"""使用bcrypt加密密码"""
# bcrypt需要bytes类型的密码
password_bytes = password.encode('utf-8')
# 生成salt并加密
salt = bcrypt.gensalt()
hashed = bcrypt.hashpw(password_bytes, salt)
# 返回字符串形式
return hashed.decode('utf-8')
# 定义初始用户
USERS = [
{
"name": "超级管理员",
"username": "admin",
"email": "admin@test.com",
"password": "admin123",
"role": "super_admin",
"full_name": "系统超级管理员",
"is_active": True,
"is_admin": True,
"subscription_tier": "enterprise",
"balance": 100000,
"credit_limit": 500000,
"status": "active"
},
{
"name": "渠道管理员",
"username": "channel",
"email": "channel@test.com",
"password": "channel123",
"role": "channel_admin",
"full_name": "渠道管理员",
"is_active": True,
"is_admin": False,
"subscription_tier": "business",
"balance": 50000,
"credit_limit": 100000,
"status": "active"
},
{
"name": "供应商管理员",
"username": "provider",
"email": "provider@test.com",
"password": "provider123",
"role": "provider_admin",
"full_name": "供应商管理员",
"is_active": True,
"is_admin": False,
"subscription_tier": "business",
"balance": 50000,
"credit_limit": 100000,
"status": "active"
},
{
"name": "测试用户",
"username": "user",
"email": "user@test.com",
"password": "user123",
"role": "user",
"full_name": "普通测试用户",
"is_active": True,
"is_admin": False,
"subscription_tier": "free",
"balance": 100,
"credit_limit": 1000,
"status": "active"
},
{
"name": "计费管理员",
"username": "billing",
"email": "billing@test.com",
"password": "billing123",
"role": "billing_admin",
"full_name": "计费管理员",
"is_active": True,
"is_admin": False,
"subscription_tier": "business",
"balance": 10000,
"credit_limit": 50000,
"status": "active"
},
{
"name": "运维管理员",
"username": "operations",
"email": "operations@test.com",
"password": "operations123",
"role": "operations_admin",
"full_name": "运维管理员",
"is_active": True,
"is_admin": False,
"subscription_tier": "business",
"balance": 10000,
"credit_limit": 50000,
"status": "active"
}
]
def generate_sql():
"""生成SQL插入语句"""
print("-- 初始用户数据")
print("-- 自动生成时间:", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
print("-- 注意: 此脚本会在用户不存在时插入初始用户\n")
for user in USERS:
user_id = str(uuid.uuid4())
password_hash = hash_password(user["password"])
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"-- 创建用户: {user['name']} ({user['email']})")
print(f"INSERT INTO users (")
print(f" id, created_at, updated_at,")
print(f" name, username, email,")
print(f" password_hash, hashed_password,")
print(f" role, full_name,")
print(f" is_active, is_admin,")
print(f" subscription_tier, balance, credit_limit,")
print(f" status, discount")
print(f") VALUES (")
print(f" '{user_id}', '{now}', '{now}',")
print(f" '{user['name']}', '{user['username']}', '{user['email']}',")
print(f" '{password_hash}', '{password_hash}',")
print(f" '{user['role']}', '{user['full_name']}',")
print(f" {user['is_active']}, {user['is_admin']},")
print(f" '{user['subscription_tier']}', {user['balance']}, {user['credit_limit']},")
print(f" '{user['status']}', 0")
print(f") ON CONFLICT (email) DO NOTHING;")
print()
def main():
print("="*80)
print("Taiji AI-PAD 初始用户SQL生成器")
print("="*80)
print()
generate_sql()
print("\n-- 用户信息汇总:")
print("-- " + "="*76)
for user in USERS:
print(f"-- {user['name']:15} | {user['email']:25} | 密码: {user['password']}")
print("-- " + "="*76)
print("\n-- 提示: 将上述SQL语句添加到 scripts/init.sql 文件末尾")
print("-- 或者创建新的 scripts/init_users.sql 文件")
if __name__ == "__main__":
main()
+462
View File
@@ -0,0 +1,462 @@
#!/usr/bin/env python3
"""
完整工作流测试脚本
测试:超级管理员创建渠道、创建计费/运维管理员、创建租户、权限验证
"""
import requests
import sys
import os
import time
import subprocess
from typing import Optional, Dict, Any
BASE_URL = "http://localhost:8002"
# 测试账户信息
SUPER_ADMIN = {
"email": "superadmin@taiji-ai.com",
"password": "Admin@123456",
"role": "super_admin"
}
BILLING_ADMIN = {
"email": "newbilling@test.com",
"password": "Billing@123456",
"role": "billing_admin"
}
OPS_ADMIN = {
"email": "newops@test.com",
"password": "Ops@123456",
"role": "operations_admin"
}
CHANNEL_ADMIN = {
"email": "channel-a@test.com",
"password": "ChannelA@123456",
"role": "channel_admin"
}
def print_header(title: str):
"""打印标题"""
print("\n" + "="*80)
print(f" {title}")
print("="*80)
def print_step(step: str):
"""打印步骤"""
print(f"\n[步骤] {step}")
print("-" * 80)
def wait_for_service(url: str, max_retries: int = 30, delay: int = 2) -> bool:
"""等待服务启动"""
print(f"等待服务启动: {url}")
for i in range(max_retries):
try:
resp = requests.get(f"{url}/health", timeout=2)
if resp.status_code == 200:
print(f" ✓ 服务已启动")
return True
except:
pass
if i < max_retries - 1:
print(f" 等待中... ({i+1}/{max_retries})")
time.sleep(delay)
print(f" ✗ 服务启动超时")
return False
def login(email: str, password: str, role: str = None) -> Optional[str]:
"""登录并获取token"""
try:
data = {"email": email, "password": password}
if role:
data["role"] = role
resp = requests.post(
f"{BASE_URL}/api/auth/login",
json=data,
timeout=10
)
if resp.status_code == 200:
result = resp.json()
token = result.get("data", {}).get("token")
if token:
print(f" ✓ 登录成功: {email}")
return token
else:
print(f" ✗ 登录失败: 响应中未找到token")
return None
else:
error = resp.json().get("detail", resp.text)
print(f" ✗ 登录失败: {error}")
return None
except Exception as e:
print(f" ✗ 登录出错: {e}")
return None
def test_create_channel(token: str) -> Optional[str]:
"""测试创建渠道"""
print_step("测试:超级管理员创建渠道")
channel_data = {
"name": "测试渠道A",
"email": "channel-a-test@test.com",
"password": "Channel@123456",
"commissionRate": 15.0
}
try:
resp = requests.post(
f"{BASE_URL}/api/admin/channels/create",
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
},
json=channel_data,
timeout=10
)
if resp.status_code == 200:
result = resp.json()
channel_id = result.get("data", {}).get("id")
print(f" ✓ 渠道创建成功: {channel_data['name']} (ID: {channel_id})")
return channel_id
else:
error = resp.json().get("detail", resp.text)
if "邮箱已被使用" in error or "already exists" in error.lower():
print(f" ⚠ 渠道已存在: {channel_data['email']}")
# 尝试获取已存在的渠道
return get_channel_by_email(token, channel_data["email"])
else:
print(f" ✗ 创建失败: {error}")
return None
except Exception as e:
print(f" ✗ 创建出错: {e}")
return None
def get_channel_by_email(token: str, email: str) -> Optional[str]:
"""通过邮箱获取渠道ID"""
try:
resp = requests.get(
f"{BASE_URL}/api/admin/channels",
headers={"Authorization": f"Bearer {token}"},
timeout=10
)
if resp.status_code == 200:
result = resp.json()
channels = result.get("data", {}).get("channels", [])
for channel in channels:
if channel.get("email") == email:
return channel.get("id")
return None
except:
return None
def test_list_channels(token: str) -> bool:
"""测试获取渠道列表"""
print_step("测试:获取渠道列表")
try:
resp = requests.get(
f"{BASE_URL}/api/admin/channels",
headers={"Authorization": f"Bearer {token}"},
timeout=10
)
if resp.status_code == 200:
result = resp.json()
channels = result.get("data", {}).get("channels", [])
print(f" ✓ 获取成功,共 {len(channels)} 个渠道")
for ch in channels[:3]: # 只显示前3个
print(f" - {ch.get('name')} ({ch.get('email')})")
return True
else:
error = resp.json().get("detail", resp.text)
print(f" ✗ 获取失败: {error}")
return False
except Exception as e:
print(f" ✗ 获取出错: {e}")
return False
def test_create_admin(token: str, admin_info: dict, channel_id: str = None) -> bool:
"""测试创建管理员"""
role_name = {
"billing_admin": "计费管理员",
"operations_admin": "运维管理员"
}.get(admin_info["role"], admin_info["role"])
print_step(f"测试:创建{role_name}")
admin_data = {
"name": admin_info.get("name", role_name),
"email": admin_info["email"],
"password": admin_info["password"],
"role": admin_info["role"]
}
if channel_id:
admin_data["channelId"] = channel_id
try:
resp = requests.post(
f"{BASE_URL}/api/admin/admins/create",
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
},
json=admin_data,
timeout=10
)
if resp.status_code == 200:
result = resp.json()
print(f" ✓ {role_name}创建成功: {admin_info['email']}")
return True
else:
error = resp.json().get("detail", resp.text)
if "邮箱已被使用" in error or "already exists" in error.lower():
print(f" ⚠ {role_name}已存在: {admin_info['email']}")
return True # 已存在也算成功
else:
print(f" ✗ 创建失败: {error}")
return False
except Exception as e:
print(f" ✗ 创建出错: {e}")
return False
def test_list_admins(token: str) -> bool:
"""测试获取管理员列表"""
print_step("测试:获取管理员列表")
try:
resp = requests.get(
f"{BASE_URL}/api/admin/admins",
headers={"Authorization": f"Bearer {token}"},
timeout=10
)
if resp.status_code == 200:
result = resp.json()
admins = result.get("data", {}).get("admins", [])
print(f" ✓ 获取成功,共 {len(admins)} 个管理员")
for admin in admins[:5]: # 只显示前5个
print(f" - {admin.get('name')} ({admin.get('email')}) - {admin.get('role')}")
return True
else:
error = resp.json().get("detail", resp.text)
print(f" ✗ 获取失败: {error}")
return False
except Exception as e:
print(f" ✗ 获取出错: {e}")
return False
def test_create_tenant(token: str, channel_token: str = None) -> Optional[str]:
"""测试创建租户"""
print_step("测试:创建租户")
# 使用渠道管理员token或提供的token
use_token = channel_token or token
tenant_data = {
"name": "测试租户A",
"email": "tenant-a@test.com",
"password": "Tenant@123456",
"subscriptionTier": "pro"
}
try:
resp = requests.post(
f"{BASE_URL}/api/channel/tenants/create",
headers={
"Authorization": f"Bearer {use_token}",
"Content-Type": "application/json"
},
json=tenant_data,
timeout=10
)
if resp.status_code == 200:
result = resp.json()
tenant_id = result.get("data", {}).get("id")
print(f" ✓ 租户创建成功: {tenant_data['name']} (ID: {tenant_id})")
return tenant_id
else:
error = resp.json().get("detail", resp.text)
if "邮箱已被使用" in error or "already exists" in error.lower():
print(f" ⚠ 租户已存在: {tenant_data['email']}")
return "existing"
else:
print(f" ✗ 创建失败: {error}")
return None
except Exception as e:
print(f" ✗ 创建出错: {e}")
return None
def test_list_tenants(token: str) -> bool:
"""测试获取租户列表"""
print_step("测试:获取租户列表")
try:
resp = requests.get(
f"{BASE_URL}/api/channel/tenants",
headers={"Authorization": f"Bearer {token}"},
timeout=10
)
if resp.status_code == 200:
result = resp.json()
tenants = result.get("data", {}).get("tenants", [])
print(f" ✓ 获取成功,共 {len(tenants)} 个租户")
for tenant in tenants[:5]: # 只显示前5个
print(f" - {tenant.get('name')} ({tenant.get('email')}) - {tenant.get('subscriptionTier')}")
return True
else:
error = resp.json().get("detail", resp.text)
print(f" ✗ 获取失败: {error}")
return False
except Exception as e:
print(f" ✗ 获取出错: {e}")
return False
def test_permission_verification():
"""测试权限验证"""
print_header("权限验证测试")
# 测试1: 计费管理员不能创建渠道
print_step("测试1: 计费管理员尝试创建渠道(应该失败)")
billing_token = login(BILLING_ADMIN["email"], BILLING_ADMIN["password"], BILLING_ADMIN["role"])
if billing_token:
try:
resp = requests.post(
f"{BASE_URL}/api/admin/channels/create",
headers={
"Authorization": f"Bearer {billing_token}",
"Content-Type": "application/json"
},
json={
"name": "未授权渠道",
"email": "unauthorized@test.com",
"password": "Test@123456",
"commissionRate": 10.0
},
timeout=10
)
if resp.status_code == 403:
print(f" ✓ 权限验证正确:计费管理员无法创建渠道")
else:
print(f" ✗ 权限验证失败:计费管理员不应该能创建渠道")
except Exception as e:
print(f" ✗ 测试出错: {e}")
# 测试2: 运维管理员不能创建租户
print_step("测试2: 运维管理员尝试创建租户(应该失败)")
ops_token = login(OPS_ADMIN["email"], OPS_ADMIN["password"], OPS_ADMIN["role"])
if ops_token:
try:
resp = requests.post(
f"{BASE_URL}/api/channel/tenants/create",
headers={
"Authorization": f"Bearer {ops_token}",
"Content-Type": "application/json"
},
json={
"name": "未授权租户",
"email": "unauthorized-tenant@test.com",
"password": "Test@123456",
"subscriptionTier": "free"
},
timeout=10
)
if resp.status_code == 403:
print(f" ✓ 权限验证正确:运维管理员无法创建租户")
else:
print(f" ✗ 权限验证失败:运维管理员不应该能创建租户")
except Exception as e:
print(f" ✗ 测试出错: {e}")
# 测试3: 计费管理员可以创建租户
print_step("测试3: 计费管理员尝试创建租户(应该成功)")
if billing_token:
tenant_id = test_create_tenant(billing_token, billing_token)
if tenant_id:
print(f" ✓ 权限验证正确:计费管理员可以创建租户")
else:
print(f" ⚠ 创建租户失败(可能是其他原因)")
# 测试4: 超级管理员可以访问所有资源
print_step("测试4: 超级管理员访问所有资源(应该成功)")
super_token = login(SUPER_ADMIN["email"], SUPER_ADMIN["password"], SUPER_ADMIN["role"])
if super_token:
success = True
success = success and test_list_channels(super_token)
success = success and test_list_admins(super_token)
if success:
print(f" ✓ 权限验证正确:超级管理员可以访问所有资源")
else:
print(f" ✗ 部分资源访问失败")
def main():
"""主函数"""
print_header("完整工作流测试")
# 步骤1: 等待服务启动
print_step("等待服务启动")
if not wait_for_service(BASE_URL):
print(" ✗ 服务未启动,请先启动Docker服务")
return
# 步骤2: 登录超级管理员
print_step("登录超级管理员")
super_token = login(SUPER_ADMIN["email"], SUPER_ADMIN["password"], SUPER_ADMIN["role"])
if not super_token:
print(" ✗ 无法登录超级管理员,请先运行 test_cre_admins.py 创建管理员")
return
# 步骤3: 测试创建渠道
channel_id = test_create_channel(super_token)
# 步骤4: 测试获取渠道列表
test_list_channels(super_token)
# 步骤5: 测试创建管理员
if channel_id:
test_create_admin(super_token, BILLING_ADMIN, channel_id)
test_create_admin(super_token, OPS_ADMIN, channel_id)
# 步骤6: 测试获取管理员列表
test_list_admins(super_token)
# 步骤7: 测试创建租户(使用计费管理员)
print_header("租户管理测试")
billing_token = login(BILLING_ADMIN["email"], BILLING_ADMIN["password"], BILLING_ADMIN["role"])
if billing_token:
test_create_tenant(billing_token, billing_token)
test_list_tenants(billing_token)
# 步骤8: 权限验证
test_permission_verification()
# 总结
print_header("测试完成")
print("\n所有测试已完成!")
if __name__ == "__main__":
main()
+5 -5
View File
@@ -402,10 +402,10 @@ async def list_channels(
"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),
"commissionRate": float(channel.commission_rate) if channel.commission_rate is not None else 0.0,
"channelCredit": float(channel.channel_credit) if channel.channel_credit is not None else 0.0,
"customAgentCpu": float(channel.custom_agent_cpu) if channel.custom_agent_cpu is not None else 2.0,
"customAgentMemory": float(channel.custom_agent_memory) if channel.custom_agent_memory is not None else 4.0,
"status": channel.status,
"createdAt": channel.created_at.isoformat(),
}
@@ -520,7 +520,7 @@ async def update_channel(
"id": str(channel.id),
"name": channel.name,
"email": channel.email,
"commissionRate": float(channel.commission_rate),
"commissionRate": float(channel.commission_rate) if channel.commission_rate is not None else 0.0,
"status": channel.status,
},
message="渠道信息更新成功"
+214 -82
View File
@@ -58,6 +58,9 @@ def _verify_permission(principal: dict, permission: str):
def _get_channel_id(principal: dict) -> Optional[uuid.UUID]:
"""获取当前用户的渠道ID(如果是渠道下的管理员)"""
role = _get_role(principal)
# 超级管理员没有固定的channel_id,可以访问所有渠道
if role == "super_admin":
return None
if role in ["billing_admin", "operations_admin", "channel_admin"]:
channel_id_str = principal.get("claims", {}).get("channelId")
if channel_id_str:
@@ -77,20 +80,25 @@ async def list_tenants(
):
"""
获取渠道下的租户列表
权限:view:tenants (channel_admin, billing_admin, operations_admin)
权限:view:tenants (channel_admin, billing_admin, operations_admin, super_admin)
"""
_verify_permission(principal, "view:tenants")
channel_id = _get_channel_id(principal)
role = _get_role(principal)
if not channel_id:
# 超级管理员可以查看所有租户,其他角色只能查看自己渠道的租户
if role == "super_admin":
result = await db.execute(select(User).where(User.role == "user"))
elif channel_id:
result = await db.execute(
select(User).where(User.channel_id == channel_id)
)
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
result = await db.execute(
select(User).where(User.channel_id == channel_id)
)
tenants = result.scalars().all()
data = [
@@ -118,15 +126,26 @@ async def create_tenant(
):
"""
创建租户
权限:manage:tenants (channel_admin, billing_admin)
权限:manage:tenants (channel_admin, billing_admin, super_admin)
注意:超级管理员创建租户时,需要提供channelId参数
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if not channel_id:
# 超级管理员可以通过请求参数指定channelId
if role == "super_admin" and hasattr(req, 'channelId') and req.channelId:
try:
channel_id = uuid.UUID(req.channelId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
detail="无法获取渠道ID,超级管理员创建租户时需要提供channelId参数"
)
# 检查邮箱是否已存在
@@ -182,17 +201,25 @@ async def allocate_tenant_resources(
权限:manage:resources (channel_admin, billing_admin)
"""
_verify_permission(principal, "manage:resources")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 验证租户属于该渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
# 验证租户存在
if role == "super_admin":
# 超级管理员可以访问所有租户
result = await db.execute(
select(User).where(User.id == tenant_id)
)
else:
# 其他角色只能访问自己渠道的租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
@@ -258,17 +285,25 @@ async def update_tenant_billing(
权限:manage:billing (channel_admin, billing_admin)
"""
_verify_permission(principal, "manage:billing")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 验证租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
# 验证租户存在
if role == "super_admin":
# 超级管理员可以访问所有租户
result = await db.execute(
select(User).where(User.id == tenant_id)
)
else:
# 其他角色只能访问自己渠道的租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
@@ -298,17 +333,25 @@ async def recharge_tenant(
权限:manage:billing (channel_admin, billing_admin)
"""
_verify_permission(principal, "manage:billing")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 验证租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
# 验证租户存在
if role == "super_admin":
# 超级管理员可以访问所有租户
result = await db.execute(
select(User).where(User.id == tenant_id)
)
else:
# 其他角色只能访问自己渠道的租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
@@ -356,17 +399,25 @@ async def set_tenant_credit_limit(
权限:manage:billing (channel_admin, billing_admin)
"""
_verify_permission(principal, "manage:billing")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 验证租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
# 验证租户存在
if role == "super_admin":
# 超级管理员可以访问所有租户
result = await db.execute(
select(User).where(User.id == tenant_id)
)
else:
# 其他角色只能访问自己渠道的租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
@@ -401,17 +452,25 @@ async def delete_tenant(
权限:manage:tenants (channel_admin, billing_admin)
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 验证租户属于该渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
# 验证租户存在
if role == "super_admin":
# 超级管理员可以访问所有租户
result = await db.execute(
select(User).where(User.id == tenant_id)
)
else:
# 其他角色只能访问自己渠道的租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
@@ -457,17 +516,25 @@ async def update_tenant_status(
权限:manage:tenants (channel_admin, billing_admin)
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 验证租户属于该渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
# 验证租户存在
if role == "super_admin":
# 超级管理员可以访问所有租户
result = await db.execute(
select(User).where(User.id == tenant_id)
)
else:
# 其他角色只能访问自己渠道的租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
@@ -510,17 +577,25 @@ async def update_tenant_permissions(
权限:manage:tenants (channel_admin, billing_admin)
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 验证租户属于该渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
# 验证租户存在
if role == "super_admin":
# 超级管理员可以访问所有租户
result = await db.execute(
select(User).where(User.id == tenant_id)
)
else:
# 其他角色只能访问自己渠道的租户
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
@@ -582,12 +657,22 @@ async def create_channel_admin(
权限:manage:admins (channel_admin)
"""
_verify_permission(principal, "manage:admins")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if not channel_id:
# 超级管理员可以通过请求参数指定channelId
if role == "super_admin" and hasattr(req, 'channelId') and req.channelId:
try:
channel_id = uuid.UUID(req.channelId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
detail="无法获取渠道ID,超级管理员创建管理员时需要提供channelId参数"
)
# 验证渠道存在
@@ -653,24 +738,34 @@ async def list_channel_admins(
权限:view:admins (channel_admin)
"""
_verify_permission(principal, "view:admins")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if not channel_id:
# 超级管理员可以查看所有管理员,其他角色只能查看自己渠道的管理员
if role == "super_admin":
result = await db.execute(
select(User).where(
and_(
User.role.in_(["channel_admin", "billing_admin", "operations_admin"]),
User.status == "active"
)
)
)
elif channel_id:
result = await db.execute(
select(User).where(
and_(
User.channel_id == channel_id,
User.role.in_(["billing_admin", "operations_admin", "channel_admin"]),
User.status == "active"
)
)
)
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 查询渠道下的管理员(billing_admin和operations_admin)
result = await db.execute(
select(User).where(
and_(
User.channel_id == channel_id,
User.role.in_(["billing_admin", "operations_admin", "channel_admin"]),
User.status == "active"
)
)
)
admins = result.scalars().all()
data = [
@@ -701,12 +796,22 @@ async def apply_for_resources(
权限:view:applications (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:applications")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if not channel_id:
# 超级管理员申请资源时需要提供channelId参数(通过请求体)
if role == "super_admin" and hasattr(req, 'channelId') and req.channelId:
try:
channel_id = uuid.UUID(req.channelId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
detail="无法获取渠道ID,超级管理员申请资源时需要提供channelId参数"
)
# 创建申请
@@ -856,9 +961,13 @@ async def list_available_providers(
权限:view:resources (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if not channel_id:
# 超级管理员可以查看所有供应商,其他角色需要channel_id
if role == "super_admin":
channel_id = None # 超级管理员不需要channel_id限制
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
@@ -926,12 +1035,22 @@ async def apply_for_provider(
权限:view:applications (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:applications")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if not channel_id:
# 超级管理员申请供应商时需要提供channelId参数(通过请求体)
if role == "super_admin" and hasattr(req, 'channelId') and req.channelId:
try:
channel_id = uuid.UUID(req.channelId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
detail="无法获取渠道ID,超级管理员申请供应商时需要提供channelId参数"
)
# 检查供应商是否存在
@@ -1019,18 +1138,27 @@ async def list_provider_applications(
权限:view:applications (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:applications")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if not channel_id:
# 超级管理员可以查看所有申请,其他角色只能查看自己渠道的申请
if role == "super_admin":
query = select(ProviderApplication)
elif channel_id:
query = select(ProviderApplication).where(
ProviderApplication.channel_id == channel_id
)
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 构建查询
query = select(ProviderApplication).where(
ProviderApplication.channel_id == channel_id
)
# 构建查询(如果还没有构建)
if role != "super_admin" and not channel_id:
query = select(ProviderApplication).where(
ProviderApplication.channel_id == channel_id
)
if status_filter:
query = query.where(ProviderApplication.status == status_filter)
@@ -1076,9 +1204,13 @@ async def list_provider_access(
权限:view:resources (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if not channel_id:
# 超级管理员可以查看所有供应商,其他角色需要channel_id
if role == "super_admin":
channel_id = None # 超级管理员不需要channel_id限制
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
@@ -28,9 +28,10 @@ from models import (
ProviderModel,
Tenant,
Tool,
User,
)
from monitoring import system_monitor
from ..auth import create_access_token, ensure_user, get_password_hash, verify_password
from ..auth import create_access_token, ensure_user, get_password_hash, verify_password, require_auth
router = APIRouter(prefix="/api", tags=["frontend-integration"])
@@ -493,14 +494,47 @@ async def channel_agents_available(db: AsyncSession = Depends(get_db)) -> Dict[s
@router.get("/channel/tenants")
async def channel_tenants(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
tenants = (await db.execute(select(Tenant))).scalars().all()
async def channel_tenants(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
) -> Dict[str, Any]:
"""获取渠道下的租户列表(需要认证)"""
# 从token中获取渠道ID
role = principal.get("claims", {}).get("role", "")
channel_id_str = principal.get("claims", {}).get("channelId")
if not channel_id_str:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
try:
channel_id = uuid.UUID(channel_id_str)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
# 查询该渠道下的所有租户(role="user"的用户)
result = await db.execute(
select(User).where(
User.channel_id == channel_id,
User.role == "user"
)
)
tenants = result.scalars().all()
items = [
{
"id": str(t.id),
"name": t.name,
"subscriptionTier": t.subscription_tier,
"discount": t.discount,
"name": t.name or t.full_name,
"email": t.email,
"subscriptionTier": getattr(t, "subscription_tier", "free"),
"balance": float(getattr(t, "balance", 0)),
"creditLimit": float(getattr(t, "credit_limit", 0)),
"status": getattr(t, "status", "active"),
"channelId": str(t.channel_id) if t.channel_id else None,
}
for t in tenants
@@ -509,23 +543,72 @@ async def channel_tenants(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
@router.post("/channel/tenants/create")
async def channel_create_tenant(payload: Dict[str, Any], db: AsyncSession = Depends(get_db)) -> Dict[str, Any]:
async def channel_create_tenant(
payload: Dict[str, Any],
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
) -> Dict[str, Any]:
"""创建租户(需要认证,自动关联到当前渠道)"""
# 从token中获取渠道ID
role = principal.get("claims", {}).get("role", "")
channel_id_str = principal.get("claims", {}).get("channelId")
if not channel_id_str:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
try:
channel_id = uuid.UUID(channel_id_str)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
# 验证必需字段
if not payload.get("name"):
raise HTTPException(status_code=400, detail="name is required")
tenant = Tenant(
if not payload.get("email"):
raise HTTPException(status_code=400, detail="email is required")
if not payload.get("password"):
raise HTTPException(status_code=400, detail="password is required")
# 检查邮箱是否已存在
result = await db.execute(
select(User).where(User.email == payload["email"])
)
if result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="邮箱已被使用"
)
# 创建租户(User with role="user")
password_hash = get_password_hash(payload["password"])
tenant = User(
name=payload["name"],
email=payload["email"],
password_hash=password_hash,
hashed_password=password_hash,
username=payload["email"].split("@")[0],
full_name=payload["name"],
role="user",
channel_id=channel_id,
subscription_tier=payload.get("subscriptionTier", "free"),
discount=payload.get("discount", 0),
channel_id=uuid.UUID(payload["channelId"]) if payload.get("channelId") else None,
status="active",
balance=0,
credit_limit=0,
)
db.add(tenant)
await db.commit()
await db.refresh(tenant)
return {
"id": str(tenant.id),
"name": tenant.name,
"subscriptionTier": tenant.subscription_tier,
"discount": tenant.discount,
"name": tenant.name or tenant.full_name,
"email": tenant.email,
"subscriptionTier": getattr(tenant, "subscription_tier", "free"),
}
@@ -738,8 +821,15 @@ async def admin_login(payload: Dict[str, str], db: AsyncSession = Depends(get_db
user.is_admin = True
db.add(user)
await db.commit()
# 使用用户的实际角色,而不是硬编码 super_admin
token = create_access_token({"sub": str(user.id), "email": email, "role": user.role})
# 使用用户的实际角色,并在token中包含channelId(如果存在)
token_data = {
"sub": str(user.id),
"email": email,
"role": user.role
}
if user.channel_id:
token_data["channelId"] = str(user.channel_id)
token = create_access_token(token_data)
return {"token": token, "tokenType": "bearer", "email": email, "expiresIn": 60 * 60}
+2
View File
@@ -257,6 +257,7 @@ class TenantCreateRequest(BaseModel):
email: EmailStr
password: str
subscriptionTier: str = Field("free", pattern="^(free|pro|enterprise)$")
channelId: Optional[str] = None # 超级管理员创建租户时需要提供
class ResourceAgentAllocation(BaseModel):
@@ -315,6 +316,7 @@ class SetCreditLimitResponse(BaseModel):
class ResourceApplicationRequest(BaseModel):
channelId: Optional[str] = None # 超级管理员申请资源时需要提供
"""资源申请请求"""
type: str = Field(..., pattern="^(model|agent)$")
modelName: Optional[str] = None
+122
View File
@@ -0,0 +1,122 @@
#!/usr/bin/env python3
"""
创建超级管理员账号
直接通过数据库创建,不需要认证
"""
import sys
import os
import asyncio
# 添加services/mcp-server到路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'services', 'mcp-server'))
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select
from models import User
from config import settings
import bcrypt
import uuid
def get_password_hash(password: str) -> str:
"""加密密码(使用bcrypt)"""
password_bytes = password.encode('utf-8')
salt = bcrypt.gensalt()
hashed = bcrypt.hashpw(password_bytes, salt)
return hashed.decode('utf-8')
async def create_super_admin():
"""创建超级管理员"""
try:
# 准备数据库URL
database_url = settings.database_url
if database_url.startswith("postgresql://") and "+asyncpg" not in database_url:
database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
engine = create_async_engine(database_url, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async with AsyncSessionLocal() as session:
# 检查用户是否已存在
result = await session.execute(
select(User).where(User.email == "superadmin@taiji-ai.com")
)
existing_user = result.scalar_one_or_none()
if existing_user:
print(f" ⚠ 用户已存在: superadmin@taiji-ai.com")
# 更新密码和角色
password_hash = get_password_hash("Admin@123456")
existing_user.password_hash = password_hash
existing_user.hashed_password = password_hash
existing_user.role = "super_admin"
existing_user.is_admin = True
existing_user.is_active = True
existing_user.name = "超级管理员"
existing_user.username = "superadmin"
existing_user.full_name = "超级管理员"
await session.commit()
print(f" ✓ 更新成功: superadmin@taiji-ai.com")
return True
# 创建新用户
password_hash = get_password_hash("Admin@123456")
user = User(
name="超级管理员",
email="superadmin@taiji-ai.com",
password_hash=password_hash,
hashed_password=password_hash,
username="superadmin",
full_name="超级管理员",
role="super_admin",
is_active=True,
is_admin=True,
status="active",
balance=0,
credit_limit=0,
)
session.add(user)
await session.commit()
await session.refresh(user)
print(f" ✓ 创建成功: superadmin@taiji-ai.com (角色: super_admin)")
return True
except Exception as e:
print(f" ✗ 创建失败: {e}")
import traceback
traceback.print_exc()
return False
finally:
if engine:
await engine.dispose()
async def main():
"""主函数"""
print("="*80)
print("创建超级管理员账号")
print("="*80)
print()
print("创建超级管理员...")
success = await create_super_admin()
if success:
print("\n" + "="*80)
print("✓ 超级管理员创建成功!")
print("="*80)
print("\n账号信息:")
print(" 邮箱: superadmin@taiji-ai.com")
print(" 密码: Admin@123456")
print(" 角色: super_admin")
sys.exit(0)
else:
print("\n" + "="*80)
print("✗ 超级管理员创建失败!")
print("="*80)
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
+4
View File
@@ -88,6 +88,10 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]:
try:
yield session
except Exception as e:
# HTTPException是正常的业务异常,不应该被当作数据库错误
from fastapi import HTTPException
if isinstance(e, HTTPException):
raise
logger.error(f"数据库会话错误: {e}")
await session.rollback()
raise
+830
View File
@@ -0,0 +1,830 @@
#!/usr/bin/env python3
"""
测试管理平台端和渠道端的所有接口
使用超级管理员账号进行测试
"""
import requests
import json
import sys
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
BASE_URL = "http://localhost:8002"
# 超级管理员账号
SUPER_ADMIN = {
"email": "superadmin@taiji-ai.com",
"password": "Admin@123456",
"role": "super_admin"
}
@dataclass
class TestResult:
"""测试结果"""
name: str
success: bool
message: str
status_code: int = 0
response_data: Any = None
class APITester:
"""API测试类"""
def __init__(self, base_url: str = BASE_URL):
self.base_url = base_url
self.token: Optional[str] = None
self.test_results: List[TestResult] = []
self.created_resources: Dict[str, Any] = {} # 存储创建的资源ID,用于清理
def login(self, email: str, password: str, role: str) -> bool:
"""登录并获取token"""
try:
resp = requests.post(
f"{self.base_url}/api/auth/login",
json={"email": email, "password": password, "role": role},
timeout=10
)
if resp.status_code == 200:
data = resp.json()
self.token = data.get("data", {}).get("token")
if self.token:
return True
return False
except Exception as e:
print(f"登录失败: {e}")
return False
def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
"""发送HTTP请求"""
headers = kwargs.pop("headers", {})
if self.token:
headers["Authorization"] = f"Bearer {self.token}"
url = f"{self.base_url}{endpoint}"
if method == "GET":
return requests.get(url, headers=headers, timeout=10, **kwargs)
elif method == "POST":
return requests.post(url, headers=headers, timeout=10, **kwargs)
elif method == "PUT":
return requests.put(url, headers=headers, timeout=10, **kwargs)
elif method == "DELETE":
return requests.delete(url, headers=headers, timeout=10, **kwargs)
else:
raise ValueError(f"不支持的HTTP方法: {method}")
def test_api(self, name: str, method: str, endpoint: str,
expected_status: int = 200, **kwargs) -> TestResult:
"""测试API接口"""
try:
resp = self._make_request(method, endpoint, **kwargs)
success = resp.status_code == expected_status
try:
response_data = resp.json() if resp.text else None
except:
response_data = resp.text
message = f"状态码: {resp.status_code}"
if not success:
error_msg = response_data.get("detail", str(response_data)) if isinstance(response_data, dict) else str(response_data)
message = f"失败 - {message}, 错误: {error_msg[:200]}"
result = TestResult(
name=name,
success=success,
message=message,
status_code=resp.status_code,
response_data=response_data
)
self.test_results.append(result)
return result
except Exception as e:
result = TestResult(
name=name,
success=False,
message=f"请求异常: {str(e)}",
status_code=0
)
self.test_results.append(result)
return result
def print_result(self, result: TestResult):
"""打印测试结果"""
icon = "✓" if result.success else "✗"
print(f" {icon} {result.name}: {result.message}")
# ==================== 管理平台端接口测试 ====================
def test_admin_dashboard_stats(self):
"""1. 获取平台统计"""
result = self.test_api(
"获取平台统计",
"GET",
"/api/admin/dashboard/stats"
)
self.print_result(result)
return result
def test_admin_admins_list(self):
"""2. 获取管理员列表"""
result = self.test_api(
"获取管理员列表",
"GET",
"/api/admin/admins"
)
self.print_result(result)
return result
def test_admin_create_admin(self):
"""3. 创建管理员"""
result = self.test_api(
"创建管理员",
"POST",
"/api/admin/admins/create",
json={
"name": "测试计费管理员",
"email": f"test-billing-{datetime.now().timestamp()}@test.com",
"password": "Test@123456",
"role": "billing_admin",
"channelId": self.created_resources.get("channel_id")
}
)
self.print_result(result)
if result.success and result.response_data:
admin_id = result.response_data.get("data", {}).get("id")
if admin_id:
if "admin_ids" not in self.created_resources:
self.created_resources["admin_ids"] = []
self.created_resources["admin_ids"].append(admin_id)
return result
def test_admin_channels_list(self):
"""5. 获取渠道列表"""
result = self.test_api(
"获取渠道列表",
"GET",
"/api/admin/channels"
)
self.print_result(result)
# 保存第一个渠道ID用于后续测试
if result.success and result.response_data:
channels = result.response_data.get("data", {}).get("channels", [])
if channels and not self.created_resources.get("channel_id"):
self.created_resources["channel_id"] = channels[0].get("id")
return result
def test_admin_create_channel(self):
"""6. 创建渠道"""
result = self.test_api(
"创建渠道",
"POST",
"/api/admin/channels/create",
json={
"name": f"测试渠道-{datetime.now().strftime('%Y%m%d%H%M%S')}",
"email": f"test-channel-{datetime.now().timestamp()}@test.com",
"password": "Channel@123456",
"commissionRate": 10.0
}
)
self.print_result(result)
if result.success and result.response_data:
channel_id = result.response_data.get("data", {}).get("id")
if channel_id:
self.created_resources["channel_id"] = channel_id
return result
def test_admin_update_channel(self):
"""7. 更新渠道信息"""
channel_id = self.created_resources.get("channel_id")
if not channel_id:
result = TestResult("更新渠道信息", False, "跳过: 没有可用的渠道ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"更新渠道信息",
"PUT",
f"/api/admin/channels/{channel_id}",
json={
"name": "测试渠道(已更新)",
"commissionRate": 12.0
}
)
self.print_result(result)
return result
def test_admin_get_channel_resources(self):
"""9. 获取渠道资源分配"""
channel_id = self.created_resources.get("channel_id")
if not channel_id:
result = TestResult("获取渠道资源分配", False, "跳过: 没有可用的渠道ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"获取渠道资源分配",
"GET",
f"/api/admin/channels/{channel_id}/resources"
)
self.print_result(result)
return result
def test_admin_update_channel_resources(self):
"""10. 统一管理渠道资源"""
channel_id = self.created_resources.get("channel_id")
if not channel_id:
result = TestResult("统一管理渠道资源", False, "跳过: 没有可用的渠道ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"统一管理渠道资源",
"PUT",
f"/api/admin/channels/{channel_id}/resources",
json={
"models": [],
"agents": [],
"customAgentResources": {
"cpu": 2.0,
"memory": 4.0
},
"channelCredit": 10000.0
}
)
self.print_result(result)
return result
def test_admin_applications_list(self):
"""11. 获取所有申请"""
result = self.test_api(
"获取所有申请",
"GET",
"/api/admin/channels/applications"
)
self.print_result(result)
return result
def test_admin_resources_models(self):
"""13. 获取所有模型供应商"""
result = self.test_api(
"获取所有模型供应商",
"GET",
"/api/admin/resources/models"
)
self.print_result(result)
# 保存第一个供应商ID用于后续测试
if result.success and result.response_data:
providers = result.response_data.get("data", {}).get("providers", [])
if providers and not self.created_resources.get("provider_id"):
self.created_resources["provider_id"] = providers[0].get("id")
return result
def test_admin_resources_agents(self):
"""14. 获取所有Agent资源"""
result = self.test_api(
"获取所有Agent资源",
"GET",
"/api/admin/resources/agents"
)
self.print_result(result)
# 保存第一个Agent ID用于后续测试
if result.success and result.response_data:
agents = result.response_data.get("data", {}).get("agents", [])
if agents and not self.created_resources.get("agent_id"):
self.created_resources["agent_id"] = agents[0].get("id")
return result
def test_admin_update_agent_config(self):
"""16. 更新Agent资源配置"""
agent_id = self.created_resources.get("agent_id")
if not agent_id:
result = TestResult("更新Agent资源配置", False, "跳过: 没有可用的Agent ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"更新Agent资源配置",
"PUT",
f"/api/admin/resources/agents/{agent_id}/config",
json={
"cpu": 2.0,
"memory": 4.0,
"maxInstances": 10
}
)
self.print_result(result)
return result
def test_admin_monitoring_agents(self):
"""17. 监控Agent健康状态"""
result = self.test_api(
"监控Agent健康状态",
"GET",
"/api/admin/monitoring/agents"
)
self.print_result(result)
return result
def test_admin_billing_overview(self):
"""18. 获取三维度计费统计"""
end_time = datetime.now()
start_time = end_time - timedelta(days=30)
result = self.test_api(
"获取三维度计费统计",
"GET",
f"/api/admin/billing/overview?startTime={start_time.isoformat()}&endTime={end_time.isoformat()}"
)
self.print_result(result)
return result
def test_admin_providers_applications(self):
"""19. 获取供应商申请列表(管理员视图)"""
result = self.test_api(
"获取供应商申请列表(管理员视图)",
"GET",
"/api/admin/providers/applications"
)
self.print_result(result)
return result
def test_admin_providers_access(self):
"""21. 获取所有渠道供应商授权列表"""
result = self.test_api(
"获取所有渠道供应商授权列表",
"GET",
"/api/admin/providers/access"
)
self.print_result(result)
return result
def test_admin_roles(self):
"""25. 获取可用角色列表"""
result = self.test_api(
"获取可用角色列表",
"GET",
"/api/admin/roles"
)
self.print_result(result)
return result
def test_admin_providers_stats(self):
"""26. 供应商统计(展示用)"""
result = self.test_api(
"供应商统计(展示用)",
"GET",
"/api/admin/providers/stats"
)
self.print_result(result)
return result
def test_admin_channels_backend_stats(self):
"""27. 后台简易渠道统计"""
result = self.test_api(
"后台简易渠道统计",
"GET",
"/api/admin/channels/backend/stats"
)
self.print_result(result)
return result
def test_admin_channel_admins(self):
"""24. 获取渠道管理员列表"""
channel_id = self.created_resources.get("channel_id")
if not channel_id:
result = TestResult("获取渠道管理员列表", False, "跳过: 没有可用的渠道ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"获取渠道管理员列表",
"GET",
f"/api/admin/channels/{channel_id}/admins"
)
self.print_result(result)
return result
# ==================== 渠道端接口测试 ====================
def test_channel_tenants_list(self):
"""1. 获取租户列表"""
result = self.test_api(
"获取租户列表",
"GET",
"/api/channel/tenants"
)
self.print_result(result)
# 保存第一个租户ID用于后续测试
if result.success and result.response_data:
tenants = result.response_data.get("data", {}).get("tenants", [])
if tenants and not self.created_resources.get("tenant_id"):
self.created_resources["tenant_id"] = tenants[0].get("id")
return result
def test_channel_create_tenant(self):
"""2. 创建租户"""
channel_id = self.created_resources.get("channel_id")
if not channel_id:
result = TestResult("创建租户", False, "跳过: 没有可用的渠道ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"创建租户",
"POST",
"/api/channel/tenants/create",
json={
"name": f"测试租户-{datetime.now().strftime('%Y%m%d%H%M%S')}",
"email": f"test-tenant-{datetime.now().timestamp()}@test.com",
"password": "Tenant@123456",
"subscriptionTier": "pro",
"channelId": str(channel_id) # 超级管理员需要提供channelId
}
)
self.print_result(result)
if result.success and result.response_data:
tenant_id = result.response_data.get("data", {}).get("id")
if tenant_id:
self.created_resources["tenant_id"] = tenant_id
return result
def test_channel_assign_tenant_resources(self):
"""3. 分配租户资源"""
tenant_id = self.created_resources.get("tenant_id")
if not tenant_id:
result = TestResult("分配租户资源", False, "跳过: 没有可用的租户ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"分配租户资源",
"PUT",
f"/api/channel/tenants/{tenant_id}/resources",
json={
"agents": [],
"models": [],
"customAgentResources": {
"cpu": 1.0,
"memory": 2.0
}
}
)
self.print_result(result)
return result
def test_channel_update_tenant_billing(self):
"""4. 更新租户计费设置"""
tenant_id = self.created_resources.get("tenant_id")
if not tenant_id:
result = TestResult("更新租户计费设置", False, "跳过: 没有可用的租户ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"更新租户计费设置",
"PUT",
f"/api/channel/tenants/{tenant_id}/billing",
json={
"subscriptionTier": "enterprise",
"discount": 10.0
}
)
self.print_result(result)
return result
def test_channel_tenant_recharge(self):
"""5. 为租户充值"""
tenant_id = self.created_resources.get("tenant_id")
if not tenant_id:
result = TestResult("为租户充值", False, "跳过: 没有可用的租户ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"为租户充值",
"POST",
f"/api/channel/tenants/{tenant_id}/recharge",
json={
"amount": 1000.0
}
)
self.print_result(result)
return result
def test_channel_tenant_credit(self):
"""6. 设置租户授信额度"""
tenant_id = self.created_resources.get("tenant_id")
if not tenant_id:
result = TestResult("设置租户授信额度", False, "跳过: 没有可用的租户ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"设置租户授信额度",
"PUT",
f"/api/channel/tenants/{tenant_id}/credit",
json={
"creditLimit": 5000.0
}
)
self.print_result(result)
return result
def test_channel_update_tenant_status(self):
"""8. 更新租户状态"""
tenant_id = self.created_resources.get("tenant_id")
if not tenant_id:
result = TestResult("更新租户状态", False, "跳过: 没有可用的租户ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"更新租户状态",
"PUT",
f"/api/channel/tenants/{tenant_id}/status",
json={
"status": "active"
}
)
self.print_result(result)
return result
def test_channel_update_tenant_permissions(self):
"""9. 更新租户权限"""
tenant_id = self.created_resources.get("tenant_id")
if not tenant_id:
result = TestResult("更新租户权限", False, "跳过: 没有可用的租户ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"更新租户权限",
"PUT",
f"/api/channel/tenants/{tenant_id}/permissions",
json={
"permissions": ["use:platform_agents", "read:billing"]
}
)
self.print_result(result)
return result
def test_channel_admins_list(self):
"""11. 获取渠道下的管理员列表"""
result = self.test_api(
"获取渠道下的管理员列表",
"GET",
"/api/channel/admins"
)
self.print_result(result)
return result
def test_channel_create_admin(self):
"""10. 创建渠道下的管理员"""
channel_id = self.created_resources.get("channel_id")
if not channel_id:
result = TestResult("创建渠道下的管理员", False, "跳过: 没有可用的渠道ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"创建渠道下的管理员",
"POST",
"/api/channel/admins/create",
json={
"name": "测试计费管理员",
"email": f"test-channel-billing-{datetime.now().timestamp()}@test.com",
"password": "Test@123456",
"role": "billing_admin",
"channelId": str(channel_id) # 超级管理员需要提供channelId
}
)
self.print_result(result)
return result
def test_channel_apply_resources(self):
"""12. 申请资源"""
channel_id = self.created_resources.get("channel_id")
if not channel_id:
result = TestResult("申请资源", False, "跳过: 没有可用的渠道ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"申请资源",
"POST",
"/api/channel/resources/apply",
json={
"type": "model",
"modelName": "gpt-4",
"rpm": 100,
"tpm": 100000,
"reason": "测试申请资源",
"channelId": str(channel_id) # 超级管理员需要提供channelId
}
)
self.print_result(result)
return result
def test_channel_billing_stats(self):
"""12. 获取渠道计费统计"""
end_time = datetime.now()
start_time = end_time - timedelta(days=30)
result = self.test_api(
"获取渠道计费统计",
"GET",
f"/api/channel/billing/stats?startTime={start_time.isoformat()}&endTime={end_time.isoformat()}"
)
self.print_result(result)
return result
def test_channel_providers_list(self):
"""13. 获取可用供应商列表"""
result = self.test_api(
"获取可用供应商列表",
"GET",
"/api/channel/providers"
)
self.print_result(result)
# 保存第一个供应商ID用于后续测试
if result.success and result.response_data:
providers = result.response_data.get("data", {}).get("providers", [])
if providers and not self.created_resources.get("provider_id"):
self.created_resources["provider_id"] = providers[0].get("id")
return result
def test_channel_providers_apply(self):
"""14. 申请使用供应商"""
provider_id = self.created_resources.get("provider_id")
if not provider_id:
result = TestResult("申请使用供应商", False, "跳过: 没有可用的供应商ID")
self.test_results.append(result)
self.print_result(result)
return result
result = self.test_api(
"申请使用供应商",
"POST",
"/api/channel/providers/apply",
json={
"providerId": provider_id,
"requestedRpm": 1000,
"requestedTpm": 50000,
"reason": "测试申请使用供应商"
}
)
self.print_result(result)
return result
def test_channel_providers_applications(self):
"""15. 获取供应商申请列表"""
result = self.test_api(
"获取供应商申请列表",
"GET",
"/api/channel/providers/applications"
)
self.print_result(result)
return result
def test_channel_providers_access(self):
"""16. 获取已授权供应商列表"""
result = self.test_api(
"获取已授权供应商列表",
"GET",
"/api/channel/providers/access"
)
self.print_result(result)
return result
def run_all_tests(self):
"""运行所有测试"""
print("="*80)
print("管理平台端和渠道端接口测试")
print("="*80)
print()
# 登录
print("步骤1: 尝试登录超级管理员账号...")
if not self.login(SUPER_ADMIN["email"], SUPER_ADMIN["password"], SUPER_ADMIN["role"]):
print(" ⚠ 超级管理员登录失败,尝试默认管理员账号...")
# 尝试默认管理员账号
if not self.login("admin@taiji-ai.com", "admin123", "super_admin"):
print(" ✗ 所有管理员账号登录失败,无法继续测试")
print(" 提示: 请先运行 test_cre_admins.py 创建管理员账号")
return False
print(" ✓ 使用默认管理员账号登录成功")
else:
print(" ✓ 超级管理员登录成功")
print()
# 管理平台端接口测试
print("="*80)
print("管理平台端接口测试")
print("="*80)
print()
self.test_admin_dashboard_stats()
self.test_admin_admins_list()
self.test_admin_channels_list()
self.test_admin_create_channel()
self.test_admin_create_admin()
self.test_admin_update_channel()
self.test_admin_get_channel_resources()
self.test_admin_update_channel_resources()
self.test_admin_applications_list()
self.test_admin_resources_models()
self.test_admin_resources_agents()
self.test_admin_update_agent_config()
self.test_admin_monitoring_agents()
self.test_admin_billing_overview()
self.test_admin_providers_applications()
self.test_admin_providers_access()
self.test_admin_channel_admins()
self.test_admin_roles()
self.test_admin_providers_stats()
self.test_admin_channels_backend_stats()
print()
# 渠道端接口测试
print("="*80)
print("渠道端接口测试")
print("="*80)
print()
self.test_channel_tenants_list()
self.test_channel_create_tenant()
self.test_channel_assign_tenant_resources()
self.test_channel_update_tenant_billing()
self.test_channel_tenant_recharge()
self.test_channel_tenant_credit()
self.test_channel_update_tenant_status()
self.test_channel_update_tenant_permissions()
self.test_channel_admins_list()
self.test_channel_create_admin()
self.test_channel_apply_resources()
self.test_channel_billing_stats()
self.test_channel_providers_list()
self.test_channel_providers_apply()
self.test_channel_providers_applications()
self.test_channel_providers_access()
# 汇总结果
print()
print("="*80)
print("测试结果汇总")
print("="*80)
print()
total = len(self.test_results)
passed = sum(1 for r in self.test_results if r.success)
failed = total - passed
print(f"总计: {total} 个接口")
print(f"通过: {passed} 个")
print(f"失败: {failed} 个")
print()
if failed > 0:
print("失败的接口:")
print("-" * 80)
for result in self.test_results:
if not result.success:
print(f" ✗ {result.name}: {result.message}")
print()
return failed == 0
def main():
"""主函数"""
tester = APITester()
success = tester.run_all_tests()
if success:
print("🎉 所有接口测试通过!")
sys.exit(0)
else:
print("⚠️ 有接口测试失败,请检查错误信息")
sys.exit(1)
if __name__ == "__main__":
main()