"""销售计划查询视图（任务 2.1 第二域：QueryView 迁移 BaseQueryView）
``SalePlanQueryView`` 是 ``BaseQueryView`` 的「宿主驱动」渲染器：它复用 ``BaseQueryView``
的 ``EnhancedTable`` + 分页栏，但**不**使用 ``BaseQueryView`` 自带的 ``FilterBar`` /
搜索/刷新按钮——这些高级筛选（快查预设、特殊值、视图列、保存筛选）由宿主 ``QueryView``
自己的 ``FilterBar`` 承担，并通过 ``run_query`` 注入条件与翻页控制。
渲染保真：移植 ``QueryView.display_results`` 的证书字段回退 / 计划日期归一 / 生产状态
图标逻辑，使迁移后显示与既有销售计划查询完全一致（纯收敛，行为不变）。
首屏加载延迟到 ``showEvent``（由宿主决定），避免构造期在无事件循环环境阻塞。
"""
from __future__ import annotations
from typing import Any
from PySide6.QtCore import Signal
from PySide6.QtWidgets import (
QComboBox,
QHBoxLayout,
QHeaderView,
QLabel,
QPushButton,
QSpinBox,
QTableWidgetItem,
QVBoxLayout,
)
from certflow.utils.date_utils import normalize_plan_date
from certflow.views.bases.base_query_view import BaseQueryView
from certflow.views.bases.datasources.sale_plan import SalePlanDataSource
from certflow.widgets.enhanced_table import EnhancedTable
# 纯展示字段（非数据库列，不可用于排序）
_DISPLAY_ONLY_FIELDS = frozenset({"progress_display", "ready_display"})
# 证书字段回退映射（cert_* 为空时回退合同字段）
_CERT_FALLBACK = {
"cert_product_name": "product_name",
"cert_product_model": "product_model",
"cert_product_spec": "product_spec",
}
# 仅存于关联 Certificate 的字段
_CERT_ONLY_FIELDS = frozenset(
{
"print_status",
"engrave_status",
"certificate_no",
"print_time",
"engraved_time",
"is_engraved",
}
)
[文档]
class SalePlanQueryView(BaseQueryView):
"""销售计划结果区：BaseQueryView 宿主驱动渲染器。
信号：
selection_changed：选中行变化（供宿主更新「选中 N 条」）。
"""
selection_changed = Signal()
def __init__(
self,
session: Any = None,
db_manager: Any = None,
parent: Any = None,
) -> None:
super().__init__(session=session, db_manager=db_manager, parent=parent)
self._view_columns: list[dict] = []
self._records: list[Any] = []
# ============================================================
# UI：仅保留表格 + 分页（隐藏 BaseQueryView 自带筛选条/搜索/刷新按钮）
# ============================================================
def _setup_ui(self) -> None:
layout = QVBoxLayout(self)
self.table = EnhancedTable()
self.table.setColumnCount(0)
self.table.sort_requested.connect(self._on_sort_requested)
self.table.itemSelectionChanged.connect(self.selection_changed)
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_controller(self, controller: Any) -> None:
"""装配 QueryController 数据源（不触发查询；首屏由宿主 showEvent 触发）。"""
self.data_source = SalePlanDataSource(controller)
columns = self.data_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)
[文档]
def set_view_columns(self, columns: list[dict]) -> None:
"""按视图模板列配置表头（含展示字段不可排序处理），并恢复列状态/排序指示。"""
self._view_columns = list(columns)
headers = [col.get("label", col.get("field", "")) for col in columns]
self.table.setColumnCount(len(headers))
self.table.setHorizontalHeaderLabels(headers)
for col, col_def in enumerate(columns):
field = col_def.get("field", "")
if field in _DISPLAY_ONLY_FIELDS:
field = ""
self.table.set_header_field(col, field)
header = self.table.horizontalHeader()
header.setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
# ============================================================
# 排序（覆盖基类：SalePlanQueryView 无 filter_bar，不调用 _do_search）
# ============================================================
def _on_sort_requested(self, field: str, desc: bool) -> None:
"""列头点击排序：发射信号由宿主处理（不调用基类 _do_search）。"""
# 由宿主 QueryView 监听 sort_requested 信号来处理排序
pass
# ============================================================
# 查询（由宿主注入条件 / 排序 / 翻页）
# ============================================================
[文档]
def run_query(
self,
conditions: dict,
or_filters: Any = None,
order_by: str | None = None,
order_desc: bool = False,
page: int = 1,
page_size: int = 50,
reset_page: bool = True,
) -> None:
"""宿主驱动查询：经 SalePlanDataSource 走服务端分页/排序。"""
if self.data_source is None:
return None
if reset_page:
page = 1
self.current_page = page
self.page_size = page_size
result = self.data_source.query(
conditions,
self.current_page,
self.page_size,
order_by,
order_desc,
or_filters,
)
self.display(
result["records"],
result["total"],
result["total_pages"],
result["page"],
)
return result
[文档]
def display(
self,
records: list[Any],
total: int,
total_pages: int,
page: int,
columns: list[dict] | None = None,
) -> None:
"""填充表格并更新分页（渲染保真见 ``_render``）。"""
if columns is not None:
self._view_columns = list(columns)
self._records = records
self._total = total
self._total_pages = total_pages
self.current_page = page
self._render(records)
self._update_pagination()
# ============================================================
# 保真渲染（移植自 QueryView.display_results）
# ============================================================
@staticmethod
def _extract_cert_values(record: Any, cert_only_fields: frozenset[str]) -> dict[str, Any]:
cert_values: dict[str, Any] = {}
certs = getattr(record, "certificates", None)
if not certs:
return cert_values
for cert in certs:
cert_dict = cert.__dict__
for field_name in cert_only_fields:
if field_name not in cert_values:
v = cert_dict.get(field_name)
if v is not None:
cert_values[field_name] = v
return cert_values
@staticmethod
def _resolve_cell_value(
col: int,
field_name: str,
row_data: dict[str, Any],
cert_values: dict[str, Any],
cert_col_indices: set[int],
fallback_cols: dict[int, str],
plan_date_col: int | None,
) -> str:
value = cert_values.get(field_name) if col in cert_col_indices else row_data.get(field_name)
if col in fallback_cols and not value:
value = row_data.get(fallback_cols[col], "")
if col == plan_date_col:
value = normalize_plan_date(value) or ""
if isinstance(value, bool):
return str(value)
return str(value or "")
def _render(self, records: list[Any]) -> None:
columns = self._view_columns
col_field_names = [col_def["field"] for col_def in columns]
cert_col_indices = {i for i, fn in enumerate(col_field_names) if fn in _CERT_ONLY_FIELDS}
fallback_cols = {
i: _CERT_FALLBACK[fn] for i, fn in enumerate(col_field_names) if fn in _CERT_FALLBACK
}
plan_date_col = next((i for i, fn in enumerate(col_field_names) if fn == "plan_date"), None)
status_col = next(
(i for i, fn in enumerate(col_field_names) if fn == "production_status"), None
)
show_raw_tooltip = getattr(self, "_show_raw_tooltip", True)
show_icon = getattr(self, "_show_icon", True)
self.table.setUpdatesEnabled(False)
header = self.table.horizontalHeader()
header.setSectionResizeMode(QHeaderView.ResizeMode.Fixed)
self.table.setRowCount(len(records))
for row_idx, record in enumerate(records):
row_data = {k: v for k, v in record.__dict__.items() if not k.startswith("_")}
for fn in col_field_names:
if fn not in row_data and fn not in cert_col_indices:
row_data[fn] = getattr(record, fn, None)
if "product_code" in col_field_names:
display = getattr(record, "product_code_display", None)
if display:
row_data["product_code"] = display
cert_values = (
self._extract_cert_values(record, _CERT_ONLY_FIELDS) if cert_col_indices else {}
)
for col_idx, field_name in enumerate(col_field_names):
text = self._resolve_cell_value(
col_idx,
field_name,
row_data,
cert_values,
cert_col_indices,
fallback_cols,
plan_date_col,
)
item = QTableWidgetItem(text)
if (
status_col is not None
and col_idx == status_col
and show_raw_tooltip
and getattr(record, "execution_status", None)
):
raw_exec = str(getattr(record, "execution_status", ""))
if raw_exec and raw_exec != getattr(record, "production_status", ""):
item.setToolTip(f"原始执行情况: {raw_exec}")
if show_icon:
item.setText(f"📋 {text}")
self.table.setItem(row_idx, col_idx, item)
header.setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
self.table.setUpdatesEnabled(True)
# ============================================================
# 选中 / 记录访问（供宿主上下文菜单、批量动作、打印跳转）
# ============================================================
@property
def records(self) -> list[Any]:
return self._records
[文档]
def selected_rows(self) -> set[int]:
return {i.row() for i in self.table.selectedIndexes()}
[文档]
def selected_records(self) -> list[Any]:
rows = self.selected_rows()
return [self._records[r] for r in rows if r < len(self._records)]
[文档]
def selected_sale_plan_ids(self) -> list[int]:
return [rec.id for rec in self.selected_records() if getattr(rec, "id", None) is not None]
# ============================================================
# 列状态（委托 EnhancedTable；键由宿主按视图管理）
# ============================================================
[文档]
def save_column_state(self) -> Any:
return self.table.save_state()
[文档]
def restore_column_state(self, state: Any) -> None:
if state:
self.table.restore_state(state)