# src/certflow/views/query_view.py
"""查询界面 - 参数化查询（配置驱动版）
从 ui.yaml 读取视图模板、操作符定义、字段映射等配置。
"""
from __future__ import annotations
import calendar
import time
from datetime import date, datetime
from typing import Any, override
from PySide6.QtCore import QStringListModel, Qt, QTimer, Signal
from PySide6.QtGui import QAction
from PySide6.QtWidgets import (
QApplication,
QCheckBox,
QComboBox,
QCompleter,
QDateEdit,
QDialog,
QFileDialog,
QGroupBox,
QHBoxLayout,
QInputDialog,
QLabel,
QLineEdit,
QMenu,
QMessageBox,
QPushButton,
QTableWidget,
QTableWidgetSelectionRange,
QToolBar,
QVBoxLayout,
QWidget,
)
from certflow.actions import dispatch
from certflow.config.settings import (
_get_user_config,
cfg,
get_ui_config,
get_views_config,
merge_user_config,
)
from certflow.controllers.query_controller import QueryController
from certflow.utils.date_utils import normalize_plan_date
from certflow.utils.logger import logger
from certflow.views.bases.lookup_table_manager_view import FieldSpec, LookupRecordEditDialog
from certflow.views.common.qdate_utils import parse_qdate, safe_qdate
from certflow.views.sale_plan_query_view import SalePlanQueryView
from certflow.widgets.filter_bar import FilterBar
from certflow.widgets.filter_row import FilterRowWidget
from certflow.widgets.quick_search_preset import QuickSearchPresetMixin
# 合格证信息订正表单（仅 cert_* + 合格证备注，不动合同字段），蓝图 §4.2.3
_CERT_EDIT_FIELDS = [
FieldSpec("cert_product_name", "合格证产品名称", placeholder="留空则打印时回退合同值"),
FieldSpec("cert_product_model", "合格证型号", placeholder="留空则打印时回退合同值"),
FieldSpec("cert_product_spec", "合格证规格", placeholder="留空则打印时回退合同值"),
FieldSpec("certificate_remarks", "合格证备注", placeholder="合格证清单备注"),
]
# 合格证记录订正表单（仅 Certificate 打印字段，不动编号/状态），蓝图 §4.3 步骤 5
_CERT_RECORD_EDIT_FIELDS = [
FieldSpec("product_name", "产品名称", placeholder="合格证实际打印名称"),
FieldSpec("product_model", "产品型号", placeholder="合格证实际打印型号"),
FieldSpec("product_spec", "产品规格", placeholder="合格证实际打印规格"),
FieldSpec("customer", "订货单位", placeholder="订货单位"),
FieldSpec("project_name", "项目名称", placeholder="项目名称"),
FieldSpec("test_standard", "试压标准", placeholder="试压标准号"),
FieldSpec("supplier", "供货商", placeholder="供货商"),
FieldSpec("remarks", "备注", placeholder="合格证备注"),
]
[文档]
class QueryView(QuickSearchPresetMixin, QWidget):
"""销售计划查询视图 - 配置驱动"""
# 信号：请求跳转到打印合格证页面，携带选中的 SalePlan ID 列表
navigate_to_print = Signal(list)
# 新增信号：请求返回首页
back_to_home = Signal()
def __init__(
self, session: Any, db_manager: Any = None, controller: QueryController | None = None
) -> None:
"""初始化查询视图
Args:
session: SQLAlchemy Session 实例，用于数据库查询
db_manager: 数据库管理器实例（可选）
controller: 查询控制器实例（可选，不传则自动创建）
"""
super().__init__()
self._shown_once = False
self.session = session
# 初始化控制器（所有业务逻辑委托给 Controller，View 不直接操作 Service/ORM）
self.controller = controller or QueryController(session, db_manager, parent=self)
self.db_manager = db_manager
# 从 ui.yaml 加载视图配置（canonical UI 源，Route A Phase 1）
self._views = get_views_config()
self._operators = cfg("query_page.operators", {})
self._field_op_map = cfg("query_page.field_operator_map", {})
self._debounce_ms = cfg("query_page.search_debounce_ms", 300)
self._page_size_options = cfg(
"query_page.pagination.page_size_options", [20, 50, 100, 200, 500]
)
self._default_date_range = cfg("query_page.default_date_range", {})
self._date_range_enabled = self._default_date_range.get("enabled", False)
self._plan_date_range_cfg = cfg("query_page.plan_date_range", {})
self._plan_date_range_enabled = self._plan_date_range_cfg.get("enabled", False)
self._plan_date_range_field = self._plan_date_range_cfg.get("field", "plan_date")
self._op_log_level = cfg("query_page.operation_log", "off")
self._number_prefix_override: str | None = None
# 快速筛选按钮映射
self._quick_filter_buttons: dict[str, QPushButton] = {}
# 状态 UI 配置
self._load_status_ui_config()
self.current_page = 1
self.page_size = cfg("query_page.pagination.default_page_size", 50)
self.current_view = list(self._views.keys())[0] if self._views else "all"
self._search_timer: QTimer | None = None
# 连接控制器信号
self.controller.data_updated.connect(self.refresh)
self.controller.operation_completed.connect(self._on_operation_completed)
self.controller.field_values_loaded.connect(self._on_field_values_loaded)
self.controller.field_values_load_failed.connect(self._on_field_values_load_failed)
# 异步加载字段值的 worker 引用（防止 GC 回收）
self._field_value_workers: dict[str, object] = {}
self.setup_ui()
if self._views:
self.switch_view(self.current_view)
# ============================================================
# 配置加载
# ============================================================
def _load_status_ui_config(self) -> None:
"""从配置加载状态 UI 配置（config.yaml production_status.ui）"""
config = cfg("production_status.ui", {})
self._show_raw_tooltip = config.get("show_raw_execution_tooltip", True)
self._show_icon = config.get("show_icon_for_inferred", True)
self._inferred_icon = config.get("inferred_icon", "📋")
# ============================================================
# UI 构建（保持不变）
# ============================================================
[文档]
def setup_ui(self) -> None:
"""初始化整体界面布局"""
main_layout = QVBoxLayout(self)
self.setup_view_toolbar(main_layout)
self.setup_filter_section(main_layout)
self.setup_result_table(main_layout)
self.setup_pagination(main_layout)
[文档]
def setup_filter_section(self, parent_layout: Any) -> None:
"""设置筛选条件区域"""
self.filter_group = QGroupBox("查询条件")
self.filter_layout = QVBoxLayout(self.filter_group)
self.setup_quick_filter_bar(self.filter_layout)
# 筛选行容器：统一委托 FilterBar（§1.7 落地组件），数据源无关。
self._presets = cfg("query_page.presets", [])
self.filter_bar = FilterBar(
columns=self._get_view_columns(),
field_op_map=self._field_op_map,
operators=self._operators,
is_dark_theme_fn=self._is_dark_theme,
load_combo_values_fn=self._load_combo_values,
open_date_picker_fn=self._open_date_picker,
date_button_style_fn=self._get_date_button_style,
grid_cols=cfg("query_page.filter_layout.columns", 2),
# E1：快查框 + preset 下拉（方案 A/B）
show_quick_search=True,
quick_search_placeholder="输入关键字快速筛选…",
quick_search_callback=self.qsp_on_quick_search,
quick_search_typing_callback=self.qsp_on_typing,
show_preset_selector=bool(self._presets),
presets=self._presets,
preset_changed_callback=self.qsp_on_preset_changed,
)
self.filter_layout.addWidget(self.filter_bar)
# E1：默认选中首个预设并刷新快查框提示/候选项
# 稳定的快查候选项基础条件（按视图默认范围取全量，不随快查临时条件收窄）
self._view_default_conditions: dict[str, Any] = {}
# 接入公共快查逻辑（QuickSearchPresetMixin）：默认选中首个预设、刷新提示/候选项、统一交互
self.setup_quick_search_preset(self._presets, self._get_view_columns())
btn_layout = QHBoxLayout()
search_btn = QPushButton("🔍 搜索")
search_btn.clicked.connect(self.search)
search_btn.setStyleSheet("background-color: #2c7be5; color: white; padding: 8px;")
btn_layout.addWidget(search_btn)
reset_btn = QPushButton("🔄 重置")
reset_btn.clicked.connect(self.reset_filters)
btn_layout.addWidget(reset_btn)
add_filter_btn = QPushButton("＋ 添加条件")
add_filter_btn.clicked.connect(self.add_filter_row)
btn_layout.addWidget(add_filter_btn)
stats_btn = QPushButton("📊 统计")
stats_btn.clicked.connect(self.show_statistics)
btn_layout.addWidget(stats_btn)
export_btn = QPushButton("📎 导出")
export_btn.clicked.connect(self.export_results)
btn_layout.addWidget(export_btn)
if self._default_date_range.get("enabled"):
self.date_range_check = QCheckBox("近一年")
self.date_range_check.setChecked(True)
self.date_range_check.setToolTip(
"勾选时自动限制计划日期范围：去年1月1日至今\n取消勾选可查询所有历史数据"
)
self.date_range_check.stateChanged.connect(lambda: self._on_filter_value_changed())
btn_layout.addWidget(self.date_range_check)
else:
self.date_range_check = None
btn_layout.addStretch()
# 「按计划日期范围筛选」：勾选后用两个日期框限定 plan_date 区间（默认关闭）
if self._plan_date_range_enabled:
self.plan_date_range_check = QCheckBox("按计划日期范围筛选")
self.plan_date_range_check.setToolTip(
"勾选时自动填充当前月 plan_date 区间；取消则清空区间"
)
self.plan_date_range_check.stateChanged.connect(self._on_date_range_toggled)
btn_layout.addWidget(self.plan_date_range_check)
self.date_start = QDateEdit()
self.date_start.setCalendarPopup(True)
self.date_start.setEnabled(False)
self.date_start.setDisplayFormat("yyyy-MM-dd")
btn_layout.addWidget(self.date_start)
self.date_end = QDateEdit()
self.date_end.setCalendarPopup(True)
self.date_end.setEnabled(False)
self.date_end.setDisplayFormat("yyyy-MM-dd")
btn_layout.addWidget(self.date_end)
else:
self.plan_date_range_check = None
self.date_start = None
self.date_end = None
self._pending_toolbar_layout = btn_layout
self.filter_layout.addLayout(btn_layout)
parent_layout.addWidget(self.filter_group)
parent_layout.setStretchFactor(self.filter_group, 0)
# ============================================================
# E1：快速查询（preset 下拉 + 方案 A 快查框）
# ============================================================
@staticmethod
def _build_conditions_from_list(filters: list[dict]) -> dict[str, Any]:
"""把默认筛选条件列表（[{field, operator, value}, ...]）转为查询服务用的条件字典。
返回格式与 ``get_filters`` 一致：``{field: {'value': ..., 'operator': ...}}``。
用于构造「稳定基础条件」，使快查候选项不随临时快查行收窄。
"""
conditions: dict[str, Any] = {}
for f in filters or []:
field = f.get("field")
if field:
conditions[field] = {
"value": f.get("value"),
"operator": f.get("operator", "eq"),
}
return conditions
@staticmethod
def _field_label(field: str) -> str:
"""字段中文标签（仅用于快查框占位提示展示）。"""
_map = {
"sales_order_no": "销售订单号",
"production_order_no": "生产令号",
"plan_date": "计划日期",
"customer": "订货单位",
"project_name": "项目名称",
"plan_no": "计划单号",
}
return _map.get(field, field)
[文档]
def qsp_get_quick_search_items(self, fields: list[str]) -> list[str]:
"""快查候选数据源（重写 mixin 钩子）：以视图默认范围（``_view_default_conditions``）作为
稳定基础抽出去重组合，不随快查临时行收窄（与历史行为一致）。
"""
try:
conditions = getattr(self, "_view_default_conditions", {})
return self.controller.get_distinct_combos(fields, conditions)
except Exception as e: # 兜底：查询异常时清空候选项，不影响主查询
logger.error(f"重算快查候选项失败: {e}")
return []
[文档]
def qsp_get_field_label(self, field: str) -> str:
"""字段中文标签（重写 mixin 钩子，仅用于快查框占位提示）。"""
return self._field_label(field)
[文档]
def on_quick_search(self, or_filters: Any = None) -> None:
"""快查确认入口（mixin 回调）：桥接到 View 的 ``search()``；
mixin 在 AND 分支以无参调用、OR 分支带 ``or_filters`` 调用，二者都能正确转发。
"""
self.search(or_filters=or_filters)
[文档]
def setup_result_table(self, parent_layout: Any) -> None:
"""设置结果表格（迁移到 BaseQueryView：SalePlanQueryView 宿主驱动渲染器）。
保留 QueryView 自身的高级 FilterBar / 快查预设 / 统计 / 导出 / 打印跳转，
仅把结果表格 + 分页展示委托给 ``SalePlanQueryView``；其内置分页条隐藏，
翻页由 QueryView 自有分页控件经 ``run_query`` 驱动。
"""
self._query_view = SalePlanQueryView(
session=self.session, db_manager=self.db_manager, parent=self
)
self._query_view.set_controller(self.controller)
# 状态图标渲染所需的配置透传给渲染器
self._query_view._show_raw_tooltip = self._show_raw_tooltip
self._query_view._show_icon = self._show_icon
# 隐藏 SalePlanQueryView 内置分页条（QueryView 自有分页控件驱动）
for w in (
self._query_view.prev_btn,
self._query_view.next_btn,
self._query_view.page_spin,
self._query_view.page_label,
self._query_view.total_label,
self._query_view.page_size_combo,
):
w.hide()
self._query_view.setMinimumHeight(300)
self._query_view.table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self._query_view.table.customContextMenuRequested.connect(self._show_context_menu)
self._query_view.table.column_config_changed.connect(self._save_column_state)
self._query_view.table.sort_requested.connect(self._on_server_sort_requested)
self._setup_toolbar_actions(self._pending_toolbar_layout)
parent_layout.addWidget(self._query_view)
parent_layout.setStretchFactor(self._query_view, 1)
# ============================================================
# 视图切换（保持不变）
# ============================================================
[文档]
def switch_view(self, view_key: str) -> None:
"""切换当前查询视图"""
if view_key not in self._views:
return
self._save_view_filters()
self.current_view = view_key
self._refresh_quick_filter_bar()
template = self._views[view_key]
for key, action in self.view_actions.items():
action.setChecked(key == view_key)
columns = template.get("columns", [])
self._rebuild_table_header(columns)
self._restore_column_state()
default_filters = template.get("default_filters", [])
saved_conditions = self._load_view_filters(view_key)
self._merge_filter_rows(default_filters, saved_conditions)
# 记录视图默认范围作为快查候选项的「稳定基础」（查询数据库全量，不随快查临时条件收窄）
self._view_default_conditions = self._build_conditions_from_list(default_filters)
# 视图默认范围已变化：按新的全量重算快查候选项（替代原「每次刷新重算」，避免重置选中态）
self._qsp_recompute_items()
self._refresh_toolbar()
self.current_page = 1
# 首屏加载延迟到 showEvent（避免构造期在无事件循环环境阻塞）；切换视图在已显示后即时加载
if self._shown_once:
self.search()
# ============================================================
# 搜索与显示（保持不变）
# ============================================================
def _refresh_toolbar(self) -> None:
"""刷新工具栏按钮（视图切换时调用）"""
template = self._views.get(self.current_view, {})
items = template.get("toolbar_actions", [])
self._status_combo = None
for i in range(self.filter_layout.count()):
child = self.filter_layout.itemAt(i)
if child and isinstance(child, QHBoxLayout):
self._clear_toolbar_widgets(child)
if items:
self._rebuild_toolbar_items(child, items)
break
def _clear_toolbar_widgets(self, layout: QHBoxLayout) -> None:
"""清除工具栏旧控件，保留 date_range_check"""
preserved_widgets = []
while layout.count():
item = layout.takeAt(0)
widget = item.widget()
if widget is not None:
if widget is self.date_range_check:
preserved_widgets.append((widget, item))
else:
widget.deleteLater()
for widget, _ in preserved_widgets:
layout.addWidget(widget)
def _rebuild_toolbar_items(self, layout: QHBoxLayout, items: list) -> None:
"""重新构建工具栏控件"""
self._selection_label = QLabel("选中 0 条")
layout.insertWidget(0, self._selection_label)
layout.insertSpacing(1, 8)
insert_idx = 2
for item_cfg in items:
if item_cfg.get("action") == "separator":
layout.insertSpacing(insert_idx, 8)
insert_idx += 1
continue
if item_cfg.get("type") == "combobox":
insert_idx = self._insert_combobox_item(layout, item_cfg, insert_idx)
continue
btn = QPushButton(item_cfg["label"])
btn.setFixedHeight(28)
btn.clicked.connect(lambda checked, i=item_cfg: self._handle_toolbar_action(i))
layout.insertWidget(insert_idx, btn)
insert_idx += 1
# 添加筛选条件按钮（在所有工具栏按钮之后）
add_filter_btn = QPushButton("＋ 添加条件")
add_filter_btn.setFixedHeight(28)
add_filter_btn.clicked.connect(self.add_filter_row)
layout.insertWidget(insert_idx, add_filter_btn)
insert_idx += 1
# 统计按钮
stats_btn = QPushButton("📊 统计")
stats_btn.setFixedHeight(28)
stats_btn.clicked.connect(self.show_statistics)
layout.insertWidget(insert_idx, stats_btn)
insert_idx += 1
# 导出按钮
export_btn = QPushButton("📎 导出")
export_btn.setFixedHeight(28)
export_btn.clicked.connect(self.export_results)
layout.insertWidget(insert_idx, export_btn)
insert_idx += 1
layout.addStretch()
def _insert_combobox_item(self, layout: QHBoxLayout, item_cfg: dict, insert_idx: int) -> int:
"""插入 combobox 类型的工具栏控件，返回新的 insert_idx"""
label = QLabel(item_cfg.get("label", ""))
label.setFixedHeight(28)
layout.insertWidget(insert_idx, label)
insert_idx += 1
combo = QComboBox()
combo.setFixedHeight(28)
combo.addItems(item_cfg.get("options", []))
default_val = item_cfg.get("default", "")
if default_val:
idx = combo.findText(default_val)
if idx >= 0:
combo.setCurrentIndex(idx)
combo.setMinimumWidth(90)
layout.insertWidget(insert_idx, combo)
insert_idx += 1
if item_cfg.get("action") == "status_combo":
self._status_combo = combo
return insert_idx
[文档]
def search(self, or_filters: Any = None, reset_page: bool = True) -> None:
"""执行搜索
Args:
or_filters: 可选 OR 条件列表（``[{field, operator, value}, ...]``），
与筛选行条件以 OR 组合，用于快查「跨字段任一命中」。
reset_page: 是否将当前页复位到第 1 页。新查询（筛选变更 / 查询按钮 /
预设 / 切换视图）应为 ``True``；翻页与改页大小由调用方显式维持页码，
传 ``False``，避免上一页残留页码超出新结果范围导致显示为空。
"""
self._pending_or_filters = or_filters
if reset_page:
self.current_page = 1
if self._search_timer is None:
self._do_search()
else:
self._search_timer.stop()
self._search_timer.start(200)
[文档]
def refresh(self) -> None:
"""刷新当前视图数据（保留当前页码）"""
self.search(reset_page=False)
[文档]
@override
def showEvent(self, event: Any) -> None:
"""首次显示时加载数据（避免构造期在无事件循环的测试/无头环境触发阻塞式查询）。"""
super().showEvent(event)
if not self._shown_once:
self._shown_once = True
self.search()
def _do_search(self) -> None:
"""实际执行搜索"""
try:
conditions = self.get_filters()
or_filters = getattr(self, "_pending_or_filters", None)
self._pending_or_filters = None
logger.info(f"执行搜索: {conditions}" + (f" (OR: {or_filters})" if or_filters else ""))
template = self._views.get(self.current_view, {})
sort_cfg = template.get("default_sort", {})
sort_field = sort_cfg.get("field", "sort_group")
sort_desc = sort_cfg.get("desc", False)
t_query = time.perf_counter()
result = self._query_view.run_query(
conditions=conditions,
or_filters=or_filters,
order_by=sort_field,
order_desc=sort_desc,
page=self.current_page,
page_size=self.page_size,
reset_page=False,
)
logger.debug(
f"数据库查询完成: {len(result['records'])} 条, 耗时 {time.perf_counter() - t_query:.3f}s"
)
self.page_label.setText(f"第 {result['page']} / {result['total_pages']} 页")
self.status_label.setText(f"共 {result['total']} 条记录")
self.prev_btn.setEnabled(result["page"] > 1)
self.next_btn.setEnabled(result["page"] < result["total_pages"])
except Exception as e:
logger.error(f"搜索失败: {e}")
QMessageBox.critical(self, "错误", f"搜索失败: {e}")
@staticmethod
def _extract_cert_values(record: Any, cert_only_fields: frozenset[str]) -> dict[str, Any]:
"""从记录的 certificates 关联中提取字段值"""
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 ""
# 布尔值特殊处理：False 不应被 value or "" 吞掉
if isinstance(value, bool):
return str(value)
return str(value or "")
def _build_row_values(
self,
record: Any,
col_field_names: list[str],
cert_col_indices: set[int],
cert_only_fields: frozenset[str],
fallback_cols: dict[int, str],
plan_date_col: int | None,
) -> list[str]:
"""从单条记录构建一行的值列表"""
row_data = {k: v for k, v in record.__dict__.items() if not k.startswith("_")}
# hybrid_property 不存储在 __dict__ 中，需通过 getattr 获取
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)
# product_code 显示值：优先从关联 Certificate 动态拼接，fallback 到 Column 值
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 {}
)
row_values: list[str] = []
for col, field_name in enumerate(col_field_names):
text = self._resolve_cell_value(
col,
field_name,
row_data,
cert_values,
cert_col_indices,
fallback_cols,
plan_date_col,
)
row_values.append(text)
return row_values
[文档]
def display_results(self, result: dict[str, Any]) -> None:
"""将查询结果填充到表格中显示（委托 SalePlanQueryView，渲染保真见其 ``_render``）。
注意：不在每次结果刷新时重算快查候选项——否则 QuickSearch 的选中态（currentIndex）
会被重置回首项，导致键盘 ↓ 永远回到第一个候选、无法翻页切换（与 report_view 行为不一致）。
候选仅在初始化 / preset 切换 / 视图默认范围变化时重算。
"""
template = self._views.get(self.current_view, {})
columns = template.get("columns", [])
self._query_view.display(
result["records"], result["total"], result["total_pages"], result["page"], columns
)
self.page_label.setText(f"第 {result['page']} / {result['total_pages']} 页")
self.status_label.setText(f"共 {result['total']} 条记录")
self.prev_btn.setEnabled(result["page"] > 1)
self.next_btn.setEnabled(result["page"] < result["total_pages"])
# ============================================================
# 右键菜单 - 删除记录（委托 Controller）
# ============================================================
@property
def _filter_rows(self) -> list[dict]:
"""兼容桥接：返回 FilterBar 各行 row_data 持久字典的列表（只读）。
迁移期 query_view 的快筛/保存/加载逻辑仍按旧 ``_filter_rows`` 列表结构访问，
此处桥接到 ``FilterBar.rows_data()``；写操作已由各 ``*_filter_row*`` 方法改为
委托 FilterBar API。
"""
return self.filter_bar.rows_data()
[文档]
def get_filters(self) -> dict[str, Any]:
"""获取当前所有筛选条件（委托 FilterBar，统一条件契约）。
遍历所有筛选行，收集非空的筛选条件，构造成查询服务可用的格式。
如果启用了「按计划日期范围筛选」且用户已勾选，自动附加 plan_date 区间条件。
Returns:
筛选条件字典，格式为 {field_name: {'value': ..., 'operator': ...}}
区间条件 value 为 [min, max] 列表
"""
conditions = self.filter_bar.get_conditions()
# 按计划日期范围筛选：用户勾选后附加 plan_date 区间条件
self._apply_plan_date_range(conditions)
return conditions
@staticmethod
def _is_valid_date(date_str: str) -> bool:
"""验证日期字符串是否为有效的 YYYY-MM-DD 格式
Args:
date_str: 待验证的日期字符串
Returns:
是否为有效日期
"""
import re
pattern = r"^\d{4}-\d{2}-\d{2}$"
if not re.match(pattern, date_str):
return False
try:
datetime.strptime(date_str, "%Y-%m-%d")
return True
except ValueError:
return False
# ============================================================
# 按计划日期范围筛选（安全构造 QDate，避免回退到 1752-09-14）
# ============================================================
def _on_date_range_toggled(self, checked: int) -> None:
"""「按计划日期范围筛选」复选框状态切换。
Args:
checked: Qt.CheckState（0/2）或 bool，统一转成布尔处理
"""
enabled = bool(checked)
self.date_range_enabled = enabled
if self.date_start is not None:
self.date_start.setEnabled(enabled)
if self.date_end is not None:
self.date_end.setEnabled(enabled)
if enabled:
self._auto_fill_month_date_range()
else:
self.date_range_start = None
self.date_range_end = None
self._on_filter_value_changed()
def _resolve_workbook_year(self) -> int | None:
"""从当前结果/视图推断计划日期所用年份。
优先使用当前视图下第一条记录的 plan_date 年份；
若无法解析则返回 None（调用方回退到当前年）。
Returns:
int 年份，或 None
"""
try:
saved = self._load_view_filters(self.current_view)
for cond in saved:
if cond.get("field") == self._plan_date_range_field:
ym = normalize_plan_date(cond.get("value"))
if ym:
return int(ym[:4])
except Exception: # noqa: BLE001 - 推断失败不应阻断 UI
pass
return None
def _auto_fill_month_date_range(self, month: int | None = None) -> None:
"""填充当前（或指定）月份的 plan_date 区间到两个日期框。
使用经校验的 safe_qdate 构造 QDate，解析/构造失败时不会把
QDateEdit 静默回退到 Qt 零日期 1752-09-14，而是保留原值并提示。
"""
if self.date_start is None or self.date_end is None:
return
year = self._resolve_workbook_year() or date.today().year
if month is None:
month = date.today().month
if not (1 <= month <= 12):
logger.warning(f"无效月份 {month}，无法填充日期区间")
return
_, last_day = calendar.monthrange(year, month)
start_str = f"{year:04d}-{month:02d}-01"
end_str = f"{year:04d}-{month:02d}-{last_day:02d}"
start_qd = parse_qdate(start_str) or safe_qdate(year, month, 1)
end_qd = parse_qdate(end_str) or safe_qdate(year, month, last_day)
if start_qd is None or end_qd is None:
logger.warning(f"日期区间构造失败: {start_str} ~ {end_str}")
QMessageBox.warning(self, "日期错误", f"无法构造有效日期区间：{start_str} ~ {end_str}")
return
self.date_start.setDate(start_qd)
self.date_end.setDate(end_qd)
self.date_range_start = start_str
self.date_range_end = end_str
def _apply_plan_date_range(self, conditions: dict[str, Any]) -> None:
"""将「按计划日期范围筛选」的区间条件写入 conditions。
Args:
conditions: 当前已收集的筛选条件字典（原地修改）
"""
if not self._plan_date_range_enabled:
return
if self.plan_date_range_check is None or not self.plan_date_range_check.isChecked():
return
if self.date_start is None or self.date_end is None:
return
start_str = self.date_start.date().toString("yyyy-MM-dd")
end_str = self.date_end.date().toString("yyyy-MM-dd")
# 若某个日期框停在 Qt 零日期(1752-09-14)，视为未正确填充，跳过
if (
self.date_start.date().year(),
self.date_start.date().month(),
self.date_start.date().day(),
) == (1752, 9, 14):
logger.warning("计划日期区间起点仍为无效零日期，已跳过区间筛选")
return
field = self._plan_date_range_field
conditions[field] = {
"value": [start_str, end_str],
"operator": "between",
}
def _apply_default_date_range(self, conditions: dict[str, Any]) -> None:
"""如果默认时间范围启用且用户未手动设置该字段，附加 gte 条件
Args:
conditions: 当前已收集的筛选条件字典（原地修改）
"""
if not self._date_range_enabled:
return
if self.date_range_check is None or not self.date_range_check.isChecked():
return
date_field = self._default_date_range.get("field", "plan_date")
if date_field in conditions:
return # 用户已手动设置该字段，不覆盖
conditions[date_field] = {
"value": self._get_default_date_start(),
"operator": "gte",
}
# ============================================================
# 批量操作工具栏（配置驱动）
# ============================================================
def _get_default_date_start(self) -> str:
"""计算默认时间范围的起始日期
根据配置中的 start 值解析：
- "last_year_0101": 去年1月1日
Returns:
日期字符串，格式 YYYY-MM-DD
"""
start_cfg = self._default_date_range.get("start", "last_year_0101")
today = date.today()
if start_cfg == "last_year_0101":
return date(today.year - 1, 1, 1).isoformat()
return start_cfg
[文档]
def add_filter_row(self) -> None:
"""添加一行空筛选条件（公共入口）
创建一个包含字段选择、操作符选择、值输入框和删除按钮的筛选行。
"""
self._add_filter_row()
def _add_filter_row(
self, field_name: str = None, operator: str = None, value: str | list | None = None
) -> None:
"""添加一行筛选条件（委托 FilterBar，兼容旧调用签名）。"""
self._create_filter_row(field_name, operator, value)
def _create_filter_row(
self, field_name: str = None, operator: str = None, value: str | list | None = None
) -> FilterRowWidget:
"""创建一行筛选行（委托 FilterBar）并连接手动编辑检测信号。
值变化 → 防抖搜索；结构与值变化 → 快筛脱离（对齐旧 _connect_filter_row_signals）。
"""
row = self.filter_bar.add_row(field_name=field_name, operator=operator, value=value)
row.value_changed.connect(self._on_filter_value_changed)
row.value_changed.connect(lambda: self._on_manual_edit_detected(row.row_data()))
row.structure_changed.connect(lambda: self._on_manual_edit_detected(row.row_data()))
return row
# 以下旧筛选行构造辅助方法（_create_single_value_page / _create_range_inputs /
# _update_operators / _update_value_stack / _update_value_inputs / _on_field_changed）
# 已迁移至 FilterRowWidget（§1.7 落地组件），由 FilterBar 统一管理，此处不再保留。
def _open_date_picker(self, target_edit: QLineEdit) -> None:
"""打开日期选择对话框，将选中日期写入目标输入框
Args:
target_edit: 目标 QLineEdit，选择的日期会写入该输入框
"""
from PySide6.QtWidgets import QCalendarWidget, QDialog, QDialogButtonBox, QVBoxLayout
dlg = QDialog(self)
dlg.setWindowTitle("选择日期")
dlg.setMinimumSize(300, 250)
layout = QVBoxLayout(dlg)
calendar = QCalendarWidget()
calendar.setGridVisible(True)
# 如果输入框已有有效日期，定位到该日期
current_text = target_edit.text().strip()
if current_text:
try:
current_date = datetime.strptime(current_text, "%Y-%m-%d").date()
calendar.setSelectedDate(current_date)
except ValueError:
pass
layout.addWidget(calendar)
btn_box = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel
)
btn_box.accepted.connect(dlg.accept)
btn_box.rejected.connect(dlg.reject)
layout.addWidget(btn_box)
if dlg.exec() == QDialog.DialogCode.Accepted:
selected = calendar.selectedDate()
target_edit.setText(selected.toString("yyyy-MM-dd"))
def _ensure_at_least_one_filter_row(self) -> None:
"""确保至少有一个筛选行（为空时添加一个空行，委托 FilterBar）。"""
self.filter_bar.ensure_at_least_one_row()
def _save_view_filters(self) -> None:
"""保存当前视图的筛选条件到 userconfig.state.yaml（按段增量写）"""
conditions = self._collect_current_conditions()
merge_user_config({"query_filters": {self.current_view: conditions}})
def _load_view_filters(self, view_key: str) -> list[dict]:
"""从 userconfig.yaml 加载指定视图的筛选条件
Args:
view_key: 视图标识符
Returns:
筛选条件列表，无保存条件时返回空列表
"""
return _get_user_config().get("query_filters", {}).get(view_key, [])
def _save_column_state(self) -> None:
"""保存列状态（宽度、可见性）到 userconfig.state.yaml（按段增量写）"""
state = self._query_view.save_column_state()
merge_user_config({"table_state": {self.current_view: state}})
def _restore_column_state(self) -> None:
"""从 userconfig.yaml 恢复列状态（宽度、可见性）"""
state = _get_user_config().get("table_state", {}).get(self.current_view)
self._query_view.restore_column_state(state)
def _merge_filter_rows(self, default_filters: list[dict], user_conditions: list[dict]) -> None:
"""合并默认筛选条件和用户手动筛选条件
清空并重建筛选行：
- 先添加用户手动设置的条件
- 再添加默认条件中用户未手动设置的（避免重复）
用户"手动设置"的判断标准：条件不在默认过滤器中，或虽然字段相同但值/操作符不同。
Args:
default_filters: 新视图的默认筛选条件
user_conditions: 用户当前设置的筛选条件
"""
# 清空现有筛选行（委托 FilterBar，并按当前视图刷新可用字段）
self.filter_bar.set_columns(self._get_view_columns())
self.filter_bar.clear_rows()
# 获取当前视图所有可用字段
view_columns = self._get_view_columns()
available_fields = {col["field"] for col in view_columns}
# 先添加用户手动条件（只保留当前视图支持的字段）
manual_conditions = self._get_manual_conditions(user_conditions, default_filters)
for condition in manual_conditions:
if condition.get("field") not in available_fields:
continue # 当前视图不支持该字段，跳过
self._add_filter_row(
field_name=condition.get("field"),
operator=condition.get("operator"),
value=condition.get("value"),
)
# 再添加默认条件（跳过用户手动修改过的字段）
# 使用已添加的 manual_conditions 字段集合，而非 user_conditions：
# user_conditions 可能包含与默认完全一致的条件，如果使用它，
# 这些"一致"的条件会阻止默认条件被添加，导致默认条件丢失。
manual_fields = {cond.get("field") for cond in manual_conditions}
for condition in default_filters:
if condition.get("field") not in manual_fields:
self._add_filter_row(
field_name=condition.get("field"),
operator=condition.get("operator"),
value=condition.get("value"),
)
# 如果完全没有筛选行，添加一个空行
if not self._filter_rows:
self._add_filter_row()
def _get_manual_conditions(
self, user_conditions: list[dict], default_filters: list[dict]
) -> list[dict]:
"""判断哪些用户条件是手动添加的（不在默认条件中）
判断标准：
- 字段不在默认条件中：认为是手动添加
- 字段在默认条件中但操作符或值不同：认为是手动修改
Args:
user_conditions: 用户当前所有条件
default_filters: 新视图的默认条件
Returns:
手动条件的列表
"""
# 构建默认条件字典 {field: {operator, value}}
default_dict = {}
for cond in default_filters:
default_dict[cond["field"]] = {
"operator": cond.get("operator"),
"value": cond.get("value"),
}
manual = []
for cond in user_conditions:
field = cond["field"]
if field not in default_dict:
# 字段不在默认条件中，认为是手动添加
manual.append(cond)
else:
default_cond = default_dict[field]
# 比较操作符和值
if cond["operator"] != default_cond["operator"] or cond.get(
"value"
) != default_cond.get("value"):
# 用户修改过，保留用户的值
manual.append(cond)
# 否则与默认一致，不添加（后面会由默认条件补充）
return manual
[文档]
def reset_filters(self) -> None:
"""重置筛选条件
恢复为当前视图的默认筛选条件（或空行），恢复默认时间范围复选框，重置页码并重新搜索。
"""
template = self._views.get(self.current_view, {})
self._rebuild_filter_rows(template.get("default_filters", []))
if self.date_range_check is not None:
self.date_range_check.setChecked(True)
self.current_page = 1
self.search()
def _rebuild_filter_rows(self, default_filters: list[dict]) -> None:
"""重建筛选行
清空现有筛选行，根据默认筛选条件或空行重新构建。
Args:
default_filters: 默认筛选条件列表，每项包含 field、operator、value
"""
# ========== 清理快速筛选按钮标记（桥接 row_data 持久字典） ==========
for row in self.filter_bar.rows_data():
row.pop("_quick_filter_button", None)
row.pop("_is_manual", None)
self.filter_bar.clear_rows()
if default_filters:
for condition in default_filters:
self._add_filter_row(
field_name=condition.get("field"),
operator=condition.get("operator"),
value=condition.get("value"),
)
else:
self._add_filter_row()
# ============================================================
# 动态筛选行
# ============================================================
def _rebuild_table_header(self, columns: list[dict]) -> None:
"""重建表格列头（委托 SalePlanQueryView，含展示字段不可排序处理 + 排序指示）。"""
self._query_view.set_view_columns(columns)
template = self._views.get(self.current_view, {})
sort_cfg = template.get("default_sort", {})
sort_field = sort_cfg.get("field")
sort_desc = sort_cfg.get("desc", False)
if sort_field:
self._query_view.table.update_sort_indicator(sort_field, sort_desc)
def _on_server_sort_requested(self, field: str, desc: bool) -> None:
"""处理服务端排序请求
用户点击列头时触发，更新当前视图的排序配置并重新从数据库查询排序后的数据。
Args:
field: 排序字段名
desc: True=降序, False=升序
"""
template = self._views.get(self.current_view, {})
template["default_sort"] = {"field": field, "desc": desc}
self.current_page = 1
self.search()
[文档]
def setup_quick_filter_bar(self, parent_layout: Any) -> None:
"""设置快速筛选按钮栏"""
quick_filters, filter_groups = self._get_quick_filter_config()
if not quick_filters and not filter_groups:
return
is_dark = self._is_dark_theme()
self._quick_bar = QWidget()
quick_layout = QHBoxLayout(self._quick_bar)
quick_layout.setContentsMargins(0, 0, 0, 4)
quick_layout.addWidget(QLabel("快速筛选:"))
self._build_quick_filter_buttons(quick_filters, quick_layout, is_dark)
self._build_filter_group_button(filter_groups, quick_layout, is_dark)
self._build_clear_button(quick_layout, is_dark)
quick_layout.addStretch()
parent_layout.insertWidget(0, self._quick_bar)
def _get_quick_filter_config(self) -> Any:
"""获取快速筛选配置"""
template = self._views.get(self.current_view, {})
quick_filters = template.get("quick_filters")
if quick_filters is None:
quick_filters = cfg("query_page.quick_filters", [])
filter_groups = cfg("query_page.filter_groups", [])
return quick_filters, filter_groups
def _build_quick_filter_buttons(
self, quick_filters: list, quick_layout: QHBoxLayout, is_dark: bool
) -> None:
"""构建快速筛选按钮"""
for qf in quick_filters:
btn = QPushButton(f"{qf.get('icon', '')} {qf['name']}")
btn.setFixedHeight(26)
btn.setCheckable(True)
btn.setProperty("filter_group", qf.get("group", ""))
btn.setStyleSheet(self._get_filter_button_style(is_dark))
# 三态 toggle 配置
if qf.get("toggle"):
btn.setProperty("_toggle", True)
btn.setProperty("_toggle_state", 0) # 0=未选中, 1=正向, 2=反向
btn.setProperty("_toggle_config", qf)
else:
btn.setProperty("_toggle", False)
conditions = qf["conditions"]
btn.clicked.connect(
lambda checked, c=conditions, b=btn, q=qf: self._apply_quick_filter(c, b, q)
)
quick_layout.addWidget(btn)
self._quick_filter_buttons[qf["name"]] = btn
def _build_filter_group_button(
self, filter_groups: list, quick_layout: QHBoxLayout, is_dark: bool
) -> None:
"""构建筛选组下拉按钮"""
if not filter_groups:
return
group_btn = QPushButton("📂 筛选组")
group_btn.setFixedHeight(26)
group_btn.setStyleSheet(self._get_group_button_style(is_dark))
group_menu = QMenu(self)
if is_dark:
group_menu.setStyleSheet(self._get_menu_style(is_dark))
for group in filter_groups:
action = group_menu.addAction(f"{group.get('icon', '')} {group['name']}")
action.setData(group["fields"])
action.triggered.connect(
lambda checked, f=group["fields"]: self._open_filter_group_dialog(f)
)
group_btn.setMenu(group_menu)
quick_layout.addWidget(group_btn)
def _build_clear_button(self, quick_layout: QHBoxLayout, is_dark: bool) -> None:
"""构建清除按钮"""
clear_btn = QPushButton("× 清除")
clear_btn.setFixedHeight(26)
clear_btn.setStyleSheet(self._get_clear_button_style(is_dark))
clear_btn.clicked.connect(self._clear_quick_filters)
quick_layout.addWidget(clear_btn)
def _refresh_quick_filter_bar(self) -> None:
"""切换视图时重建快速筛选按钮栏
不同视图可能有不同的 quick_filters，需要销毁旧按钮并重建。
"""
if not hasattr(self, "_quick_bar") or self._quick_bar is None:
return
# 清除筛选行中的快速筛选标记
for row in self._filter_rows:
row.pop("_quick_filter_button", None)
row["_is_manual"] = True
# 销毁旧按钮栏，重新构建
self._quick_bar.deleteLater()
self._quick_bar = None
self._quick_filter_buttons = {}
self.setup_quick_filter_bar(self.filter_layout)
def _apply_quick_filter(
self, conditions: list[dict], button: QPushButton, qf_config: dict = None
) -> None:
"""应用或移除快速筛选条件（支持三态切换）
普通按钮：选中→应用条件，取消→移除条件
三态按钮：未选中→正向条件→反向条件→移除（循环）
Args:
conditions: 正向筛选条件列表
button: 触发操作的快速筛选按钮
qf_config: 按钮的完整配置字典（含 toggle/reverse_conditions 等）
"""
is_toggle = button.property("_toggle") is True
if is_toggle and qf_config:
self._apply_toggle_filter(button, qf_config)
return
# 普通按钮逻辑
# 取消选中 → 移除条件
if not button.isChecked():
self._remove_quick_filter_conditions(button)
self._on_filter_value_changed()
return
# 组内互斥：取消同组其他按钮
group = button.property("filter_group")
if group:
for _name, btn in self._quick_filter_buttons.items():
if btn != button and btn.property("filter_group") == group and btn.isChecked():
btn.blockSignals(True)
btn.setChecked(False)
btn.blockSignals(False)
self._remove_quick_filter_conditions(btn)
# 添加条件到筛选行
for condition in conditions:
self._apply_condition_to_row(condition, button)
self.current_page = 1
self._on_filter_value_changed()
def _apply_toggle_filter(self, button: QPushButton, qf_config: dict) -> None:
"""三态切换逻辑：未选中(0) → 正向(1) → 反向(2) → 未选中(0)
Args:
button: 三态按钮
qf_config: 按钮配置（含 conditions/reverse_conditions/reverse_name/reverse_icon）
"""
current_state = button.property("_toggle_state") or 0
next_state = (current_state + 1) % 3
is_dark = self._is_dark_theme()
if next_state == 0:
# 状态 0：移除条件，恢复正常样式
self._remove_quick_filter_conditions(button)
button.setText(f"{qf_config.get('icon', '')} {qf_config['name']}")
button.setStyleSheet(self._get_filter_button_style(is_dark))
button.setChecked(False)
elif next_state == 1:
# 状态 1：正向条件
button.setText(f"{qf_config.get('icon', '')} {qf_config['name']}")
button.setStyleSheet(self._get_filter_button_style(is_dark))
button.setChecked(True)
self._replace_toggle_conditions(qf_config["conditions"], button)
elif next_state == 2:
# 状态 2：反向条件
reverse_name = qf_config.get("reverse_name", qf_config["name"])
reverse_icon = qf_config.get("reverse_icon", qf_config.get("icon", ""))
button.setText(f"{reverse_icon} {reverse_name}")
button.setStyleSheet(self._get_toggle_reverse_button_style(is_dark))
button.setChecked(True)
self._replace_toggle_conditions(qf_config["reverse_conditions"], button)
button.setProperty("_toggle_state", next_state)
self.current_page = 1
self._on_filter_value_changed()
def _replace_toggle_conditions(self, conditions: list[dict], button: QPushButton) -> None:
"""替换三态按钮对应的筛选行（先移除旧的，再添加新的）
Args:
conditions: 新的筛选条件列表
button: 关联的按钮
"""
self._remove_quick_filter_conditions(button)
for condition in conditions:
self._apply_condition_to_row(condition, button)
def _apply_condition_to_row(self, condition: dict, button: QPushButton) -> None:
"""将单个快速筛选条件应用到筛选行（新建或更新已有行）
Args:
condition: 包含 field/operator/value 的条件字典
button: 关联的快速筛选按钮
"""
field = condition["field"]
operator = condition.get("operator", "eq")
value = self._resolve_special_value(condition.get("value"))
existing_row = self._find_filter_row_by_field(field)
if existing_row:
op_type = self._field_op_map.get(field, "text")
ve = existing_row["value_edit"]
if op_type == "exact" and not isinstance(ve, QComboBox):
self._remove_filter_row_widget(existing_row)
existing_row = None
if existing_row:
op_idx = existing_row["op_combo"].findData(operator)
if op_idx >= 0:
existing_row["op_combo"].setCurrentIndex(op_idx)
if value is not None:
ve = existing_row["value_edit"]
if isinstance(ve, QComboBox):
idx = ve.findText(str(value))
if idx >= 0:
ve.setCurrentIndex(idx)
else:
ve.setCurrentText(str(value))
else:
ve.setText(str(value))
existing_row["_quick_filter_button"] = button
else:
row_data = self._add_filter_row_return(field_name=field, operator=operator, value=value)
row_data["_quick_filter_button"] = button
row_data["_is_manual"] = False
self._update_row_style(row_data)
def _remove_quick_filter_conditions(self, button: QPushButton) -> None:
"""移除指定快速筛选按钮对应的所有条件行
Args:
button: 被取消选中的快速筛选按钮
"""
rows_to_remove = []
for row in self._filter_rows:
if row.get("_quick_filter_button") == button:
rows_to_remove.append(row)
for row in rows_to_remove:
self._remove_filter_row_widget(row)
def _remove_filter_row_widget(self, row_data: dict) -> None:
"""根据行数据移除筛选行控件（委托 FilterBar）。"""
self.filter_bar.remove_row(row_data["widget"])
def _resolve_special_value(self, value: str) -> str | None:
"""解析特殊值占位符
支持的占位符：
THIS_MONTH_START → 本月1日
THIS_YEAR_START → 今年1月1日
Args:
value: 原始值
Returns:
解析后的值，如果不是特殊占位符则原样返回
"""
if value == "THIS_MONTH_START":
from datetime import date
return date.today().replace(day=1).isoformat()
if value == "THIS_YEAR_START":
from datetime import date
return date.today().replace(month=1, day=1).isoformat()
return value
def _find_filter_row_by_field(self, field_name: str) -> dict | None:
"""查找指定字段的筛选行
Args:
field_name: 数据库字段名
Returns:
找到的筛选行数据字典，未找到返回 None
"""
for row in self._filter_rows:
if row["field_combo"].currentData() == field_name:
return row
return None
def _add_filter_row_return(
self, field_name: str = None, operator: str = None, value: str = None
) -> dict:
"""添加筛选行并返回行数据引用
Args:
field_name: 字段名
operator: 操作符
value: 值
Returns:
新创建的筛选行数据字典
"""
return self._create_filter_row(field_name, operator, value).row_data()
def _on_manual_edit_detected(self, row_data: dict) -> None:
"""检测到用户手动修改快速筛选条件行
取消该行与快速筛选按钮的关联，转为普通手动条件。
Args:
row_data: 被修改的筛选行数据
"""
button = row_data.get("_quick_filter_button")
if button and button.isChecked():
button.blockSignals(True)
button.setChecked(False)
button.blockSignals(False)
row_data.pop("_quick_filter_button", None)
row_data["_is_manual"] = True
self._update_row_style(row_data)
def _update_row_style(self, row_data: dict) -> None:
"""更新筛选行的视觉样式
快速筛选条件：浅蓝色背景
手动条件：默认白色背景
Args:
row_data: 筛选行数据
"""
widget = row_data["widget"]
if row_data.get("_quick_filter_button"):
widget.setStyleSheet("""
QWidget {
background-color: #e8f0fe;
border: 1px solid #a8c7fa;
border-radius: 3px;
padding: 2px;
}
""")
else:
widget.setStyleSheet("")
def _clear_quick_filters(self) -> None:
"""清除所有快速筛选条件
取消所有快速筛选按钮的选中状态，
移除所有关联的筛选行，恢复默认状态。
"""
is_dark = self._is_dark_theme()
# 取消所有按钮选中，重置三态按钮
for btn in self._quick_filter_buttons.values():
if btn.isChecked() or btn.property("_toggle_state"):
btn.blockSignals(True)
btn.setChecked(False)
btn.blockSignals(False)
# 重置三态按钮文字和样式
if btn.property("_toggle"):
cfg = btn.property("_toggle_config")
if cfg:
btn.setText(f"{cfg.get('icon', '')} {cfg['name']}")
btn.setStyleSheet(self._get_filter_button_style(is_dark))
btn.setProperty("_toggle_state", 0)
# 移除所有筛选行中的快速筛选标记
rows_to_remove = []
for row in self._filter_rows:
if row.get("_quick_filter_button"):
rows_to_remove.append(row)
for row in rows_to_remove:
self._remove_filter_row_widget(row)
self._ensure_at_least_one_filter_row()
self._on_filter_value_changed()
def _open_filter_group_dialog(self, fields: list[str]) -> None:
"""打开筛选组对话框
弹出模态对话框，集中填写一组相关的筛选条件。
Args:
fields: 要显示的字段名列表
"""
from PySide6.QtWidgets import QDialog
dialog = QDialog(self)
dialog.setWindowTitle("组合筛选")
dialog.setMinimumWidth(420)
layout = QVBoxLayout(dialog)
layout.setSpacing(8)
inputs = {}
for field in fields:
row = QHBoxLayout()
# 获取字段标签
label_text = field
for view_cols in self._views.values():
for col in view_cols.get("columns", []):
if col["field"] == field:
label_text = col["label"]
break
row.addWidget(QLabel(f"{label_text}:"))
# 操作符选择
op_combo = QComboBox()
op_type = self._field_op_map.get(field, "text")
operators = self._operators.get(op_type, {})
for op_key, op_label in operators.items():
op_combo.addItem(op_label, op_key)
row.addWidget(op_combo)
# 值输入
value_edit = QLineEdit()
value_edit.setPlaceholderText(f"输入{label_text}...")
value_edit.setMinimumWidth(180)
row.addWidget(value_edit)
layout.addLayout(row)
inputs[field] = {"operator": op_combo, "value": value_edit}
# 按钮
btn_layout = QHBoxLayout()
btn_layout.addStretch()
apply_btn = QPushButton("应用并搜索")
apply_btn.setStyleSheet("background-color: #2c7be5; color: white; padding: 6px 16px;")
apply_btn.clicked.connect(lambda: self._apply_group_filter(inputs, dialog))
btn_layout.addWidget(apply_btn)
cancel_btn = QPushButton("取消")
cancel_btn.clicked.connect(dialog.reject)
btn_layout.addWidget(cancel_btn)
layout.addLayout(btn_layout)
dialog.exec()
def _apply_group_filter(self, inputs: dict, dialog: Any) -> None:
"""应用组合筛选条件到筛选行
Args:
inputs: 字段->控件映射字典
dialog: 要关闭的对话框
"""
for field, controls in inputs.items():
operator = controls["operator"].currentData()
value = controls["value"].text().strip()
if not value and operator not in ["is_null", "is_not_null"]:
continue
self._add_filter_row(field_name=field, operator=operator, value=value)
dialog.accept()
self.current_page = 1
self.search()
def _setup_toolbar_actions(self, layout: Any) -> None:
"""构建批量操作工具栏（根据当前视图配置）"""
# 获取当前视图的工具栏配置（来自 ui.yaml canonical 源）
template = self._views.get(self.current_view, {})
items = template.get("toolbar_actions", [])
# 选中计数标签
self._selection_label = QLabel("选中 0 条")
layout.addWidget(self._selection_label)
layout.addSpacing(8)
# 存储 status_combo 引用
self._status_combo: QComboBox | None = None
for item in items:
if item.get("action") == "separator":
layout.addSpacing(8)
continue
item_type = item.get("type", "")
if item_type == "combobox":
# 创建下拉框 + 标签
label = QLabel(item.get("label", ""))
label.setFixedHeight(28)
layout.addWidget(label)
combo = QComboBox()
combo.setFixedHeight(28)
combo.addItems(item.get("options", []))
default_val = item.get("default", "")
if default_val:
idx = combo.findText(default_val)
if idx >= 0:
combo.setCurrentIndex(idx)
combo.setMinimumWidth(90)
layout.addWidget(combo)
if item.get("action") == "status_combo":
self._status_combo = combo
continue
btn = QPushButton(item["label"])
btn.setFixedHeight(28)
# 捕获item，避免闭包问题
btn.clicked.connect(lambda checked, i=item: self._handle_toolbar_action(i))
layout.addWidget(btn)
# 监听选择变化更新计数
self._query_view.selection_changed.connect(self._update_selection_label)
def _handle_toolbar_action(self, item: dict) -> None:
"""分发工具栏动作（统一经 action 注册表派发，蓝图 §1.6 / §7.2 任务 1.2 全接入）"""
act_name = item["action"]
logger.info(f"工具栏操作: {act_name}")
if not dispatch(self, act_name, item):
logger.warning("工具栏动作未注册或视图无处理方法: %s", act_name)
self._update_selection_label()
def _update_selection_label(self) -> None:
count = len(self._query_view.selected_rows())
self._selection_label.setText(f"选中 {count} 条")
# ------------------------------------------------------------------
# action 命名方法（Phase 3.2：原 lambda 提升，统一经 actions.dispatch 派发）
# ------------------------------------------------------------------
def _action_select_all(self) -> None:
self._query_view.table.selectAll()
def _action_hide_selected(self) -> None:
self._batch_update_field("is_hidden", True)
def _action_hide_rows(self) -> None:
self._batch_update_field("is_hidden", True)
def _action_mark_inspected(self) -> None:
self._batch_update_field("inspection_status", "已检验")
def _action_mark_rejected(self) -> None:
self._batch_update_field("inspection_status", "不合格")
def _action_set_make(self) -> None:
self._batch_update_field("supply_type", "自制")
def _action_set_buy(self) -> None:
self._batch_update_field("supply_type", "外购")
def _action_set_repair(self) -> None:
self._batch_update_field("supply_type", "维修")
def _action_toggle_needs_numbering(self, item: dict | None = None) -> None:
"""切换选中行的「是否需要编号」（维度16/H：UI 覆盖免出证件件的编号开关）
当前 ``needs_numbering=True`` 的行设为 ``False``（编号时跳过），
``False`` 的行设为 ``True``。重导入会因派生覆盖，故为会话内即时覆盖。
"""
rows = self._query_view.selected_rows()
if not rows:
QMessageBox.warning(self, "提示", "请先选择行")
return
records = self._query_view.records
true_ids: list[int] = []
false_ids: list[int] = []
for r in rows:
if r >= len(records):
continue
rec = records[r]
if getattr(rec, "needs_numbering", True):
false_ids.append(rec.id)
else:
true_ids.append(rec.id)
try:
if false_ids:
self.controller.batch_update_field(false_ids, "needs_numbering", False)
if true_ids:
self.controller.batch_update_field(true_ids, "needs_numbering", True)
self._op_log(f"切换是否需要编号 | 设False={len(false_ids)} 设True={len(true_ids)}")
except Exception as e:
self._op_log(f"切换是否需要编号失败 | 错误={e}")
return
self.search()
# --- 配置驱动通用 action 方法（任务 1.2 全接入，统一经 actions.dispatch 派发）---
def _action_set_status(self, item: dict | None = None, action_name: str = "") -> None:
"""设置生产状态：优先取 item.production_status，否则从 set_status_<x> 名解析。"""
if item and item.get("production_status") is not None:
status = item["production_status"]
elif action_name.startswith("set_status_"):
status = action_name[len("set_status_") :]
else:
return
self._batch_update_field("production_status", status)
def _action_set_supply_type(self, item: dict | None = None, action_name: str = "") -> None:
"""设置供货类型：优先取 item.supply_type，否则从 set_supply_type_<x> 名解析。"""
if item and item.get("supply_type") is not None:
value = item["supply_type"]
elif action_name.startswith("set_supply_type_"):
value = {
"make": "自制",
"buy": "外购",
"repair": "维修",
}.get(action_name[len("set_supply_type_") :], "")
else:
return
self._batch_update_field("supply_type", value)
def _action_set_clear(self) -> None:
self._batch_update_field("supply_type", "")
def _action_open_lookup_table(self, item: dict | None = None) -> None:
table = item.get("table") if item else None
self._open_lookup_table(table)
def _action_view_raw_execution(self) -> None:
self._show_raw_execution_status()
def _action_mark_document_done(self, action_name: str = "") -> None:
doc_type = (
action_name[len("mark_") : -len("_done")] if action_name.endswith("_done") else ""
)
self._mark_document_done(doc_type)
[文档]
def get_selected_ids(self) -> list[int]:
"""获取当前选中的 SalePlan ID 列表
用于跨页面数据传递，将查询视图中选中的记录 ID 传递给合格证页面。
Returns:
List[int]: 选中的 SalePlan.id 列表
"""
return self._query_view.selected_sale_plan_ids()
def _go_back_home(self) -> None:
"""返回首页"""
main_window = self.window()
if main_window and hasattr(main_window, "content_stack"):
main_window.content_stack.setCurrentIndex(0)
def _invert_selection(self) -> None:
"""反选：选中未选中的行，取消已选中的行"""
table = self._query_view.table
table.setSelectionMode(QTableWidget.SelectionMode.MultiSelection)
for row in range(table.rowCount()):
is_selected = table.selectionModel().isRowSelected(row, table.rootIndex())
table.setRangeSelected(
QTableWidgetSelectionRange(row, 0, row, table.columnCount() - 1),
not is_selected,
)
table.setSelectionMode(QTableWidget.SelectionMode.ExtendedSelection)
# ============================================================
# 右键菜单（配置驱动）
# ============================================================
def _show_context_menu(self, pos: int) -> None:
"""右键菜单（根据当前视图配置）"""
menu_items = self._get_context_menu_items()
if not menu_items:
return
selected_rows = self._query_view.selected_rows()
records = self._query_view.records
has_hidden = (
any(records[r].is_hidden for r in selected_rows if r < len(records))
if selected_rows
else False
)
menu = QMenu(self)
action_map: dict[QAction, dict] = {}
self._build_context_menu(menu, menu_items, action_map, has_hidden)
action = menu.exec(self._query_view.table.viewport().mapToGlobal(pos))
if action is None:
return
item = action_map.get(action)
if not item:
return
self._handle_context_action(item)
def _get_context_menu_items(self) -> list:
"""获取当前视图的右键菜单配置
从 ui.yaml（canonical 源，``query_page.views``）读取基础菜单配置，再叠加
ui.yaml 的动态 UI 配置（§17：``UIConfig.get_view_context_menu``），最后根据
config.yaml 的 production_status.context_menu 动态追加"查看原始执行情况"菜单项。
叠加策略：ui.yaml 中基础菜单不存在的菜单项（按 action 去重）会被追加，
既有菜单项不受影响——实现"动态增"而不破坏现有功能；移除既有项作为后续增强。
"""
template = self._views.get(self.current_view, {})
items = list(template.get("context_menu", []))
# 动态 UI 配置叠加（ui.yaml → UIConfig，可编辑增删右键菜单项）
ui_cfg = get_ui_config()
if ui_cfg:
existing_actions = {it.get("action") for it in items}
for ui_item in ui_cfg.get_view_context_menu(self.current_view):
action = ui_item.get("action")
if action not in existing_actions:
items.append(ui_item)
existing_actions.add(action)
# 动态添加"查看原始执行情况"菜单（如果配置启用且有数据）
ctx_cfg = cfg("production_status.context_menu", {})
view_raw_cfg = ctx_cfg.get("view_raw_execution", {})
if view_raw_cfg.get("enabled", False) and self._has_selected_execution_status():
items.append(
{
"action": "view_raw_execution",
"label": view_raw_cfg.get("label", "📋 查看原始执行情况"),
}
)
# 动态追加"维护基础数据"子菜单（配置驱动，见 config.yaml 的
# lookup_tables.context_menu；右键即可就近打开对应对照表编辑器）
lookup_ctx = cfg("lookup_tables.context_menu", {})
if lookup_ctx.get("enabled", True):
items.append(
{
"action": "submenu",
"label": lookup_ctx.get("label", "🗂 维护基础数据"),
"submenu": lookup_ctx.get("items", []),
}
)
# 动态追加"切换是否需要编号"（维度16/H：UI 覆盖免出证件件的编号开关）
items.append(
{
"action": "toggle_needs_numbering",
"label": "🔢 切换是否需要编号",
}
)
return items
def _build_context_menu(
self, parent_menu: Any, items: list, action_map: dict, has_hidden: bool
) -> None:
"""递归构建菜单（支持 submenu）"""
for item in items:
if item.get("action") == "separator":
parent_menu.addSeparator()
continue
act_name = item["action"]
if act_name == "submenu":
sub_menu = parent_menu.addMenu(item["label"])
self._build_context_menu(sub_menu, item.get("submenu", []), action_map, has_hidden)
continue
label = self._resolve_menu_label(item, has_hidden)
act = parent_menu.addAction(label)
act.setData(item)
action_map[act] = item
def _resolve_menu_label(self, item: dict, has_hidden: bool) -> str:
"""根据菜单项配置和上下文，解析最终显示的标签文本"""
act_name = item["action"]
if (act_name == "hide_rows" or act_name == "unhide_rows") and has_hidden:
return item.get("unhide_label", "👁 取消隐藏")
if act_name in ("hide_rows", "unhide_rows"):
return item.get("label", "🙈 隐藏选中行")
return item["label"]
# 右键菜单动作 → 文档类型映射（进度跟踪）
_PROGRESS_DOC_MAP: dict[str, str] = {
"mark_certificate_done": "certificate",
"mark_nameplate_done": "nameplate",
"mark_test_report_done": "test_report",
"mark_warranty_done": "warranty",
"mark_scan_done": "scan",
}
# 进度 Boolean 字段集合（显示 ✅/⬜ 图标）
def _handle_context_action(self, item: dict) -> None:
"""分发右键菜单动作（统一经 action 注册表派发，蓝图 §7.2 任务 1.2 全接入）"""
act_name = item["action"]
if not dispatch(self, act_name, item):
logger.warning("右键动作未注册或视图无处理方法: %s", act_name)
def _preview_certificate(self) -> None:
"""预览合格证（合格证视图专用）"""
selected_ids = self.get_selected_ids()
if not selected_ids:
QMessageBox.warning(self, "提示", "请先选中记录")
return
# 获取选中的 Certificate 记录（委托 Controller）
certs = self.controller.get_certificates_by_plan_ids(selected_ids)
if not certs:
QMessageBox.warning(self, "提示", "未找到合格证记录，请先生成合格证")
return
# 弹出预览对话框
from certflow.views.print_dialog import PrintPreviewDialog
for cert in certs:
cert_data = {
"product_name": cert.product_name or "",
"product_model": cert.product_model or "",
"dn": cert.product_spec or "",
"pn": cert.pn_display or "",
"temperature": cert.working_temp or "",
"medium": cert.working_medium or "",
"check_standard": cert.test_standard or "",
"inspector_id": "",
"manufacture_date": cert.issue_date or "",
"certificate_no": cert.certificate_no or "",
}
dialog = PrintPreviewDialog(cert_data, cert.certificate_no or "预览", parent=self)
dialog.exec()
break # 只预览第一个
def _has_selected_execution_status(self) -> bool:
"""检查选中行是否有 execution_status
Returns:
True 如果至少有一个选中行有 execution_status
"""
rows = self._query_view.selected_rows()
records = self._query_view.records
for row in rows:
if row < len(records) and getattr(records[row], "execution_status", ""):
return True
return False
def _show_raw_execution_status(self) -> None:
"""右键菜单：查看选中行的原始执行情况"""
rows = self._query_view.selected_rows()
records = self._query_view.records
if not rows:
QMessageBox.information(self, "提示", "请先选中记录")
return
exec_statuses: dict[int, str] = {}
for row in rows:
if row >= len(records):
continue
record = records[row]
if getattr(record, "execution_status", None):
exec_statuses[record.id] = str(record.execution_status)
if not exec_statuses:
QMessageBox.information(self, "提示", "选中记录均无原始执行情况")
return
lines = []
for record_id, status in exec_statuses.items():
lines.append(f"ID {record_id}: {status}")
msg = "\n".join(lines)
QMessageBox.information(self, "原始执行情况", msg)
def _open_lookup_table(self, table: str | None) -> None:
"""右键菜单：打开基础数据维护对话框并定位到指定标签页
Args:
table: 标签页标识（material_grade / caliber_mapping / model_param），
由 config.yaml 的 lookup_tables.context_menu.items[].table 提供
"""
from certflow.views.lookup_tables_dialog import LookupTablesDialog
main_win = self.window()
db_manager = getattr(main_win, "db_manager", None)
session = getattr(main_win, "session", None)
dialog = LookupTablesDialog(
db_manager=db_manager, session=session, initial_table=table, parent=self
)
dialog.exec()
def _delete_selected_records(self) -> None: # noqa: C901
"""删除选中的记录（带二次确认和事务回滚）"""
rows = self._query_view.selected_rows()
all_records = self._query_view.records
if not rows:
QMessageBox.warning(self, "提示", "请先选中要删除的记录")
return
records = []
for row_idx in rows:
if row_idx < len(all_records):
record = all_records[row_idx]
records.append(
{
"id": record.id,
"sales_order_no": getattr(record, "sales_order_no", ""),
"product_name": getattr(record, "product_name", ""),
"product_model": getattr(record, "product_model", ""),
}
)
if not records:
QMessageBox.warning(self, "提示", "没有可删除的记录")
return
# 检查是否有关联合格证
ids = [r["id"] for r in records]
cert_count = self.controller.check_certificate_exists(ids)
# 用户偏好：关闭「删除前确认」时直接执行删除（级联 or 标准）
if self._try_delete_without_confirm(records, cert_count):
return
# 构建确认消息
confirm_msg = self._build_delete_confirmation(records, cert_count)
# 如果有合格证，询问是否级联删除
if cert_count > 0:
reply = QMessageBox.question(
self,
"⚠️ 关联合格证",
confirm_msg,
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No,
)
if reply != QMessageBox.StandardButton.Yes:
return
# 执行级联删除
success, deleted_count, msg = self.controller.delete_with_certificates(records)
if success:
QMessageBox.information(self, "删除成功", msg)
else:
QMessageBox.critical(self, "删除失败", msg)
return
# 无合格证，标准删除流程
reply = QMessageBox.question(
self,
"⚠️ 确认删除",
confirm_msg,
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No,
)
if reply != QMessageBox.StandardButton.Yes:
return
success, deleted_count, msg = self.controller.delete_records(records)
if success:
QMessageBox.information(self, "删除成功", f"已成功删除 {deleted_count} 条记录")
else:
QMessageBox.critical(self, "删除失败", msg)
def _try_delete_without_confirm(self, records: list[dict], cert_count: int) -> bool:
"""若用户关闭「删除前确认」偏好，则直接执行删除并返回 True。
返回 False 表示需要走常规确认流程。
"""
from certflow.config.settings import should_confirm_before_delete
if should_confirm_before_delete():
return False
if cert_count > 0:
success, deleted_count, msg = self.controller.delete_with_certificates(records)
else:
success, deleted_count, msg = self.controller.delete_records(records)
if success:
QMessageBox.information(self, "删除成功", msg)
else:
QMessageBox.critical(self, "删除失败", msg)
return True
def _build_delete_confirmation(self, records: list[dict], cert_count: int = 0) -> str:
"""构建删除确认对话框的详细信息"""
count = len(records)
preview = ""
for _i, rec in enumerate(records[:5]):
preview += f" • #{rec['id']} | {rec.get('sales_order_no', '')} | {rec.get('product_name', '')}\n"
if count > 5:
preview += f" ... 以及其他 {count - 5} 条记录\n"
msg = f"即将删除 {count} 条记录，此操作不可恢复！\n\n确认删除以下记录吗？\n\n{preview}"
if cert_count > 0:
msg += f"\n⚠️ 同时存在 {cert_count} 条关联合格证，将一并删除！"
return msg
# ============================================================
# 批量操作（委托 Controller）
# ============================================================
def _batch_update_field(self, field_name: str, value: str) -> None:
"""通用：批量更新选中行的某字段
Args:
field_name: 数据库字段名
value: 要设置的值
"""
rows = self._query_view.selected_rows()
if not rows:
QMessageBox.warning(self, "提示", "请先选择行")
return
ids = [self._query_view.records[row_idx].id for row_idx in rows]
try:
updated = self.controller.batch_update_field(ids, field_name, value)
self._op_log(f"右键批量更新完成 | 字段={field_name} | 更新数量={updated}")
except Exception as e:
self._op_log(f"右键批量更新失败 | 字段={field_name} | 错误={e}")
return
self.search()
self.status_label.setText(f"已更新 {updated} 条")
# ============================================================
# 生产状态批量操作
# ============================================================
def _export_selected(self) -> None:
"""导出选中行到 Excel
按当前视图的列定义导出选中行的数据。
委托 Controller → Service → Handler 处理 Excel 写入。
"""
rows = self._query_view.selected_rows()
if not rows:
return
file_path, _ = QFileDialog.getSaveFileName(self, "导出选中行", "", "Excel (*.xlsx)")
if not file_path:
return
ids = [self._query_view.records[r].id for r in sorted(rows)]
template = self._views.get(self.current_view, {})
columns = template.get("columns", [])
count = self.controller.export_selected(ids, columns, file_path)
self._op_log(f"导出选中行 | 数量={count} | 路径={file_path}")
self.status_label.setText(f"已导出 {count} 条")
# ============================================================
# 已发货处理（三层数据生命周期）
# ============================================================
def _shipped_archive(self) -> None:
"""标为已发货 → 归档到 sale_plans_shipped 表（Layer 1 → Layer 2）
将选中记录完整拷贝到已发货归档表，保留在数据库中可回溯查询。
"""
ids = self.get_selected_ids()
if not ids:
QMessageBox.warning(self, "提示", "请先选中记录")
return
confirm = QMessageBox.question(
self,
"确认归档",
f"确认将选中的 {len(ids)} 条记录标为已发货并归档到已发货表？\n\n"
"归档后将从当前表移除，可在已发货表中查看。",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if confirm != QMessageBox.StandardButton.Yes:
return
try:
count = self.controller.archive_to_shipped(ids)
self._op_log(f"已发货归档完成 | 数量={count}")
self.status_label.setText(f"已归档 {count} 条到已发货表")
self.search()
except Exception as e:
logger.error(f"已发货归档失败: {e}")
QMessageBox.critical(self, "错误", f"归档失败: {e}")
def _shipped_export_delete(self) -> None:
"""标为已发货 → 导出完整 Excel 并删除（Layer 1 → Layer 3）
将选中记录完整字段导出 Excel，归档到 shipped 表留痕，从主表删除。
Excel 文件名自动包含时间戳: 已发货导出_YYYYMMDD_HHMMSS.xlsx
"""
ids = self.get_selected_ids()
if not ids:
QMessageBox.warning(self, "提示", "请先选中记录")
return
confirm = QMessageBox.question(
self,
"确认导出并删除",
f"确认将选中的 {len(ids)} 条记录标为已发货、导出完整字段 Excel 并从数据库移除？\n\n"
"导出后将从数据库彻底删除，请确认已备份。",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if confirm != QMessageBox.StandardButton.Yes:
return
# 选择导出目录
from PySide6.QtWidgets import QFileDialog
output_dir = QFileDialog.getExistingDirectory(self, "选择 Excel 导出目录")
if not output_dir:
return
try:
result = self.controller.export_shipped_and_delete(ids, output_dir)
count = result["count"]
filepath = result["filepath"]
self._op_log(f"已发货导出删除完成 | 数量={count} | 路径={filepath}")
self.status_label.setText(f"已导出 {count} 条到 {filepath}")
self.search()
except Exception as e:
logger.error(f"已发货导出删除失败: {e}")
QMessageBox.critical(self, "错误", f"导出删除失败: {e}")
# ============================================================
# 快速筛选栏（新增区域）
# ============================================================
def _goto_print(self) -> None:
"""统一「打印合格证」入口：选中销售计划 → 去重编号 + 写 Certificate(待打印) → 跳打印页。
合并原 goto_print / auto_number / import_cert_list 三个入口：
委托 QueryController.generate_print_certificates（去重安全），再刷新并跳转打印页。
"""
selected_ids = self.get_selected_ids()
if not selected_ids:
QMessageBox.warning(self, "提示", "请先在表格中选中需要打印的记录")
return
# 不再要求 product_code 预先存在：控制器 generate_print_certificates 会对可编号记录
# （product_code 为空/占位符 且 needs_numbering!=False）自动编号并写 Certificate，
# 无生产令号/标记无需编号的记录在 create_certificates 内被过滤，仅提示不阻断。
# 这修复了「点击打印提示需先编号、却无自动编号按钮」的体验缺口。
numbered, cert_count = self.controller.generate_print_certificates(
selected_ids, prefix_override=self._number_prefix_override, print_status="待打印"
)
self.search()
prefix = self.controller.get_number_prefix(self._number_prefix_override)
if cert_count == 0:
QMessageBox.warning(
self,
"提示",
"选中记录未能生成合格证：可能均已生成，或缺少可编号信息"
"（生产令号 / needs_numbering）。可先点「🔢 自动编号」或「🔢 手动编号」再打印。",
)
self.status_label.setText("选中记录均未生成合格证（可能已存在或缺少可编号信息）")
return
self.status_label.setText(
f"已编号 {numbered} 条、生成 {cert_count} 份合格证（前缀={prefix}），跳转打印页面..."
)
self.navigate_to_print.emit(selected_ids)
def _print_certificates(self) -> None:
"""打印合格证：将已编号的记录写入 Certificate 表
对标 VBA print.vb 的完整打印流程：
1. 检查参数完整性 → 弹出参数补充对话框（如需）
2. 将 SalePlan 写入 Certificate 表
3. 参数补充结果应用到 Certificate
"""
rows = self._query_view.selected_rows()
if not rows:
return
unnumbered = [r for r in rows if not self._query_view.records[r].product_code]
if unnumbered:
QMessageBox.warning(self, "提示", f"有 {len(unnumbered)} 条未编号，请先编号")
return
ids = [self._query_view.records[r].id for r in rows]
# 生成 Certificate 记录（委托 Controller）
count = self.controller.create_certificates(ids)
if count == 0:
self.status_label.setText("未生成合格证（可能已存在）")
return
self._op_log(f"生成合格证完成 | 数量={count}")
# === 参数补充（对标 VBA InputBox）===
# View 负责 UI 对话框，Controller 负责业务逻辑
try:
cert = self.controller.get_latest_certificate(ids)
if cert:
from certflow.views.param_supplement_dialog import ParamSupplementDialog
supp_result = self.controller.supplement_certificate_params(
cert,
parent=self,
supplement_callback=ParamSupplementDialog.get_parameters,
)
if supp_result.get("supplemented"):
if supp_result.get("skipped"):
self._op_log("参数补充已跳过（留空）")
else:
self._op_log("参数补充完成")
except Exception as e:
logger.warning(f"参数补充跳过: {e}")
self.status_label.setText(f"已生成 {count} 份合格证")
def _get_record_by_id(self, record_id: int) -> Any:
"""根据 ID 获取记录"""
for record in self._query_view.records:
if record.id == record_id:
return record
return None
# ============================================================
# 复制（委托 Controller）
# ============================================================
def _copy_cert_info(self) -> None:
"""将合同信息（产品名称/型号/规格）复制到合格证打印字段
仅覆盖 cert_* 为空的字段，保留已有的手动编辑值。
"""
rows = self._query_view.selected_rows()
if not rows:
return
ids = [self._query_view.records[r].id for r in rows]
try:
updated = self.controller.copy_cert_info_to_plan(ids)
self._op_log(f"复制合同信息到合格证 | 数量={updated}")
except Exception as e:
self._op_log(f"复制合同信息失败 | 错误={e}")
return
self.search()
self.status_label.setText(f"已复制 {updated} 条")
def _edit_cert_fields(self) -> None:
"""订正合格证信息（仅 cert_* + certificate_remarks，不动合同字段）
蓝图 §4.2.3：打开仅含合格证字段的编辑对话框，保存走
``QueryController.update_cert_fields``，只 UPDATE cert_* 列。
"""
rows = self._query_view.selected_rows()
if not rows:
QMessageBox.information(self, "提示", "请先选择一条记录")
return
if len(rows) > 1:
QMessageBox.information(self, "提示", "请仅选择一条记录进行合格证信息订正")
return
r = next(iter(rows))
if r >= len(self._query_view.records):
return
record = self._query_view.records[r]
dlg = LookupRecordEditDialog(_CERT_EDIT_FIELDS, record, "✏️ 订正合格证信息", self)
if dlg.exec() != QDialog.DialogCode.Accepted:
return
try:
data = dlg.get_data()
except ValueError as e:
QMessageBox.warning(self, "校验失败", str(e))
return
try:
self.controller.update_cert_fields(record.id, data)
self._op_log(f"订正合格证信息 | id={record.id} | 字段={list(data.keys())}")
except Exception as e:
self._op_log(f"订正合格证信息失败 | id={record.id} | 错误={e}")
QMessageBox.warning(self, "订正失败", f"保存合格证信息失败：{e}")
return
self.search()
def _edit_certificate(self) -> None:
"""订正合格证记录（仅 Certificate 打印字段，不动编号/状态，对称步骤 4）
在「合格证清单」视图勾选单条 Certificate，打开 ``LookupRecordEditDialog``
订正 product_name/model/spec 等打印字段，保存走 ``update_certificate_fields``。
"""
rows = self._query_view.selected_rows()
if not rows:
QMessageBox.information(self, "提示", "请先选择一条合格证记录")
return
if len(rows) > 1:
QMessageBox.information(self, "提示", "请仅选择一条合格证记录进行订正")
return
r = next(iter(rows))
if r >= len(self._query_view.records):
return
record = self._query_view.records[r]
dlg = LookupRecordEditDialog(_CERT_RECORD_EDIT_FIELDS, record, "✏️ 订正合格证记录", self)
if dlg.exec() != QDialog.DialogCode.Accepted:
return
try:
data = dlg.get_data()
except ValueError as e:
QMessageBox.warning(self, "校验失败", str(e))
return
try:
self.controller.update_certificate_fields(record.id, data)
self._op_log(f"订正合格证记录 | cert_id={record.id} | 字段={list(data.keys())}")
except Exception as e:
self._op_log(f"订正合格证记录失败 | cert_id={record.id} | 错误={e}")
QMessageBox.warning(self, "订正失败", f"保存合格证记录失败：{e}")
return
self.search()
def _batch_fill_plan_no(self) -> None:
"""批量填充计划单号
弹出输入框让用户输入序号，自动补全为 YYYYMM_序号 格式，
仅填充 plan_no 为空的行。
"""
text, ok = QInputDialog.getText(self, "计划单号", "输入序号（自动补全为 YYYYMM_序号）：")
if not ok or not text.strip():
return
rows = self._query_view.selected_rows()
ids = [self._query_view.records[r].id for r in rows]
try:
updated = self.controller.fill_plan_no(ids, text.strip())
self._op_log(f"填充计划单号完成 | 数量={updated}")
except Exception as e:
self._op_log(f"填充计划单号失败 | 错误={e}")
return
self.search()
self.status_label.setText(f"已填充 {updated} 条计划单号")
def _set_production_status(self, status: str) -> None:
"""设置选中行的生产状态（供右键菜单子菜单调用）
Args:
status: 生产状态值，如 "待生产"、"生产中"、"已完成"、"已发货"、"部分发货"
"""
self._batch_update_field("production_status", status)
def _batch_update_production_status(self) -> None:
"""批量更新生产状态（供工具栏下拉框 + 更新按钮调用）"""
if self._status_combo is None:
QMessageBox.warning(self, "提示", "未找到状态下拉框")
return
target_status = self._status_combo.currentText()
if not target_status:
QMessageBox.warning(self, "提示", "请选择目标状态")
return
rows = self._query_view.selected_rows()
if not rows:
QMessageBox.warning(self, "提示", "请先选择要更新的行")
return
batch_config = cfg("query_page.batch_status_update", {})
if not self._confirm_status_update(rows, target_status, batch_config):
return
update_relations = batch_config.get("update_relations", [])
ids = [self._query_view.records[r].id for r in rows]
try:
updated = self.controller.batch_update_production_status(
ids, target_status, update_relations
)
self._op_log(f"批量更新生产状态完成 | 目标={target_status} | 数量={updated}")
except Exception as e:
self._op_log(f"批量更新生产状态失败 | 目标={target_status} | 错误={e}")
return
self.search()
self.status_label.setText(f"已更新 {updated} 条生产状态 → {target_status}")
def _confirm_status_update(self, rows: set, target_status: str, batch_config: dict) -> bool:
"""确认批量更新操作，返回 True 表示继续"""
if not batch_config.get("confirm_before_update", True):
return True
msg = f"将 {len(rows)} 行的生产状态更新为「{target_status}」"
if batch_config.get("show_affected_count", True):
msg += f"\n\n共选中 {len(rows)} 条记录"
msg += "\n\n确定吗？"
reply = QMessageBox.question(
self,
"确认更新",
msg,
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
return reply == QMessageBox.StandardButton.Yes
def _set_year_month(self) -> None:
"""将选中行的计划日期设为编号前缀年月
提取第一条选中记录的 plan_date，格式化为 YYMM 作为编号前缀。
若无有效日期则重置为当前年月。
"""
rows = self._query_view.selected_rows()
if not rows:
return
record = self._query_view.records[list(rows)[0]]
if record.plan_date and len(record.plan_date) >= 7:
self._number_prefix_override = record.plan_date[2:4] + record.plan_date[5:7]
self.status_label.setText(f"编号年月已设为 {self._number_prefix_override}")
else:
self._number_prefix_override = None
self.status_label.setText("已重置为当前年月")
def _manual_number(self) -> None:
"""手动编号：弹出对话框输入前缀/起始序号/每件数量/后缀"""
rows = self._query_view.selected_rows()
if not rows:
return
record = self._query_view.records[list(rows)[0]]
prefix, ok = QInputDialog.getText(
self,
"手动编号",
"输入前缀（如 V2506）：",
text=self._number_prefix_override or f"V{datetime.now().strftime('%y%m')}",
)
if not ok:
return
start, ok = QInputDialog.getInt(self, "手动编号", "起始序号：", 1, 1)
if not ok:
return
quantity, ok = QInputDialog.getInt(self, "手动编号", "每件数量：", record.quantity or 1, 1)
if not ok:
return
suffix, ok = QInputDialog.getText(self, "手动编号", "后缀（外购=Y，维修=XL）：")
if not ok:
return
ids = [self._query_view.records[r].id for r in rows]
updated = self.controller.manual_number(ids, prefix, start, quantity, suffix or "")
self.search()
self._op_log(f"手动编号完成 | 数量={updated}")
self.status_label.setText(f"手动编号 {updated} 条")
def _auto_number(self) -> None:
"""自动编号（工具栏「🔢 自动编号」）：为选中行按「本批台数」生成一批产品编号（G4）。
每个选中的销售计划弹窗输入「本批台数」（默认 = 剩余可发 ``remaining_quantity``，
上限 = 剩余可发），确认后委托 ``QueryController.generate_batch`` 调
``CertificateNumberService.generate_batch``：合同总数从 ``SalePlan.quantity`` 读取、
用 ``shipped_quantity`` 判剩余/满发、超额拒绝。替代原全量 ``auto_number`` 路径，
从根上消除 ``(4,1)`` 类参数歧义（见 OPERATOR_WORKFLOW §11.4.5）。
"""
rows = self._query_view.selected_rows()
if not rows:
return
records = [self._query_view.records[r] for r in rows if r < len(self._query_view.records)]
if not records:
return
done = 0
for rec in records:
remaining = int(getattr(rec, "remaining_quantity", 0) or 0)
default_qty = remaining if remaining > 0 else int(getattr(rec, "quantity", 1) or 1)
max_qty = max(remaining, 1)
qty, ok = QInputDialog.getInt(
self,
"本批台数（G4）",
f"订单：{getattr(rec, 'product_model', '') or ''} / "
f"{getattr(rec, 'customer', '') or ''}\n"
f"合同总数 {getattr(rec, 'quantity', 0)} | 已发 "
f"{getattr(rec, 'shipped_quantity', 0)} | 剩余 {remaining}\n\n"
f"请输入本批发货台数：",
default_qty,
1,
max_qty,
)
if not ok:
continue
try:
self.controller.generate_batch(rec.id, qty, ym=self._number_prefix_override)
done += 1
except ValueError as e:
QMessageBox.warning(
self, "本批编号失败", f"{getattr(rec, 'product_model', '')}: {e}"
)
except Exception as e: # noqa: BLE001
QMessageBox.critical(self, "错误", f"本批编号异常 | {e}")
self.search()
self._op_log(f"本批编号完成 | 计划数={done}/{len(records)}")
self.status_label.setText(f"本批编号完成 {done} 个计划（G4）")
def _copy_number(self) -> None:
"""复制编号：直接复制已有编号到选中行"""
rows = self._query_view.selected_rows()
if not rows:
return
record = self._query_view.records[list(rows)[0]]
source, ok = QInputDialog.getText(
self, "复制编号", "输入要复制的编号：", text=record.product_code or ""
)
if not ok or not source.strip():
return
ids = [self._query_view.records[r].id for r in rows]
updated = self.controller.copy_number(ids, source.strip())
self.search()
self._op_log(f"复制编号完成 | 数量={updated} | 编号={source.strip()}")
self.status_label.setText(f"复制编号 {updated} 条")
def _export_to_access(self) -> None:
"""导出到 Access 数据库"""
rows = self._query_view.selected_rows()
if not rows:
return
ids = [self._query_view.records[r].id for r in rows]
count = self.controller.export_to_access(ids)
self.status_label.setText(f"已导出 {count} 条到 Access")
def _mark_document_done(self, doc_type: str) -> None:
"""标记选中行的文档完成
Args:
doc_type: 文档类型 (certificate/nameplate/test_report/warranty/scan)
"""
rows = self._query_view.selected_rows()
if not rows:
QMessageBox.warning(self, "提示", "请先选中记录")
return
ids = [self._query_view.records[r].id for r in rows if r < len(self._query_view.records)]
result = self.controller.mark_document_done(ids, doc_type)
self.search()
self.status_label.setText(f"已标记 {result.get('success', 0)} 条")
[文档]
def show_statistics(self) -> None:
"""显示统计信息弹窗"""
conditions = self.get_filters()
stats = self.controller.get_statistics(conditions)
msg = f"""
📊 统计信息
===============
总记录数: {stats["total"]}
按生产状态:
{self._format_stats(stats["status_stats"])}
"""
QMessageBox.information(self, "统计信息", msg)
@staticmethod
def _format_stats(stats: dict[str, int]) -> str:
if not stats:
return " 无数据"
return "\n".join([f" {k}: {v} 条" for k, v in stats.items()])
# ============================================================
# 导出结果（委托 Controller）
# ============================================================
[文档]
def export_results(self) -> None:
"""导出当前查询结果为 Excel 文件（委托 Controller → Service → Handler）"""
file_path, _ = QFileDialog.getSaveFileName(self, "导出结果", "", "Excel文件 (*.xlsx)")
if not file_path:
return
try:
max_export = cfg("query_page.export.max_records", 10000)
conditions = self.get_filters()
result = self.controller.query(conditions, 1, max_export, "", False)
template = self._views.get(self.current_view, {})
columns = template.get("columns", [])
count = self.controller.export_selected(
[r.id for r in result["records"]], columns, file_path
)
QMessageBox.information(self, "导出成功", f"已导出 {count} 条记录")
except Exception as e:
logger.error(f"导出失败: {e}")
QMessageBox.critical(self, "导出失败", str(e))
# ============================================================
# 分页 ← 添加这部分
# ============================================================
[文档]
def prev_page(self) -> None:
"""方法说明。
Returns:
None
"""
if self.current_page > 1:
self.current_page -= 1
self.search(reset_page=False)
[文档]
def next_page(self) -> None:
"""方法说明。
Returns:
None
"""
self.current_page += 1
self.search(reset_page=False)
[文档]
def on_page_size_changed(self) -> None:
"""方法说明。
Returns:
None
"""
self.page_size = int(self.page_size_combo.currentText())
self.current_page = 1
self.search(reset_page=False)
# ============================================================
# 获取字段值（委托 Controller）
# ============================================================
def _load_combo_values(
self, combo: QComboBox, field_name: str, pending_value: str | None = None
) -> None:
"""为 exact 类型字段加载 QComboBox 下拉选项（异步）
通过后台线程从数据库查询该字段的所有已有值，完成后通过信号
_on_field_values_loaded 更新 QComboBox。
Args:
combo: 目标 QComboBox 实例（可编辑模式）
field_name: 数据库字段名
pending_value: 异步加载完成后需要设置的值（用于解决 _add_filter_row 中
setCurrentText 在空 QComboBox 上不生效的问题）
"""
# 存储 pending_value 供信号回调使用
self._field_value_pending: dict[str, str | None] = getattr(self, "_field_value_pending", {})
self._field_value_pending[field_name] = pending_value
# 记录所有等待该字段值的 QComboBox（可能有多个筛选行使用同一字段）
self._field_value_combos: dict[str, list[QComboBox]] = getattr(
self, "_field_value_combos", {}
)
self._field_value_combos.setdefault(field_name, []).append(combo)
# 通过 Controller 异步加载（后台线程，不阻塞 UI）
worker = self.controller.load_field_values_async(field_name, order_desc=True)
self._field_value_workers[field_name] = worker
def _on_field_values_loaded(self, field_name: str, values: list) -> None:
"""Controller 信号回调：字段可选值异步加载完成，更新所有等待的 QComboBox"""
logger.debug(f"加载 exact 下拉选项: {field_name}, 共 {len(values)} 个值")
pending_map: dict = getattr(self, "_field_value_pending", {})
pending_value = pending_map.pop(field_name, None)
combo_map: dict = getattr(self, "_field_value_combos", {})
combos = combo_map.pop(field_name, [])
for combo in combos:
try:
current = combo.currentText()
combo.blockSignals(True)
combo.clear()
combo.addItem("") # 空选项（不筛选）
for v in values:
combo.addItem(str(v))
# 优先使用 pending_value，其次恢复之前的值
target = pending_value if pending_value else current
if target:
idx = combo.findText(target)
if idx >= 0:
combo.setCurrentIndex(idx)
combo.blockSignals(False)
except RuntimeError:
# QComboBox 可能已被删除（筛选行被移除）
pass
# 清理 worker 引用
self._field_value_workers.pop(field_name, None)
def _on_field_values_load_failed(self, field_name: str, error_msg: str) -> None:
"""Controller 信号回调：字段可选值异步加载失败"""
logger.warning(f"加载下拉选项失败: {field_name}, {error_msg}")
pending_map: dict = getattr(self, "_field_value_pending", {})
pending_map.pop(field_name, None)
combo_map: dict = getattr(self, "_field_value_combos", {})
combos = combo_map.pop(field_name, [])
for combo in combos:
try:
combo.blockSignals(True)
combo.clear()
combo.addItem("加载失败")
combo.blockSignals(False)
except RuntimeError:
pass
self._field_value_workers.pop(field_name, None)
def _load_completer(self, completer: QCompleter, field_name: str) -> None:
"""为 exact 类型字段加载自动补全选项
从数据库中查询该字段的所有已有值，填充到 QCompleter 中，
用户在输入框中键入时自动弹出匹配选项。
Args:
completer: 目标 QCompleter 实例
field_name: 数据库字段名
"""
try:
values = self.controller.get_field_values(field_name, order_desc=True)
model = QStringListModel()
model.setStringList([str(v) for v in values])
completer.setModel(model)
except Exception as e:
logger.warning(f"加载下拉选项失败: {field_name}, {e}")
def _on_filter_value_changed(self) -> None:
"""筛选条件变化时触发搜索（防抖）
通过 QTimer 单次触发实现防抖，避免用户快速输入时频繁查询。
防抖间隔由 ui.yaml 的 query_page.search_debounce_ms 配置（默认 300ms）。
"""
if self._search_timer is None:
self._search_timer = QTimer()
self._search_timer.setSingleShot(True)
self._search_timer.timeout.connect(self._do_search)
else:
self._search_timer.stop()
self._search_timer.start(self._debounce_ms)
# ============================================================
# 筛选条件获取
# ============================================================
def _is_dark_theme(self) -> bool:
"""检测当前是否为深色主题"""
app = QApplication.instance()
if hasattr(app, "theme_manager"):
return app.theme_manager.current_theme == "dark"
return False
def _get_date_button_style(self, is_dark: bool) -> str:
"""获取日期按钮样式（主题感知）"""
# 添加 Emoji 支持的字体
font_family = "Segoe UI Emoji, Apple Color Emoji, Noto Color Emoji, sans-serif"
if is_dark:
return f"""
QPushButton {{
background-color: transparent;
border: 1px solid #45475a;
border-radius: 4px;
color: #cdd6f4;
font-size: 14px;
font-family: {font_family};
}}
QPushButton:hover {{
background-color: #313244;
}}
"""
return f"""
QPushButton {{
background-color: transparent;
border: 1px solid #ced4da;
border-radius: 4px;
color: #2c3e50;
font-size: 14px;
font-family: {font_family};
}}
QPushButton:hover {{
background-color: #e9ecef;
}}
"""
def _get_filter_button_style(self, is_dark: bool) -> str:
"""获取快速筛选按钮样式"""
if is_dark:
return """
QPushButton {
background-color: #313244;
border: 1px solid #45475a;
border-radius: 3px;
padding: 2px 8px;
font-size: 12px;
color: #cdd6f4;
}
QPushButton:hover {
background-color: #45475a;
border-color: #89b4fa;
}
QPushButton:checked {
background-color: #89b4fa;
border-color: #74a8f7;
color: #1e1e2e;
font-weight: bold;
}
"""
return """
QPushButton {
background-color: #e8f0fe;
border: 1px solid #c4d7f2;
border-radius: 3px;
padding: 2px 8px;
font-size: 12px;
color: #1a1a1a;
}
QPushButton:hover {
background-color: #d2e3fc;
border-color: #8ab4f8;
}
QPushButton:checked {
background-color: #4d8bf5;
border-color: #2c7be5;
color: #ffffff;
font-weight: bold;
}
"""
def _get_toggle_reverse_button_style(self, is_dark: bool) -> str:
"""获取三态按钮反向状态样式（橙色系，区别于正向蓝色）"""
if is_dark:
return """
QPushButton {
background-color: #f5a623;
border: 1px solid #e09520;
border-radius: 3px;
padding: 2px 8px;
font-size: 12px;
color: #1e1e2e;
font-weight: bold;
}
QPushButton:hover {
background-color: #f7b84e;
border-color: #e09520;
}
QPushButton:checked {
background-color: #f5a623;
border-color: #d4891a;
color: #1e1e2e;
font-weight: bold;
}
"""
return """
QPushButton {
background-color: #f5a623;
border: 1px solid #e09520;
border-radius: 3px;
padding: 2px 8px;
font-size: 12px;
color: #ffffff;
font-weight: bold;
}
QPushButton:hover {
background-color: #f7b84e;
border-color: #d4891a;
}
QPushButton:checked {
background-color: #f5a623;
border-color: #d4891a;
color: #ffffff;
font-weight: bold;
}
"""
def _get_group_button_style(self, is_dark: bool) -> str:
"""获取筛选组按钮样式"""
if is_dark:
return """
QPushButton {
background-color: #313244;
border: 1px solid #45475a;
border-radius: 3px;
padding: 2px 8px;
font-size: 12px;
color: #cdd6f4;
}
QPushButton:hover {
background-color: #45475a;
border-color: #89b4fa;
}
"""
return """
QPushButton {
background-color: #fef7e0;
border: 1px solid #fce8b2;
border-radius: 3px;
padding: 2px 8px;
font-size: 12px;
}
QPushButton:hover { background-color: #fef0c8; }
"""
def _get_clear_button_style(self, is_dark: bool) -> str:
"""获取清除按钮样式"""
if is_dark:
return """
QPushButton {
background-color: #45475a;
border: 1px solid #f38ba8;
border-radius: 3px;
padding: 2px 8px;
font-size: 12px;
color: #f38ba8;
}
QPushButton:hover {
background-color: #585b70;
border-color: #f38ba8;
}
"""
return """
QPushButton {
background-color: #fce8e6;
border: 1px solid #f5c6cb;
border-radius: 3px;
padding: 2px 8px;
font-size: 12px;
}
QPushButton:hover { background-color: #f8d7da; }
"""
def _get_menu_style(self, is_dark: bool) -> str:
"""获取菜单样式"""
if is_dark:
return """
QMenu {
background-color: #313244;
color: #cdd6f4;
border: 1px solid #45475a;
}
QMenu::item:selected {
background-color: #89b4fa;
color: #1e1e2e;
}
"""
return ""
def _op_log(self, msg: str) -> None:
"""根据配置级别输出操作日志
off: 不输出
file: 仅写文件（DEBUG 级别，文件配置为 DEBUG 才会落盘）
console: INFO 级别，同时输出到控制台和文件
"""
if self._op_log_level == "off":
return
if self._op_log_level == "console":
logger.info(msg)
else:
logger.debug(msg)
def _on_operation_completed(self, msg: str) -> None:
"""操作完成回调"""
self.status_label.setText(msg)
# ============================================================
# 以下方法保持不变（UI 构建、筛选行、快速筛选栏等）
# 由于篇幅限制，此处省略，实际应保留所有 UI 相关方法
# ============================================================
# ... 此处保留以下方法（与原代码相同）:
def _collect_current_conditions(self) -> list[dict]:
"""收集当前所有筛选行的条件（委托 FilterBar，列表形式 {field, operator, value}）。
将当前界面上用户设置的所有筛选条件收集起来，用于视图切换时保留。
"""
return self.filter_bar.collect_conditions()
# 旧筛选行辅助方法（_connect_filter_row_signals / _set_filter_row_initial_values /
# _ensure_combo_value）已迁移至 FilterRowWidget（§1.7 落地组件），由 FilterBar 统一管理。
# 纯展示字段（非数据库列，不可用于筛选/排序）
_DISPLAY_ONLY_FIELDS = frozenset({"progress_display", "ready_display"})
def _get_view_columns(self) -> list[dict]:
"""获取当前视图的列定义
Returns:
当前视图的列定义列表，每项包含 label 和 field 等属性。
排除 id 字段（主键）和纯展示字段（不可筛选/排序）。
"""
template = self._views.get(self.current_view, {})
columns = template.get("columns", [])
return [
col
for col in columns
if col.get("field") != "id" and col.get("field") not in self._DISPLAY_ONLY_FIELDS
]
_PROGRESS_BOOL_FIELDS = frozenset(
{
"certificate_done",
"nameplate_done",
"test_report_done",
"warranty_done",
"cert_scan_done",
}
)
@staticmethod
def _to_bool(value: object) -> bool:
"""将各种类型值安全转换为 Python bool。
兼容 SQLite 中可能存为字符串 'False'/'True' 或整数 0/1 的情况。
"""
if isinstance(value, bool):
return value
if isinstance(value, int):
return value != 0
if isinstance(value, str):
return value.strip().lower() in ("true", "1")
return bool(value)