深入Dify服务层模块：业务逻辑与领域服务架构

 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

class CodeExecutor:
    """
    安全代码执行器
    支持Python、Node.js、Go等多种语言的沙箱执行
    """
    
    SUPPORTED_LANGUAGES = {
        "python": PythonCodeExecutor,
        "javascript": NodeJSCodeExecutor, 
        "typescript": NodeJSCodeExecutor,
        "go": GoCodeExecutor
    }
    
    def execute_code(
        self,
        language: str,
        code: str,
        inputs: dict,
        timeout: int = 30
    ) -> CodeExecutionResult:
        """
        安全执行用户代码
        
        Args:
            language: 编程语言
            code: 待执行的代码
            inputs: 输入变量
            timeout: 执行超时时间
            
        Returns:
            CodeExecutionResult: 执行结果
        """
        # 1. 语言支持检查
        if language not in self.SUPPORTED_LANGUAGES:
            raise UnsupportedLanguageError(f"不支持的语言: {language}")
        
        # 2. 代码安全检查
        security_checker = CodeSecurityChecker()
        if not security_checker.is_safe(code, language):
            raise UnsafeCodeError("代码包含不安全的操作")
        
        # 3. 创建隔离执行环境
        executor = self.SUPPORTED_LANGUAGES[language](
            timeout=timeout,
            memory_limit="128MB",
            cpu_limit="1"
        )
        
        # 4. 执行代码
        return executor.run(code, inputs)

  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

class SecureCodeExecutor:
    """
    安全代码执行器实现
    基于容器化和资源限制的多重安全防护
    """
    
    def __init__(self):
        self.security_policies = {
            "python": PythonSecurityPolicy(),
            "javascript": JavaScriptSecurityPolicy(),
            "typescript": TypeScriptSecurityPolicy()
        }
        
        # 安全检查器
        self.security_checker = CodeSecurityChecker()
        
        # 资源监控器
        self.resource_monitor = ResourceMonitor()

    def execute_with_sandbox(
        self,
        language: str,
        code: str,
        inputs: dict,
        timeout: int = 30
    ) -> CodeExecutionResult:
        """
        沙箱化代码执行
        
        Args:
            language: 编程语言
            code: 用户代码
            inputs: 输入变量
            timeout: 执行超时
            
        Returns:
            CodeExecutionResult: 执行结果
        """
        # 1. 代码静态安全分析
        security_report = self.security_checker.analyze_code(code, language)
        if not security_report.is_safe:
            raise UnsafeCodeError(f"代码安全检查失败: {security_report.violations}")
        
        # 2. 创建隔离执行环境
        sandbox = self._create_sandbox(language)
        
        try:
            # 3. 资源限制设置
            self._apply_resource_limits(sandbox, {
                "memory_limit": "128MB",
                "cpu_limit": "100m",
                "disk_limit": "10MB",
                "network_access": False,
                "file_system_access": "read-only"
            })
            
            # 4. 执行代码
            with self.resource_monitor.monitor():
                result = sandbox.execute(
                    code=code,
                    inputs=inputs,
                    timeout=timeout
                )
            
            # 5. 结果安全检查
            sanitized_result = self._sanitize_output(result)
            
            return CodeExecutionResult(
                success=True,
                output=sanitized_result,
                execution_time=result.execution_time,
                memory_usage=result.memory_usage,
                security_violations=[]
            )
            
        except SecurityViolationError as e:
            logger.error(f"代码执行安全违规: {e}")
            return CodeExecutionResult(
                success=False,
                error=f"安全违规: {str(e)}",
                security_violations=[str(e)]
            )
        
        finally:
            # 6. 清理沙箱环境
            sandbox.cleanup()

    def _create_sandbox(self, language: str) -> CodeSandbox:
        """
        创建语言特定的沙箱环境
        """
        sandbox_configs = {
            "python": {
                "base_image": "python:3.11-alpine",
                "allowed_imports": [
                    "json", "math", "datetime", "re", "uuid",
                    "pandas", "numpy", "requests"
                ],
                "blocked_imports": [
                    "os", "sys", "subprocess", "socket", 
                    "threading", "multiprocessing"
                ]
            },
            
            "javascript": {
                "runtime": "node:18-alpine",
                "allowed_modules": [
                    "lodash", "moment", "axios", "crypto"
                ],
                "blocked_modules": [
                    "fs", "child_process", "cluster", "os"
                ]
            }
        }
        
        config = sandbox_configs.get(language)
        if not config:
            raise UnsupportedLanguageError(f"不支持的语言: {language}")
        
        return DockerSandbox(config)


