"""基础字典表 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)
"""
from __future__ import annotations
import csv
from pathlib import Path
from typing import Any
from sqlalchemy import inspect
# 项目根：src/certflow/services/dict_csv_sync.py → parents[3]
_PROJECT_ROOT = Path(__file__).resolve().parents[3]
[文档]
def resolve_dict_csv_path(dict_key: str) -> Path | None:
"""从 ``config/dictionaries.yaml`` 解析 ``dictionary.<dict_key>.csv`` 的绝对路径。
Args:
dict_key: 字典键名，如 ``material_grades`` / ``caliber_mappings`` / ``model_param_mappings``
Returns:
Path | None: CSV 绝对路径；配置缺失或解析失败返回 None
"""
yaml_path = _PROJECT_ROOT / "config" / "dictionaries.yaml"
if not yaml_path.exists():
return None
try:
import yaml
data = yaml.safe_load(yaml_path.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001 - 配置不可用时不阻塞 UI
return None
rel = (data or {}).get("dictionary", {}).get(dict_key, {}).get("csv")
if not rel:
return None
p = Path(rel)
return p if p.is_absolute() else _PROJECT_ROOT / rel
[文档]
class DictCsvSync:
"""通用字典表 CSV 双向同步器（列定义驱动，不绑定具体表）。
Attributes:
model_class: SQLAlchemy 模型类
csv_columns: CSV 列序（模型字段名）
field_types: 字段名 → 类型（text/int/float/bool），用于导入时类型还原
unique_keys: 唯一键字段列表（用于 upsert 定位）
default_source: 导入时新行的来源标记（model/caliber 用 import）
"""
def __init__(
self,
model_class: Any,
csv_columns: list[str],
field_types: dict[str, str],
unique_keys: list[str],
default_source: str = "import",
) -> None:
self.model_class = model_class
self.csv_columns = csv_columns
self.field_types = field_types
self.unique_keys = unique_keys
self.default_source = default_source
# ============================================================
# 导出（DB → CSV）
# ============================================================
[文档]
def export_csv(self, session: Any, path: str | Path) -> int:
"""把当前表整表导出为 CSV 快照，返回行数。
Args:
session: SQLAlchemy 会话
path: 输出 CSV 路径
"""
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
rows = session.query(self.model_class).order_by(self.model_class.id).all()
with open(path, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f, lineterminator="\n")
writer.writerow(self.csv_columns)
for obj in rows:
writer.writerow(
[self._serialize(getattr(obj, col, None)) for col in self.csv_columns]
)
return len(rows)
@staticmethod
def _serialize(value: Any) -> str:
"""Python 值 → CSV 单元格文本（bool→true/false，None→空串）。"""
if value is None:
return ""
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
# ============================================================
# 导入（CSV → DB，upsert）
# ============================================================
[文档]
def import_csv(self, session: Any, path: str | Path) -> dict[str, int]: # noqa: C901
"""把 CSV 内容 upsert 回 DB，返回统计。
唯一键命中则更新非键字段（保留 source/usage_count 等）；未命中则插入。
必填唯一键缺失的行跳过。
Args:
session: SQLAlchemy 会话
path: 输入 CSV 路径
Returns:
dict: {inserted, updated, skipped, errors}
"""
stats = {"inserted": 0, "updated": 0, "skipped": 0, "errors": 0}
path = Path(path)
if not path.exists():
stats["errors"] = -1 # 文件缺失标记
return stats
with open(path, newline="", encoding="utf-8-sig") as f:
reader = csv.reader(f)
header_seen = False
for row in reader:
if not row:
continue
if not header_seen:
header_seen = True
continue
record = self._coerce(row)
if record is None:
stats["skipped"] += 1
continue
filters = {k: record[k] for k in self.unique_keys}
existing = session.query(self.model_class).filter_by(**filters).first()
try:
if existing is None:
data = dict(record)
if self._has_column("source") and "source" not in data:
data["source"] = self.default_source
session.add(self.model_class(**data))
stats["inserted"] += 1
else:
changed = False
for col in self.csv_columns:
if col in self.unique_keys:
continue
new_val = record[col]
if getattr(existing, col) != new_val:
setattr(existing, col, new_val)
changed = True
if changed:
stats["updated"] += 1
else:
stats["skipped"] += 1
except Exception: # noqa: BLE001 - 单行失败不中断整批
stats["errors"] += 1
try:
session.commit()
except Exception: # noqa: BLE001
session.rollback()
stats["errors"] += 1
return stats
def _coerce(self, row: list[str]) -> dict[str, Any] | None:
"""CSV 行 → 记录字典（按列序 + 类型还原）；必填唯一键缺失返回 None。"""
record: dict[str, Any] = {}
for i, col in enumerate(self.csv_columns):
raw = row[i].strip() if i < len(row) else ""
ft = self.field_types.get(col, "text")
if ft == "bool":
record[col] = raw.lower() in ("true", "1", "yes") if raw else False
elif ft == "int":
record[col] = int(raw) if raw else None
elif ft == "float":
record[col] = float(raw) if raw else None
else:
record[col] = raw or None
for k in self.unique_keys:
if not record.get(k):
return None
return record
def _has_column(self, name: str) -> bool:
"""判断模型是否含指定列（避免对无 source 字段的表误写）。"""
try:
return name in inspect(self.model_class).columns
except Exception: # noqa: BLE001
return hasattr(self.model_class, name)