feat(models,utils): implement data models and utility classes

lsfx-mock-server/utils/__init__.py

@@ -0,0 +1 @@
+# Utils package

lsfx-mock-server/utils/error_simulator.py

@@ -0,0 +1,49 @@
+from typing import Dict, Optional
+import re
+
+
+class ErrorSimulator:
+    """错误场景模拟器"""
+
+    # 错误码映射表
+    ERROR_CODES = {
+        "40101": {"code": "40101", "message": "appId错误"},
+        "40102": {"code": "40102", "message": "appSecretCode错误"},
+        "40104": {"code": "40104", "message": "可使用项目次数为0，无法创建项目"},
+        "40105": {"code": "40105", "message": "只读模式下无法新建项目"},
+        "40106": {"code": "40106", "message": "错误的分析类型，不在规定的取值范围内"},
+        "40107": {"code": "40107", "message": "当前系统不支持的分析类型"},
+        "40108": {"code": "40108", "message": "当前用户所属行社无权限"},
+        "501014": {"code": "501014", "message": "无行内流水文件"},
+    }
+
+    @staticmethod
+    def detect_error_marker(value: str) -> Optional[str]:
+        """检测字符串中的错误标记
+
+        规则：如果字符串包含 error_XXXX，则返回 XXXX
+        例如：
+        - "project_error_40101" -> "40101"
+        - "test_error_501014" -> "501014"
+        """
+        if not value:
+            return None
+
+        pattern = r'error_(\d+)'
+        match = re.search(pattern, value)
+        if match:
+            return match.group(1)
+        return None
+
+    @staticmethod
+    def build_error_response(error_code: str) -> Optional[Dict]:
+        """构建错误响应"""
+        if error_code in ErrorSimulator.ERROR_CODES:
+            error_info = ErrorSimulator.ERROR_CODES[error_code]
+            return {
+                "code": error_info["code"],
+                "message": error_info["message"],
+                "status": error_info["code"],
+                "successResponse": False
+            }
+        return None

lsfx-mock-server/utils/response_builder.py

@@ -0,0 +1,69 @@
+import json
+from pathlib import Path
+from typing import Dict, Any
+import copy
+
+
+class ResponseBuilder:
+    """响应构建器"""
+
+    TEMPLATE_DIR = Path(__file__).parent.parent / "config" / "responses"
+
+    @staticmethod
+    def load_template(template_name: str) -> Dict:
+        """加载 JSON 模板
+
+        Args:
+            template_name: 模板名称（不含.json扩展名）
+
+        Returns:
+            模板字典
+        """
+        file_path = ResponseBuilder.TEMPLATE_DIR / f"{template_name}.json"
+        with open(file_path, 'r', encoding='utf-8') as f:
+            return json.load(f)
+
+    @staticmethod
+    def replace_placeholders(template: Dict, **kwargs) -> Dict:
+        """递归替换占位符
+
+        Args:
+            template: 模板字典
+            **kwargs: 占位符键值对
+
+        Returns:
+            替换后的字典
+        """
+        def replace_value(value):
+            if isinstance(value, str):
+                result = value
+                for key, val in kwargs.items():
+                    placeholder = f"{{{key}}}"
+                    if placeholder in result:
+                        result = result.replace(placeholder, str(val))
+                return result
+            elif isinstance(value, dict):
+                return {k: replace_value(v) for k, v in value.items()}
+            elif isinstance(value, list):
+                return [replace_value(item) for item in value]
+            return value
+
+        # 深拷贝模板，避免修改原始数据
+        return replace_value(copy.deepcopy(template))
+
+    @staticmethod
+    def build_success_response(template_name: str, **kwargs) -> Dict:
+        """构建成功响应
+
+        Args:
+            template_name: 模板名称
+            **kwargs: 占位符键值对
+
+        Returns:
+            响应字典
+        """
+        template = ResponseBuilder.load_template(template_name)
+        return ResponseBuilder.replace_placeholders(
+            template["success_response"],
+            **kwargs
+        )