深入Dify控制器模块：三层API架构与RESTful接口设计

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

# API路由配置示例
class APIGateway:
    """
    API网关统一配置
    实现请求路由、认证、限流和监控
    """
    
    def __init__(self):
        self.routes = {
            "/console/api": ConsoleAPIHandler,
            "/v1": ServiceAPIHandler, 
            "/api": WebAPIHandler,
            "/inner_api": InnerAPIHandler
        }
        
        # 中间件链配置
        self.middleware_chain = [
            AuthenticationMiddleware(),
            RateLimitingMiddleware(),
            RequestLoggingMiddleware(),
            CORSMiddleware(),
            ResponseFormattingMiddleware()
        ]
    
    def route_request(self, request: Request) -> Response:
        """
        智能路由请求到相应的API处理器
        """
        # 1. 中间件预处理
        for middleware in self.middleware_chain:
            request = middleware.pre_process(request)
        
        # 2. 路由匹配
        handler = self._match_handler(request.path)
        
        # 3. 处理请求
        response = handler.handle(request)
        
        # 4. 中间件后处理
        for middleware in reversed(self.middleware_chain):
            response = middleware.post_process(response)
        
        return response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 企业部署API配置优化
SERVICE_API_URL = "https://your-domain.com"  # 服务API基础URL
WEB_API_CORS_ALLOW_ORIGINS = ["https://your-frontend.com"]  # CORS配置
API_RATE_LIMIT_ENABLED = True                # 启用API限流
API_RATE_LIMIT_REQUESTS_PER_MINUTE = 1000    # 每分钟请求限制

# API安全配置
API_KEY_ENCRYPTION_ENABLED = True            # API Key加密存储
API_REQUEST_SIGNATURE_REQUIRED = False       # 请求签名验证
API_IP_WHITELIST_ENABLED = False            # IP白名单功能

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

// API分层职责定义
interface APILayerResponsibilities {
  // Console API - 管理控制台（/console/api）
  console: {
    target_users: ["系统管理员", "应用开发者", "租户管理员"];
    authentication: "基于会话的用户认证";
    primary_functions: [
      "应用配置和管理",
      "数据集创建和维护", 
      "用户和权限管理",
      "系统监控和统计",
      "模型提供者配置",
      "工作流设计和调试"
    ];
    response_format: "同步响应为主，部分支持流式";
    security_level: "高级 - 完整权限控制";
  };
  
  // Service API - 应用服务（/v1）
  service_api: {
    target_users: ["外部开发者", "第三方系统", "企业集成"];
    authentication: "API Key认证";
    primary_functions: [
      "应用运行时调用",
      "对话和文本生成",
      "数据集检索服务",
      "文件处理服务",
      "工作流执行"
    ];
    response_format: "支持同步和流式响应";
    security_level: "中等 - API Key隔离";
  };
  
  // Web API - 前端应用（/api）
  web_api: {
    target_users: ["最终用户", "Web应用", "移动应用"];
    authentication: "Bearer Token或会话认证";
    primary_functions: [
      "用户对话交互",
      "应用界面数据",
      "文件上传下载",
      "实时通信",
      "用户账户管理"
    ];
    response_format: "优化的前端响应格式";
    security_level: "标准 - 用户级权限";
  };
  
  // Inner API - 内部服务
  inner_api: {
    target_users: ["内部微服务", "系统组件"];
    authentication: "内部服务认证";
    primary_functions: [
      "服务间通信",
      "内部数据同步",
      "系统集成"
    ];
    response_format: "轻量级内部协议";
    security_level: "内部 - 服务间信任";
  };
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498

# Console API Blueprint配置
bp = Blueprint("console", __name__, url_prefix="/console/api")

api = ExternalApi(
    bp,
    version="1.0", 
    title="Console API",
    description="Console management APIs for app configuration, monitoring, and administration",
)

# Console命名空间
console_ns = Namespace("console", description="Console management API operations", path="/")


class ConsoleAPIRouter:
    """
    Console API路由管理器
    负责组织和注册Console API的所有路由
    """
    
    def __init__(self, api: ExternalApi):
        """
        初始化Console API路由
        
        Args:
            api: ExternalApi实例
        """
        self.api = api
        self._register_core_routes()
        self._register_app_routes() 
        self._register_dataset_routes()
        self._register_workspace_routes()
        self._register_auth_routes()