class CodeSecurityChecker:
    """
    代码安全检查器
    基于AST分析和规则引擎的安全检查
    """
    
    def __init__(self):
        self.security_rules = self._load_security_rules()
        self.ast_analyzer = ASTSecurityAnalyzer()
    
    def analyze_code(self, code: str, language: str) -> SecurityReport:
        """
        分析代码安全性
        
        Args:
            code: 源代码
            language: 编程语言
            
        Returns:
            SecurityReport: 安全分析报告
        """
        violations = []
        
        # 1. AST分析
        if language == "python":
            violations.extend(self._analyze_python_ast(code))
        elif language in ["javascript", "typescript"]:
            violations.extend(self._analyze_js_ast(code))
        
        # 2. 字符串模式匹配
        violations.extend(self._pattern_based_analysis(code, language))
        
        # 3. 动态分析（如果启用）
        if self.security_rules.get("enable_dynamic_analysis", False):
            violations.extend(self._dynamic_analysis(code, language))
        
        return SecurityReport(
            is_safe=len(violations) == 0,
            violations=violations,
            severity_score=self._calculate_severity_score(violations)
        )
    
    def _analyze_python_ast(self, code: str) -> list[SecurityViolation]:
        """Python AST安全分析"""
        violations = []
        
        try:
            tree = ast.parse(code)
            
            for node in ast.walk(tree):
                # 检查危险函数调用
                if isinstance(node, ast.Call):
                    if hasattr(node.func, 'id'):
                        func_name = node.func.id
                        if func_name in self.security_rules["python"]["blocked_functions"]:
                            violations.append(SecurityViolation(
                                type="dangerous_function_call",
                                message=f"禁止调用函数: {func_name}",
                                line_number=node.lineno
                            ))
                
                # 检查导入模块
                elif isinstance(node, ast.Import):
                    for alias in node.names:
                        if alias.name in self.security_rules["python"]["blocked_imports"]:
                            violations.append(SecurityViolation(
                                type="blocked_import",
                                message=f"禁止导入模块: {alias.name}",
                                line_number=node.lineno
                            ))
        
        except SyntaxError as e:
            violations.append(SecurityViolation(
                type="syntax_error",
                message=f"代码语法错误: {str(e)}",
                line_number=e.lineno
            ))
        
        return violations

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 企业级部署优化配置
CELERY_WORKER_MAX_MEMORY_PER_CHILD = 200000  # 工作进程最大内存
CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000     # 工作进程最大任务数

# API工具配置优化
API_TOOL_DEFAULT_CONNECT_TIMEOUT = 10        # 连接超时
API_TOOL_DEFAULT_READ_TIMEOUT = 60           # 读取超时

# 中国区部署优化
PIP_MIRROR_URL = "https://pypi.tuna.tsinghua.edu.cn/simple"  # 使用清华镜像源

  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
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676

