"""报告生成视图模块
提供报告生成的界面组件：模板/语言选择、数据选择（来自 certflow.db 的销售计划）、
报告生成（后台线程，避免界面阻塞）与结果展示。
生成链路：``ReportView`` → ``ReportController.generate_report`` →
``report_fill_service.generate``（填充 12 模板之一）→ ``ReportOutputService.emit``
（双存根 xlsx+pdf + 文件夹 + 报告目录）。
"""
from __future__ import annotations
from datetime import date
from typing import Any
from loguru import logger
from PySide6.QtCore import Qt, QThread, QTimer, Signal
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QDialog,
QGroupBox,
QHBoxLayout,
QHeaderView,
QLabel,
QMessageBox,
QPushButton,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from certflow.config.settings import (
cfg,
get_report_page_options,
get_report_save_path,
set_user_preference,
should_export_report_pdf,
)
from certflow.controllers.report_controller import ReportController
from certflow.widgets.filter_bar import FilterBar
from certflow.widgets.quick_search_preset import QuickSearchPresetMixin
# 全库检索上限（远大于旧 500 上限；超出视为异常大数据，仅用于兜底避免一次性渲染卡死）
_MAX_FETCH = 100_000
# 报告筛选字段（复用查询引擎的统一条件契约；操作符类型由 ui.yaml 的 field_operator_map 决定）
_REPORT_FILTER_COLUMNS = [
{"label": "订货单位", "field": "customer"},
{"label": "销售订单号", "field": "sales_order_no"},
{"label": "生产令号", "field": "production_order_no"},
{"label": "计划单号", "field": "plan_no"},
{"label": "产品型号", "field": "product_model"},
{"label": "产品名称", "field": "product_name"},
{"label": "项目名称", "field": "project_name"},
{"label": "计划日期", "field": "plan_date"},
{"label": "是否已编号", "field": "product_code"},
]
# 操作符 / 字段类型映射：直接复用查询页同一份 ui.yaml 配置
_FIELD_OP_MAP = cfg("query_page.field_operator_map", {}) or {}
_OPERATORS = cfg("query_page.operators", {}) or {}
# 快查预设（按订单 / 按销售订单号 / 按生产令号）：与查询页同源 ui.yaml 配置
_PRESETS = cfg("query_page.presets", []) or []
# 无值操作符（scope 展示时只显示操作符，不显示空值）
_NO_VALUE_OPS = ("is_null", "is_not_null")
[文档]
class ReportWorker(QThread):
"""报告生成工作线程（后台执行，避免阻塞主界面）。"""
finished = Signal(dict)
progress = Signal(int, str)
def __init__(
self,
controller: ReportController,
sale_plan_ids: list[int],
report_kind: str,
language: str,
records_per_page: int,
export_pdf: bool | None = None,
) -> None:
super().__init__()
self.controller = controller
self.sale_plan_ids = sale_plan_ids
self.report_kind = report_kind
self.language = language
self.records_per_page = records_per_page
self.export_pdf = export_pdf
[文档]
def run(self) -> None:
result = self.controller.generate_report(
sale_plan_ids=self.sale_plan_ids,
report_kind=self.report_kind,
language=self.language,
records_per_page=self.records_per_page,
export_pdf=self.export_pdf,
)
self.finished.emit(result)
[文档]
class ReportView(QuickSearchPresetMixin, QWidget):
"""报告生成视图。
提供报告类型/语言/每页记录数选择、可勾选的数据列表（来自数据库销售计划）、
生成按钮与结果提示。
"""
_KIND_MAP = {"质保书": "quality", "试压报告": "pressure", "材质报告": "material"}
def __init__(
self, controller: ReportController | None = None, parent: QWidget | None = None
) -> None:
super().__init__(parent)
self.controller = controller or ReportController()
self._last_folder: str | None = None
self._id_to_pno: dict[int, str] = {}
self._qs_service = self.controller
self._search_timer = QTimer(self)
self._search_timer.setSingleShot(True)
self._search_timer.timeout.connect(self.on_quick_search)
self.setup_ui()
self.load_data()
# ------------------------------------------------------------------ #
# 配置驱动：每页行数候选（按报告类型从 config/report.yaml 读取）
# ------------------------------------------------------------------ #
[文档]
def populate_page_options(self, report_kind: str) -> None:
"""按报告类型填充「每页行数」候选（配置可选，而非数字递增）。"""
self.page_option_combo.blockSignals(True)
self.page_option_combo.clear()
opts = get_report_page_options().get(report_kind, [6])
for o in opts:
self.page_option_combo.addItem(f"{o} 行", int(o))
# 默认选中首个候选
self.page_option_combo.setCurrentIndex(0)
self.page_option_combo.blockSignals(False)
# ------------------------------------------------------------------ #
# UI 构建
# ------------------------------------------------------------------ #
[文档]
def setup_ui(self) -> None:
layout = QVBoxLayout(self)
template_group = self.create_template_group()
layout.addWidget(template_group)
data_group = self.create_data_group()
layout.addWidget(data_group)
preview_group = self.create_preview_group()
layout.addWidget(preview_group)
button_bar = self.create_button_bar()
layout.addLayout(button_bar)
[文档]
def create_template_group(self) -> QGroupBox:
# 顶部选项单行紧凑排布：选项文字均较短，横向一行即可，显著降低高度
group = QGroupBox("报告模板")
layout = QHBoxLayout(group)
layout.addWidget(QLabel("类型:"))
self.report_type_combo = QComboBox()
self.report_type_combo.addItems(["质保书", "试压报告", "材质报告"])
self.report_type_combo.currentTextChanged.connect(self.on_report_type_changed)
layout.addWidget(self.report_type_combo)
layout.addWidget(QLabel("语言:"))
self.lang_combo = QComboBox()
self.lang_combo.addItems(["中文", "English"])
layout.addWidget(self.lang_combo)
layout.addWidget(QLabel("每页行数:"))
self.page_option_combo = QComboBox()
layout.addWidget(self.page_option_combo)
self.export_pdf_check = QCheckBox("同时导出 PDF")
self.export_pdf_check.setChecked(should_export_report_pdf())
self.export_pdf_check.setToolTip(
"勾选：xlsx 生成后用 Excel 派生 PDF（需 Windows+Excel）；\n"
"不勾（默认）：仅生成 xlsx 可手改档，xlsx 异常则不导 PDF"
)
self.export_pdf_check.toggled.connect(
lambda checked: set_user_preference("report_export_pdf", checked)
)
layout.addWidget(self.export_pdf_check)
layout.addStretch()
# 依据当前报告类型填充「每页行数」候选
self.populate_page_options(
self._KIND_MAP.get(self.report_type_combo.currentText(), "quality")
)
return group
[文档]
def create_data_group(self) -> QGroupBox:
group = QGroupBox("数据选择（来自数据库销售计划，支持全库快速查询）")
layout = QVBoxLayout(group)
# ---- 快速筛选按钮（一键常用范围，免手动搭条件）----
quick_layout = QHBoxLayout()
quick_layout.addWidget(QLabel("快速筛选:"))
self.qf_month_btn = QPushButton("本月计划")
self.qf_month_btn.clicked.connect(
lambda: self._apply_quick_filter([("plan_date", "gte", self._month_start())])
)
quick_layout.addWidget(self.qf_month_btn)
self.qf_year_btn = QPushButton("本年计划")
self.qf_year_btn.clicked.connect(
lambda: self._apply_quick_filter([("plan_date", "gte", self._year_start())])
)
quick_layout.addWidget(self.qf_year_btn)
self.qf_numbered_btn = QPushButton("已编号")
self.qf_numbered_btn.clicked.connect(
lambda: self._apply_quick_filter([("product_code", "is_not_null", None)])
)
quick_layout.addWidget(self.qf_numbered_btn)
self.qf_unnumbered_btn = QPushButton("未编号")
self.qf_unnumbered_btn.clicked.connect(
lambda: self._apply_quick_filter([("product_code", "is_null", None)])
)
quick_layout.addWidget(self.qf_unnumbered_btn)
self.qf_clear_btn = QPushButton("清空条件")
self.qf_clear_btn.clicked.connect(self._clear_filters)
quick_layout.addWidget(self.qf_clear_btn)
quick_layout.addStretch()
layout.addLayout(quick_layout)
# ---- 多条件筛选条（复用查询子模块 FilterBar，统一条件契约）----
self.filter_bar = FilterBar(
columns=_REPORT_FILTER_COLUMNS,
field_op_map=_FIELD_OP_MAP,
operators=_OPERATORS,
load_combo_values_fn=self._load_combo_values,
grid_cols=2,
# 与查询页一致的「按订单快速查询」：快查框 + 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(_PRESETS),
presets=_PRESETS,
preset_changed_callback=self.qsp_on_preset_changed,
)
self.filter_bar.conditions_changed.connect(self._on_filter_changed_debounced)
# 初始化一个空筛选行，并抑制其首发的 conditions_changed（避免启动即触发搜索）
self.filter_bar.blockSignals(True)
self.filter_bar.add_row()
self.filter_bar.blockSignals(False)
layout.addWidget(self.filter_bar)
# 初始化「按订单快速查询」公共逻辑（preset 下拉 + 快查框），默认选中首个预设
self.setup_quick_search_preset(_PRESETS, _REPORT_FILTER_COLUMNS)
# ---- 查询 / 加载全部 ----
btn_row = QHBoxLayout()
self.search_btn = QPushButton("查询")
self.search_btn.clicked.connect(self.on_quick_search)
btn_row.addWidget(self.search_btn)
self.load_all_btn = QPushButton("加载全部")
self.load_all_btn.clicked.connect(self.load_all)
btn_row.addWidget(self.load_all_btn)
btn_row.addStretch()
layout.addLayout(btn_row)
self.data_table = QTableWidget()
self.data_table.setColumnCount(5)
self.data_table.setHorizontalHeaderLabels(
["选择", "产品型号", "产品编号", "生产令号", "订货单位"]
)
self.data_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
self.data_table.setColumnWidth(0, 50)
layout.addWidget(self.data_table)
btn_layout = QHBoxLayout()
self.select_all_btn = QPushButton("全选")
self.select_all_btn.clicked.connect(self.select_all)
btn_layout.addWidget(self.select_all_btn)
self.clear_all_btn = QPushButton("清空")
self.clear_all_btn.clicked.connect(self.clear_all)
btn_layout.addWidget(self.clear_all_btn)
self.refresh_btn = QPushButton("刷新数据")
self.refresh_btn.clicked.connect(self.load_data)
btn_layout.addWidget(self.refresh_btn)
self.preview_bom_btn = QPushButton("按生产令号预览 BOM 材质")
self.preview_bom_btn.setToolTip(
"生成前先按选中行的生产令号预览 BOM 材质归类结果（命中主材位 + 待核对明细），"
"复用查询中心同源 BOM 查询，只读不修改数据"
)
self.preview_bom_btn.clicked.connect(self.on_preview_bom)
btn_layout.addWidget(self.preview_bom_btn)
btn_layout.addStretch()
layout.addLayout(btn_layout)
return group
[文档]
def create_preview_group(self) -> QGroupBox:
group = QGroupBox("报告信息")
layout = QVBoxLayout(group)
self.info_label = QLabel("请选择模板和数据后点击「生成报告」")
self.info_label.setStyleSheet(
"padding: 20px; background-color: #f8f9fa; border-radius: 5px;"
)
layout.addWidget(self.info_label)
return group
# ------------------------------------------------------------------ #
# 数据加载（来自 certflow.db）
# ------------------------------------------------------------------ #
[文档]
def load_data(self) -> None:
"""从数据库加载销售计划到选择表格（全库，无 500 上限）。"""
self.load_all()
[文档]
def load_all(self) -> None:
"""加载全部销售计划（全库检索，不限制 500 条）。"""
self._fetch_and_fill({})
[文档]
def on_quick_search(self, or_filters: Any = None) -> None:
"""取 FilterBar 统一条件契约，对全库做检索并刷新勾选表格。
Args:
or_filters: 可选 OR 条件列表（快查模糊输入无分隔符时传入），
与筛选行条件以 OR 组合。
"""
conditions = self.filter_bar.get_conditions()
if not conditions and not or_filters:
self.update_info("请设置至少一条有效筛选条件后点击「查询」（或点「加载全部」）")
return
self._fetch_and_fill(conditions, or_filters=or_filters)
# ============================================================
# 快速筛选 / 多条件筛选辅助
# ============================================================
def _month_start(self) -> str:
"""本月 1 号（YYYY-MM-DD）。"""
return date.today().replace(day=1).strftime("%Y-%m-%d")
def _year_start(self) -> str:
"""本年 1 月 1 号（YYYY-MM-DD）。"""
return date.today().replace(month=1, day=1).strftime("%Y-%m-%d")
def _apply_quick_filter(self, rows: list[tuple[str, str, Any]]) -> None:
"""用一组 (field, operator, value) 替换当前筛选行并立即查询。"""
self.filter_bar.clear_rows()
for field, operator, value in rows:
self.filter_bar.add_row(field_name=field, operator=operator, value=value)
self.on_quick_search()
def _clear_filters(self) -> None:
"""清空所有筛选条件并回到全库。"""
self.filter_bar.clear_rows()
self.filter_bar.add_row()
self.load_all()
def _on_filter_changed_debounced(self) -> None:
"""筛选条件变化：防抖触发查询（避免拖拽/连点频繁检索）。"""
self._search_timer.start(300)
def _load_combo_values(
self, combo: Any, field_name: str, pending_value: str | None = None
) -> None:
"""exact 字段枚举下拉加载（兜底适配器：报告筛选字段多为 text/numeric，
通常不会触发；此处用 QueryService 去重值同步填充，保持 FilterBar 兼容。"""
try:
values = self._qs_service.get_distinct_field_combos([field_name])
except Exception as e: # 加载失败不阻塞，仅记录
logger.warning(f"报告筛选下拉加载失败: {field_name}, {e}")
return
combo.blockSignals(True)
combo.clear()
combo.addItem("")
for v in values:
combo.addItem(str(v))
if pending_value:
idx = combo.findText(str(pending_value))
if idx >= 0:
combo.setCurrentIndex(idx)
combo.blockSignals(False)
# ============================================================
# 按订单快速查询（preset 下拉 + 快查框）
# 逻辑已抽到 certflow.widgets.quick_search_preset.QuickSearchPresetMixin
# （含「严格字段数 / 空段保留」修复），本视图仅通过 setup_quick_search_preset
# 初始化并把 FilterBar 的回调指向 qsp_on_* 方法。
# ============================================================
def _fetch_and_fill(self, conditions: dict, or_filters: Any = None) -> None:
"""统一取数入口：复用查询引擎对全库检索，结果填入勾选表格。"""
try:
result = self._qs_service.query(
conditions=conditions,
page=1,
page_size=_MAX_FETCH,
or_filters=or_filters,
)
rows = result.get("records") or []
except Exception as e: # 取数失败不崩溃，仅提示
self.update_info(f"加载数据失败: {e}")
return
data = [
{
"id": r.id,
"product_model": r.product_model or "",
"product_code": r.product_code or "",
"production_order_no": r.production_order_no or "",
"customer": r.customer or "",
}
for r in rows
]
self.update_data_table(data)
if not conditions:
scope = "全库"
else:
op_label = {**(_OPERATORS.get("text", {})), **(_OPERATORS.get("exact", {}))}
parts = []
for k, v in conditions.items():
op = v.get("operator", "contains")
val = v.get("value")
if op in _NO_VALUE_OPS or val is None:
parts.append(f"{k} {op_label.get(op, op)}")
elif isinstance(val, list):
parts.append(f"{k}={val[0]}~{val[1]}")
else:
parts.append(f"{k}={val}")
scope = " AND ".join(parts)
self.update_info(f"已加载 {len(data)} 条销售计划（{scope}），请勾选需要生成报告的数据")
[文档]
def on_report_type_changed(self, report_type: str) -> None:
"""报告类型切换时刷新「每页行数」候选（配置驱动）。"""
kind = self._KIND_MAP.get(report_type, "quality")
self.populate_page_options(kind)
[文档]
def select_all(self) -> None:
for row in range(self.data_table.rowCount()):
item = self.data_table.item(row, 0)
if item:
item.setCheckState(Qt.CheckState.Checked)
[文档]
def clear_all(self) -> None:
for row in range(self.data_table.rowCount()):
item = self.data_table.item(row, 0)
if item:
item.setCheckState(Qt.CheckState.Unchecked)
[文档]
def get_selected_ids(self) -> list[int]:
selected_ids: list[int] = []
for row in range(self.data_table.rowCount()):
checkbox = self.data_table.item(row, 0)
if checkbox and checkbox.checkState() == Qt.CheckState.Checked:
id_value = checkbox.data(Qt.ItemDataRole.UserRole)
if id_value:
selected_ids.append(int(id_value))
return selected_ids
[文档]
def update_data_table(self, data: list[dict[str, Any]]) -> None:
self._id_to_pno = {item.get("id", 0): item.get("production_order_no", "") for item in data}
self.data_table.setRowCount(len(data))
for row, item in enumerate(data):
checkbox = QTableWidgetItem()
checkbox.setCheckState(Qt.CheckState.Unchecked)
checkbox.setFlags(Qt.ItemFlag.ItemIsUserCheckable | Qt.ItemFlag.ItemIsEnabled)
self.data_table.setItem(row, 0, checkbox)
checkbox.setData(Qt.ItemDataRole.UserRole, item.get("id", 0))
self.data_table.setItem(row, 1, QTableWidgetItem(item.get("product_model", "")))
self.data_table.setItem(row, 2, QTableWidgetItem(item.get("product_code", "")))
self.data_table.setItem(row, 3, QTableWidgetItem(item.get("production_order_no", "")))
self.data_table.setItem(row, 4, QTableWidgetItem(item.get("customer", "")))
[文档]
def update_info(self, text: str) -> None:
self.info_label.setText(text)
# ------------------------------------------------------------------ #
# 生成
# ------------------------------------------------------------------ #
[文档]
def on_generate(self) -> None:
ids = self.get_selected_ids()
if not ids:
QMessageBox.warning(self, "提示", "请先勾选至少一条数据")
return
report_kind = self._KIND_MAP.get(self.report_type_combo.currentText(), "quality")
language = "en" if self.lang_combo.currentText() == "English" else "zh"
rpp = int(self.page_option_combo.currentData() or 6)
export_pdf = self.export_pdf_check.isChecked()
self.update_info(
f"正在生成 {len(ids)} 条数据的报告…（PDF: {'是' if export_pdf else '否'}）"
)
self.generate_btn.setEnabled(False)
self.worker = ReportWorker(self.controller, ids, report_kind, language, rpp, export_pdf)
self.worker.finished.connect(self.show_result)
self.worker.start()
[文档]
def show_result(self, result: dict[str, Any]) -> None:
self.generate_btn.setEnabled(True)
if result.get("success"):
msg = "✅ 报告生成成功!\n\n"
msg += f"文件夹: {result.get('folder', '')}\n"
msg += f"xlsx: {result.get('xlsx', '')}\n"
if result.get("pdf"):
msg += f"pdf: {result.get('pdf')}\n"
else:
msg += f"pdf: 未生成（{result.get('pdf_note', 'xlsx 已生成')}）\n"
msg += f"已嵌章: {'是' if result.get('sealed') else '否'}"
# 阶段 C：BOM 材质命中/待核对计数提示（C4.3）
if "bom_matched" in result:
msg += (
f"\nBOM 材质：命中 {result.get('bom_matched', 0)} 项 / "
f"待核对 {result.get('bom_detail_count', 0)} 项"
)
self.update_info(msg)
self._last_folder = result.get("folder") or None
QMessageBox.information(self, "生成成功", msg)
else:
QMessageBox.critical(
self, "生成失败", f"报告生成失败: {result.get('error', '未知错误')}"
)
self.update_info(f"生成失败: {result.get('error', '未知错误')}")
[文档]
def on_open_folder(self) -> None:
folder = self._last_folder
if not folder:
# 尚未生成报告：回退到配置的报告根目录
folder = get_report_save_path() or cfg("paths.report_archive.root", "output/项目资料")
if folder:
ok = self.controller.open_report_folder(folder)
if not ok:
QMessageBox.warning(self, "提示", f"无法打开目录（可能不存在）:\n{folder}")
# ------------------------------------------------------------------ #
# 阶段 C（C5）：按生产令号预览 BOM 材质归类（只读，复用查询中心同源查询）
# ------------------------------------------------------------------ #
[文档]
def on_preview_bom(self) -> None:
"""对首个选中行取生产令号，预览 BOM 材质归类（命中主材位 + 待核对明细）。"""
ids = self.get_selected_ids()
if not ids:
QMessageBox.information(self, "提示", "请先勾选至少一条数据（以取其生产令号）")
return
pno = self._id_to_pno.get(ids[0])
if not pno:
QMessageBox.information(self, "提示", "该行无生产令号，无法预览 BOM 材质")
return
try:
result = self.controller.enrich_bom(pno)
except Exception as e: # 查询失败给出提示，不崩溃
logger.exception(f"BOM 材质预览失败: {e}")
QMessageBox.warning(self, "预览失败", f"BOM 材质查询失败: {e}")
return
self._show_bom_preview(pno, result)
def _show_bom_preview(self, pno: str, result: Any) -> None:
"""以对话框展示 BOM 材质归类结果（命中主材位 + 待核对明细）。"""
dlg = QDialog(self)
dlg.setWindowTitle(f"BOM 材质预览 - {pno}")
dlg.resize(580, 440)
layout = QVBoxLayout(dlg)
summary = QLabel(
f"生产令号 {pno}：命中 {result.matched} 项主材位，待核对 {result.detail_count} 项"
)
summary.setStyleSheet("font-weight: bold; padding: 4px;")
layout.addWidget(summary)
table = QTableWidget()
table.setColumnCount(4)
table.setHorizontalHeaderLabels(["零件名称", "材料牌号", "归类/状态", "备注"])
rows: list[tuple[str, str, str, str]] = []
for _slot, mp in result.classified.items():
rows.append((mp.零件名称, mp.材料, "主材位命中", ""))
for d in result.detail:
rows.append(
(
d.get("part_name", ""),
d.get("material_grade", ""),
d.get("reason", ""),
d.get("note", ""),
)
)
table.setRowCount(len(rows))
for r, (a, b, c, dd) in enumerate(rows):
table.setItem(r, 0, QTableWidgetItem(str(a)))
table.setItem(r, 1, QTableWidgetItem(str(b)))
table.setItem(r, 2, QTableWidgetItem(str(c)))
table.setItem(r, 3, QTableWidgetItem(str(dd)))
table.horizontalHeader().setStretchLastSection(True)
table.resizeColumnsToContents()
layout.addWidget(table)
close_btn = QPushButton("关闭")
close_btn.clicked.connect(dlg.accept)
layout.addWidget(close_btn)
dlg.exec()