certflow.views.bom_import_widget 源代码
"""BOM 零件牌号导入控件（目录扫描入库）
作为「计划导入」堆叠页的第二个标签页（与销售计划导入并列），复用
``certflow.services.bom_material_service.scan_and_import`` 做目录递归扫描入库。
对应 CLI ``certflow bom import``；``--mode`` 对应本控件的「导入模式」四选一。
提供四种导入模式，UI 上明确告知用法与去重风险：
- 清空重导（clear）：清除全表后重导，结果干净、无重复；重建全表用；
- 直接追加（append）：不清空直接插入，同一文件重复导入会产生重复行；
- 按文件刷新（refresh）：已导入文件删旧行写最新内容，新文件追加；
- 增量跳过已导入（skip）：只导未入库文件，已存在文件（即便已更新）跳过。
默认目录与默认模式由配置 ``bom_import.default_folder`` / ``bom_import.default_mode``
（config/import.yaml，经 !include 并入）驱动；本控件打开时自动预填目录、预选模式，
日常只扫新文件把 ``default_mode`` 设为 ``skip`` 即可，需重建全表时再改回 ``clear``
或在 UI 手动切换，无需改代码。
导入在后台线程执行，避免阻塞 UI；通过 ``import_started`` / ``import_finished``
信号通知外层（用于锁定导航）。
蓝图关联：``docs/BLUEPRINT_QUERY_UNIFY.md`` 阶段 B1（BOM 导入 tab）。
"""
from __future__ import annotations
import contextlib
from pathlib import Path
from typing import Any
from PySide6.QtCore import QThread, Signal
from PySide6.QtWidgets import (
QFileDialog,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QProgressBar,
QPushButton,
QRadioButton,
QTextEdit,
QVBoxLayout,
QWidget,
)
from certflow.config.settings import cfg
from certflow.controllers.bom_material_controller import BomMaterialController
from certflow.utils.logger import logger
# 合法的导入模式键（与 scan_and_import 的 BomImportMode 对应）
_VALID_MODES = ("clear", "append", "refresh", "skip")
class _BomScanWorker(QThread):
"""后台扫描入库线程。"""
log = Signal(str)
done = Signal(dict)
failed = Signal(str)
def __init__(
self,
db_manager: Any,
session: Any,
folder: str,
mode: str,
) -> None:
super().__init__()
self._db = db_manager
self._session = session
self._folder = folder
self._mode = mode
def run(self) -> None: # noqa: D401
session = None
try:
session = self._db.get_session() if self._db is not None else self._session
self.log.emit(f"开始扫描目录: {self._folder}（模式: {self._mode}）")
stat = BomMaterialController.scan_and_import(self._folder, session, mode=self._mode)
self.log.emit(
f"导入统计: {stat.get('files')} 文件 / "
f"{stat.get('rows')} 行零件 / {stat.get('orders')} 个生产令号"
)
if stat.get("new_files"):
self.log.emit(f" 新增文件: {stat['new_files']}")
if stat.get("refreshed_files"):
self.log.emit(f" 刷新文件: {stat['refreshed_files']}")
if stat.get("skipped_files"):
self.log.emit(f" 跳过已导入: {stat['skipped_files']}")
self.done.emit(stat)
except Exception as e: # noqa: BLE001
logger.error(f"BOM 扫描入库失败: {e}")
self.failed.emit(str(e))
finally:
if self._db is not None and session is not None:
with contextlib.suppress(Exception):
session.close()
[文档]
class BomImportWidget(QWidget):
"""BOM 导入控件：目录选择 + 追加/清除选项 + 扫描入库 + 进度/日志。"""
import_started = Signal()
import_finished = Signal()
def __init__(
self,
db_manager: Any = None,
session: Any = None,
parent: QWidget | None = None,
) -> None:
super().__init__(parent)
self._db_manager = db_manager
self._session = session
self._worker: _BomScanWorker | None = None
self._init_ui()
def _init_ui(self) -> None:
layout = QVBoxLayout(self)
layout.setContentsMargins(12, 12, 12, 12)
layout.setSpacing(10)
tip = QLabel(
"扫描「生产-下料计划 TAO」材料清单目录（含子目录递归 .xls/.xlsx），"
"解析零件牌号并入库到 bom_materials 表，供报告生成时按生产令号查询阀体/阀杆等材质。"
)
tip.setWordWrap(True)
tip.setStyleSheet("font-size: 11px; color: #666;")
layout.addWidget(tip)
# —— 目录选择 ——
dir_row = QHBoxLayout()
dir_row.addWidget(QLabel("材料清单目录:"))
self._folder_edit = QLineEdit()
self._folder_edit.setPlaceholderText("点击「浏览」选择材料清单根目录")
# 配置驱动：bom_import.default_folder 预填默认目录（留空则不预填）
default_folder = (str(cfg("bom_import.default_folder", "")) or "").strip()
if default_folder:
self._folder_edit.setText(default_folder)
dir_row.addWidget(self._folder_edit, stretch=1)
self._browse_btn = QPushButton("📁 浏览...")
self._browse_btn.clicked.connect(self._browse)
dir_row.addWidget(self._browse_btn)
layout.addLayout(dir_row)
# —— 导入模式（四选一，明确告知用法与去重风险）——
mode_box = QGroupBox("导入模式（决定旧数据如何处理）")
mode_layout = QVBoxLayout(mode_box)
mode_layout.setSpacing(4)
self._mode_radios: dict[str, QRadioButton] = {}
self._mode_descs: dict[str, str] = {}
_mode_defs = [
(
"clear",
"清空重导（推荐）",
"清除全部旧数据后重新导入，结果干净、无重复。适合首次导入或目录整体更新。",
),
(
"append",
"直接追加",
"不清空，所有扫描到的零件直接插入。⚠ 同一文件重复导入会产生重复行（如需去重请用「按文件刷新」）。",
),
(
"refresh",
"按文件刷新",
"不清空；已导入过的文件删除其旧零件再写入最新内容（某文件 12→10 行则删 2、12→15 行则加 5），新文件追加。库与目录当前内容一致。",
),
(
"skip",
"增量跳过已导入",
"不清空；只导入尚未入库的新文件，已导入文件（即便内容已更新）跳过。最安全最快；想刷新旧文件请改用「清空重导」。",
),
]
for key, title, desc in _mode_defs:
rb = QRadioButton(title)
rb.setToolTip(desc)
rb.toggled.connect(lambda checked, k=key: self._on_mode_toggled(k, checked))
mode_layout.addWidget(rb)
self._mode_radios[key] = rb
self._mode_descs[key] = desc
self._mode_help = QLabel()
self._mode_help.setWordWrap(True)
self._mode_help.setStyleSheet(
"font-size: 11px; color: #555; padding: 6px 8px; "
"background: #f5f5f5; border-radius: 4px;"
)
mode_layout.addWidget(self._mode_help)
self._mode_radios["clear"].setChecked(True)
# 配置驱动：bom_import.default_mode 决定打开时预选模式（默认 skip）
default_mode = (str(cfg("bom_import.default_mode", "skip")) or "skip").strip().lower()
if default_mode not in _VALID_MODES:
default_mode = "skip"
self._mode_radios[default_mode].setChecked(True)
self._on_mode_toggled(default_mode, True) # 初始化帮助文本
layout.addWidget(mode_box)
# —— 扫描按钮 ——
btn_row = QHBoxLayout()
self._scan_btn = QPushButton("🚀 扫描入库")
self._scan_btn.setStyleSheet("background-color: #28a745; color: white; padding: 8px 20px;")
self._scan_btn.clicked.connect(self._scan)
btn_row.addWidget(self._scan_btn)
btn_row.addStretch()
layout.addLayout(btn_row)
# —— 进度 ——
self._progress = QProgressBar()
self._progress.setVisible(False)
self._progress.setMinimum(0)
self._progress.setMaximum(0)
self._progress.setFormat("扫描中...")
layout.addWidget(self._progress)
# —— 日志 ——
self._log_area = QTextEdit()
self._log_area.setReadOnly(True)
self._log_area.setPlaceholderText("导入日志将显示在这里...")
self._log_area.setMaximumHeight(160)
layout.addWidget(self._log_area)
layout.addStretch()
def _browse(self) -> None:
start = self._folder_edit.text() or ""
folder = QFileDialog.getExistingDirectory(self, "选择材料清单目录", start)
if folder:
self._folder_edit.setText(folder)
def _scan(self) -> None:
folder = self._folder_edit.text().strip()
if not folder or not Path(folder).exists():
from PySide6.QtWidgets import QMessageBox
QMessageBox.warning(self, "提示", "请先选择有效的材料清单目录")
return
mode = next(k for k, rb in self._mode_radios.items() if rb.isChecked())
self._set_ui_enabled(False)
self._progress.setVisible(True)
self.import_started.emit()
self._worker = _BomScanWorker(self._db_manager, self._session, folder, mode)
self._worker.log.connect(self._log_area.append)
self._worker.done.connect(self._on_done)
self._worker.failed.connect(self._on_failed)
self._worker.finished.connect(self._worker.deleteLater)
self._worker.start()
def _on_mode_toggled(self, key: str, checked: bool) -> None:
"""选中某模式时更新帮助文本（告知用法/去重风险）。"""
if checked:
self._mode_help.setText(self._mode_descs.get(key, ""))
def _on_done(self, stat: dict) -> None:
self._log_area.append(
f"✅ 导入完成: 写入文件 {stat.get('files')} "
f"（新增 {stat.get('new_files', 0)} / 刷新 {stat.get('refreshed_files', 0)} "
f"/ 跳过 {stat.get('skipped_files', 0)}），"
f"{stat.get('rows')} 行零件, {stat.get('orders')} 个生产令号"
)
self._finish()
def _on_failed(self, msg: str) -> None:
self._log_area.append(f"❌ 导入失败: {msg}")
self._finish()
def _finish(self) -> None:
self._progress.setVisible(False)
self._set_ui_enabled(True)
self.import_finished.emit()
def _set_ui_enabled(self, enabled: bool) -> None:
self._browse_btn.setEnabled(enabled)
self._folder_edit.setEnabled(enabled)
for rb in self._mode_radios.values():
rb.setEnabled(enabled)
self._scan_btn.setEnabled(enabled)