195 lines
6.0 KiB
Python
195 lines
6.0 KiB
Python
|
|
"""
|
|||
|
|
字段业务模型
|
|||
|
|
定义字段的数据结构和业务逻辑
|
|||
|
|
"""
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
from datetime import datetime
|
|||
|
|
from typing import Optional, List
|
|||
|
|
from config import FieldType
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class Field:
|
|||
|
|
"""
|
|||
|
|
字段模型类
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
id: 字段ID
|
|||
|
|
full_name: 完整名称(如 gis.map.location)
|
|||
|
|
domain: 域名(如 gis)
|
|||
|
|
sub_domains: 子域列表(如 ['map'])
|
|||
|
|
name: 字段名(如 location)
|
|||
|
|
type: 字段类型
|
|||
|
|
range_min: 最小值
|
|||
|
|
range_max: 最大值
|
|||
|
|
default_value: 默认值
|
|||
|
|
unit: 单位
|
|||
|
|
description: 描述信息
|
|||
|
|
created_time: 创建时间
|
|||
|
|
updated_time: 更新时间
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
full_name: str
|
|||
|
|
domain: str
|
|||
|
|
name: str
|
|||
|
|
type: FieldType
|
|||
|
|
id: Optional[int] = None
|
|||
|
|
sub_domains: Optional[List[str]] = field(default_factory=list)
|
|||
|
|
range_min: Optional[str] = None
|
|||
|
|
range_max: Optional[str] = None
|
|||
|
|
default_value: Optional[str] = None
|
|||
|
|
unit: Optional[str] = None
|
|||
|
|
description: Optional[str] = None
|
|||
|
|
created_time: Optional[datetime] = None
|
|||
|
|
updated_time: Optional[datetime] = None
|
|||
|
|
|
|||
|
|
def __post_init__(self):
|
|||
|
|
"""初始化后处理"""
|
|||
|
|
# 确保type是FieldType枚举
|
|||
|
|
if isinstance(self.type, str):
|
|||
|
|
self.type = FieldType(self.type)
|
|||
|
|
|
|||
|
|
# 从full_name解析domain、sub_domains和name
|
|||
|
|
if self.full_name and not self.domain:
|
|||
|
|
self._parse_full_name()
|
|||
|
|
|
|||
|
|
def _parse_full_name(self):
|
|||
|
|
"""从完整名称解析域、子域和字段名"""
|
|||
|
|
parts = self.full_name.split('.')
|
|||
|
|
if len(parts) >= 2:
|
|||
|
|
self.domain = parts[0]
|
|||
|
|
self.name = parts[-1]
|
|||
|
|
if len(parts) > 2:
|
|||
|
|
self.sub_domains = parts[1:-1]
|
|||
|
|
elif len(parts) == 1:
|
|||
|
|
self.domain = "default"
|
|||
|
|
self.name = parts[0]
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def build_full_name(domain: str, name: str, sub_domains: Optional[List[str]] = None) -> str:
|
|||
|
|
"""
|
|||
|
|
构建完整字段名
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
domain: 域名
|
|||
|
|
name: 字段名
|
|||
|
|
sub_domains: 子域列表
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
完整字段名
|
|||
|
|
"""
|
|||
|
|
parts = [domain]
|
|||
|
|
if sub_domains:
|
|||
|
|
parts.extend(sub_domains)
|
|||
|
|
parts.append(name)
|
|||
|
|
return '.'.join(parts)
|
|||
|
|
|
|||
|
|
def get_sub_domains_str(self) -> str:
|
|||
|
|
"""获取子域的字符串表示(用于数据库存储)"""
|
|||
|
|
if self.sub_domains:
|
|||
|
|
return '.'.join(self.sub_domains)
|
|||
|
|
return ''
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def parse_sub_domains(sub_domains_str: str) -> List[str]:
|
|||
|
|
"""解析子域字符串"""
|
|||
|
|
if sub_domains_str:
|
|||
|
|
return sub_domains_str.split('.')
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
def validate(self) -> tuple[bool, str]:
|
|||
|
|
"""
|
|||
|
|
验证字段数据的有效性
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(是否有效, 错误信息)
|
|||
|
|
"""
|
|||
|
|
if not self.full_name:
|
|||
|
|
return False, "字段完整名称不能为空"
|
|||
|
|
|
|||
|
|
if not self.domain:
|
|||
|
|
return False, "字段域不能为空"
|
|||
|
|
|
|||
|
|
if not self.name:
|
|||
|
|
return False, "字段名不能为空"
|
|||
|
|
|
|||
|
|
if not self.type:
|
|||
|
|
return False, "字段类型不能为空"
|
|||
|
|
|
|||
|
|
# 验证范围值
|
|||
|
|
if self.range_min and self.range_max:
|
|||
|
|
try:
|
|||
|
|
if self.type in [FieldType.INT, FieldType.FLOAT, FieldType.DOUBLE]:
|
|||
|
|
min_val = float(self.range_min)
|
|||
|
|
max_val = float(self.range_max)
|
|||
|
|
if min_val > max_val:
|
|||
|
|
return False, "最小值不能大于最大值"
|
|||
|
|
except ValueError:
|
|||
|
|
return False, "范围值格式不正确"
|
|||
|
|
|
|||
|
|
return True, ""
|
|||
|
|
|
|||
|
|
def to_dict(self) -> dict:
|
|||
|
|
"""转换为字典"""
|
|||
|
|
return {
|
|||
|
|
'id': self.id,
|
|||
|
|
'full_name': self.full_name,
|
|||
|
|
'domain': self.domain,
|
|||
|
|
'sub_domains': self.get_sub_domains_str(),
|
|||
|
|
'name': self.name,
|
|||
|
|
'type': self.type.value if isinstance(self.type, FieldType) else self.type,
|
|||
|
|
'range_min': self.range_min,
|
|||
|
|
'range_max': self.range_max,
|
|||
|
|
'default_value': self.default_value,
|
|||
|
|
'unit': self.unit,
|
|||
|
|
'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,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def from_dict(cls, data: dict) -> 'Field':
|
|||
|
|
"""从字典创建对象"""
|
|||
|
|
# 解析子域
|
|||
|
|
sub_domains_str = data.get('sub_domains', '')
|
|||
|
|
sub_domains = cls.parse_sub_domains(sub_domains_str) if sub_domains_str else []
|
|||
|
|
|
|||
|
|
# 解析时间
|
|||
|
|
created_time = None
|
|||
|
|
if data.get('created_time'):
|
|||
|
|
if isinstance(data['created_time'], str):
|
|||
|
|
created_time = datetime.fromisoformat(data['created_time'])
|
|||
|
|
else:
|
|||
|
|
created_time = data['created_time']
|
|||
|
|
|
|||
|
|
updated_time = None
|
|||
|
|
if data.get('updated_time'):
|
|||
|
|
if isinstance(data['updated_time'], str):
|
|||
|
|
updated_time = datetime.fromisoformat(data['updated_time'])
|
|||
|
|
else:
|
|||
|
|
updated_time = data['updated_time']
|
|||
|
|
|
|||
|
|
return cls(
|
|||
|
|
id=data.get('id'),
|
|||
|
|
full_name=data['full_name'],
|
|||
|
|
domain=data['domain'],
|
|||
|
|
sub_domains=sub_domains,
|
|||
|
|
name=data['name'],
|
|||
|
|
type=data['type'],
|
|||
|
|
range_min=data.get('range_min'),
|
|||
|
|
range_max=data.get('range_max'),
|
|||
|
|
default_value=data.get('default_value'),
|
|||
|
|
unit=data.get('unit'),
|
|||
|
|
description=data.get('description'),
|
|||
|
|
created_time=created_time,
|
|||
|
|
updated_time=updated_time,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def __str__(self) -> str:
|
|||
|
|
"""字符串表示"""
|
|||
|
|
return f"Field(id={self.id}, full_name={self.full_name}, type={self.type.value})"
|
|||
|
|
|
|||
|
|
def __repr__(self) -> str:
|
|||
|
|
"""详细字符串表示"""
|
|||
|
|
return self.__str__()
|