"""型号参数对照表服务
收口 ModelParamMapping 的 ORM 类与 CRUD 逻辑，供控制器层（Controller）调用，
消除控制器直接依赖 ``certflow.models.*`` 的跨层边。视图（View）只经控制器门面
访问，不直接触碰本服务或模型层。
架构层次：View → Controller (ModelParamMappingController)
→ Service (ModelParamMappingService) → Model (ModelParamMapping)
"""
from __future__ import annotations
from typing import Any
from sqlalchemy.orm import Session
from certflow.models.model_param_mapping import ModelParamMapping
from certflow.services.dict_csv_sync import DictCsvSync
[文档]
class ModelParamMappingService:
"""型号参数对照表服务 — 控制器与模型之间的业务调度层。
CRUD 逻辑从原 ``ModelParamMappingController`` 下沉至此，控制器仅做事务提交。
"""
# CSV 列序 / 类型 / 唯一键（与 dictionaries.yaml 的 model_param_mappings 段一致）
CSV_COLUMNS = [
"product_model",
"pressure_value",
"test_standard",
"working_temp",
"working_medium",
]
CSV_FIELD_TYPES = {
"product_model": "text",
"pressure_value": "text",
"test_standard": "text",
"working_temp": "text",
"working_medium": "text",
}
CSV_UNIQUE_KEYS = ["product_model"]
def __init__(self, session: Session) -> None:
"""初始化服务；绑定数据库会话。
Args:
session: SQLAlchemy 会话（通常由 BaseController.get_service 注入）。
"""
self.session = session
self._csv_sync = DictCsvSync(
ModelParamMapping,
csv_columns=self.CSV_COLUMNS,
field_types=self.CSV_FIELD_TYPES,
unique_keys=self.CSV_UNIQUE_KEYS,
)
[文档]
def list_all(self) -> list[ModelParamMapping]:
"""返回全部型号参数对照记录（按 id 升序）。"""
return self.session.query(ModelParamMapping).order_by(ModelParamMapping.id).all()
[文档]
def get_by_id(self, mapping_id: int) -> ModelParamMapping | None:
"""按主键获取记录。"""
return self.session.get(ModelParamMapping, mapping_id)
[文档]
def get_by_model(self, product_model: str) -> ModelParamMapping | None:
"""按产品型号查询对照记录。"""
return self.session.query(ModelParamMapping).filter_by(product_model=product_model).first()
[文档]
def create(self, data: dict[str, Any]) -> ModelParamMapping:
"""新增记录（不提交，交由控制器统一 commit）。"""
obj = ModelParamMapping(**data)
self.session.add(obj)
self.session.flush()
return obj
[文档]
def update(self, mapping_id: int, data: dict[str, Any]) -> ModelParamMapping | None:
"""更新记录（不提交）；id 不存在返回 None。"""
obj = self.session.get(ModelParamMapping, 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(ModelParamMapping, mapping_id)
if obj is None:
return False
self.session.delete(obj)
return True
[文档]
def seed_common_models(self) -> int:
"""幂等灌入常见型号→压力示例，库非空则跳过（不破坏已有数据）。"""
if self.session.query(ModelParamMapping).first() is not None:
return 0
samples = [
("Z41H-16C", "1.6MPa", "GB/T 13927-2008", "≤200℃", "水、蒸汽", "manual"),
("Z41H-25C", "2.5MPa", "GB/T 13927-2008", "≤425℃", "水、蒸汽", "manual"),
("Z41H-40C", "4.0MPa", "GB/T 13927-2008", "≤425℃", "水、蒸汽", "manual"),
]
for (
product_model,
pressure_value,
test_standard,
working_temp,
working_medium,
source,
) in samples:
self.session.add(
ModelParamMapping(
product_model=product_model,
pressure_value=pressure_value,
test_standard=test_standard,
working_temp=working_temp,
working_medium=working_medium,
source=source,
usage_count=0,
)
)
return len(samples)
[文档]
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 → 按策略合并。
"""
obj = self.session.query(ModelParamMapping).filter_by(product_model=product_model).first()
if obj is None:
obj = ModelParamMapping(
product_model=product_model, source="auto-learned", usage_count=0
)
self.session.add(obj)
elif obj.source == "manual" and not overwrite_manual:
return # 人工权威，绝不自动覆盖
if strategy == "last_write_wins":
for col, val in updates.items():
if val:
setattr(obj, col, val)
else: # conservative / field_level：仅补缺
for col, val in updates.items():
if not val:
continue
cur = getattr(obj, col, None) or ""
if not cur:
setattr(obj, col, val)
obj.usage_count = (obj.usage_count or 0) + 1
if obj.source != "manual":
obj.source = "auto-learned"
[文档]
def export_csv(self, path: str | Any) -> int:
"""把当前 model_param_mappings 表整表导出为 CSV 快照，返回行数。"""
return self._csv_sync.export_csv(self.session, path)
[文档]
def import_csv(self, path: str | Any) -> dict[str, int]:
"""把 CSV 快照 upsert 回 model_param_mappings 表，返回统计。"""
return self._csv_sync.import_csv(self.session, path)
# 供控制器/其他模块从服务层间接引用模型类，避免 controller 直连 certflow.models。
__all__ = ["ModelParamMapping", "ModelParamMappingService"]