certflow.services.caliber_mapping_service 源代码

"""口径映射服务

收口 CaliberMapping 的 ORM 类与 CRUD 逻辑，供控制器层（Controller）调用，
消除控制器直接依赖 ``certflow.models.*`` 的跨层边。视图（View）只经控制器门面
访问，不直接触碰本服务或模型层。

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

from __future__ import annotations

from typing import Any

from sqlalchemy.orm import Session

from certflow.models.caliber_mapping import CaliberMapping
from certflow.services.dict_csv_sync import DictCsvSync




class CaliberMappingService:
    """口径映射服务 — 控制器与模型之间的业务调度层。

    CRUD 逻辑从原 ``CaliberMappingController`` 下沉至此，控制器仅做事务提交。
    """

    # CSV 列序 / 类型 / 唯一键（与 dictionaries.yaml 的 caliber_mappings 段一致）
    CSV_COLUMNS = ["raw_text", "caliber_value", "is_imperial"]
    CSV_FIELD_TYPES = {
        "raw_text": "text",
        "caliber_value": "text",
        "is_imperial": "bool",
    }
    CSV_UNIQUE_KEYS = ["raw_text"]

    def __init__(self, session: Session) -> None:
        """初始化服务；绑定数据库会话。

        Args:
            session: SQLAlchemy 会话（通常由 BaseController.get_service 注入）。
        """
        self.session = session
        self._csv_sync = DictCsvSync(
            CaliberMapping,
            csv_columns=self.CSV_COLUMNS,
            field_types=self.CSV_FIELD_TYPES,
            unique_keys=self.CSV_UNIQUE_KEYS,
        )



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




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




    def create(self, data: dict[str, Any]) -> CaliberMapping:
        """新增记录（不提交，交由控制器统一 commit）。"""
        obj = CaliberMapping(**data)
        self.session.add(obj)
        self.session.flush()
        return obj




    def update(self, mapping_id: int, data: dict[str, Any]) -> CaliberMapping | None:
        """更新记录（不提交）；id 不存在返回 None。"""
        obj = self.session.get(CaliberMapping, mapping_id)
        if obj is None:
            return None
        for key, value in data.items():
            setattr(obj, key, value)
        return obj




    def delete(self, mapping_id: int) -> bool:
        """删除记录（不提交）；id 不存在返回 False。"""
        obj = self.session.get(CaliberMapping, mapping_id)
        if obj is None:
            return False
        self.session.delete(obj)
        return True




    def seed_common_calibers(self) -> int:
        """幂等灌入常见口径映射示例，库非空则跳过（不破坏已有数据）。"""
        if self.session.query(CaliberMapping).first() is not None:
            return 0
        samples = [
            ("DN15", "15", False, "manual"),
            ("DN50", "50", False, "manual"),
            ("DN100", "100", False, "manual"),
            ('2"', "2", True, "manual"),
            ('1/2"', "1/2", True, "manual"),
        ]
        for raw_text, caliber_value, is_imperial, source in samples:
            self.session.add(
                CaliberMapping(
                    raw_text=raw_text,
                    caliber_value=caliber_value,
                    is_imperial=is_imperial,
                    source=source,
                    usage_count=0,
                )
            )
        return len(samples)




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




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




# 供控制器/其他模块从服务层间接引用模型类，避免 controller 直连 certflow.models。
__all__ = ["CaliberMapping", "CaliberMappingService"]