"""通用对照表管理界面基类
把「材质牌号 / 口径映射 / 型号→压力」等基础字典表的 CRUD 维护界面统一到
一套**列定义驱动 + 共享控件**的框架中，避免每个表重复手写工具栏、表格、
增删改与编辑表单。
共享控件（``certflow.widgets``）：
- ``ActionToolbar``：按钮簇工具栏（按内容定宽）
- ``EnhancedTable``：列元数据 → 表头 + 选择/编辑行为 + 列头点击客户端排序
- ``FilterBar``：统一条件契约的字段级筛选条
- ``ConfirmDialog``：危险操作二次确认
- ``DialogButtonBar``：确认 / 取消按钮栏
字段类型（``field_type``）：
- ``text`` 普通文本（QLineEdit）
- ``int`` 整数（空→None；必填时空值抛 ``ValueError``）
- ``float`` 浮点数（同上）
- ``bool`` 布尔（QCheckBox，可设 ``default``）
- ``choice`` 下拉单选（QComboBox，``choices`` 列表，可含空串表示允许为空）
使用方式：子类只需声明 ``_COLUMNS``（表格列）与 ``_EDIT_FIELDS``（编辑表单），
并提供一个实现标准 CRUD 接口的控制器（``list_all`` / ``get_by_id`` /
``create`` / ``update`` / ``delete`` / ``close``）即可。
"""
from __future__ import annotations
import contextlib
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, override
from PySide6.QtCore import Qt, QTimer
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QDialog,
QFileDialog,
QFormLayout,
QHBoxLayout,
QLabel,
QLineEdit,
QMessageBox,
QPushButton,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from certflow.utils.logger import logger
from certflow.views.bases.query_data_source import client_filter_records, client_sort_records
from certflow.widgets import ActionToolbar, ConfirmDialog, DialogButtonBar
from certflow.widgets.enhanced_table import EnhancedTable
from certflow.widgets.filter_bar import FilterBar
# 基础数据表字段多为文本，复用统一条件契约的客户端过滤算子
_TEXT_OPERATORS = {
"text": {
"contains": "包含",
"eq": "等于",
"ne": "不等于",
"is_null": "为空",
"is_not_null": "非空",
}
}
[文档]
@dataclass
class ColumnSpec:
"""表格列定义。
Attributes:
header: 表头显示文本
field: 对应 ORM 模型字段名（用于 ``getattr`` 取值）
editable: 是否可编辑（当前仅作语义标记，表格本身只读）
field_type: 字段类型，决定表格中的展示方式（bool → ✅/❌）
"""
header: str
field: str
editable: bool = True
field_type: str = "text"
[文档]
@dataclass
class FieldSpec:
"""编辑表单字段定义。
Attributes:
field: 模型字段名
label: 表单标签（可含 ``*`` 提示必填）
field_type: 字段类型（text/int/float/bool/choice）
placeholder: 输入框占位提示
choices: choice 类型的下拉选项列表（可含空串表示允许为空）
required: 是否必填（空值抛 ``ValueError``）
empty_as_none: 文本为空时返回 None（而非空串）
default: bool 类型的默认勾选值；其他类型暂未使用
"""
field: str
label: str
field_type: str = "text"
placeholder: str = ""
choices: list[str] | None = None
required: bool = False
empty_as_none: bool = False
default: Any = None
[文档]
class LookupRecordEditDialog(QDialog):
"""通用编辑子对话框（新增/编辑共用），由 ``FieldSpec`` 列表驱动构建表单。"""
def __init__(
self,
edit_fields: list[FieldSpec],
record: Any = None,
title: str = "编辑记录",
parent: QWidget | None = None,
) -> None:
"""初始化编辑对话框
Args:
edit_fields: 字段定义列表，决定表单内容
record: 待编辑的 ORM 实例；为 None 表示新增
title: 对话框标题
parent: 父窗口
"""
super().__init__(parent)
self._edit_fields = edit_fields
self.setWindowTitle(title)
self.setMinimumWidth(440)
self._widgets: dict[str, object] = {}
self._build_form()
if record is not None:
self._fill(record)
def _build_form(self) -> None:
layout = QVBoxLayout(self)
form = QFormLayout()
form.setSpacing(8)
for spec in self._edit_fields:
if spec.field_type == "choice":
w = QComboBox()
w.addItems(spec.choices or [])
self._widgets[spec.field] = w
form.addRow(spec.label, w)
elif spec.field_type == "bool":
w = QCheckBox(spec.label)
w.setChecked(bool(spec.default) if spec.default is not None else True)
self._widgets[spec.field] = w
form.addRow("", w)
else:
w = QLineEdit()
w.setPlaceholderText(spec.placeholder)
self._widgets[spec.field] = w
form.addRow(spec.label, w)
# 按钮（共享件 DialogButtonBar：确认/取消，定宽右对齐）
bar = DialogButtonBar(self, confirm_text="保存", cancel_text="取消")
bar.accepted.connect(self.accept)
bar.rejected.connect(self.reject)
layout.addLayout(form)
layout.addWidget(bar)
def _fill(self, record: Any) -> None:
for spec in self._edit_fields:
value = getattr(record, spec.field, None)
w = self._widgets[spec.field]
if spec.field_type == "choice":
idx = w.findText("" if value is None else str(value))
if idx >= 0:
w.setCurrentIndex(idx)
elif spec.field_type == "bool":
w.setChecked(bool(value))
else:
w.setText("" if value is None else str(value))
[文档]
def get_data(self) -> dict[str, Any]:
"""收集表单数据，按字段类型转换；必填为空抛 ``ValueError``。
Returns:
dict[str, Any]: 字段名 → 值（空文本按类型转为 None 或空串）
"""
data: dict[str, Any] = {}
for spec in self._edit_fields:
w = self._widgets[spec.field]
if spec.field_type == "choice":
text = w.currentText()
if spec.required and not text:
raise ValueError(f"「{spec.label}」为必填项")
data[spec.field] = text or None
elif spec.field_type == "bool":
data[spec.field] = w.isChecked()
elif spec.field_type in ("int", "float"):
text = w.text().strip()
if not text:
if spec.required:
raise ValueError(f"「{spec.label}」为必填项")
data[spec.field] = None
else:
try:
data[spec.field] = int(text) if spec.field_type == "int" else float(text)
except ValueError as err:
raise ValueError(f"「{spec.label}」需为{spec.field_type}数值") from err
else: # text
text = w.text().strip()
if spec.required and not text:
raise ValueError(f"「{spec.label}」为必填项")
data[spec.field] = (text or None) if spec.empty_as_none else text
return data
[文档]
class LookupTableManagerView(QDialog):
"""通用对照表管理对话框（列定义驱动 + 共享控件）。
子类通过传入 ``controller`` / ``columns`` / ``edit_fields`` / ``title``
即可获得完整的「工具栏 + 表格 + 增删改 + 编辑表单」能力，无需重复造轮子。
"""
def __init__(
self,
*,
controller: Any,
columns: list[ColumnSpec],
edit_fields: list[FieldSpec],
title: str,
window_title: str | None = None,
extra_actions: list[tuple[str, Callable[[], None], bool]] | None = None,
dict_key: str | None = None,
quick_filters: list[dict] | None = None,
table_key: str | None = None,
parent: QWidget | None = None,
) -> None:
"""初始化通用管理对话框
Args:
controller: 实现标准 CRUD 接口的控制器实例
columns: 表格列定义列表
edit_fields: 编辑表单字段定义列表
title: 表名（用于提示文案）
window_title: 窗口标题；缺省同 ``title``
extra_actions: 额外工具栏动作列表，元素为
``(按钮文本, 回调, 是否危险操作)``
dict_key: 字典键名（``dictionary.<key>.csv``）；传入则在工具栏追加
「导出CSV / 导入CSV」按钮，实现 DB ↔ CSV 双向同步
quick_filters: 常用筛选快速按钮（显式覆盖）；元素为
``{"label", "field", "operator", "value"}``。为 ``None`` 时改从
YAML 配置 ``lookup_tables.quick_filters.<table_key>`` 读取（数据驱动，
便于增减，无需改代码）。
table_key: 在 ``lookup_tables`` 配置中的标识（与 ``_TAB_REGISTRY`` 的 key
一致），用于从 YAML 拉取该表的快速筛选按钮。
parent: 父窗口
"""
super().__init__(parent)
self._controller = controller
self._columns = columns
self._edit_fields = edit_fields
self._title = title
self._dict_key = dict_key
# 显式 quick_filters 优先；否则按 table_key 从 YAML 配置读取
self._quick_filters = (
list(quick_filters)
if quick_filters is not None
else self._load_quick_filters_from_config(table_key)
)
# choice / bool 列的静态枚举，用于筛选条渲染下拉（避免文本框误输入）
self._field_choices_map: dict[str, list[str]] = {}
for f in self._edit_fields:
if f.field_type == "choice" and f.choices:
self._field_choices_map[f.field] = list(f.choices)
elif f.field_type == "bool":
self._field_choices_map[f.field] = ["", "True", "False"]
# 客户端筛选/排序状态
self._all_rows: list[Any] = []
self._order_by: str | None = None
self._order_desc: bool = False
self.setWindowTitle(window_title or title)
self.setMinimumSize(1000, 600)
self.resize(1100, 680)
self._init_ui(extra_actions or [])
self.refresh()
# ============================================================
# UI 构建
# ============================================================
@staticmethod
def _load_quick_filters_from_config(table_key: str | None) -> list[dict]:
"""从 YAML 配置 ``lookup_tables.quick_filters.<table_key>`` 读取快速按钮。
配置不可用时（如测试环境未加载）安全回退为 ``[]``，不影响主流程。
"""
if not table_key:
return []
try:
from certflow.config.settings import cfg
quick_filters_cfg = cfg("lookup_tables.quick_filters", {}) or {}
except Exception: # pragma: no cover - 防御：配置缺失不阻断界面
return []
rows = quick_filters_cfg.get(table_key)
return list(rows) if rows else []
def _init_ui(self, extra_actions: list[tuple[str, Callable[[], None], bool]]) -> None:
main_layout = QVBoxLayout(self)
main_layout.setContentsMargins(12, 12, 12, 12)
main_layout.setSpacing(10)
# 工具栏（共享件 ActionToolbar：按内容定宽，刷新按钮右对齐）
toolbar = ActionToolbar()
toolbar.add_button("➕ 新增", self.on_add)
toolbar.add_button("✏️ 编辑", self.on_edit)
toolbar.add_button("🗑️ 删除", self.on_delete, danger=True)
for label, cb, danger in extra_actions:
toolbar.add_button(label, cb, danger=danger)
toolbar.add_stretch()
if self._dict_key:
toolbar.add_button("📤 导出CSV", self.on_export_csv)
toolbar.add_button("📥 导入CSV", self.on_import_csv)
toolbar.add_button("🧹 清除筛选", self._on_clear_filter)
toolbar.add_button("🔄 刷新", self.refresh)
main_layout.addWidget(toolbar)
# 字段筛选条（复用统一 FilterBar；客户端 client_filter_records 过滤）
self._filter_bar = FilterBar(
columns=[{"label": c.header, "field": c.field} for c in self._columns],
field_op_map={},
field_choices_map=self._field_choices_map,
operators=_TEXT_OPERATORS,
grid_cols=3,
)
self._filter_bar.conditions_changed.connect(self._on_filter_changed_debounced)
main_layout.addWidget(self._filter_bar)
# 配置驱动的常用筛选快速按钮（数据驱动，可整体迁移到 YAML）
if self._quick_filters:
quick_layout = QHBoxLayout()
quick_layout.setContentsMargins(0, 0, 0, 0)
quick_label = QLabel("快速筛选：")
quick_label.setStyleSheet("font-size: 11px; color: #666;")
quick_layout.addWidget(quick_label)
for spec in self._quick_filters:
btn = QPushButton(spec.get("label", spec.get("field", "筛选")))
btn.setFixedHeight(24)
btn.clicked.connect(lambda _checked, s=spec: self._apply_quick_filter(s))
quick_layout.addWidget(btn)
quick_layout.addStretch()
main_layout.addLayout(quick_layout)
# 表格（EnhancedTable：列头点击触发客户端排序，行 UserRole 携带 obj.id）
self._table = EnhancedTable()
self._table.setSelectionMode(QTableWidget.SelectionMode.SingleSelection)
self._table.setColumnCount(len(self._columns))
self._table.setHorizontalHeaderLabels([c.header for c in self._columns])
for idx, col in enumerate(self._columns):
self._table.set_header_field(idx, col.field)
self._table.doubleClicked.connect(self._on_double_click)
self._table.sort_requested.connect(self._on_sort_requested)
main_layout.addWidget(self._table, stretch=1)
# 底部提示（显示筛选后数量）
self._hint_label = QLabel("")
self._hint_label.setStyleSheet("font-size: 11px; color: #666;")
main_layout.addWidget(self._hint_label)
# 筛选输入防抖定时器
self._filter_timer = QTimer(self)
self._filter_timer.setSingleShot(True)
self._filter_timer.timeout.connect(self._apply_view)
# ============================================================
# 数据加载
# ============================================================
[文档]
def refresh(self) -> None:
"""从数据库重新加载全量数据并应用当前筛选/排序。"""
try:
self._all_rows = self._controller.list_all()
except Exception as e:
logger.error(f"加载{self._title}失败: {e}")
QMessageBox.critical(self, "错误", f"加载失败: {e}")
return
self._apply_view()
# ============================================================
# 筛选 / 排序 / 填充
# ============================================================
def _apply_view(self) -> None:
"""按当前筛选条件过滤 + 列头排序，重填表格并更新计数提示。"""
conditions = self._filter_bar.get_conditions()
rows = client_filter_records(self._all_rows, conditions)
rows = client_sort_records(rows, self._order_by, self._order_desc)
self._populate(rows)
total = len(self._all_rows)
shown = len(rows)
if conditions:
self._hint_label.setText(
f"筛选后 {shown} 条／总计 {total} 双击行可编辑；删除前会二次确认。"
f"当前维护：{self._title} 对照表。"
)
else:
self._hint_label.setText(
f"共 {total} 条 双击行可编辑；删除前会二次确认。当前维护：{self._title} 对照表。"
)
def _populate(self, rows: list[Any]) -> None:
"""把记录列表填入表格（每行用 UserRole 携带 obj.id 供增删改定位）。"""
self._table.setRowCount(len(rows))
for r, obj in enumerate(rows):
for c, col in enumerate(self._columns):
value = getattr(obj, col.field, None)
if col.field_type == "bool":
value = "✅" if value else "❌"
item = QTableWidgetItem("" if value is None else str(value))
item.setData(Qt.ItemDataRole.UserRole, obj.id)
self._table.setItem(r, c, item)
self._table.resizeColumnsToContents()
def _on_filter_changed_debounced(self) -> None:
"""筛选条件变化：防抖触发重算（避免逐字符重排卡顿）。"""
self._filter_timer.start(200)
def _on_sort_requested(self, field: str, desc: bool) -> None:
"""列头点击：本地对全量数据排序后重填（EnhancedTable 已更新箭头）。"""
self._order_by = field
self._order_desc = desc
self._apply_view()
def _on_clear_filter(self) -> None:
"""清空所有筛选条件与排序，回到全量。"""
self._filter_bar.clear_rows()
self._filter_bar.ensure_at_least_one_row()
self._order_by = None
self._order_desc = False
self._table.horizontalHeader().setSortIndicator(-1, Qt.SortOrder.AscendingOrder)
self._apply_view()
def _apply_quick_filter(self, spec: dict) -> None:
"""点击快速筛选按钮：在筛选条追加一条预置条件并立即应用。"""
self._filter_bar.add_row(spec.get("field"), spec.get("operator", "eq"), spec.get("value"))
# conditions_changed 已触发防抖重算，这里再立即触发一次保证即时生效
self._filter_timer.start(0)
# ============================================================
# 增删改
# ============================================================
[文档]
def on_add(self) -> None:
"""新增记录"""
dlg = LookupRecordEditDialog(self._edit_fields, None, f"新增{self._title}", self)
if dlg.exec() != QDialog.DialogCode.Accepted:
return
try:
self._controller.create(dlg.get_data())
self.refresh()
except ValueError as e:
QMessageBox.warning(self, "校验失败", str(e))
[文档]
def on_edit(self) -> None:
"""编辑选中记录"""
rid = self._selected_id()
if rid is None:
QMessageBox.information(self, "提示", "请先选择一行")
return
obj = self._controller.get_by_id(rid)
if obj is None:
return
dlg = LookupRecordEditDialog(self._edit_fields, obj, f"编辑{self._title}", self)
if dlg.exec() != QDialog.DialogCode.Accepted:
return
try:
self._controller.update(rid, dlg.get_data())
self.refresh()
except ValueError as e:
QMessageBox.warning(self, "校验失败", str(e))
[文档]
def on_delete(self) -> None:
"""删除选中记录（共享件 ConfirmDialog 统一二次确认）"""
rid = self._selected_id()
if rid is None:
QMessageBox.information(self, "提示", "请先选择一行")
return
if not ConfirmDialog.danger(self, "确认删除", "确定删除该记录？此操作不可撤销。"):
return
if self._controller.delete(rid):
self.refresh()
def _on_double_click(self, index: int) -> None:
self.on_edit()
# ============================================================
# DB ↔ CSV 双向同步
# ============================================================
[文档]
def on_export_csv(self) -> None:
"""把当前 DB 表整表导出为 CSV 快照（与界面编辑保持同步）。"""
if not self._dict_key:
return
path = self._controller.resolve_dict_csv_path(self._dict_key)
if path is None:
QMessageBox.warning(self, "提示", f"未配置 {self._dict_key} 的 CSV 路径")
return
try:
n = self._controller.export_csv(path)
except Exception as e:
logger.error(f"导出CSV失败: {e}")
QMessageBox.critical(self, "错误", f"导出失败: {e}")
return
QMessageBox.information(self, "成功", f"已导出 {n} 行到：\n{path}")
self.refresh()
[文档]
def on_import_csv(self) -> None:
"""从 CSV 快照 upsert 回 DB（外部编辑的 CSV 同步进库）。"""
if not self._dict_key:
return
default_dir = ""
resolved = self._controller.resolve_dict_csv_path(self._dict_key)
if resolved is not None:
default_dir = str(resolved.parent)
file_path, _ = QFileDialog.getOpenFileName(
self, "选择 CSV 文件", default_dir, "CSV 文件 (*.csv)"
)
if not file_path:
return
try:
stats = self._controller.import_csv(file_path)
except Exception as e:
logger.error(f"导入CSV失败: {e}")
QMessageBox.critical(self, "错误", f"导入失败: {e}")
return
if stats.get("errors", 0) < 0:
QMessageBox.warning(self, "提示", f"文件不存在：\n{file_path}")
return
QMessageBox.information(
self,
"成功",
f"已从 {file_path} 同步：\n"
f"新增 {stats['inserted']} 行，更新 {stats['updated']} 行，"
f"跳过 {stats['skipped']} 行，错误 {stats['errors']} 行。",
)
self.refresh()
def _selected_id(self) -> int | None:
row = self._table.currentRow()
if row < 0:
return None
id_item = self._table.item(row, 0)
if id_item is None:
return None
return id_item.data(Qt.ItemDataRole.UserRole)
# ============================================================
# 资源清理
# ============================================================
[文档]
@override
def closeEvent(self, event: Any) -> None:
"""关闭时释放控制器会话"""
with contextlib.suppress(Exception): # pragma: no cover - 防御性
self._controller.close()
super().closeEvent(event)