    def _register_core_routes(self):
        """注册核心路由"""
        
        # 系统管理路由
        self.api.add_resource(SetupApi, "/setup")
        self.api.add_resource(PingApi, "/ping") 
        self.api.add_resource(VersionApi, "/version")
        self.api.add_resource(FeatureApi, "/features")
        
        # 文件管理路由
        self.api.add_resource(FileApi, "/files/upload")
        self.api.add_resource(FilePreviewApi, "/files/<uuid:file_id>/preview")
        self.api.add_resource(FileSupportTypeApi, "/files/support-type")
        
        # 远程文件路由
        self.api.add_resource(RemoteFileInfoApi, "/remote-files/<path:url>")
        self.api.add_resource(RemoteFileUploadApi, "/remote-files/upload")

    def _register_app_routes(self):
        """注册应用管理路由"""
        
        # 应用CRUD操作
        from .app import app, workflow, message, conversation
        
        # 应用基础管理
        self.api.add_resource(AppListApi, "/apps")
        self.api.add_resource(AppApi, "/apps/<uuid:app_id>")
        self.api.add_resource(AppCreateApi, "/apps")
        
        # 工作流管理
        self.api.add_resource(WorkflowListApi, "/apps/<uuid:app_id>/workflows")
        self.api.add_resource(WorkflowApi, "/apps/<uuid:app_id>/workflows/<uuid:workflow_id>")
        
        # 消息和对话管理
        self.api.add_resource(MessageListApi, "/apps/<uuid:app_id>/messages")
        self.api.add_resource(ConversationListApi, "/apps/<uuid:app_id>/conversations")

    def _register_dataset_routes(self):
        """注册数据集管理路由"""
        
        from .datasets import datasets, datasets_document, datasets_segments
        
        # 数据集CRUD
        self.api.add_resource(DatasetListApi, "/datasets")
        self.api.add_resource(DatasetApi, "/datasets/<uuid:dataset_id>")
        
