certflow.services.dict_csv_sync module

基础字典表 DB ↔ CSV 双向同步

为「材质牌号 / 口径映射 / 型号→压力」三个管理界面提供统一的 CSV 快照同步能力， 使界面内编辑（已实时落库）与可提交的 CSV 快照保持一致：

• export_csv：把当前 DB 表整表导出为 CSV（UTF-8-SIG + 表头 + LF 行尾）。

• import_csv：把 CSV 内容 upsert 回 DB（按唯一键插入/更新，保留 source/usage_count）。

CSV 路径从 config/dictionaries.yaml 的 dictionary.<key>.csv 解析， 与脚本（B2/B3）共用同一份路径配置，避免漂移。

用法（在控制器内）：

self._csv_sync = DictCsvSync(

MaterialGrade, csv_columns=[...], field_types={...}, unique_keys=["grade", "standard_code", "element"],

) self._csv_sync.export_csv(session, path) self._csv_sync.import_csv(session, path)

certflow.services.dict_csv_sync.resolve_dict_csv_path(dict_key)

从 config/dictionaries.yaml 解析 dictionary.<dict_key>.csv 的绝对路径。

参数:

dict_key (str) -- 字典键名，如 material_grades / caliber_mappings / model_param_mappings

返回:

CSV 绝对路径；配置缺失或解析失败返回 None

返回类型:

Path | None

class certflow.services.dict_csv_sync.DictCsvSync(model_class, csv_columns, field_types, unique_keys, default_source='import')

基类：object

通用字典表 CSV 双向同步器（列定义驱动，不绑定具体表）。

参数:

• model_class (Any)

• csv_columns (list[str])

• field_types (dict[str, str])

• unique_keys (list[str])

• default_source (str)

model_class

SQLAlchemy 模型类

csv_columns

CSV 列序（模型字段名）

field_types

字段名 → 类型（text/int/float/bool），用于导入时类型还原

unique_keys

唯一键字段列表（用于 upsert 定位）

default_source

导入时新行的来源标记（model/caliber 用 import）

export_csv(session, path)

把当前表整表导出为 CSV 快照，返回行数。

参数:

• session (Any) -- SQLAlchemy 会话

• path (str | Path) -- 输出 CSV 路径

返回类型:

int

import_csv(session, path)

把 CSV 内容 upsert 回 DB，返回统计。

唯一键命中则更新非键字段（保留 source/usage_count 等）；未命中则插入。 必填唯一键缺失的行跳过。

参数:

• session (Any) -- SQLAlchemy 会话

• path (str | Path) -- 输入 CSV 路径

返回:

{inserted, updated, skipped, errors}

返回类型:

dict