Admin
cf4d6afbc8
All checks were successful
continuous-integration/drone/push Build is passing
- 后端: App 模型添加 config_schema 字段，支持配置项定义 - 后端: apps API 支持 config_schema 的增删改查 - 前端: 应用管理页面支持定义配置项（text/radio/select/switch类型） - 前端: 租户订阅页面根据应用 schema 动态渲染对应表单控件 - 支持设置选项、默认值、必填等属性
314 lines
9.4 KiB
Python
314 lines
9.4 KiB
Python
"""应用管理路由"""
|
||
import json
|
||
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.app import App
|
||
from ..models.tenant_app import TenantApp
|
||
from .auth import get_current_user, require_operator
|
||
from ..models.user import User
|
||
|
||
router = APIRouter(prefix="/apps", tags=["应用管理"])
|
||
|
||
|
||
# ============ Schemas ============
|
||
|
||
class ToolItem(BaseModel):
|
||
"""工具项"""
|
||
code: str
|
||
name: str
|
||
path: str
|
||
|
||
|
||
class ConfigSchemaItem(BaseModel):
|
||
"""配置项定义"""
|
||
key: str # 配置键
|
||
label: str # 显示标签
|
||
type: str # text | radio | select | switch
|
||
options: Optional[List[str]] = None # radio/select 的选项值
|
||
option_labels: Optional[dict] = None # 选项显示名称 {"value": "显示名"}
|
||
default: Optional[str] = None # 默认值
|
||
placeholder: Optional[str] = None # 输入提示（text类型）
|
||
required: bool = False # 是否必填
|
||
|
||
|
||
class AppCreate(BaseModel):
|
||
"""创建应用"""
|
||
app_code: str
|
||
app_name: str
|
||
base_url: Optional[str] = None
|
||
description: Optional[str] = None
|
||
tools: Optional[List[ToolItem]] = None
|
||
config_schema: Optional[List[ConfigSchemaItem]] = None
|
||
require_jssdk: bool = False
|
||
|
||
|
||
class AppUpdate(BaseModel):
|
||
"""更新应用"""
|
||
app_name: Optional[str] = None
|
||
base_url: Optional[str] = None
|
||
description: Optional[str] = None
|
||
tools: Optional[List[ToolItem]] = None
|
||
config_schema: Optional[List[ConfigSchemaItem]] = None
|
||
require_jssdk: Optional[bool] = None
|
||
status: Optional[int] = None
|
||
|
||
|
||
class GenerateUrlRequest(BaseModel):
|
||
"""生成链接请求"""
|
||
tenant_id: str
|
||
app_code: str
|
||
tool_code: Optional[str] = None # 不传则生成应用首页链接
|
||
|
||
|
||
# ============ API Endpoints ============
|
||
|
||
@router.get("")
|
||
async def list_apps(
|
||
page: int = Query(1, ge=1),
|
||
size: int = Query(20, ge=1, le=1000),
|
||
status: Optional[int] = None,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取应用列表"""
|
||
query = db.query(App)
|
||
|
||
if status is not None:
|
||
query = query.filter(App.status == status)
|
||
|
||
total = query.count()
|
||
apps = query.order_by(App.id.asc()).offset((page - 1) * size).limit(size).all()
|
||
|
||
return {
|
||
"total": total,
|
||
"page": page,
|
||
"size": size,
|
||
"items": [format_app(app) for app in apps]
|
||
}
|
||
|
||
|
||
@router.get("/all")
|
||
async def list_all_apps(
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取所有启用的应用（用于下拉选择）"""
|
||
apps = db.query(App).filter(App.status == 1).order_by(App.id.asc()).all()
|
||
return [{"app_code": app.app_code, "app_name": app.app_name} for app in apps]
|
||
|
||
|
||
@router.get("/{app_id}")
|
||
async def get_app(
|
||
app_id: int,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取应用详情"""
|
||
app = db.query(App).filter(App.id == app_id).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用不存在")
|
||
|
||
return format_app(app)
|
||
|
||
|
||
@router.post("")
|
||
async def create_app(
|
||
data: AppCreate,
|
||
user: User = Depends(require_operator),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""创建应用"""
|
||
# 检查 app_code 是否重复
|
||
exists = db.query(App).filter(App.app_code == data.app_code).first()
|
||
if exists:
|
||
raise HTTPException(status_code=400, detail="应用代码已存在")
|
||
|
||
app = App(
|
||
app_code=data.app_code,
|
||
app_name=data.app_name,
|
||
base_url=data.base_url,
|
||
description=data.description,
|
||
tools=json.dumps([t.model_dump() for t in data.tools], ensure_ascii=False) if data.tools else None,
|
||
config_schema=json.dumps([c.model_dump() for c in data.config_schema], ensure_ascii=False) if data.config_schema else None,
|
||
require_jssdk=1 if data.require_jssdk else 0,
|
||
status=1
|
||
)
|
||
db.add(app)
|
||
db.commit()
|
||
db.refresh(app)
|
||
|
||
return {"success": True, "id": app.id}
|
||
|
||
|
||
@router.put("/{app_id}")
|
||
async def update_app(
|
||
app_id: int,
|
||
data: AppUpdate,
|
||
user: User = Depends(require_operator),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""更新应用"""
|
||
app = db.query(App).filter(App.id == app_id).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用不存在")
|
||
|
||
update_data = data.model_dump(exclude_unset=True)
|
||
|
||
# 处理 tools JSON
|
||
if 'tools' in update_data:
|
||
if update_data['tools']:
|
||
update_data['tools'] = json.dumps([t.model_dump() if hasattr(t, 'model_dump') else t for t in update_data['tools']], ensure_ascii=False)
|
||
else:
|
||
update_data['tools'] = None
|
||
|
||
# 处理 config_schema JSON
|
||
if 'config_schema' in update_data:
|
||
if update_data['config_schema']:
|
||
update_data['config_schema'] = json.dumps([c.model_dump() if hasattr(c, 'model_dump') else c for c in update_data['config_schema']], ensure_ascii=False)
|
||
else:
|
||
update_data['config_schema'] = None
|
||
|
||
# 处理 require_jssdk
|
||
if 'require_jssdk' in update_data:
|
||
update_data['require_jssdk'] = 1 if update_data['require_jssdk'] 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_app(
|
||
app_id: int,
|
||
user: User = Depends(require_operator),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""删除应用"""
|
||
app = db.query(App).filter(App.id == app_id).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用不存在")
|
||
|
||
# 检查是否有租户在使用
|
||
tenant_count = db.query(TenantApp).filter(TenantApp.app_code == app.app_code).count()
|
||
if tenant_count > 0:
|
||
raise HTTPException(status_code=400, detail=f"有 {tenant_count} 个租户正在使用此应用，无法删除")
|
||
|
||
db.delete(app)
|
||
db.commit()
|
||
|
||
return {"success": True}
|
||
|
||
|
||
@router.post("/generate-url")
|
||
async def generate_url(
|
||
data: GenerateUrlRequest,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""
|
||
生成访问链接
|
||
|
||
返回完整的可直接使用的 URL（使用静态 token，长期有效）
|
||
"""
|
||
# 获取应用信息
|
||
app = db.query(App).filter(App.app_code == data.app_code, App.status == 1).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用不存在或已禁用")
|
||
|
||
if not app.base_url:
|
||
raise HTTPException(status_code=400, detail="应用未配置基础URL")
|
||
|
||
# 获取租户配置
|
||
tenant_app = db.query(TenantApp).filter(
|
||
TenantApp.tenant_id == data.tenant_id,
|
||
TenantApp.app_code == data.app_code,
|
||
TenantApp.status == 1
|
||
).first()
|
||
|
||
if not tenant_app:
|
||
raise HTTPException(status_code=404, detail="租户未配置此应用")
|
||
|
||
if not tenant_app.access_token:
|
||
raise HTTPException(status_code=400, detail="租户应用未配置访问令牌")
|
||
|
||
# 构建基础 URL
|
||
base_url = app.base_url.rstrip('/')
|
||
if data.tool_code:
|
||
# 查找工具路径
|
||
tools = json.loads(app.tools) if app.tools else []
|
||
tool = next((t for t in tools if t.get('code') == data.tool_code), None)
|
||
if tool:
|
||
base_url = f"{base_url}{tool.get('path', '')}"
|
||
else:
|
||
base_url = f"{base_url}/{data.tool_code}"
|
||
|
||
# 构建参数（静态 token，长期有效）
|
||
params = {
|
||
"tid": data.tenant_id,
|
||
"token": tenant_app.access_token
|
||
}
|
||
|
||
# 组装 URL
|
||
query_string = "&".join([f"{k}={v}" for k, v in params.items()])
|
||
full_url = f"{base_url}?{query_string}"
|
||
|
||
return {
|
||
"success": True,
|
||
"url": full_url,
|
||
"params": params,
|
||
"note": "静态链接，长期有效"
|
||
}
|
||
|
||
|
||
@router.get("/{app_code}/tools")
|
||
async def get_app_tools(
|
||
app_code: str,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取应用的工具列表（用于配置权限时选择）"""
|
||
app = db.query(App).filter(App.app_code == app_code).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用不存在")
|
||
|
||
tools = json.loads(app.tools) if app.tools else []
|
||
return tools
|
||
|
||
|
||
@router.get("/{app_code}/config-schema")
|
||
async def get_app_config_schema(
|
||
app_code: str,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取应用的配置项定义（用于租户订阅时渲染表单）"""
|
||
app = db.query(App).filter(App.app_code == app_code).first()
|
||
if not app:
|
||
raise HTTPException(status_code=404, detail="应用不存在")
|
||
|
||
config_schema = json.loads(app.config_schema) if app.config_schema else []
|
||
return config_schema
|
||
|
||
|
||
def format_app(app: App) -> dict:
|
||
"""格式化应用数据"""
|
||
return {
|
||
"id": app.id,
|
||
"app_code": app.app_code,
|
||
"app_name": app.app_name,
|
||
"base_url": app.base_url,
|
||
"description": app.description,
|
||
"tools": json.loads(app.tools) if app.tools else [],
|
||
"config_schema": json.loads(app.config_schema) if app.config_schema else [],
|
||
"require_jssdk": bool(app.require_jssdk),
|
||
"status": app.status,
|
||
"created_at": app.created_at,
|
||
"updated_at": app.updated_at
|
||
}
|