certflow.controllers.model_param_mapping_controller 源代码

"""型号参数对照表控制器

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

架构层次：View (ModelParamMappingManagerView) → Controller
         (ModelParamMappingController) → Service (ModelParamMappingService)
         → Model (ModelParamMapping)
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

from certflow.controllers.base_controller import BaseController
from certflow.services.model_param_mapping_service import (
    ModelParamMapping,
    ModelParamMappingService,
)




class ModelParamMappingController(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(ModelParamMappingService)



    def list_all(self) -> list[ModelParamMapping]:
        """返回全部型号参数对照记录（按 id 升序）

        Returns:
            list[ModelParamMapping]: 对照记录列表
        """
        return self._service.list_all()




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

        Args:
            mapping_id: 记录主键

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




    def get_by_model(self, product_model: str) -> ModelParamMapping | None:
        """按产品型号查询对照记录（阶段 C：收口 print_view 的裸 session.query）

        视图层不再直接剥出 Controller 的 session 自己写 ORM 查询，统一经此方法。

        Args:
            product_model: 产品型号字符串

        Returns:
            ModelParamMapping | None: 匹配的记录；不存在时返回 None
        """
        return self._service.get_by_model(product_model)




    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_models(self) -> int:
        """幂等灌入常见型号→压力示例，库非空则跳过（不破坏已有数据）"""
        count = self._service.seed_common_models()
        if count:
            self.commit()
        return count




    def upsert_for_learnback(
        self,
        product_model: str,
        updates: dict[str, str],
        strategy: str = "conservative",
        overwrite_manual: bool = False,
    ) -> None:
        """打印成功后学回型号字典（保守合并语义，见 docs/PRINT_BLUEPRINT.md §2.4）。

        - 不存在 → 新建 source='auto-learned'，应用全部非空 updates。
        - 已存在且 source=='manual' 且 overwrite_manual=False → 整条不动（人工权威）。
        - 已存在且非 manual → 按策略合并：
            conservative / field_level：仅补缺（DB 为空、表单非空才写；已非空保持）。
            last_write_wins：所有非空表单字段覆盖 DB。
        - 无论新建/补缺，usage_count++。
        """
        self._service.upsert_for_learnback(
            product_model,
            updates,
            strategy=strategy,
            overwrite_manual=overwrite_manual,
        )
        self.commit()




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




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