        # 文档管理
        self.api.add_resource(DocumentListApi, "/datasets/<uuid:dataset_id>/documents")
        self.api.add_resource(DocumentApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
        
        # 段落管理
        self.api.add_resource(SegmentListApi, "/datasets/<uuid:dataset_id>/segments")

    def _register_workspace_routes(self):
        """注册工作空间管理路由"""
        
        from .workspace import workspace, account, model_providers
        
        # 工作空间管理
        self.api.add_resource(WorkspaceApi, "/workspaces/current")
        self.api.add_resource(AccountApi, "/account")
        
        # 模型提供者配置
        self.api.add_resource(ModelProviderListApi, "/workspaces/current/model-providers")
        self.api.add_resource(ModelProviderApi, "/workspaces/current/model-providers/<string:provider>")

    def _register_auth_routes(self):
        """注册认证相关路由"""
        
        from .auth import login, oauth, forgot_password
        
        # 用户认证
        self.api.add_resource(LoginApi, "/login")
        self.api.add_resource(LogoutApi, "/logout")
        self.api.add_resource(ForgotPasswordApi, "/forgot-password")
        
        # OAuth认证
        self.api.add_resource(OAuthApi, "/oauth")


@console_ns.route("/apps")
class AppListApi(Resource):
    """
    应用列表API
    提供应用的分页查询、过滤和管理功能
    """
    
    @api.doc("list_apps")
    @api.doc(description="获取应用列表，支持分页和多种过滤条件")
    @api.expect(
        api.parser()
        .add_argument("page", type=int, location="args", help="页码 (1-99999)", default=1)
        .add_argument("limit", type=int, location="args", help="页大小 (1-100)", default=20)
        .add_argument(
            "mode", type=str, location="args",
            choices=["completion", "chat", "advanced-chat", "workflow", "agent-chat", "channel", "all"],
            default="all", help="应用模式过滤"
        )
        .add_argument("name", type=str, location="args", help="按应用名称过滤")
        .add_argument("tag_ids", type=str, location="args", help="逗号分隔的标签ID")
        .add_argument("is_created_by_me", type=bool, location="args", help="只显示我创建的应用")
    )
    @api.response(200, "Success", app_pagination_fields)
    @setup_required
    @login_required
    @account_initialization_required  
    @enterprise_license_required
    def get(self):
        """
        获取应用列表
        支持多维度过滤和分页查询
        
        Returns:
            分页的应用列表，包含应用基本信息和统计数据
        """
        # 参数解析和验证
        parser = reqparse.RequestParser()
        parser.add_argument("page", type=inputs.int_range(1, 99999), default=1)
        parser.add_argument("limit", type=inputs.int_range(1, 100), default=20)
        parser.add_argument("mode", type=str, choices=ALLOWED_APP_MODES, default="all")
        parser.add_argument("name", type=str)
        parser.add_argument("tag_ids", type=self._parse_uuid_list)
        parser.add_argument("is_created_by_me", type=inputs.boolean)
        
        args = parser.parse_args()

        # 业务逻辑调用
        app_service = AppService()
        app_pagination = app_service.get_paginate_apps(
            user_id=current_user.id,
            tenant_id=current_user.current_tenant_id,
            args=args
        )
        
        # 返回格式化结果
        if not app_pagination:
            return {
                "data": [], 
                "total": 0, 
                "page": 1, 
                "limit": 20, 
                "has_more": False
            }
        
        return marshal(app_pagination, app_pagination_fields)

    @api.doc("create_app")
    @api.doc(description="创建新应用")
    @api.expect(app_create_fields)
    @api.response(201, "应用创建成功", app_detail_fields)
    @api.response(400, "请求参数错误")
    @setup_required
    @login_required
    @account_initialization_required
    @cloud_edition_billing_resource_check("apps")
    def post(self):
        """
        创建新应用
        支持多种应用类型的创建
        
        Returns:
            创建的应用详细信息
        """
        # 参数解析
        parser = reqparse.RequestParser()
        parser.add_argument("name", type=str, required=True, help="应用名称")
        parser.add_argument("icon", type=str, help="应用图标")
        parser.add_argument("icon_background", type=str, help="图标背景色")
        parser.add_argument(
            "mode", type=str, required=True,
            choices=ALLOW_CREATE_APP_MODES,
            help="应用模式"
        )
        parser.add_argument(
            "description", type=self._validate_description_length,
            help="应用描述"
        )
        
        args = parser.parse_args()

        # 创建应用
        try:
            app_service = AppService()
            app = app_service.create_app(
                tenant_id=current_user.current_tenant_id,
                app_name=args["name"],
                app_mode=args["mode"],
                icon=args.get("icon"),
                icon_background=args.get("icon_background"),
                description=args.get("description"),
                account=current_user
            )
            
            return marshal(app, app_detail_fields), 201
            
        except ValueError as e:
            raise BadRequest(str(e))
        except Exception as e:
            logger.exception("创建应用失败")
            raise InternalServerError("应用创建失败")

    def _validate_description_length(self, description: str) -> str:
        """
        验证描述长度
        
        Args:
            description: 应用描述
            
        Returns:
            str: 验证后的描述
            
        Raises:
            ValueError: 描述长度超限
        """
        if description and len(description) > 400:
            raise ValueError("应用描述不能超过400个字符")
        return description

    def _parse_uuid_list(self, value: str) -> list[str]:
        """
        解析UUID列表
        
        Args:
            value: 逗号分隔的UUID字符串
            
        Returns:
            list[str]: UUID列表
            
        Raises:
            ValueError: UUID格式错误
        """
        try:
            return [str(uuid.UUID(v)) for v in value.split(",")]
        except ValueError:
            abort(400, message="标签ID格式错误")


@console_ns.route("/apps/<uuid:app_id>")
class AppApi(Resource):
    """
    单个应用API
    提供应用的详细信息查询、更新和删除功能
    """
    
    @api.doc("get_app")
    @api.doc(description="获取应用详细信息")
    @api.response(200, "Success", app_detail_fields_with_site)
    @api.response(404, "应用不存在")
    @setup_required
    @login_required
    @account_initialization_required
    @get_app_model
    def get(self, app_model: App):
        """
        获取应用详细信息
        包含应用配置、统计数据和站点信息
        
        Args:
            app_model: 应用模型实例（通过装饰器注入）
            
        Returns:
            应用详细信息，包含所有配置和统计数据
        """
        try:
            # 获取应用服务
            app_service = AppService()
            
            # 获取应用详细信息
            app_detail = app_service.get_app_detail(
                app_id=app_model.id,
                user_id=current_user.id,
                tenant_id=current_user.current_tenant_id
            )
            
            # 获取应用站点信息
            site_info = app_service.get_app_site_info(app_model.id)
            
            # 合并信息
            app_data = marshal(app_detail, app_detail_fields)
            if site_info:
                app_data.update(marshal(site_info, site_fields))
            
            return app_data, 200
            
        except Exception as e:
            logger.exception(f"获取应用 {app_model.id} 详情失败")
            raise InternalServerError("获取应用信息失败")

    @api.doc("update_app")
    @api.doc(description="更新应用信息")
    @api.expect(app_update_fields)
    @api.response(200, "更新成功", app_detail_fields)
    @setup_required
    @login_required
    @account_initialization_required
    @get_app_model
    def put(self, app_model: App):
        """
        更新应用信息
        支持应用名称、描述、图标等基础信息的更新
        
        Args:
            app_model: 应用模型实例
            
        Returns:
            更新后的应用信息
        """
        # 参数解析
        parser = reqparse.RequestParser()
        parser.add_argument("name", type=str, help="应用名称")
        parser.add_argument("description", type=self._validate_description, help="应用描述")
        parser.add_argument("icon", type=str, help="应用图标")
        parser.add_argument("icon_background", type=str, help="图标背景色")
        
        args = parser.parse_args()

        try:
            # 更新应用
            app_service = AppService()
            updated_app = app_service.update_app(
                app_id=app_model.id,
                tenant_id=current_user.current_tenant_id,
                account=current_user,
                **args
            )
            
            return marshal(updated_app, app_detail_fields), 200
            
        except ValueError as e:
            raise BadRequest(str(e))
        except Exception as e:
            logger.exception(f"更新应用 {app_model.id} 失败")
            raise InternalServerError("更新应用失败")

    @api.doc("delete_app")
    @api.doc(description="删除应用")
    @api.response(204, "删除成功")
    @api.response(404, "应用不存在")
    @setup_required
    @login_required
    @account_initialization_required
    @get_app_model
    def delete(self, app_model: App):
        """
        删除应用
        软删除应用，保留历史数据
        
        Args:
            app_model: 应用模型实例
        """
        try:
            app_service = AppService()
            app_service.delete_app(
                app_id=app_model.id,
                user_id=current_user.id,
                tenant_id=current_user.current_tenant_id
            )
            
            return "", 204
            
        except Exception as e:
            logger.exception(f"删除应用 {app_model.id} 失败")
            raise InternalServerError("删除应用失败")


class ConsoleAuthMiddleware:
    """
    Console API认证中间件
    处理用户会话认证和权限验证
    """
    
    @staticmethod
    def setup_required(f):
        """
        系统初始化检查装饰器
        确保系统已完成初始设置
        """
        def decorated(*args, **kwargs):
            # 检查系统是否已初始化
            if not SystemSetupService.is_setup_completed():
                return {"message": "系统未完成初始化设置"}, 412
            
            return f(*args, **kwargs)
        
        return decorated

    @staticmethod
    def login_required(f):
        """
        登录认证装饰器
        确保用户已登录
        """
        def decorated(*args, **kwargs):
            if not current_user.is_authenticated:
                return {"message": "未登录或会话已过期"}, 401
            
            return f(*args, **kwargs)
        
        return decorated

    @staticmethod
    def account_initialization_required(f):
        """
        账户初始化检查装饰器
        确保用户账户已完成初始化
        """
        def decorated(*args, **kwargs):
            if not current_user.is_initialized:
                return {"message": "账户未完成初始化"}, 400
            
            return f(*args, **kwargs)
        
        return decorated

    @staticmethod
    def enterprise_license_required(f):
        """
        企业许可证检查装饰器
        检查企业功能的许可证状态
        """
        def decorated(*args, **kwargs):
            if not EnterpriseService.check_license_valid():
                return {"message": "需要有效的企业许可证"}, 403
            
            return f(*args, **kwargs)
        
        return decorated

    @staticmethod
    def cloud_edition_billing_resource_check(resource_type: str):
        """
        云版本计费资源检查装饰器
        检查租户的资源使用限额
        
        Args:
            resource_type: 资源类型（如"apps", "datasets"等）
        """
        def decorator(f):
            def decorated(*args, **kwargs):
                # 检查资源配额
                if not BillingService.check_resource_quota(
                    tenant_id=current_user.current_tenant_id,
                    resource_type=resource_type
                ):
                    return {"message": f"已达到{resource_type}资源配额限制"}, 429
                
                return f(*args, **kwargs)
            
            return decorated
        return decorator

    @staticmethod  
    def admin_required(f):
        """
        管理员权限检查装饰器
        确保当前用户具有管理员权限
        """
        def decorated(*args, **kwargs):
            if not current_user.is_admin:
                return {"message": "需要管理员权限"}, 403
            
            return f(*args, **kwargs)
        
        return decorated

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349

# Service API Blueprint配置
bp = Blueprint("service_api", __name__, url_prefix="/v1")

api = ExternalApi(
    bp,
    version="1.0",
    title="Service API", 
    description="API for application services",
)

service_api_ns = Namespace("service_api", description="Service operations", path="/")


@service_api_ns.route("/completion-messages")
class CompletionApi(Resource):
    """
    文本完成API
    提供基于提示的文本生成服务
    """
    
    @service_api_ns.expect(completion_parser)
    @service_api_ns.doc("create_completion")
    @service_api_ns.doc(description="基于给定提示创建文本完成")
    @service_api_ns.doc(
        responses={
            200: "完成创建成功",
            400: "请求参数错误",
            401: "API令牌无效",
            404: "应用不存在",
            500: "内部服务器错误",
        }
    )
    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
    def post(self, app_model: App, end_user: EndUser):
        """
        创建文本完成
        根据提供的输入和查询生成文本完成结果
        
        Args:
            app_model: 应用模型实例（通过装饰器注入）
            end_user: 最终用户实例（通过装饰器注入）
            
        Returns:
            文本完成结果，支持同步和流式两种模式
        """
        # 验证应用类型
        if app_model.mode != AppMode.COMPLETION:
            raise AppUnavailableError("应用类型不是completion")

        # 解析请求参数
        args = completion_parser.parse_args()
        
        # 参数验证
        self._validate_completion_params(args)

        try:
            # 调用应用生成服务
            response = AppGenerateService.generate(
                app_model=app_model,
                user=end_user,
                args=args,
                invoke_from=InvokeFrom.SERVICE_API,
                streaming=args["response_mode"] == "streaming"
            )

            # 根据响应模式返回结果
            if args["response_mode"] == "streaming":
                # 流式响应
                return Response(
                    stream_with_context(response),
                    status=200,
                    mimetype="text/event-stream",
                    headers={
                        "Cache-Control": "no-cache",
                        "Connection": "keep-alive",
                        "Access-Control-Allow-Origin": "*",
                    }
                )
            else:
                # 阻塞式响应
                return response, 200

        except CompletionRequestError as e:
            raise BadRequest(str(e))
        except ProviderNotInitializeError as e:
            raise BadRequest(f"模型提供者未初始化: {e}")
        except ProviderQuotaExceededError as e:
            raise BadRequest(f"提供者配额超出: {e}")
        except Exception as e:
            logger.exception("文本完成API调用失败")
            raise InternalServerError("服务暂时不可用")

    def _validate_completion_params(self, args: dict):
        """
        验证完成参数
        
        Args:
            args: 请求参数字典
            
        Raises:
            BadRequest: 参数验证失败
        """
        # 检查必填参数
        if not args.get("inputs"):
            raise BadRequest("inputs参数是必需的")
        
        # 检查文件参数格式
        if args.get("files"):
            if not isinstance(args["files"], list):
                raise BadRequest("files参数必须是列表格式")
            
            # 验证文件格式
            for file_info in args["files"]:
                if not isinstance(file_info, dict):
                    raise BadRequest("文件信息格式错误")
                
                required_fields = ["type", "transfer_method"]
                for field in required_fields:
                    if field not in file_info:
                        raise BadRequest(f"文件信息缺少必填字段: {field}")


@service_api_ns.route("/chat-messages")
class ChatApi(Resource):
    """
    对话API  
    提供多轮对话交互服务
    """
    
    @service_api_ns.expect(chat_parser)
    @service_api_ns.doc("create_chat_message")
    @service_api_ns.doc(description="创建新的对话消息")
    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
    def post(self, app_model: App, end_user: EndUser):
        """
        创建对话消息
        支持多轮对话上下文管理和流式响应
        
        Args:
            app_model: 应用模型实例
            end_user: 最终用户实例
            
        Returns:
            对话响应，包含AI回复和对话状态
        """
        # 验证应用类型
        if app_model.mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
            raise NotChatAppError("应用类型不支持对话")

        args = chat_parser.parse_args()
        
        # 参数验证
        self._validate_chat_params(args)

        try:
            # 处理对话ID（新建或续接对话）
            conversation_id = args.get("conversation_id")
            if conversation_id:
                # 验证对话存在性和权限
                self._validate_conversation_access(conversation_id, end_user.id)

            # 调用对话生成服务
            response = AppGenerateService.generate_chat(
                app_model=app_model,
                user=end_user,
                conversation_id=conversation_id,
                query=args["query"],
                inputs=args["inputs"],
                files=args.get("files", []),
                response_mode=args["response_mode"],
                auto_generate_name=args.get("auto_generate_name", True),
                invoke_from=InvokeFrom.SERVICE_API
            )

            # 处理流式响应
            if args["response_mode"] == "streaming":
                return self._create_streaming_response(response)
            else:
                return response, 200

        except ConversationCompletedError:
            raise BadRequest("对话已结束，无法继续")
        except InvokeRateLimitError:
            raise InvokeRateLimitHttpError("调用频率超限，请稍后重试")
        except Exception as e:
            logger.exception("对话API调用失败")
            raise InternalServerError("对话服务暂时不可用")

    def _validate_chat_params(self, args: dict):
        """验证对话参数"""
        if not args.get("query"):
            raise BadRequest("query参数是必需的")
        
        if not args.get("inputs"):
            raise BadRequest("inputs参数是必需的")

    def _validate_conversation_access(self, conversation_id: str, user_id: str):
        """验证对话访问权限"""
        conversation_service = ConversationService()
        conversation = conversation_service.get_conversation(conversation_id)
        
        if not conversation:
            raise NotFound("对话不存在")
        
        if conversation.from_end_user_id != user_id:
            raise Forbidden("无权限访问该对话")

    def _create_streaming_response(self, response_generator):
        """创建流式响应"""
        def generate():
            try:
                for chunk in response_generator:
                    yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
                yield "data: [DONE]\n\n"
            except Exception as e:
                error_data = {"error": str(e)}
                yield f"data: {json.dumps(error_data, ensure_ascii=False)}\n\n"

        return Response(
            generate(),
            status=200,
            mimetype="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "Access-Control-Allow-Origin": "*",
            }
        )


class ServiceAPIAuthMiddleware:
    """
    Service API认证中间件
    基于API Key的认证和授权机制
    """
    
