191 lines
6.1 KiB
Python
191 lines
6.1 KiB
Python
"""
|
||
消息业务模型
|
||
定义消息的数据结构和业务逻辑
|
||
"""
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from typing import Optional, List
|
||
from config import ProtocolType, SerializationType
|
||
from models.field import Field
|
||
|
||
|
||
@dataclass
|
||
class Message:
|
||
"""
|
||
消息模型类
|
||
|
||
Attributes:
|
||
id: 消息ID
|
||
full_name: 完整名称（如 Radar.Receiver.v3）
|
||
system_name: 系统名（如 Radar）
|
||
message_type: 消息类型（如 Receiver）
|
||
version: 版本号（如 v3）
|
||
fields: 关联的字段列表
|
||
description: 描述信息
|
||
protocol: 传输协议
|
||
serialization: 序列化格式
|
||
created_time: 创建时间
|
||
updated_time: 更新时间
|
||
"""
|
||
|
||
full_name: str
|
||
system_name: str
|
||
message_type: str
|
||
version: str
|
||
protocol: ProtocolType
|
||
serialization: SerializationType
|
||
id: Optional[int] = None
|
||
fields: List[Field] = field(default_factory=list)
|
||
description: Optional[str] = None
|
||
created_time: Optional[datetime] = None
|
||
updated_time: Optional[datetime] = None
|
||
|
||
def __post_init__(self):
|
||
"""初始化后处理"""
|
||
# 确保protocol和serialization是枚举类型
|
||
if isinstance(self.protocol, str):
|
||
self.protocol = ProtocolType(self.protocol)
|
||
|
||
if isinstance(self.serialization, str):
|
||
self.serialization = SerializationType(self.serialization)
|
||
|
||
# 从full_name解析system_name、message_type和version
|
||
if self.full_name and not self.system_name:
|
||
self._parse_full_name()
|
||
|
||
def _parse_full_name(self):
|
||
"""从完整名称解析系统名、消息类型和版本"""
|
||
parts = self.full_name.split('.')
|
||
if len(parts) >= 3:
|
||
self.system_name = parts[0]
|
||
self.message_type = parts[1]
|
||
self.version = parts[2]
|
||
elif len(parts) == 2:
|
||
self.system_name = parts[0]
|
||
self.message_type = parts[1]
|
||
self.version = "v1"
|
||
|
||
@staticmethod
|
||
def build_full_name(system_name: str, message_type: str, version: str) -> str:
|
||
"""
|
||
构建完整消息名
|
||
|
||
Args:
|
||
system_name: 系统名
|
||
message_type: 消息类型
|
||
version: 版本号
|
||
|
||
Returns:
|
||
完整消息名
|
||
"""
|
||
return f"{system_name}.{message_type}.{version}"
|
||
|
||
def add_field(self, field_obj: Field):
|
||
"""添加字段"""
|
||
if field_obj not in self.fields:
|
||
self.fields.append(field_obj)
|
||
|
||
def remove_field(self, field_obj: Field):
|
||
"""移除字段"""
|
||
if field_obj in self.fields:
|
||
self.fields.remove(field_obj)
|
||
|
||
def get_field_by_name(self, full_name: str) -> Optional[Field]:
|
||
"""根据完整名称获取字段"""
|
||
for f in self.fields:
|
||
if f.full_name == full_name:
|
||
return f
|
||
return None
|
||
|
||
def validate(self) -> tuple[bool, str]:
|
||
"""
|
||
验证消息数据的有效性
|
||
|
||
Returns:
|
||
(是否有效, 错误信息)
|
||
"""
|
||
if not self.full_name:
|
||
return False, "消息完整名称不能为空"
|
||
|
||
if not self.system_name:
|
||
return False, "系统名不能为空"
|
||
|
||
if not self.message_type:
|
||
return False, "消息类型不能为空"
|
||
|
||
if not self.version:
|
||
return False, "版本号不能为空"
|
||
|
||
if not self.fields or len(self.fields) == 0:
|
||
return False, "消息至少需要包含一个字段"
|
||
|
||
if not self.protocol:
|
||
return False, "传输协议不能为空"
|
||
|
||
if not self.serialization:
|
||
return False, "序列化格式不能为空"
|
||
|
||
return True, ""
|
||
|
||
def to_dict(self) -> dict:
|
||
"""转换为字典"""
|
||
return {
|
||
'id': self.id,
|
||
'full_name': self.full_name,
|
||
'system_name': self.system_name,
|
||
'message_type': self.message_type,
|
||
'version': self.version,
|
||
'description': self.description,
|
||
'protocol': self.protocol.value if isinstance(self.protocol, ProtocolType) else self.protocol,
|
||
'serialization': self.serialization.value if isinstance(self.serialization,
|
||
SerializationType) else self.serialization,
|
||
'fields': [f.to_dict() for f in self.fields],
|
||
'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) -> 'Message':
|
||
"""从字典创建对象"""
|
||
# 解析字段列表
|
||
fields = []
|
||
if 'fields' in data:
|
||
fields = [Field.from_dict(f) if isinstance(f, dict) else f for f in data['fields']]
|
||
|
||
# 解析时间
|
||
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'],
|
||
system_name=data['system_name'],
|
||
message_type=data['message_type'],
|
||
version=data['version'],
|
||
protocol=data['protocol'],
|
||
serialization=data['serialization'],
|
||
fields=fields,
|
||
description=data.get('description'),
|
||
created_time=created_time,
|
||
updated_time=updated_time,
|
||
)
|
||
|
||
def __str__(self) -> str:
|
||
"""字符串表示"""
|
||
return f"Message(id={self.id}, full_name={self.full_name}, fields={len(self.fields)})"
|
||
|
||
def __repr__(self) -> str:
|
||
"""详细字符串表示"""
|
||
return self.__str__()
|