class AppService:
    """
    应用管理服务
    负责应用的完整生命周期管理
    """

    def create_app(self, tenant_id: str, args: dict, account: Account) -> App:
        """
        创建应用
        实现应用的初始化和配置
        
        Args:
            tenant_id: 租户ID
            args: 创建参数
            account: 创建者账户
            
        Returns:
            App: 创建的应用实例
        """
        # 1. 验证应用模式
        app_mode = AppMode.value_of(args["mode"])
        app_template = default_app_templates[app_mode]

        # 2. 获取默认模型配置
        default_model_config = app_template.get("model_config", {}).copy()
        
        if default_model_config and "model" in default_model_config:
            try:
                # 获取默认模型实例
                model_manager = ModelManager()
                model_instance = model_manager.get_default_model_instance(
                    tenant_id=tenant_id, 
                    model_type=ModelType.LLM
                )
                
                # 更新模型配置
                default_model_config.update({
                    "provider": model_instance.provider,
                    "model": model_instance.model,
                })
                
            except Exception as e:
                logger.exception("获取默认模型失败")
                # 使用模板默认配置
                pass

        # 3. 创建应用实例
        app = App(
            id=str(uuid.uuid4()),
            tenant_id=tenant_id,
            name=args["name"],
            mode=app_mode,
            icon=args.get("icon"),
            icon_background=args.get("icon_background"),
            description=args.get("description", ""),
            created_by=account.id,
            updated_by=account.id,
            created_at=naive_utc_now(),
            updated_at=naive_utc_now()
        )

        # 4. 创建应用模型配置
        app_model_config = AppModelConfig(
            id=str(uuid.uuid4()),
            app_id=app.id,
            provider=default_model_config.get("provider"),
            model_id=default_model_config.get("model"),
            configs=default_model_config,
            created_by=account.id,
            updated_by=account.id
        )

        # 5. 保存到数据库
        db.session.add(app)
        db.session.add(app_model_config)
        db.session.commit()

        # 6. 触发应用创建事件
        app_was_created.send(app, account=account)

        return app

    def get_paginate_apps(
        self, 
        user_id: str, 
        tenant_id: str, 
        args: dict
    ) -> Optional[Pagination]:
        """
        分页获取应用列表
        支持多种过滤条件的应用查询
        
        Args:
            user_id: 用户ID
            tenant_id: 租户ID
            args: 查询参数
            
        Returns:
            Optional[Pagination]: 分页结果
        """
        # 构建过滤条件
        filters = [
            App.tenant_id == tenant_id,
            App.is_universal == False
        ]

        # 应用模式过滤
        mode_filters = {
            "workflow": App.mode == AppMode.WORKFLOW,
            "completion": App.mode == AppMode.COMPLETION,
            "chat": App.mode == AppMode.CHAT,
            "advanced-chat": App.mode == AppMode.ADVANCED_CHAT,
            "agent-chat": App.mode == AppMode.AGENT_CHAT,
        }
        
        if args["mode"] in mode_filters:
            filters.append(mode_filters[args["mode"]])

        # 创建者过滤
        if args.get("is_created_by_me", False):
            filters.append(App.created_by == user_id)

        # 名称模糊搜索
        if args.get("name"):
            name = args["name"][:30]  # 限制搜索长度
            filters.append(App.name.ilike(f"%{name}%"))

        # 标签过滤
        if args.get("tag_ids") and len(args["tag_ids"]) > 0:
            target_ids = TagService.get_target_ids_by_tag_ids(
                "app", tenant_id, args["tag_ids"]
            )
            if target_ids:
                filters.append(App.id.in_(target_ids))
            else:
                return None

        # 执行分页查询
        return db.paginate(
            db.select(App).where(*filters).order_by(App.created_at.desc()),
            page=args["page"],
            per_page=args["limit"],
            error_out=False,
        )

    def update_app_model_config(
        self, 
        app_id: str, 
        app_model_config: dict, 
        account: Account
    ) -> AppModelConfig:
        """
        更新应用模型配置
        
        Args:
            app_id: 应用ID
            app_model_config: 新的模型配置
            account: 操作账户
            
        Returns:
            AppModelConfig: 更新后的配置
        """
        # 验证配置
        self._validate_model_config(app_model_config)
        
        # 获取当前配置
        current_config = AppModelConfig.query.filter_by(app_id=app_id).first()
        if not current_config:
            raise ValueError("应用配置不存在")

        # 更新配置
        current_config.configs = app_model_config
        current_config.updated_by = account.id
        current_config.updated_at = naive_utc_now()
        
        db.session.commit()
        
        # 触发配置更新事件
        from events.app_event import app_model_config_was_updated
        app_model_config_was_updated.send(current_config)
        
        return current_config

    def _validate_model_config(self, config: dict):
        """验证模型配置"""
        required_fields = ["provider", "model"]
        for field in required_fields:
            if field not in config:
                raise ValueError(f"模型配置缺少必填字段: {field}")