    @staticmethod
    def validate_app_token(fetch_user_arg: FetchUserArg):
        """
        应用Token验证装饰器
        验证API Key的有效性并获取关联的应用和用户信息
        
        Args:
            fetch_user_arg: 用户获取参数配置
        """
        def decorator(f):
            def decorated(*args, **kwargs):
                # 获取API Key
                api_key = request.headers.get("Authorization")
                if not api_key:
                    return {"message": "缺少Authorization头"}, 401
                
                # 移除Bearer前缀
                if api_key.startswith("Bearer "):
                    api_key = api_key[7:]
                
                try:
                    # 验证API Key并获取应用信息
                    app_service = AppService()
                    app_model, end_user = app_service.validate_api_key(
                        api_key=api_key,
                        fetch_user_arg=fetch_user_arg
                    )
                    
                    # 检查应用状态
                    if not app_model.is_enabled:
                        return {"message": "应用已禁用"}, 403
                    
                    # 将应用和用户信息注入到参数中
                    kwargs["app_model"] = app_model
                    kwargs["end_user"] = end_user
                    
                    return f(*args, **kwargs)
                    
                except InvalidAPIKeyError:
                    return {"message": "无效的API Key"}, 401
                except AppNotFoundError:
                    return {"message": "应用不存在"}, 404
                except Exception as e:
                    logger.exception("API Key验证失败")
                    return {"message": "认证服务异常"}, 500
            
