certflow.views.history_view 源代码

# src/certflow/views/history_view.py
"""打印历史视图 - 显示打印日志记录

支持:日期区间筛选(合格证打印时间/计划日期)、产品型号模糊搜索、
自由文本快速筛选,以及将当前结果导出为 UTF-8 BOM 的 CSV(Excel 中文不乱码)。
"""

from __future__ import annotations

from datetime import datetime
from typing import Any

from PySide6.QtCore import QDate
from PySide6.QtWidgets import (
    QFileDialog,
    QHBoxLayout,
    QLineEdit,
    QMessageBox,
    QPushButton,
    QTableWidgetItem,
    QVBoxLayout,
    QWidget,
)

from certflow.utils.logger import logger
from certflow.widgets.enhanced_table import EnhancedTable
from certflow.widgets.filter_bar import FilterBar

# 历史视图筛选字段定义(复用通用 FilterBar,数据源无关)
_HISTORY_FILTER_COLUMNS = [
    {"label": "打印时间", "field": "print_date"},
    {"label": "产品型号", "field": "model"},
]
_HISTORY_FIELD_OP_MAP = {
    # 日期选择器由字段名 _date 后缀触发;默认区间
    "print_date": "date",
    # 默认包含(模糊)
    "model": "text",
}
_HISTORY_OPERATORS = {
    "date": {
        "between": "区间",
        "eq": "等于",
        "is_null": "为空",
        "is_not_null": "非空",
    },
    "text": {
        "contains": "包含",
        "eq": "等于",
        "is_null": "为空",
        "is_not_null": "非空",
    },
}


