""" 映射模型 定义映射关系的数据结构和业务逻辑 """ from dataclasses import dataclass, field from datetime import datetime from typing import Optional, List, Union from enum import Enum from config import OperatorType, LanguageType from models.field import Field class MappingType(Enum): """映射类型枚举""" OPERATOR = "operator" CODE = "code" COMPOSITE = "composite" @dataclass class OperatorMapping: """ 操作符映射 Attributes: id: 映射ID source_field: 源字段 target_field: 目标字段 operator: 操作符类型 operand: 操作数 description: 描述信息 """ source_field: Field target_field: Field operator: OperatorType id: Optional[int] = None operand: Optional[str] = None description: Optional[str] = None created_time: Optional[datetime] = None updated_time: Optional[datetime] = None def __post_init__(self): """初始化后处理""" if isinstance(self.operator, str): self.operator = OperatorType(self.operator) def validate(self) -> tuple[bool, str]: """验证映射的有效性""" if not self.source_field: return False, "源字段不能为空" if not self.target_field: return False, "目标字段不能为空" if not self.operator: return False, "操作符不能为空" # 验证操作符与字段类型的兼容性 if self.operator in [OperatorType.ADD, OperatorType.SUBTRACT, OperatorType.MULTIPLY, OperatorType.DIVIDE]: from config import FieldType numeric_types = [FieldType.INT, FieldType.FLOAT, FieldType.DOUBLE] if self.source_field.type not in numeric_types: return False, f"操作符 {self.operator.value} 需要数值类型的源字段" return True, "" def to_dict(self) -> dict: """转换为字典""" return { 'id': self.id, 'mapping_type': MappingType.OPERATOR.value, 'source_field': self.source_field.to_dict() if self.source_field else None, 'target_field': self.target_field.to_dict() if self.target_field else None, 'operator': self.operator.value if isinstance(self.operator, OperatorType) else self.operator, 'operand': self.operand, 'description': self.description, 'created_time': self.created_time.isoformat() if self.created_time else None, 'updated_time': self.updated_time.isoformat() if self.updated_time else None, } @dataclass class CodeMapping: """ 代码映射 Attributes: id: 映射ID source_fields: 源字段列表(可多个) target_field: 目标字段 language: 编程语言 code: 转换代码 description: 描述信息 """ source_fields: List[Field] target_field: Field language: LanguageType code: str id: Optional[int] = None description: Optional[str] = None created_time: Optional[datetime] = None updated_time: Optional[datetime] = None def __post_init__(self): """初始化后处理""" if isinstance(self.language, str): self.language = LanguageType(self.language) def validate(self) -> tuple[bool, str]: """验证映射的有效性""" if not self.source_fields or len(self.source_fields) == 0: return False, "源字段列表不能为空" if not self.target_field: return False, "目标字段不能为空" if not self.language: return False, "编程语言不能为空" if not self.code or not self.code.strip(): return False, "转换代码不能为空" return True, "" def to_dict(self) -> dict: """转换为字典""" return { 'id': self.id, 'mapping_type': MappingType.CODE.value, 'source_fields': [f.to_dict() for f in self.source_fields], 'target_field': self.target_field.to_dict() if self.target_field else None, 'language': self.language.value if isinstance(self.language, LanguageType) else self.language, 'code': self.code, 'description': self.description, 'created_time': self.created_time.isoformat() if self.created_time else None, 'updated_time': self.updated_time.isoformat() if self.updated_time else None, } @dataclass class CompositeMapping: """ 复合映射 Attributes: id: 映射ID name: 映射名称 mappings: 子映射列表 description: 描述信息 """ name: str mappings: List[Union[OperatorMapping, CodeMapping]] id: Optional[int] = None description: Optional[str] = None created_time: Optional[datetime] = None updated_time: Optional[datetime] = None def validate(self) -> tuple[bool, str]: """验证映射的有效性""" if not self.name: return False, "映射名称不能为空" if not self.mappings or len(self.mappings) == 0: return False, "子映射列表不能为空" # 验证所有子映射 for mapping in self.mappings: is_valid, error_msg = mapping.validate() if not is_valid: return False, f"子映射验证失败: {error_msg}" return True, "" def to_dict(self) -> dict: """转换为字典""" return { 'id': self.id, 'mapping_type': MappingType.COMPOSITE.value, 'name': self.name, 'mappings': [m.to_dict() for m in self.mappings], 'description': self.description, 'created_time': self.created_time.isoformat() if self.created_time else None, 'updated_time': self.updated_time.isoformat() if self.updated_time else None, } # 映射类型联合 Mapping = Union[OperatorMapping, CodeMapping, CompositeMapping]