            return decorated
        return decorator

    @staticmethod
    def rate_limit(calls_per_minute: int = 60):
        """
        API调用频率限制装饰器
        
        Args:
            calls_per_minute: 每分钟允许的调用次数
        """
        def decorator(f):
            def decorated(*args, **kwargs):
                # 获取客户端标识
                client_id = request.headers.get("X-Client-ID") or request.remote_addr
                
                # 检查调用频率
                rate_limiter = RateLimiter()
                if not rate_limiter.allow_request(client_id, calls_per_minute):
                    return {
                        "message": f"调用频率超限，每分钟最多{calls_per_minute}次"
                    }, 429
                
                return f(*args, **kwargs)
            
            return decorated
        return decorator


class RateLimiter:
    """
    API调用频率限制器
    基于Redis的滑动窗口算法实现
    """
    
    def __init__(self):
        self.redis_client = redis_client

    def allow_request(self, client_id: str, limit: int, window: int = 60) -> bool:
        """
        检查是否允许请求
        
        Args:
            client_id: 客户端标识
            limit: 限制次数
            window: 时间窗口（秒）
            
        Returns:
            bool: 是否允许请求
        """
        current_time = int(time.time())
        key = f"rate_limit:{client_id}:{current_time // window}"
        
        # 获取当前窗口的请求计数
        current_count = self.redis_client.get(key)
        current_count = int(current_count) if current_count else 0
        
