"""统一查询视图基类（阶段 A）
承载 FilterBar + EnhancedTable + 分页 + 导出，由 ``QueryDataSource`` 协议驱动，
使销售计划 / 校正队列 / 基础数据 / BOM 查询 共用一套 UX，不在视图里写 ``if`` 分支。
排序：监听 EnhancedTable 的 ``sort_requested`` 信号——
- server 数据源（销售计划 / 校正队列）：带 order_by 重新 query（现有服务端排序行为）。
- client 数据源（基础数据 / BOM）：本地对 records 排序后重填（无分页冲突）。
分页：翻页 / 改页大小保留当前页码（与补丁 11 一致）；新查询复位第 1 页。
"""
from __future__ import annotations
from typing import Any
from PySide6.QtWidgets import (
QApplication,
QComboBox,
QFileDialog,
QHBoxLayout,
QLabel,
QPushButton,
QSpinBox,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from certflow.views.bases.query_data_source import QueryDataSource
from certflow.widgets.enhanced_table import EnhancedTable
from certflow.widgets.filter_bar import FilterBar
# 文本字段默认操作符（FilterBar 行下拉；对齐客户端过滤统一契约）
_TEXT_OPERATORS = {
"text": {
"contains": "包含",
"eq": "等于",
"ne": "不等于",
"is_null": "为空",
"is_not_null": "非空",
}
}
def _ensure_app() -> None:
"""测试 / 非 GUI 入口下确保 QApplication 存在。"""
if QApplication.instance() is None:
QApplication([]) # noqa: F841
[文档]
class BaseQueryView(QWidget):
"""统一查询视图基类（数据源驱动）。"""
def __init__(
self,
session: Any = None,
db_manager: Any = None,
parent: Any = None,
) -> None:
super().__init__(parent)
self.session = session
self.db_manager = db_manager
self.data_source: QueryDataSource | None = None
self.page_size = 50
self.current_page = 1
self._order_by: str | None = None
self._order_desc = False
self._total = 0
self._total_pages = 1
self._records: list[Any] = []
self._setup_ui()
def _setup_ui(self) -> None:
layout = QVBoxLayout(self)
# 筛选条（列在 set_data_source 时按数据源配置）
self.filter_bar = FilterBar(
columns=[], field_op_map={}, operators=_TEXT_OPERATORS, grid_cols=2
)
self.filter_bar.setVisible(False)
layout.addWidget(self.filter_bar)
# 操作按钮行
btn_row = QHBoxLayout()
self.search_btn = QPushButton("查询")
self.search_btn.clicked.connect(lambda: self._do_search(reset_page=True))
btn_row.addWidget(self.search_btn)
self.export_btn = QPushButton("导出CSV")
self.export_btn.clicked.connect(self._export_csv)
btn_row.addWidget(self.export_btn)
btn_row.addStretch()
self.refresh_btn = QPushButton("刷新")
self.refresh_btn.clicked.connect(lambda: self._do_search(reset_page=False))
btn_row.addWidget(self.refresh_btn)
layout.addLayout(btn_row)
# 表格（复用 EnhancedTable，已支持排序信号）
self.table = EnhancedTable()
self.table.sort_requested.connect(self._on_sort_requested)
layout.addWidget(self.table, stretch=1)
# 分页栏
pg = QHBoxLayout()
self.prev_btn = QPushButton("上一页")
self.prev_btn.clicked.connect(self._prev_page)
pg.addWidget(self.prev_btn)
self.page_spin = QSpinBox()
self.page_spin.setMinimum(1)
self.page_spin.valueChanged.connect(self._on_page_spin)
pg.addWidget(self.page_spin)
self.page_label = QLabel("/ 1 页")
pg.addWidget(self.page_label)
self.next_btn = QPushButton("下一页")
self.next_btn.clicked.connect(self._next_page)
pg.addWidget(self.next_btn)
pg.addWidget(QLabel("每页"))
self.page_size_combo = QComboBox()
self.page_size_combo.addItems(["20", "50", "100", "200"])
self.page_size_combo.setCurrentText("50")
self.page_size_combo.currentTextChanged.connect(self._on_page_size_changed)
pg.addWidget(self.page_size_combo)
self.total_label = QLabel("共 0 条")
pg.addWidget(self.total_label)
pg.addStretch()
layout.addLayout(pg)
# ============================================================
# 数据源装配
# ============================================================
[文档]
def set_data_source(self, source: QueryDataSource) -> None:
"""装入数据源：配置表头排序字段与筛选列，并触发首次查询。"""
self.data_source = source
columns = source.get_columns()
self.table.setColumnCount(len(columns))
self.table.setHorizontalHeaderLabels([h for h, _ in columns])
for idx, (_, field) in enumerate(columns):
self.table.set_header_field(idx, field)
fcols = source.get_filter_columns() if hasattr(source, "get_filter_columns") else None
if fcols:
self.filter_bar.set_columns(fcols)
self.filter_bar.setVisible(True)
else:
self.filter_bar.setVisible(False)
self.current_page = 1
self._order_by = None
self._order_desc = False
self._do_search(reset_page=True)
# ============================================================
# 查询 / 排序 / 分页
# ============================================================
def _do_search(self, reset_page: bool = True) -> None:
"""执行查询。新查询复位第 1 页；翻页/改页大小传 reset_page=False。"""
if self.data_source is None:
return
if reset_page:
self.current_page = 1
conditions = self.filter_bar.get_conditions() if self.filter_bar.isVisible() else {}
self._run_query(conditions)
def _run_query(self, conditions: dict) -> None:
assert self.data_source is not None
result = self.data_source.query(
conditions,
self.current_page,
self.page_size,
order_by=self._order_by,
order_desc=self._order_desc,
)
self._records = result.get("records", [])
self._total = result.get("total", 0)
self._total_pages = result.get("total_pages", 1)
self._populate(self._records)
self._update_pagination()
def _populate(self, records: list[Any]) -> None:
assert self.data_source is not None
columns = self.data_source.get_columns()
self.table.setRowCount(len(records))
for r, rec in enumerate(records):
for c, (_, field) in enumerate(columns):
val = rec.get(field) if isinstance(rec, dict) else getattr(rec, field, None)
self.table.setItem(r, c, QTableWidgetItem("" if val is None else str(val)))
def _update_pagination(self) -> None:
self.page_spin.blockSignals(True)
self.page_spin.setMaximum(max(1, self._total_pages))
self.page_spin.setValue(self.current_page)
self.page_spin.blockSignals(False)
self.page_label.setText(f"/ {self._total_pages} 页")
self.total_label.setText(f"共 {self._total} 条")
self.prev_btn.setEnabled(self.current_page > 1)
self.next_btn.setEnabled(self.current_page < self._total_pages)
def _on_sort_requested(self, field: str, desc: bool) -> None:
"""列头点击：server 源重查、client 源本地排序，均回到第 1 页。"""
self._order_by = field
self._order_desc = desc
self._do_search(reset_page=True)
def _prev_page(self) -> None:
if self.current_page > 1:
self.current_page -= 1
self._do_search(reset_page=False)
def _next_page(self) -> None:
if self.current_page < self._total_pages:
self.current_page += 1
self._do_search(reset_page=False)
def _on_page_spin(self, val: int) -> None:
self.current_page = val
self._do_search(reset_page=False)
def _on_page_size_changed(self, text: str) -> None:
self.page_size = int(text)
self.current_page = 1
self._do_search(reset_page=False)
[文档]
def refresh(self) -> None:
"""刷新当前结果（保留当前页/筛选，重跑查询）。供宿主或子类在写操作后调用。"""
self._do_search(reset_page=False)
@property
def total_count(self) -> int:
"""当前查询结果总条数。"""
return self._total
def _export_csv(self) -> None:
if self.data_source is None:
return
path, _ = QFileDialog.getSaveFileName(self, "导出CSV", "", "CSV (*.csv)")
if path:
n = self.data_source.export_csv(path)
self.total_label.setText(f"已导出 {n} 条")