171 lines
4.4 KiB
Python
171 lines
4.4 KiB
Python
"""
|
|
字段控制器
|
|
协调字段管理的UI和业务逻辑
|
|
"""
|
|
from typing import List, Optional
|
|
from pathlib import Path
|
|
|
|
from services.field_service import FieldService
|
|
from models.field import Field
|
|
from config import FieldType
|
|
from utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class FieldController:
|
|
"""字段控制器类"""
|
|
|
|
def __init__(self):
|
|
self.service = FieldService()
|
|
|
|
def create_field(self, field_data: dict) -> tuple[bool, str, Optional[int]]:
|
|
"""
|
|
创建字段
|
|
|
|
Args:
|
|
field_data: 字段数据字典
|
|
|
|
Returns:
|
|
(是否成功, 消息, 字段ID)
|
|
"""
|
|
try:
|
|
field = Field.from_dict(field_data)
|
|
return self.service.create_field(field)
|
|
except Exception as e:
|
|
logger.error(f"Controller: Failed to create field: {e}")
|
|
return False, f"创建字段失败: {str(e)}", None
|
|
|
|
def update_field(self, field_data: dict) -> tuple[bool, str]:
|
|
"""
|
|
更新字段
|
|
|
|
Args:
|
|
field_data: 字段数据字典
|
|
|
|
Returns:
|
|
(是否成功, 消息)
|
|
"""
|
|
try:
|
|
field = Field.from_dict(field_data)
|
|
return self.service.update_field(field)
|
|
except Exception as e:
|
|
logger.error(f"Controller: 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:
|
|
(是否成功, 消息)
|
|
"""
|
|
return self.service.delete_field(field_id)
|
|
|
|
def get_field(self, field_id: int) -> Optional[Field]:
|
|
"""
|
|
获取字段
|
|
|
|
Args:
|
|
field_id: 字段ID
|
|
|
|
Returns:
|
|
字段对象或None
|
|
"""
|
|
return self.service.get_field_by_id(field_id)
|
|
|
|
def search_fields(self, search_params: dict) -> tuple[List[Field], int]:
|
|
"""
|
|
搜索字段
|
|
|
|
Args:
|
|
search_params: 搜索参数字典
|
|
|
|
Returns:
|
|
(字段列表, 总数量)
|
|
"""
|
|
domain = search_params.get('domain')
|
|
name = search_params.get('name')
|
|
field_type_str = search_params.get('type')
|
|
field_type = FieldType(field_type_str) if field_type_str else None
|
|
page = search_params.get('page', 1)
|
|
page_size = search_params.get('page_size', 50)
|
|
|
|
return self.service.search_fields(domain, name, field_type, page, page_size)
|
|
|
|
def get_all_fields(self, page: int = 1, page_size: int = 50) -> tuple[List[Field], int]:
|
|
"""
|
|
获取所有字段
|
|
|
|
Args:
|
|
page: 页码
|
|
page_size: 每页数量
|
|
|
|
Returns:
|
|
(字段列表, 总数量)
|
|
"""
|
|
return self.service.get_all_fields(page, page_size)
|
|
|
|
def get_domains(self) -> List[str]:
|
|
"""
|
|
获取所有域名
|
|
|
|
Returns:
|
|
域名列表
|
|
"""
|
|
return self.service.get_domains()
|
|
|
|
def export_fields(self, field_ids: List[int], file_path: str,
|
|
format: str = 'json') -> tuple[bool, str]:
|
|
"""
|
|
导出字段
|
|
|
|
Args:
|
|
field_ids: 字段ID列表
|
|
file_path: 文件路径
|
|
format: 格式
|
|
|
|
Returns:
|
|
(是否成功, 消息)
|
|
"""
|
|
# 获取字段对象
|
|
fields = []
|
|
for field_id in field_ids:
|
|
field = self.service.get_field_by_id(field_id)
|
|
if field:
|
|
fields.append(field)
|
|
|
|
if not fields:
|
|
return False, "没有可导出的字段"
|
|
|
|
return self.service.export_fields(fields, Path(file_path), format)
|
|
|
|
def import_fields(self, file_path: str,
|
|
format: str = 'json') -> tuple[bool, str, List[Field]]:
|
|
"""
|
|
导入字段
|
|
|
|
Args:
|
|
file_path: 文件路径
|
|
format: 格式
|
|
|
|
Returns:
|
|
(是否成功, 消息, 字段列表)
|
|
"""
|
|
return self.service.import_fields(Path(file_path), format)
|
|
|
|
def batch_create_fields(self, fields: List[Field]) -> tuple[int, int, List[str]]:
|
|
"""
|
|
批量创建字段
|
|
|
|
Args:
|
|
fields: 字段列表
|
|
|
|
Returns:
|
|
(成功数量, 失败数量, 错误信息列表)
|
|
"""
|
|
return self.service.batch_create_fields(fields)
|