class DatasetService:
    """
    数据集管理服务
    负责知识库的完整生命周期管理
    """

    def create_dataset(
        self, 
        account: Account, 
        name: str, 
        data_source_type: str,
        indexing_technique: str = "high_quality",
        description: Optional[str] = None
    ) -> Dataset:
        """
        创建数据集
        
        Args:
            account: 创建者账户
            name: 数据集名称
            data_source_type: 数据源类型
            indexing_technique: 索引技术
            description: 可选描述
            
        Returns:
            Dataset: 创建的数据集
        """
        # 验证租户资源配额
        if not self._check_dataset_quota(account.current_tenant_id):
            raise QuotaExceededError("数据集配额已满")

        # 创建数据集实例
        dataset = Dataset(
            id=str(uuid.uuid4()),
            tenant_id=account.current_tenant_id,
            name=name,
            description=description or "",
            data_source_type=data_source_type,
            indexing_technique=indexing_technique,
            created_by=account.id,
            updated_by=account.id
        )

        # 保存到数据库
        db.session.add(dataset)
        db.session.commit()

        # 触发数据集创建事件
        from events.dataset_event import dataset_was_created
        dataset_was_created.send(dataset, account=account)

        return dataset

    def process_document(
        self,
        dataset_id: str,
        document_data: dict,
        account: Account
    ) -> Document:
        """
        处理文档
        文档的提取、分割、向量化和索引
        
        Args:
            dataset_id: 数据集ID
            document_data: 文档数据
            account: 操作账户
            
        Returns:
            Document: 处理后的文档
        """
        # 获取数据集
        dataset = self.get_dataset(dataset_id)
        if not dataset:
            raise ValueError("数据集不存在")

        # 创建文档处理任务
        from tasks.document_indexing_task import document_indexing_task
        
        task_id = str(uuid.uuid4())
        
        # 异步处理文档
        document_indexing_task.delay(
            dataset_id=dataset_id,
            document_data=document_data,
            user_id=account.id,
            task_id=task_id
        )

        return task_id

    def _check_dataset_quota(self, tenant_id: str) -> bool:
        """检查数据集配额"""
        if not dify_config.BILLING_ENABLED:
            return True
            
        return BillingService.check_resource_quota(
            tenant_id=tenant_id,
            resource_type="datasets"
        )


class WorkflowService:
    """
    工作流服务
    负责工作流的设计、执行和管理
    """

    def run_workflow(
        self,
        app_model: App,
        workflow: Workflow,
        user_inputs: dict,
        user: Account,
        invoke_from: InvokeFrom = InvokeFrom.DEBUGGER
    ) -> Generator:
        """
        运行工作流
        
        Args:
            app_model: 应用模型
            workflow: 工作流实例
            user_inputs: 用户输入
            user: 执行用户
            invoke_from: 调用来源
            
        Yields:
            工作流执行事件
        """
        # 创建工作流入口
        workflow_entry = WorkflowEntry(
            tenant_id=app_model.tenant_id,
            app_id=app_model.id,
            workflow_id=workflow.id,
            user_id=user.id,
            invoke_from=invoke_from
        )

        # 执行工作流
        try:
            yield from workflow_entry.run(
                inputs=user_inputs,
                files=[]
            )
        except Exception as e:
            logger.exception("工作流执行失败")
            raise WorkflowExecutionError(f"工作流执行失败: {e}")

    def save_draft_workflow(
        self,
        app_model: App,
        graph_config: dict,
        features_config: dict,
        account: Account
    ) -> Workflow:
        """
        保存草稿工作流
        
        Args:
            app_model: 应用模型
            graph_config: 图配置
            features_config: 功能配置
            account: 操作账户
            
        Returns:
            Workflow: 保存的工作流
        """
        # 验证图配置
        self._validate_graph_config(graph_config)
        
        # 获取或创建草稿工作流
        draft_workflow = self._get_or_create_draft_workflow(app_model)
        
        # 更新配置
        draft_workflow.graph = graph_config
        draft_workflow.features = features_config
        draft_workflow.updated_by = account.id
        draft_workflow.updated_at = naive_utc_now()
        
        db.session.commit()
        
        return draft_workflow


