"""材质牌号控制器
桥接材质牌号管理界面（View）与材质牌号服务（Service）：
- 读操作直接委托 MaterialGradeService；
- 写操作（新增/更新/删除）在此统一提交事务，失败时回滚并向上抛出。
架构层次：View (MaterialGradeManagerView) → Controller (MaterialGradeController)
→ Service (MaterialGradeService) → Model (MaterialGrade)
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from certflow.controllers.base_controller import BaseController
from certflow.services.dict_csv_sync import DictCsvSync
from certflow.services.material_grade_service import MaterialGrade, MaterialGradeService
from certflow.utils.logger import logger
[文档]
class MaterialGradeController(BaseController):
"""材质牌号控制器 — 管理界面的业务调度层"""
# CSV 列序 / 类型 / 唯一键（与 dictionaries.yaml 的 material_grades 段一致）
_CSV_COLUMNS = [
"grade",
"standard_code",
"category_large",
"category_medium",
"category",
"kind",
"element",
"standard_min",
"standard_max",
"decimal_places",
"actual_min",
"actual_max",
"is_active",
]
_CSV_FIELD_TYPES = {
"grade": "text",
"standard_code": "text",
"category_large": "text",
"category_medium": "text",
"category": "text",
"kind": "text",
"element": "text",
"standard_min": "float",
"standard_max": "float",
"decimal_places": "int",
"actual_min": "float",
"actual_max": "float",
"is_active": "bool",
}
_CSV_UNIQUE_KEYS = ["grade", "standard_code", "element"]
def __init__(self, db_session=None, db_manager=None, use_shared: bool = True):
"""初始化控制器
Args:
db_session: 外部会话（可选）；不传则按 BaseController 规则自动创建
db_manager: 数据库管理器（可选）
use_shared: 是否使用全局共享管理器
"""
super().__init__(db_session=db_session, db_manager=db_manager, use_shared=use_shared)
self._service = self.get_service(MaterialGradeService)
self._csv_sync = DictCsvSync(
MaterialGrade,
csv_columns=self._CSV_COLUMNS,
field_types=self._CSV_FIELD_TYPES,
unique_keys=self._CSV_UNIQUE_KEYS,
)
# ============================================================
# 读操作（直接委托）
# ============================================================
[文档]
def list_all(self, active_only: bool = False) -> list[Any]:
"""返回全部材质牌号记录
Args:
active_only: 为 True 时仅返回启用状态的牌号，默认为 False
Returns:
list[Any]: 材质牌号记录列表（ORM 对象或字典）
"""
return self._service.list_all(active_only=active_only)
[文档]
def get_by_id(self, grade_id: int) -> Any:
"""按主键获取记录
Args:
grade_id: 材质牌号记录主键
Returns:
Any: 匹配的记录对象；不存在时返回 None
"""
return self._service.get_by_id(grade_id)
[文档]
def get_grades_by_category(self) -> dict[str, list[str]]:
"""按阀门零件类别分组返回牌号（供打印对话框）"""
return self._service.get_grades_by_category()
[文档]
def get_data_quality_summary(self) -> dict[str, int]:
"""材质牌号表数据质量概览（总数/启用/停用/范围缺失），供管理界面状态栏。"""
return self._service.get_data_quality_summary()
# ============================================================
# 写操作（提交事务）
# ============================================================
[文档]
def create(self, data: dict[str, Any]) -> int:
"""新增记录并提交
Returns:
新记录主键 id
Raises:
ValueError: 校验失败（交给上层提示用户）
Examples:
>>> controller = MaterialGradeController()
>>> new_id = controller.create({
... "category": "阀体",
... "grade": "WCB",
... "is_active": True,
... })
>>> print(new_id)
"""
mg = self._service.create(data)
self.commit()
return mg.id
[文档]
def update(self, grade_id: int, data: dict[str, Any]) -> bool:
"""更新记录并提交
Returns:
是否更新成功（id 不存在返回 False）
"""
try:
result = self._service.update(grade_id, data)
if result is None:
return False
self.commit()
return True
except ValueError:
self.rollback()
raise
[文档]
def delete(self, grade_id: int) -> bool:
"""删除记录并提交
Returns:
是否删除成功
"""
result = self._service.delete(grade_id)
if result:
self.commit()
return result
[文档]
def seed_common_grades(self) -> int:
"""幂等灌入常见牌号示例（化学成分 + 力学性能）
Returns:
实际写入的记录条数（0 表示库已有数据，未写入）
"""
count = self._service.seed_common_grades()
if count:
self.commit()
return count
[文档]
def export_csv(self, path: str | Path) -> int:
"""把当前 material_grades 表整表导出为 CSV 快照，返回行数。"""
return self._csv_sync.export_csv(self.session, path)
[文档]
def import_csv(self, path: str | Path) -> dict[str, int]:
"""把 CSV 快照 upsert 回 material_grades 表，返回统计。"""
return self._csv_sync.import_csv(self.session, path)
[文档]
def refresh(self) -> list[Any]:
"""重新加载全部记录（提交并刷新会话）
Returns:
list[Any]: 刷新后的全部材质牌号记录列表
"""
try:
self.session.commit()
except Exception as e: # pragma: no cover - 防御性
logger.debug(f"refresh 提交失败（忽略）: {e}")
self.session.rollback()
self.session.expire_all()
return self.list_all()