305 lines
9.3 KiB
Python
305 lines
9.3 KiB
Python
|
"""
|
|||
|
|
字段管理服务
|
|||
|
|
提供字段管理的业务逻辑
|
|||
|
|
"""
|
|||
|
|
from typing import List, Optional
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
from database.repositories.field_repository import FieldRepository
|
|||
|
|
from models.field import Field
|
|||
|
|
from config import FieldType
|
|||
|
|
from utils.validator import Validator
|
|||
|
|
from utils.import_export import ImportExportManager
|
|||
|
|
from utils.logger import get_logger
|
|||
|
|
|
|||
|
|
logger = get_logger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class FieldService:
|
|||
|
|
"""字段管理服务类"""
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self.repository = FieldRepository()
|
|||
|
|
self.validator = Validator()
|
|||
|
|
self.import_export = ImportExportManager()
|
|||
|
|
|
|||
|
|
def create_field(self, field: Field) -> tuple[bool, str, Optional[int]]:
|
|||
|
|
"""
|
|||
|
|
创建字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
field: 字段对象
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(是否成功, 消息, 字段ID)
|
|||
|
|
"""
|
|||
|
|
# 验证字段数据
|
|||
|
|
is_valid, error_msg = field.validate()
|
|||
|
|
if not is_valid:
|
|||
|
|
logger.warning(f"Field validation failed: {error_msg}")
|
|||
|
|
return False, error_msg, None
|
|||
|
|
|
|||
|
|
# 验证字段名称格式
|
|||
|
|
is_valid, error_msg = self.validator.validate_field_name(field.full_name)
|
|||
|
|
if not is_valid:
|
|||
|
|
return False, error_msg, None
|
|||
|
|
|
|||
|
|
# 验证域名格式
|
|||
|
|
is_valid, error_msg = self.validator.validate_domain(field.domain)
|
|||
|
|
if not is_valid:
|
|||
|
|
return False, error_msg, None
|
|||
|
|
|
|||
|
|
# 验证范围值
|
|||
|
|
is_valid, error_msg = self.validator.validate_range(
|
|||
|
|
field.range_min, field.range_max, field.type
|
|||
|
|
)
|
|||
|
|
if not is_valid:
|
|||
|
|
return False, error_msg, None
|
|||
|
|
|
|||
|
|
# 检查字段是否已存在
|
|||
|
|
existing = self.repository.get_by_full_name(field.full_name)
|
|||
|
|
if existing:
|
|||
|
|
return False, f"字段 '{field.full_name}' 已存在", None
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
field_id = self.repository.create(field)
|
|||
|
|
logger.info(f"Field created successfully: {field.full_name}")
|
|||
|
|
return True, "字段创建成功", field_id
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to create field: {e}")
|
|||
|
|
return False, f"创建字段失败: {str(e)}", None
|
|||
|
|
|
|||
|
|
def update_field(self, field: Field) -> tuple[bool, str]:
|
|||
|
|
"""
|
|||
|
|
更新字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
field: 字段对象
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(是否成功, 消息)
|
|||
|
|
"""
|
|||
|
|
if not field.id:
|
|||
|
|
return False, "字段ID不能为空"
|
|||
|
|
|
|||
|
|
# 验证字段数据
|
|||
|
|
is_valid, error_msg = field.validate()
|
|||
|
|
if not is_valid:
|
|||
|
|
return False, error_msg
|
|||
|
|
|
|||
|
|
# 验证范围值
|
|||
|
|
is_valid, error_msg = self.validator.validate_range(
|
|||
|
|
field.range_min, field.range_max, field.type
|
|||
|
|
)
|
|||
|
|
if not is_valid:
|
|||
|
|
return False, error_msg
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
success = self.repository.update(field)
|
|||
|
|
if success:
|
|||
|
|
logger.info(f"Field updated successfully: {field.full_name}")
|
|||
|
|
return True, "字段更新成功"
|
|||
|
|
else:
|
|||
|
|
return False, "字段不存在"
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to update field: {e}")
|
|||
|
|
return False, f"更新字段失败: {str(e)}"
|
|||
|
|
|
|||
|
|
def delete_field(self, field_id: int) -> tuple[bool, str]:
|
|||
|
|
"""
|
|||
|
|
删除字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
field_id: 字段ID
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(是否成功, 消息)
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
success = self.repository.delete(field_id)
|
|||
|
|
if success:
|
|||
|
|
logger.info(f"Field deleted successfully: ID {field_id}")
|
|||
|
|
return True, "字段删除成功"
|
|||
|
|
else:
|
|||
|
|
return False, "字段不存在"
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to delete field: {e}")
|
|||
|
|
return False, f"删除字段失败: {str(e)}"
|
|||
|
|
|
|||
|
|
def get_field_by_id(self, field_id: int) -> Optional[Field]:
|
|||
|
|
"""
|
|||
|
|
根据ID获取字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
field_id: 字段ID
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
字段对象或None
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
return self.repository.get_by_id(field_id)
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to get field by id: {e}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def get_field_by_name(self, full_name: str) -> Optional[Field]:
|
|||
|
|
"""
|
|||
|
|
根据完整名称获取字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
full_name: 完整名称
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
字段对象或None
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
return self.repository.get_by_full_name(full_name)
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to get field by name: {e}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def search_fields(self, domain: Optional[str] = None,
|
|||
|
|
name: Optional[str] = None,
|
|||
|
|
field_type: Optional[FieldType] = None,
|
|||
|
|
page: int = 1, page_size: int = 50) -> tuple[List[Field], int]:
|
|||
|
|
"""
|
|||
|
|
搜索字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
domain: 域名
|
|||
|
|
name: 字段名
|
|||
|
|
field_type: 字段类型
|
|||
|
|
page: 页码（从1开始）
|
|||
|
|
page_size: 每页数量
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(字段列表, 总数量)
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
offset = (page - 1) * page_size
|
|||
|
|
fields = self.repository.search(domain, name, field_type, page_size, offset)
|
|||
|
|
total = self.repository.count(domain, name, field_type)
|
|||
|
|
return fields, total
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to search fields: {e}")
|
|||
|
|
return [], 0
|
|||
|
|
|
|||
|
|
def get_all_fields(self, page: int = 1, page_size: int = 50) -> tuple[List[Field], int]:
|
|||
|
|
"""
|
|||
|
|
获取所有字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
page: 页码
|
|||
|
|
page_size: 每页数量
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(字段列表, 总数量)
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
offset = (page - 1) * page_size
|
|||
|
|
fields = self.repository.get_all(page_size, offset)
|
|||
|
|
total = self.repository.count()
|
|||
|
|
return fields, total
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to get all fields: {e}")
|
|||
|
|
return [], 0
|
|||
|
|
|
|||
|
|
def get_domains(self) -> List[str]:
|
|||
|
|
"""
|
|||
|
|
获取所有域名
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
域名列表
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
return self.repository.get_domains()
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to get domains: {e}")
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
def export_fields(self, fields: List[Field], file_path: Path,
|
|||
|
|
format: str = 'json') -> tuple[bool, str]:
|
|||
|
|
"""
|
|||
|
|
导出字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
fields: 字段列表
|
|||
|
|
file_path: 文件路径
|
|||
|
|
format: 格式（json/csv/xml）
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(是否成功, 消息)
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
if format.lower() == 'json':
|
|||
|
|
success = self.import_export.export_fields_to_json(fields, file_path)
|
|||
|
|
elif format.lower() == 'csv':
|
|||
|
|
success = self.import_export.export_fields_to_csv(fields, file_path)
|
|||
|
|
elif format.lower() == 'xml':
|
|||
|
|
success = self.import_export.export_fields_to_xml(fields, file_path)
|
|||
|
|
else:
|
|||
|
|
return False, f"不支持的格式: {format}"
|
|||
|
|
|
|||
|
|
if success:
|
|||
|
|
return True, f"成功导出 {len(fields)} 个字段"
|
|||
|
|
else:
|
|||
|
|
return False, "导出失败"
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to export fields: {e}")
|
|||
|
|
return False, f"导出失败: {str(e)}"
|
|||
|
|
|
|||
|
|
def import_fields(self, file_path: Path,
|
|||
|
|
format: str = 'json') -> tuple[bool, str, List[Field]]:
|
|||
|
|
"""
|
|||
|
|
导入字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
file_path: 文件路径
|
|||
|
|
format: 格式（json/csv/xml）
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(是否成功, 消息, 字段列表)
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
if format.lower() == 'json':
|
|||
|
|
fields = self.import_export.import_fields_from_json(file_path)
|
|||
|
|
elif format.lower() == 'csv':
|
|||
|
|
fields = self.import_export.import_fields_from_csv(file_path)
|
|||
|
|
elif format.lower() == 'xml':
|
|||
|
|
fields = self.import_export.import_fields_from_xml(file_path)
|
|||
|
|
else:
|
|||
|
|
return False, f"不支持的格式: {format}", []
|
|||
|
|
|
|||
|
|
if fields:
|
|||
|
|
return True, f"成功导入 {len(fields)} 个字段", fields
|
|||
|
|
else:
|
|||
|
|
return False, "导入失败或文件为空", []
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Failed to import fields: {e}")
|
|||
|
|
return False, f"导入失败: {str(e)}", []
|
|||
|
|
|
|||
|
|
def batch_create_fields(self, fields: List[Field]) -> tuple[int, int, List[str]]:
|
|||
|
|
"""
|
|||
|
|
批量创建字段
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
fields: 字段列表
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(成功数量, 失败数量, 错误信息列表)
|
|||
|
|
"""
|
|||
|
|
success_count = 0
|
|||
|
|
fail_count = 0
|
|||
|
|
errors = []
|
|||
|
|
|
|||
|
|
for field in fields:
|
|||
|
|
success, msg, field_id = self.create_field(field)
|
|||
|
|
if success:
|
|||
|
|
success_count += 1
|
|||
|
|
else:
|
|||
|
|
fail_count += 1
|
|||
|
|
errors.append(f"{field.full_name}: {msg}")
|
|||
|
|
|
|||
|
|
logger.info(f"Batch create fields: {success_count} succeeded, {fail_count} failed")
|
|||
|
|
return success_count, fail_count, errors
|