b89d5ddee9
Some checks failed
continuous-integration/drone/push Build is failing
- Add Vue 3 frontend with Element Plus - Implement login, dashboard, tenant management - Add app configuration, logs viewer, stats pages - Add user management for admins - Update Drone CI to build and deploy frontend - Frontend ports: 3001 (test), 4001 (prod)
241 lines
7.7 KiB
Python
241 lines
7.7 KiB
Python
"""租户应用配置路由"""
|
||
import json
|
||
import secrets
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from pydantic import BaseModel
|
||
from typing import Optional, List
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..database import get_db
|
||
from ..models.tenant_app import TenantApp
|
||
from .auth import get_current_user, require_operator
|
||
from ..models.user import User
|
||
from ..services.crypto import encrypt_config, decrypt_config
|
||
|
||
router = APIRouter(prefix="/tenant-apps", tags=["应用配置"])
|
||
|
||
|
||
# Schemas
|
||
|
||
class TenantAppCreate(BaseModel):
|
||
tenant_id: str
|
||
app_code: str = "tools"
|
||
app_name: Optional[str] = None
|
||
wechat_corp_id: Optional[str] = None
|
||
wechat_agent_id: Optional[str] = None
|
||
wechat_secret: Optional[str] = None # 明文，存储时加密
|
||
token_secret: Optional[str] = None # 如果不传则自动生成
|
||
token_required: bool = False
|
||
allowed_origins: Optional[List[str]] = None
|
||
allowed_tools: Optional[List[str]] = None
|
||
|
||
|
||
class TenantAppUpdate(BaseModel):
|
||
app_name: Optional[str] = None
|
||
wechat_corp_id: Optional[str] = None
|
||
wechat_agent_id: Optional[str] = None
|
||
wechat_secret: Optional[str] = None
|
||
token_secret: Optional[str] = None
|
||
token_required: Optional[bool] = None
|
||
allowed_origins: Optional[List[str]] = None
|
||
allowed_tools: Optional[List[str]] = None
|
||
status: Optional[int] = None
|
||
|
||
|
||
# API Endpoints
|
||
|
||
@router.get("")
|
||
async def list_tenant_apps(
|
||
page: int = Query(1, ge=1),
|
||
size: int = Query(20, ge=1, le=100),
|
||
tenant_id: Optional[str] = None,
|
||
app_code: Optional[str] = None,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取应用配置列表"""
|
||
query = db.query(TenantApp)
|
||
|
||
if tenant_id:
|
||
query = query.filter(TenantApp.tenant_id == tenant_id)
|
||
if app_code:
|
||
query = query.filter(TenantApp.app_code == app_code)
|
||
|
||
total = query.count()
|
||
apps = query.order_by(TenantApp.id.desc()).offset((page - 1) * size).limit(size).all()
|
||
|
||
return {
|
||
"total": total,
|
||
"page": page,
|
||
"size": size,
|
||
"items": [format_tenant_app(app, mask_secret=True) for app in apps]
|
||
}
|
||
|
||
|
||
@router.get("/{app_id}")
|
||
async def get_tenant_app(
|
||
app_id: int,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取应用配置详情"""
|
||
app = db.query(TenantApp).filter(TenantApp.id == app_id).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用配置不存在")
|
||
|
||
return format_tenant_app(app, mask_secret=True)
|
||
|
||
|
||
@router.post("")
|
||
async def create_tenant_app(
|
||
data: TenantAppCreate,
|
||
user: User = Depends(require_operator),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""创建应用配置"""
|
||
# 检查是否重复
|
||
exists = db.query(TenantApp).filter(
|
||
TenantApp.tenant_id == data.tenant_id,
|
||
TenantApp.app_code == data.app_code
|
||
).first()
|
||
if exists:
|
||
raise HTTPException(status_code=400, detail="该租户应用配置已存在")
|
||
|
||
# 自动生成 token_secret
|
||
token_secret = data.token_secret or secrets.token_hex(32)
|
||
|
||
# 加密 wechat_secret
|
||
wechat_secret_encrypted = None
|
||
if data.wechat_secret:
|
||
wechat_secret_encrypted = encrypt_config(data.wechat_secret)
|
||
|
||
app = TenantApp(
|
||
tenant_id=data.tenant_id,
|
||
app_code=data.app_code,
|
||
app_name=data.app_name,
|
||
wechat_corp_id=data.wechat_corp_id,
|
||
wechat_agent_id=data.wechat_agent_id,
|
||
wechat_secret_encrypted=wechat_secret_encrypted,
|
||
token_secret=token_secret,
|
||
token_required=1 if data.token_required else 0,
|
||
allowed_origins=json.dumps(data.allowed_origins) if data.allowed_origins else None,
|
||
allowed_tools=json.dumps(data.allowed_tools) if data.allowed_tools else None,
|
||
status=1
|
||
)
|
||
db.add(app)
|
||
db.commit()
|
||
db.refresh(app)
|
||
|
||
return {"success": True, "id": app.id, "token_secret": token_secret}
|
||
|
||
|
||
@router.put("/{app_id}")
|
||
async def update_tenant_app(
|
||
app_id: int,
|
||
data: TenantAppUpdate,
|
||
user: User = Depends(require_operator),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""更新应用配置"""
|
||
app = db.query(TenantApp).filter(TenantApp.id == app_id).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用配置不存在")
|
||
|
||
update_data = data.model_dump(exclude_unset=True)
|
||
|
||
# 处理 wechat_secret
|
||
if 'wechat_secret' in update_data:
|
||
if update_data['wechat_secret']:
|
||
app.wechat_secret_encrypted = encrypt_config(update_data['wechat_secret'])
|
||
del update_data['wechat_secret']
|
||
|
||
# 处理 JSON 字段
|
||
if 'allowed_origins' in update_data:
|
||
update_data['allowed_origins'] = json.dumps(update_data['allowed_origins']) if update_data['allowed_origins'] else None
|
||
if 'allowed_tools' in update_data:
|
||
update_data['allowed_tools'] = json.dumps(update_data['allowed_tools']) if update_data['allowed_tools'] else None
|
||
|
||
# 处理 token_required
|
||
if 'token_required' in update_data:
|
||
update_data['token_required'] = 1 if update_data['token_required'] else 0
|
||
|
||
for key, value in update_data.items():
|
||
setattr(app, key, value)
|
||
|
||
db.commit()
|
||
return {"success": True}
|
||
|
||
|
||
@router.delete("/{app_id}")
|
||
async def delete_tenant_app(
|
||
app_id: int,
|
||
user: User = Depends(require_operator),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""删除应用配置"""
|
||
app = db.query(TenantApp).filter(TenantApp.id == app_id).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用配置不存在")
|
||
|
||
db.delete(app)
|
||
db.commit()
|
||
|
||
return {"success": True}
|
||
|
||
|
||
@router.post("/{app_id}/regenerate-token")
|
||
async def regenerate_token(
|
||
app_id: int,
|
||
user: User = Depends(require_operator),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""重新生成 token_secret"""
|
||
app = db.query(TenantApp).filter(TenantApp.id == app_id).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用配置不存在")
|
||
|
||
new_token = secrets.token_hex(32)
|
||
app.token_secret = new_token
|
||
db.commit()
|
||
|
||
return {"success": True, "token_secret": new_token}
|
||
|
||
|
||
@router.get("/{app_id}/wechat-secret")
|
||
async def get_wechat_secret(
|
||
app_id: int,
|
||
user: User = Depends(require_operator),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取解密的 wechat_secret（仅操作员以上）"""
|
||
app = db.query(TenantApp).filter(TenantApp.id == app_id).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用配置不存在")
|
||
|
||
secret = None
|
||
if app.wechat_secret_encrypted:
|
||
secret = decrypt_config(app.wechat_secret_encrypted)
|
||
|
||
return {"wechat_secret": secret}
|
||
|
||
|
||
def format_tenant_app(app: TenantApp, mask_secret: bool = True) -> dict:
|
||
"""格式化应用配置"""
|
||
result = {
|
||
"id": app.id,
|
||
"tenant_id": app.tenant_id,
|
||
"app_code": app.app_code,
|
||
"app_name": app.app_name,
|
||
"wechat_corp_id": app.wechat_corp_id,
|
||
"wechat_agent_id": app.wechat_agent_id,
|
||
"has_wechat_secret": bool(app.wechat_secret_encrypted),
|
||
"token_secret": "******" if mask_secret and app.token_secret else app.token_secret,
|
||
"token_required": bool(app.token_required),
|
||
"allowed_origins": json.loads(app.allowed_origins) if app.allowed_origins else [],
|
||
"allowed_tools": json.loads(app.allowed_tools) if app.allowed_tools else [],
|
||
"status": app.status,
|
||
"created_at": app.created_at,
|
||
"updated_at": app.updated_at
|
||
}
|
||
return result
|