class AppGenerateService:
    """
    应用生成服务
    处理各种应用类型的内容生成
    """

    @classmethod
    def generate(
        cls,
        app_model: App,
        user: Union[Account, EndUser],
        args: dict,
        invoke_from: InvokeFrom,
        streaming: bool = True
    ) -> Generator:
        """
        生成应用响应
        根据应用类型调用相应的生成器
        
        Args:
            app_model: 应用模型
            user: 用户实例
            args: 生成参数
            invoke_from: 调用来源
            streaming: 是否流式输出
            
        Yields:
            生成过程事件
        """
        # 频率限制检查
        cls._check_rate_limit(user, app_model)
        
        # 根据应用模式选择生成器
        generator_mapping = {
            AppMode.COMPLETION: CompletionAppGenerator,
            AppMode.CHAT: ChatAppGenerator,
            AppMode.AGENT_CHAT: AgentChatAppGenerator,
            AppMode.WORKFLOW: WorkflowAppGenerator,
            AppMode.ADVANCED_CHAT: AdvancedChatAppGenerator,
        }
        
        generator_class = generator_mapping.get(app_model.mode)
        if not generator_class:
            raise ValueError(f"不支持的应用模式: {app_model.mode}")

        # 创建生成器实例
        generator = generator_class(
            app_model=app_model,
            user=user,
            invoke_from=invoke_from
        )

        # 执行生成
        try:
            yield from generator.generate(
                inputs=args.get("inputs", {}),
                query=args.get("query", ""),
                files=args.get("files", []),
                conversation_id=args.get("conversation_id"),
                streaming=streaming
            )
        except Exception as e:
            logger.exception("应用生成失败")
            raise AppGenerateError(f"生成失败: {e}")

    @classmethod
    def _check_rate_limit(cls, user: Union[Account, EndUser], app_model: App):
        """检查频率限制"""
        # 系统级频率限制
        if not cls.system_rate_limiter.check_request_limit(user.id):
            raise InvokeRateLimitError("系统调用频率超限")
        
        # 应用级频率限制
        app_rate_limit = RateLimit.from_app_config(app_model)
        if app_rate_limit and not app_rate_limit.check_request_limit(user.id):
            raise InvokeRateLimitError("应用调用频率超限")


class AccountService:
    """
    账户管理服务
    处理用户账户和租户的完整管理
    """

    def create_account(
        self,
        email: str,
        name: str,
        password: str,
        interface_language: str = "en-US",
        timezone: str = "UTC"
    ) -> Account:
        """
        创建用户账户
        
        Args:
            email: 邮箱地址
            name: 用户名
            password: 密码
            interface_language: 界面语言
            timezone: 时区
            
        Returns:
            Account: 创建的账户
        """
        # 1. 验证邮箱是否已存在
        if self._email_exists(email):
            raise AccountRegisterError("邮箱地址已注册")

        # 2. 验证密码强度
        if not self._validate_password_strength(password):
            raise AccountPasswordError("密码强度不足")

        # 3. 创建账户
        account = Account(
            id=str(uuid.uuid4()),
            email=email,
            name=name,
            password=hash_password(password),
            interface_language=interface_language,
            timezone=timezone,
            status=AccountStatus.ACTIVE,
            created_at=naive_utc_now()
        )

        # 4. 创建默认租户
        tenant = self._create_default_tenant(account)
        
        # 5. 建立账户租户关联
        tenant_account_join = TenantAccountJoin(
            tenant_id=tenant.id,
            account_id=account.id,
            role=TenantAccountRole.OWNER,
            created_at=naive_utc_now()
        )

        # 6. 保存到数据库
        db.session.add_all([account, tenant, tenant_account_join])
        db.session.commit()

        # 7. 触发账户创建事件
        from events.tenant_event import tenant_was_created
        tenant_was_created.send(tenant, account=account)

        return account

    def authenticate(self, email: str, password: str) -> Optional[Account]:
        """
        用户认证
        
        Args:
            email: 邮箱
            password: 密码
            
        Returns:
            Optional[Account]: 认证成功的账户
        """
        # 查找账户
        account = Account.query.filter_by(email=email).first()
        if not account:
            return None

        # 验证密码
        if not compare_password(password, account.password):
            return None

        # 检查账户状态
        if account.status != AccountStatus.ACTIVE:
            raise AccountLoginError("账户已被禁用")

        # 更新最后登录时间
        account.last_login_at = naive_utc_now()
        db.session.commit()

        return account

    def _create_default_tenant(self, account: Account) -> Tenant:
        """创建默认租户"""
        return Tenant(
            id=str(uuid.uuid4()),
            name=f"{account.name}'s workspace",
            status=TenantStatus.NORMAL,
            created_at=naive_utc_now()
        )


