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

更新模型计费

Browse Source
This commit is contained in:
zhanggangyong
2026-01-08 15:03:52 +00:00
parent c94367994e
commit 625afdf441
12 changed files with 5694 additions and 730 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/mcp-server/app/auth.py
+76 -3
View File
@@ -16,7 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from config import settings
from database import get_db
from models import APIKey, User
from models import APIKey, User, TokenBlacklist
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
bearer_scheme = HTTPBearer(auto_error=False)
@@ -32,8 +32,12 @@ def get_password_hash(password: str) -> str:
def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=settings.jwt_expire_minutes))
to_encode.update({"exp": expire})
now = datetime.utcnow()
expire = now + (expires_delta or timedelta(minutes=settings.jwt_expire_minutes))
to_encode.update({
"exp": expire,
"iat": now, # 添加签发时间，用于登出验证
})
encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.jwt_algorithm)
return encoded_jwt
@@ -89,6 +93,59 @@ async def _check_api_key(key: str, db: AsyncSession) -> Optional[APIKey]:
return None
async def _is_user_logged_out(user_id: str, token_iat: Optional[int], db: AsyncSession) -> bool:
"""
检查用户是否已登出
通过检查 token_blacklist 表中是否存在该用户的登出记录，
且登出时间在 token 签发时间之后。
Args:
user_id: 用户ID
token_iat: Token 签发时间戳 (iat claim)
db: 数据库会话
Returns:
True 如果用户已登出，False 否则
"""
try:
# 查询该用户的登出记录
result = await db.execute(
select(TokenBlacklist).where(
TokenBlacklist.user_id == user_id,
TokenBlacklist.reason == "logout",
TokenBlacklist.expires_at > datetime.utcnow() # 只查询未过期的记录
).order_by(TokenBlacklist.created_at.desc()).limit(1)
)
blacklist_entry = result.scalar_one_or_none()
if not blacklist_entry:
return False
# 如果有登出记录，检查 token 签发时间
if token_iat:
token_issued_at = datetime.utcfromtimestamp(token_iat)
# 从 token_jti 中提取登出时间戳
# 格式: logout_{user_id}_{timestamp}
try:
logout_timestamp = float(blacklist_entry.token_jti.split("_")[-1])
logout_time = datetime.utcfromtimestamp(logout_timestamp)
# 如果 token 是在登出之前签发的，则认为已登出
if token_issued_at < logout_time:
return True
except (ValueError, IndexError):
# 无法解析登出时间，保守起见认为已登出
return True
else:
# 没有 iat claim，保守起见认为已登出
return True
return False
except Exception:
# 查询失败时不阻止认证
return False
async def require_auth(
request: Request,
credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
@@ -106,6 +163,7 @@ async def require_auth(
"/api/channel/auth/login",
"/api/admin/auth/login",
"/api/providers/auth/login",
"/api/auth/login", # 添加统一登录接口
"/agents/templates", # 模板列表公开访问
}
# 允许公开路径和非 API/agents 路径
@@ -132,8 +190,15 @@ async def require_auth(
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.jwt_algorithm])
user_id: str | None = payload.get("sub")
email: str | None = payload.get("email")
token_iat: int | None = payload.get("iat")
if user_id is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload")
# 检查用户是否已登出
if await _is_user_logged_out(user_id, token_iat, db):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token已失效，请重新登录")
request.state.principal = {"type": "jwt", "user_id": user_id, "email": email, "claims": payload}
return request.state.principal
except JWTError:
@@ -155,6 +220,7 @@ async def authenticate_request(request: Request, db: AsyncSession) -> Optional[D
"/api/channel/auth/login",
"/api/admin/auth/login",
"/api/providers/auth/login",
"/api/auth/login", # 添加统一登录接口
"/agents/templates", # 模板列表公开访问
}
# 允许公开路径
@@ -181,8 +247,15 @@ async def authenticate_request(request: Request, db: AsyncSession) -> Optional[D
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.jwt_algorithm])
user_id: str | None = payload.get("sub")
email: str | None = payload.get("email")
token_iat: int | None = payload.get("iat")
if user_id is None:
return None
# 检查用户是否已登出
if await _is_user_logged_out(user_id, token_iat, db):
return None
return {"type": "jwt", "user_id": user_id, "email": email, "claims": payload}
except JWTError:
return None
services/mcp-server/app/routes/channel.py
+89 -1
View File
@@ -325,8 +325,9 @@ async def allocate_tenant_resources(
)
)
# 分配Agent资源
# 分配平台 Agent 资源（更新 PlatformAgentQuota）
for agent_alloc in req.agents:
# 1. 创建 ResourceAllocation 记录（兼容旧逻辑）
allocation = ResourceAllocation(
target_id=tenant_id,
target_type="tenant",
@@ -335,6 +336,93 @@ async def allocate_tenant_resources(
quantity=agent_alloc.quantity,
)
db.add(allocation)
# 2. 更新 PlatformAgentQuota（平台 Agent 配额管理）
# agentId 在这里是模板名称（如 echo_agent）
template_name = agent_alloc.agentId
# 获取渠道的平台 Agent 配额
channel_quota_result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == channel_id,
PlatformAgentQuota.target_type == "channel",
PlatformAgentQuota.template_name == template_name
)
)
)
channel_quota = channel_quota_result.scalar_one_or_none()
if channel_quota:
# 计算已分配给该渠道其他租户的配额
other_tenants_quota_result = await db.execute(
select(func.sum(PlatformAgentQuota.pod_quota).label("total"))
.select_from(PlatformAgentQuota)
.join(User, PlatformAgentQuota.target_id == User.id)
.where(
and_(
User.channel_id == channel_id,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.template_name == template_name,
PlatformAgentQuota.target_id != tenant_id
)
)
)
other_quota = other_tenants_quota_result.scalar() or 0
# 检查是否超过渠道配额
remaining = channel_quota.pod_quota - other_quota
if agent_alloc.quantity > remaining:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"平台 Agent '{template_name}' 配额超出渠道剩余配额。渠道剩余: {remaining}，请求: {agent_alloc.quantity}"
)
# 查找或创建租户的平台 Agent 配额记录
tenant_quota_result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == tenant_id,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.template_name == template_name
)
)
)
tenant_quota = tenant_quota_result.scalar_one_or_none()
# 计算配额变化量（用于更新渠道的 pod_used）
old_tenant_quota = tenant_quota.pod_quota if tenant_quota else 0
quota_delta = agent_alloc.quantity - old_tenant_quota
if tenant_quota:
# 更新现有配额
tenant_quota.pod_quota = agent_alloc.quantity
tenant_quota.allocated_at = datetime.utcnow()
else:
# 创建新配额记录
tenant_quota = PlatformAgentQuota(
target_id=tenant_id,
target_type="tenant",
template_name=template_name,
pod_quota=agent_alloc.quantity,
pod_used=0,
allocated_by=None, # 渠道管理员分配
allocated_at=datetime.utcnow(),
)
db.add(tenant_quota)
# 更新渠道的 pod_used（分配给租户的配额视为渠道已使用的配额）
channel_quota.pod_used = (channel_quota.pod_used or 0) + quota_delta
logger.info(
f"分配平台 Agent 配额: template={template_name}, tenant={tenant_id}, "
f"quota={agent_alloc.quantity}, channel_pod_used={channel_quota.pod_used}"
)
else:
# 渠道没有该平台 Agent 的配额，记录警告但不阻止操作
logger.warning(
f"渠道没有平台 Agent '{template_name}' 的配额，仅创建 ResourceAllocation 记录"
)
# 分配模型资源（集成 LiteLLM）
# 获取渠道信息（用于 LiteLLM 集成）
services/mcp-server/app/routes/user.py
+73 -4
View File
@@ -1095,6 +1095,58 @@ async def recharge_balance(
)
def _parse_datetime(dt_str: str) -> datetime:
"""
解析日期时间字符串，支持多种格式：
- YYYY-MM-DD
- YYYY-MM-DDTHH:MM:SS
- YYYY-MM-DDTHH:MM:SSZ
- YYYY-MM-DDTHH:MM:SS+00:00
返回的 datetime 对象不包含时区信息（naive datetime），
以便与数据库中的 TIMESTAMP WITHOUT TIME ZONE 兼容。
"""
if not dt_str:
raise ValueError("日期时间字符串不能为空")
dt_str = dt_str.strip()
# 移除末尾的 Z 并替换为 +00:00
if dt_str.endswith("Z"):
dt_str = dt_str[:-1] + "+00:00"
result = None
try:
# 尝试解析完整的 ISO 格式
result = datetime.fromisoformat(dt_str)
except ValueError:
pass
if result is None:
# 尝试解析简单日期格式 YYYY-MM-DD
try:
result = datetime.strptime(dt_str, "%Y-%m-%d")
except ValueError:
pass
if result is None:
# 尝试解析带时间的格式
try:
result = datetime.strptime(dt_str, "%Y-%m-%dT%H:%M:%S")
except ValueError:
pass
if result is None:
raise ValueError(f"无法解析日期时间格式: {dt_str}")
# 移除时区信息，返回 naive datetime
if result.tzinfo is not None:
result = result.replace(tzinfo=None)
return result
@router.get("/billing/history", response_model=SuccessResponse)
async def get_billing_history(
startTime: str = Query(...),
@@ -1110,12 +1162,23 @@ async def get_billing_history(
):
"""
获取计费历史记录
支持的时间格式：
- YYYY-MM-DD (如: 2026-01-01)
- YYYY-MM-DDTHH:MM:SS (如: 2026-01-01T00:00:00)
- YYYY-MM-DDTHH:MM:SSZ (如: 2026-01-01T00:00:00Z)
"""
user_id = principal.get("user_id")
# 解析时间
start_dt = datetime.fromisoformat(startTime.replace("Z", "+00:00"))
end_dt = datetime.fromisoformat(endTime.replace("Z", "+00:00"))
try:
start_dt = _parse_datetime(startTime)
end_dt = _parse_datetime(endTime)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"时间格式错误: {str(e)}。支持的格式: YYYY-MM-DD 或 YYYY-MM-DDTHH:MM:SSZ"
)
# 构建查询
query = select(BillingRecord).where(
@@ -2060,8 +2123,14 @@ async def get_my_agent_billing_history(
user_id = principal.get("user_id")
# 解析时间
start_dt = datetime.fromisoformat(startTime.replace("Z", "+00:00"))
end_dt = datetime.fromisoformat(endTime.replace("Z", "+00:00"))
try:
start_dt = _parse_datetime(startTime)
end_dt = _parse_datetime(endTime)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"时间格式错误: {str(e)}。支持的格式: YYYY-MM-DD 或 YYYY-MM-DDTHH:MM:SSZ"
)
# 构建查询
conditions = [
services/mcp-server/monitoring.py
+23 -12
View File
@@ -31,8 +31,11 @@ class SystemMonitor:
self._service_endpoints = {
"mcp_server": "http://localhost:8000/health", # 本服务
"data_ingestion": "http://data-ingestion:8000/health", # Docker 内部网络
"api_gateway": "http://api-gateway:80/health", # Nginx
"model_gateway": "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/health", # LiteLLM 模型网关
"agent_manager": "http://127.0.0.1:8000/health", # Agent Manager 服务（本地）
}
# LiteLLM API Key
self._litellm_api_key = "sk-taiji-prod-2026"
async def get_system_health(self) -> Dict[str, Any]:
"""获取系统健康状态，包括三个微服务的状态"""
@@ -44,13 +47,13 @@ class SystemMonitor:
"services": {}
}
# 检查数据库
# 检查数据库（内部使用，不对外暴露）
try:
async with AsyncSessionLocal() as session:
await session.execute(text("SELECT 1"))
health["services"]["database"] = {"status": "healthy", "latency": 0}
# 数据库健康检查成功，但不添加到 services 中
except Exception as e:
health["services"]["database"] = {"status": "error", "latency": 0, "error": str(e)}
# 数据库不健康会影响整体状态
health["status"] = "degraded"
health["score"] -= 25
@@ -72,13 +75,14 @@ class SystemMonitor:
async def _check_microservices_health(self) -> Dict[str, Dict[str, Any]]:
"""
检查三个微服务的健康状态
检查四个微服务的健康状态
返回格式：
{
"mcp_server": { "status": "healthy", "latency": 45 },
"data_ingestion": { "status": "healthy", "latency": 32 },
"api_gateway": { "status": "healthy", "latency": 28 }
"model_gateway": { "status": "healthy", "latency": 28 },
"agent_manager": { "status": "healthy", "latency": 15 }
}
状态值：
@@ -88,12 +92,12 @@ class SystemMonitor:
"""
results = {}
async def check_service(name: str, url: str) -> Dict[str, Any]:
async def check_service(name: str, url: str, headers: Dict[str, str] = None) -> Dict[str, Any]:
"""检查单个服务的健康状态"""
start_time = time.time()
try:
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(url)
response = await client.get(url, headers=headers)
latency = int((time.time() - start_time) * 1000)
if response.status_code == 200:
@@ -119,10 +123,17 @@ class SystemMonitor:
self._service_endpoints["data_ingestion"]
)
# 检查 api_gateway (Nginx)
results["api_gateway"] = await check_service(
"api_gateway",
self._service_endpoints["api_gateway"]
# 检查 model_gateway (LiteLLM) - 需要带上 API Key
results["model_gateway"] = await check_service(
"model_gateway",
self._service_endpoints["model_gateway"],
headers={"Authorization": f"Bearer {self._litellm_api_key}"}
)
# 检查 agent_manager 服务
results["agent_manager"] = await check_service(
"agent_manager",
self._service_endpoints["agent_manager"]
)
return results