certflow.controllers.caliber_mapping_controller 源代码

"""口径映射控制器

桥接口径映射管理界面（View）与口径映射服务（Service）：读操作直接委托
CaliberMappingService；写操作在此统一提交事务，失败时回滚并向上抛出。模型类
经服务层间接引用，控制器不直接依赖 ``certflow.models.*``。

架构层次：View (CaliberMappingManagerView) → Controller (CaliberMappingController)
         → Service (CaliberMappingService) → Model (CaliberMapping)
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

from certflow.controllers.base_controller import BaseController
from certflow.services.caliber_mapping_service import (
    CaliberMapping,
    CaliberMappingService,
)




class CaliberMappingController(BaseController):
    """口径映射控制器 — 管理界面的业务调度层"""

    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(CaliberMappingService)



    def list_all(self) -> list[CaliberMapping]:
        """返回全部口径映射记录（按 id 升序）

        Returns:
            list[CaliberMapping]: 口径映射记录列表
        """
        return self._service.list_all()




    def get_by_id(self, mapping_id: int) -> CaliberMapping | None:
        """按主键获取记录

        Args:
            mapping_id: 记录主键

        Returns:
            CaliberMapping | None: 匹配的记录；不存在时返回 None
        """
        return self._service.get_by_id(mapping_id)




    def create(self, data: dict[str, Any]) -> int:
        """新增记录并提交

        Returns:
            int: 新记录主键 id
        """
        obj = self._service.create(data)
        self.commit()
        return obj.id




    def update(self, mapping_id: int, data: dict[str, Any]) -> bool:
        """更新记录并提交

        Args:
            mapping_id: 记录主键
            data: 待更新的字段字典

        Returns:
            bool: 是否更新成功（id 不存在返回 False）
        """
        obj = self._service.update(mapping_id, data)
        if obj is None:
            return False
        self.commit()
        return True




    def delete(self, mapping_id: int) -> bool:
        """删除记录并提交

        Returns:
            bool: 是否删除成功
        """
        if not self._service.delete(mapping_id):
            return False
        self.commit()
        return True




    def seed_common_calibers(self) -> int:
        """幂等灌入常见口径映射示例，库非空则跳过（不破坏已有数据）"""
        count = self._service.seed_common_calibers()
        if count:
            self.commit()
        return count




    def export_csv(self, path: str | Path) -> int:
        """把当前 caliber_mappings 表整表导出为 CSV 快照，返回行数。"""
        return self._service.export_csv(path)




    def import_csv(self, path: str | Path) -> dict[str, int]:
        """把 CSV 快照 upsert 回 caliber_mappings 表，返回统计。"""
        return self._service.import_csv(path)