class BillingService:
    """
    计费管理服务
    处理订阅、配额和计费相关功能
    """

    @staticmethod
    def check_resource_quota(tenant_id: str, resource_type: str) -> bool:
        """
        检查资源配额
        
        Args:
            tenant_id: 租户ID
            resource_type: 资源类型
            
        Returns:
            bool: 是否有剩余配额
        """
        if not dify_config.BILLING_ENABLED:
            return True

        try:
            # 获取租户订阅信息
            subscription = cls._get_tenant_subscription(tenant_id)
            if not subscription:
                return False

            # 获取当前使用量
            current_usage = cls._get_resource_usage(tenant_id, resource_type)
            
            # 获取配额限制
            quota_limit = subscription.get_quota_limit(resource_type)
            
            return current_usage < quota_limit

        except Exception as e:
            logger.exception("配额检查失败")
            return False

    @staticmethod
    def record_usage(
        tenant_id: str,
        resource_type: str,
        usage_amount: int,
        metadata: Optional[dict] = None
    ):
        """
        记录资源使用
        
        Args:
            tenant_id: 租户ID
            resource_type: 资源类型
            usage_amount: 使用量
            metadata: 可选元数据
        """
        if not dify_config.BILLING_ENABLED:
            return

        try:
            # 记录使用量到计费系统
            billing_record = BillingRecord(
                tenant_id=tenant_id,
                resource_type=resource_type,
                usage_amount=usage_amount,
                metadata=metadata or {},
                recorded_at=naive_utc_now()
            )
            
            db.session.add(billing_record)
            db.session.commit()

        except Exception as e:
            logger.exception("使用量记录失败")


class FeatureService:
    """
    功能特性服务
    管理系统和租户级别的功能开关
    """

    @classmethod
    def get_features(cls, tenant_id: str) -> FeatureModel:
        """
        获取功能特性配置
        
        Args:
            tenant_id: 租户ID
            
        Returns:
            FeatureModel: 功能特性配置
        """
        features = FeatureModel()

        # 从环境变量填充基础功能
        cls._fulfill_params_from_env(features)

        # 如果启用计费，从计费API获取功能配置
        if dify_config.BILLING_ENABLED and tenant_id:
            cls._fulfill_params_from_billing_api(features, tenant_id)

        # 如果启用企业版，应用企业功能
        if dify_config.ENTERPRISE_ENABLED:
            features.webapp_copyright_enabled = True
            cls._fulfill_params_from_workspace_info(features, tenant_id)

        return features

    @classmethod
    def _fulfill_params_from_env(cls, features: FeatureModel):
        """从环境变量填充功能参数"""
        features.can_replace_logo = dify_config.CAN_REPLACE_LOGO
        features.model_load_balancing_enabled = dify_config.MODEL_LB_ENABLED
        features.dataset_operator_enabled = dify_config.DATASET_OPERATOR_ENABLED
        features.member_invite_enabled = dify_config.INVITE_EXPIRY_HOURS > 0

  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