[文档] class HistoryView(QWidget): """打印历史视图""" def __init__(self, session: Any, db_manager: Any = None, parent: Any = None) -> None: """初始化HistoryView。 Args: session: SQLAlchemy 会话对象 db_manager: 数据库管理器实例 parent: 父控件(可选) """ super().__init__(parent) self.db_manager = db_manager self.session = session self.setup_ui() self._load_history()
[文档] def setup_ui(self) -> None: """构建界面""" layout = QVBoxLayout(self) # 筛选条(统一复用通用 FilterBar:打印时间区间 + 产品型号,数据源无关) self.filter_bar = FilterBar( parent=self, columns=_HISTORY_FILTER_COLUMNS, field_op_map=_HISTORY_FIELD_OP_MAP, operators=_HISTORY_OPERATORS, grid_cols=2, ) layout.addWidget(self.filter_bar) self._init_default_filters() # 操作按钮行 btn_row = QHBoxLayout() self.query_btn = QPushButton("查询") self.query_btn.clicked.connect(self._apply_filters) btn_row.addWidget(self.query_btn) self.export_btn = QPushButton("导出CSV") self.export_btn.clicked.connect(self._export_csv) btn_row.addWidget(self.export_btn) btn_row.addStretch() self.refresh_btn = QPushButton("刷新") self.refresh_btn.clicked.connect(self._load_history) btn_row.addWidget(self.refresh_btn) layout.addLayout(btn_row) # 自由文本快速筛选(客户端隐藏行) quick_row = QHBoxLayout() self.filter_input = QLineEdit() self.filter_input.setPlaceholderText("表格内快速筛选...") self.filter_input.textChanged.connect(self._filter_history) quick_row.addWidget(self.filter_input) layout.addLayout(quick_row) # 表格 self.history_table = EnhancedTable() layout.addWidget(self.history_table)
def _build_filters(self) -> dict: """从 FilterBar 统一条件契约构建查询过滤条件。 把 ``FilterBar.get_conditions()`` 的 ``{field:{value,operator}}`` 转换为 ``HistoryController.query_logs`` 期望的 ``{date_from, date_to, model_keyword}``。 Returns: dict: 含 date_from/date_to/model_keyword 的过滤字典 """ filters: dict[str, Any] = {} for field, spec in self.filter_bar.get_conditions().items(): op = spec.get("operator") val = spec.get("value") if field == "print_date": if op == "between" and isinstance(val, list) and len(val) == 2: d1, d2 = self._parse_date(val[0]), self._parse_date(val[1]) if d1 and d2: filters["date_from"] = datetime(d1.year, d1.month, d1.day, 0, 0, 0) filters["date_to"] = datetime(d2.year, d2.month, d2.day, 23, 59, 59) elif op == "eq" and val: d = self._parse_date(val) if d: filters["date_from"] = datetime(d.year, d.month, d.day, 0, 0, 0) filters["date_to"] = datetime(d.year, d.month, d.day, 23, 59, 59) elif field == "model" and val: filters["model_keyword"] = val return filters @staticmethod def _parse_date(text: str): """解析 YYYY-MM-DD 为 date,失败返回 None。""" try: return datetime.strptime(str(text).strip(), "%Y-%m-%d").date() except (ValueError, AttributeError): return None def _init_default_filters(self) -> None: """预填默认筛选行:打印时间近一个月区间(与原默认行为一致)。""" fmt = "yyyy-MM-dd" today = QDate.currentDate() month_ago = today.addMonths(-1) self.filter_bar.clear_rows() self.filter_bar.add_row( "print_date", "between", [month_ago.toString(fmt), today.toString(fmt)] ) def _load_history(self) -> None: """加载最近的打印历史(无额外筛选)""" if self.session is None: return try: from certflow.controllers.history_controller import HistoryController controller = HistoryController(self.session) logs = controller.get_recent_logs(200) self._populate(logs) except Exception as e: logger.error(f"加载打印历史失败: {e}") def _apply_filters(self) -> None: """按当前筛选条件查询并打印历史""" if self.session is None: return try: from certflow.controllers.history_controller import HistoryController controller = HistoryController(self.session) logs = controller.query_logs(self._build_filters()) self._populate(logs) self._filter_history(self.filter_input.text()) except Exception as e: logger.error(f"筛选打印历史失败: {e}") def _populate(self, logs: list[Any]) -> None: """将打印日志填充到表格""" self.history_table.setRowCount(len(logs) if logs else 0) if not logs: self.history_table.setColumnCount(7) self.history_table.setHorizontalHeaderLabels( ["ID", "编号", "产品型号", "打印机", "状态", "时间", "内容摘要"] ) return columns = ["ID", "编号", "产品型号", "打印机", "状态", "时间", "内容摘要"] self.history_table.setColumnCount(len(columns)) self.history_table.setHorizontalHeaderLabels(columns) for i, log in enumerate(logs): self.history_table.setItem(i, 0, QTableWidgetItem(str(log.id))) self.history_table.setItem(i, 1, QTableWidgetItem(log.SN or log.KKS or "")) self.history_table.setItem(i, 2, QTableWidgetItem(log.content_summary or "")) self.history_table.setItem(i, 3, QTableWidgetItem(log.PrinterName or "")) self.history_table.setItem(i, 4, QTableWidgetItem(log.status or "")) self.history_table.setItem(i, 5, QTableWidgetItem(str(log.hgz_PrintTime or ""))) self.history_table.setItem(i, 6, QTableWidgetItem(log.content_summary or "")) self.history_table.resizeColumnsToContents() def _export_csv(self) -> None: """将当前筛选结果导出为 CSV(UTF-8 BOM)""" if self.session is None: return path, _ = QFileDialog.getSaveFileName(self, "导出打印历史CSV", "", "CSV 文件 (*.csv)") if not path: return try: from certflow.controllers.history_controller import HistoryController controller = HistoryController(self.session) logs = controller.query_logs(self._build_filters()) count = controller.export_logs_csv(logs, path) QMessageBox.information(self, "导出完成", f"已导出 {count} 条记录到:\n{path}") except Exception as e: logger.error(f"导出CSV失败: {e}") QMessageBox.warning(self, "导出失败", f"导出失败:{e}") def _filter_history(self, text: str) -> None: """自由文本快速筛选(隐藏不匹配行)""" text = (text or "").lower() for row in range(self.history_table.rowCount()): show = False for col in range(self.history_table.columnCount()): item = self.history_table.item(row, col) if item and text in item.text().lower(): show = True break self.history_table.setRowHidden(row, not show)