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__()
|