class ServiceOrchestrator:
    """
    服务编排器
    协调多个服务之间的复杂业务流程
    """

    def __init__(self):
        """初始化服务编排器"""
        self.app_service = AppService()
        self.dataset_service = DatasetService()
        self.workflow_service = WorkflowService()
        self.account_service = AccountService()
        self.billing_service = BillingService()

    def create_complete_app(
        self,
        tenant_id: str,
        app_config: dict,
        dataset_configs: list[dict],
        workflow_config: Optional[dict],
        account: Account
    ) -> dict:
        """
        创建完整应用
        包含应用、数据集、工作流的一站式创建
        
        Args:
            tenant_id: 租户ID
            app_config: 应用配置
            dataset_configs: 数据集配置列表
            workflow_config: 可选的工作流配置
            account: 创建者账户
            
        Returns:
            dict: 创建结果
        """
        created_resources = {
            "app": None,
            "datasets": [],
            "workflow": None,
            "errors": []
        }

        try:
            # 1. 检查资源配额
            self._validate_resource_quotas(tenant_id, app_config, dataset_configs)

            # 2. 创建应用
            app = self.app_service.create_app(
                tenant_id=tenant_id,
                args=app_config,
                account=account
            )
            created_resources["app"] = app

            # 3. 创建数据集
            for dataset_config in dataset_configs:
                try:
                    dataset = self.dataset_service.create_dataset(
                        account=account,
                        **dataset_config
                    )
                    created_resources["datasets"].append(dataset)
                except Exception as e:
                    created_resources["errors"].append({
                        "type": "dataset_creation",
                        "config": dataset_config,
                        "error": str(e)
                    })

            # 4. 创建工作流（如果提供）
            if workflow_config:
                try:
                    workflow = self.workflow_service.create_workflow(
                        app_id=app.id,
                        graph_config=workflow_config,
                        account=account
                    )
                    created_resources["workflow"] = workflow
                except Exception as e:
                    created_resources["errors"].append({
                        "type": "workflow_creation",
                        "error": str(e)
                    })

            # 5. 记录资源使用
            self.billing_service.record_usage(
                tenant_id=tenant_id,
                resource_type="apps",
                usage_amount=1,
                metadata={"app_id": app.id}
            )

            return created_resources

        except Exception as e:
            # 清理已创建的资源
            self._cleanup_created_resources(created_resources)
            raise e

    def _validate_resource_quotas(
        self,
        tenant_id: str,
        app_config: dict,
        dataset_configs: list[dict]
    ):
        """验证资源配额"""
        # 检查应用配额
        if not self.billing_service.check_resource_quota(tenant_id, "apps"):
            raise QuotaExceededError("应用配额已满")

        # 检查数据集配额
        required_datasets = len(dataset_configs)
        if required_datasets > 0:
            for _ in range(required_datasets):
                if not self.billing_service.check_resource_quota(tenant_id, "datasets"):
                    raise QuotaExceededError("数据集配额不足")

    def _cleanup_created_resources(self, resources: dict):
        """清理创建的资源"""
        try:
            # 删除创建的应用
            if resources["app"]:
                db.session.delete(resources["app"])

            # 删除创建的数据集
            for dataset in resources["datasets"]:
                db.session.delete(dataset)

            # 删除创建的工作流
            if resources["workflow"]:
                db.session.delete(resources["workflow"])

            db.session.commit()

        except Exception as e:
            logger.exception("资源清理失败")