        if current_count >= limit:
            return False
        
        # 增加计数并设置过期时间
        pipe = self.redis_client.pipeline()
        pipe.incr(key)
        pipe.expire(key, window)
        pipe.execute()
        
        return True

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269

# Web API Blueprint配置
bp = Blueprint("web", __name__, url_prefix="/api")

api = ExternalApi(
    bp,
    version="1.0",
    title="Web API",
    description="Public APIs for web applications including file uploads, chat interactions, and app management",
)

web_ns = Namespace("web", description="Web application API operations", path="/")


@web_ns.route("/apps/<uuid:app_id>/completion-messages")
class WebCompletionApi(Resource):
    """
    Web文本完成API
    为前端应用优化的文本完成接口
    """
    
    @api.doc("web_create_completion")
    @api.doc(description="为Web应用创建文本完成")
    @validate_web_user_token
    def post(self, app_model: App, user: Account):
        """
        Web文本完成
        针对前端用户体验优化的文本生成接口
        
        Args:
            app_model: 应用模型
            user: 用户账户
            
        Returns:
            优化格式的完成结果
        """
        # 解析请求数据
        data = request.get_json()
        
        # 参数验证
        self._validate_web_completion_params(data)
        
        try:
            # 生成文本完成
            response = WebAppGenerateService.generate_completion(
                app_model=app_model,
                user=user,
                inputs=data["inputs"],
                query=data.get("query", ""),
                files=data.get("files", []),
                streaming=data.get("streaming", True)
            )
            
