""" 导入导出工具 支持JSON、XML、CSV格式的数据导入导出 """ import json import csv import xml.etree.ElementTree as ET from typing import List, Dict, Any from pathlib import Path from models.field import Field from models.message import Message from models.mapping import OperatorMapping, CodeMapping from utils.logger import get_logger logger = get_logger(__name__) class ImportExportManager: """导入导出管理器""" @staticmethod def export_fields_to_json(fields: List[Field], file_path: Path) -> bool: """ 导出字段到JSON文件 Args: fields: 字段列表 file_path: 文件路径 Returns: 是否成功 """ try: data = { 'fields': [field.to_dict() for field in fields] } with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) logger.info(f"Exported {len(fields)} fields to {file_path}") return True except Exception as e: logger.error(f"Failed to export fields to JSON: {e}") return False @staticmethod def import_fields_from_json(file_path: Path) -> List[Field]: """ 从JSON文件导入字段 Args: file_path: 文件路径 Returns: 字段列表 """ try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) fields = [] for field_data in data.get('fields', []): field = Field.from_dict(field_data) fields.append(field) logger.info(f"Imported {len(fields)} fields from {file_path}") return fields except Exception as e: logger.error(f"Failed to import fields from JSON: {e}") return [] @staticmethod def export_fields_to_csv(fields: List[Field], file_path: Path) -> bool: """ 导出字段到CSV文件 Args: fields: 字段列表 file_path: 文件路径 Returns: 是否成功 """ try: with open(file_path, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) # 写入表头 headers = [ 'full_name', 'domain', 'sub_domains', 'name', 'type', 'range_min', 'range_max', 'default_value', 'unit', 'description' ] writer.writerow(headers) # 写入数据 for field in fields: row = [ field.full_name, field.domain, field.get_sub_domains_str(), field.name, field.type.value, field.range_min or '', field.range_max or '', field.default_value or '', field.unit or '', field.description or '' ] writer.writerow(row) logger.info(f"Exported {len(fields)} fields to {file_path}") return True except Exception as e: logger.error(f"Failed to export fields to CSV: {e}") return False @staticmethod def import_fields_from_csv(file_path: Path) -> List[Field]: """ 从CSV文件导入字段 Args: file_path: 文件路径 Returns: 字段列表 """ try: fields = [] with open(file_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: field_data = { 'full_name': row['full_name'], 'domain': row['domain'], 'sub_domains': row.get('sub_domains', ''), 'name': row['name'], 'type': row['type'], 'range_min': row.get('range_min') or None, 'range_max': row.get('range_max') or None, 'default_value': row.get('default_value') or None, 'unit': row.get('unit') or None, 'description': row.get('description') or None, } field = Field.from_dict(field_data) fields.append(field) logger.info(f"Imported {len(fields)} fields from {file_path}") return fields except Exception as e: logger.error(f"Failed to import fields from CSV: {e}") return [] @staticmethod def export_fields_to_xml(fields: List[Field], file_path: Path) -> bool: """ 导出字段到XML文件 Args: fields: 字段列表 file_path: 文件路径 Returns: 是否成功 """ try: root = ET.Element('fields') for field in fields: field_elem = ET.SubElement(root, 'field') ET.SubElement(field_elem, 'full_name').text = field.full_name ET.SubElement(field_elem, 'domain').text = field.domain ET.SubElement(field_elem, 'sub_domains').text = field.get_sub_domains_str() ET.SubElement(field_elem, 'name').text = field.name ET.SubElement(field_elem, 'type').text = field.type.value if field.range_min: ET.SubElement(field_elem, 'range_min').text = field.range_min if field.range_max: ET.SubElement(field_elem, 'range_max').text = field.range_max if field.default_value: ET.SubElement(field_elem, 'default_value').text = field.default_value if field.unit: ET.SubElement(field_elem, 'unit').text = field.unit if field.description: ET.SubElement(field_elem, 'description').text = field.description tree = ET.ElementTree(root) ET.indent(tree, space=' ') tree.write(file_path, encoding='utf-8', xml_declaration=True) logger.info(f"Exported {len(fields)} fields to {file_path}") return True except Exception as e: logger.error(f"Failed to export fields to XML: {e}") return False @staticmethod def import_fields_from_xml(file_path: Path) -> List[Field]: """ 从XML文件导入字段 Args: file_path: 文件路径 Returns: 字段列表 """ try: tree = ET.parse(file_path) root = tree.getroot() fields = [] for field_elem in root.findall('field'): field_data = { 'full_name': field_elem.find('full_name').text, 'domain': field_elem.find('domain').text, 'sub_domains': field_elem.find('sub_domains').text or '', 'name': field_elem.find('name').text, 'type': field_elem.find('type').text, } # 可选字段 for optional in ['range_min', 'range_max', 'default_value', 'unit', 'description']: elem = field_elem.find(optional) field_data[optional] = elem.text if elem is not None else None field = Field.from_dict(field_data) fields.append(field) logger.info(f"Imported {len(fields)} fields from {file_path}") return fields except Exception as e: logger.error(f"Failed to import fields from XML: {e}") return [] @staticmethod def export_messages_to_json(messages: List[Message], file_path: Path) -> bool: """ 导出消息到JSON文件 Args: messages: 消息列表 file_path: 文件路径 Returns: 是否成功 """ try: data = { 'messages': [msg.to_dict() for msg in messages] } with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) logger.info(f"Exported {len(messages)} messages to {file_path}") return True except Exception as e: logger.error(f"Failed to export messages to JSON: {e}") return False @staticmethod def import_messages_from_json(file_path: Path) -> List[Message]: """ 从JSON文件导入消息 Args: file_path: 文件路径 Returns: 消息列表 """ try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) messages = [] for msg_data in data.get('messages', []): message = Message.from_dict(msg_data) messages.append(message) logger.info(f"Imported {len(messages)} messages from {file_path}") return messages except Exception as e: logger.error(f"Failed to import messages from JSON: {e}") return []