class TaskService:
    """
    任务管理服务
    处理异步任务的调度和监控
    """

    def __init__(self):
        """初始化任务服务"""
        self.task_registry = {}
        self.task_status_cache = {}

    def submit_async_task(
        self,
        task_type: str,
        task_params: dict,
        user_id: str,
        priority: int = 0
    ) -> str:
        """
        提交异步任务
        
        Args:
            task_type: 任务类型
            task_params: 任务参数
            user_id: 用户ID
            priority: 优先级
            
        Returns:
            str: 任务ID
        """
        task_id = str(uuid.uuid4())
        
        # 创建任务记录
        task_record = {
            "task_id": task_id,
            "task_type": task_type,
            "params": task_params,
            "user_id": user_id,
            "priority": priority,
            "status": "pending",
            "created_at": time.time()
        }

        # 根据任务类型分发到相应队列
        task_mapping = {
            "document_indexing": self._submit_document_indexing_task,
            "model_fine_tuning": self._submit_model_fine_tuning_task,
            "dataset_export": self._submit_dataset_export_task,
            "app_migration": self._submit_app_migration_task,
        }

        task_handler = task_mapping.get(task_type)
        if not task_handler:
            raise ValueError(f"不支持的任务类型: {task_type}")

        # 提交任务
        task_handler(task_record)
        
        # 缓存任务状态
        self.task_status_cache[task_id] = task_record
        
        return task_id

    def get_task_status(self, task_id: str) -> Optional[dict]:
        """
        获取任务状态
        
        Args:
            task_id: 任务ID
            
        Returns:
            Optional[dict]: 任务状态信息
        """
        # 先从缓存获取
        if task_id in self.task_status_cache:
            return self.task_status_cache[task_id]

        # 从数据库获取
        task_record = TaskRecord.query.filter_by(task_id=task_id).first()
        if task_record:
            return task_record.to_dict()

        return None

    def _submit_document_indexing_task(self, task_record: dict):
        """提交文档索引任务"""
        from tasks.document_indexing_task import document_indexing_task
        
        document_indexing_task.delay(
            task_id=task_record["task_id"],
            **task_record["params"]
        )

    def _submit_model_fine_tuning_task(self, task_record: dict):
        """提交模型微调任务"""
        from tasks.model_fine_tuning_task import model_fine_tuning_task
        
        model_fine_tuning_task.delay(
            task_id=task_record["task_id"],
            **task_record["params"]
        )

  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

class TransactionalService:
    """
    事务性服务基类
    提供数据库事务管理和回滚机制
    """

    def __init__(self):
        self.db_session = db.session

    def execute_with_transaction(self, operations: list[Callable]):
        """
        在事务中执行操作列表
        
        Args:
            operations: 操作函数列表
        """
        try:
            self.db_session.begin()
            
            for operation in operations:
                operation()
            
            self.db_session.commit()
            
        except Exception as e:
            self.db_session.rollback()
            logger.exception("事务执行失败")
            raise e

    def rollback_safe_operation(self, operation: Callable, rollback_operation: Callable):
        """
        安全操作（支持回滚）
        
        Args:
            operation: 主操作
            rollback_operation: 回滚操作
        """
        try:
            return operation()
        except Exception as e:
            try:
                rollback_operation()
            except Exception as rollback_error:
                logger.exception("回滚操作失败")
            raise e


class CacheableService:
    """
    可缓存服务基类
    提供服务级别的缓存机制
    """

    def __init__(self, cache_prefix: str, default_ttl: int = 300):
        self.cache_prefix = cache_prefix
        self.default_ttl = default_ttl
        self.redis_client = redis_client

    def get_cached_result(
        self,
        cache_key: str,
        fetch_function: Callable,
        ttl: Optional[int] = None
    ):
        """
        获取缓存结果
        
        Args:
            cache_key: 缓存键
            fetch_function: 数据获取函数
            ttl: 可选的TTL
            
        Returns:
            缓存或新获取的结果
        """
        full_key = f"{self.cache_prefix}:{cache_key}"
        
        # 尝试从缓存获取
        cached_data = self.redis_client.get(full_key)
        if cached_data:
            try:
                return json.loads(cached_data)
            except json.JSONDecodeError:
                pass

        # 缓存未命中，获取新数据
        result = fetch_function()
        
        # 写入缓存
        try:
            self.redis_client.setex(
                full_key,
                ttl or self.default_ttl,
                json.dumps(result, ensure_ascii=False, default=str)
            )
        except Exception as e:
            logger.warning(f"缓存写入失败: {e}")

        return result

    def invalidate_cache(self, cache_key: str):
        """失效缓存"""
        full_key = f"{self.cache_prefix}:{cache_key}"
        self.redis_client.delete(full_key)