            # 根据是否流式返回不同格式
            if data.get("streaming", True):
                return self._stream_response(response)
            else:
                return self._blocking_response(response)
                
        except Exception as e:
            logger.exception("Web完成API失败")
            return {"error": "服务暂时不可用"}, 500

    def _validate_web_completion_params(self, data: dict):
        """验证Web完成参数"""
        if not data:
            raise BadRequest("请求体不能为空")
        
        if "inputs" not in data:
            raise BadRequest("缺少inputs参数")

    def _stream_response(self, response_generator):
        """处理流式响应"""
        def generate():
            for chunk in response_generator:
                # 为前端优化响应格式
                web_chunk = {
                    "event": chunk.get("event", "message"),
                    "data": chunk.get("data", {}),
                    "timestamp": time.time()
                }
                yield f"data: {json.dumps(web_chunk, ensure_ascii=False)}\n\n"
            
            # 发送结束标记
            yield "data: [DONE]\n\n"

        return Response(
            generate(),
            mimetype="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",  # 禁用nginx缓冲
            }
        )

    def _blocking_response(self, response):
        """处理阻塞式响应"""
        # 为前端优化响应格式
        return {
            "result": response,
            "metadata": {
                "request_id": str(uuid.uuid4()),
                "timestamp": time.time(),
                "model_info": response.get("metadata", {})
            }
        }


@web_ns.route("/apps/<uuid:app_id>/conversations")
class WebConversationListApi(Resource):
    """
    Web对话列表API
    为前端优化的对话管理接口
    """
    
    @api.doc("get_web_conversations")
    @api.doc(description="获取用户的对话列表")
    @validate_web_user_token
    def get(self, app_model: App, user: Account):
        """
        获取对话列表
        返回用户在该应用下的所有对话
        
        Args:
            app_model: 应用模型
            user: 用户账户
            
        Returns:
            对话列表，包含对话摘要和状态
        """
        # 获取分页参数
        page = request.args.get("page", 1, type=int)
        limit = request.args.get("limit", 20, type=int)
        
        # 参数验证
        if page < 1 or limit < 1 or limit > 100:
            raise BadRequest("分页参数错误")

        try:
            # 获取对话服务
            conversation_service = WebConversationService()
            
            # 获取对话列表
            conversations = conversation_service.get_user_conversations(
                app_id=app_model.id,
                user_id=user.id,
                page=page,
                limit=limit
            )
            
            # 格式化返回结果
            return {
                "data": [
                    {
                        "id": conv.id,
                        "name": conv.name,
                        "created_at": conv.created_at.isoformat(),
                        "updated_at": conv.updated_at.isoformat(),
                        "message_count": conv.message_count,
                        "status": conv.status,
                        "is_pinned": conv.is_pinned,
                        "summary": conv.summary
                    }
                    for conv in conversations.items
                ],
                "total": conversations.total,
                "page": page,
                "limit": limit,
                "has_more": conversations.has_next
            }
            
        except Exception as e:
            logger.exception("获取对话列表失败")
            return {"error": "服务暂时不可用"}, 500


class WebAuthMiddleware:
    """
    Web API认证中间件
    支持多种认证方式的前端认证机制
    """
    
    @staticmethod
    def validate_web_user_token(f):
        """
        Web用户Token验证装饰器
        支持Bearer Token和会话认证
        """
        def decorated(*args, **kwargs):
            # 尝试Bearer Token认证
            auth_header = request.headers.get("Authorization")
            if auth_header and auth_header.startswith("Bearer "):
                token = auth_header[7:]
                
                try:
                    # 验证访问令牌
                    user_info = JWTTokenService.decode_access_token(token)
                    
                    # 获取用户和应用信息
                    user = AccountService.get_user_by_id(user_info["user_id"])
                    app_model = AppService.get_app_by_id(kwargs.get("app_id"))
                    
                    if not user or not app_model:
                        return {"message": "用户或应用不存在"}, 404
                    
                    kwargs["user"] = user
                    kwargs["app_model"] = app_model
                    
                    return f(*args, **kwargs)
                    
                except JWTTokenExpiredError:
                    return {"message": "访问令牌已过期"}, 401
                except JWTTokenInvalidError:
                    return {"message": "无效的访问令牌"}, 401
            
            # 尝试会话认证
            elif current_user.is_authenticated:
                # 使用会话中的用户信息
                app_model = AppService.get_app_by_id(kwargs.get("app_id"))
                
                if not app_model:
                    return {"message": "应用不存在"}, 404
                
                # 检查用户对应用的访问权限
                if not AppService.check_user_app_access(current_user, app_model):
                    return {"message": "无权限访问该应用"}, 403
                
                kwargs["user"] = current_user
                kwargs["app_model"] = app_model
                
                return f(*args, **kwargs)
            
            else:
                return {"message": "未提供有效的认证信息"}, 401
        
        return decorated

    @staticmethod
    def cors_handler(f):
        """
        CORS处理装饰器
        处理跨域请求
        """
        def decorated(*args, **kwargs):
            # 处理预检请求
            if request.method == "OPTIONS":
                response = make_response()
                response.headers["Access-Control-Allow-Origin"] = "*"
                response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE, OPTIONS"
                response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
                response.headers["Access-Control-Max-Age"] = "3600"
                return response
            
            # 执行实际请求
            result = f(*args, **kwargs)
            
            # 添加CORS头
            if isinstance(result, tuple):
                response_data, status_code = result
                response = make_response(response_data, status_code)
            else:
                response = make_response(result)
            
            response.headers["Access-Control-Allow-Origin"] = "*"
            response.headers["Access-Control-Allow-Credentials"] = "true"
            
            return response
        
        return decorated