# src/certflow/views/import_confirm_dialog.py
"""销售计划导入「确认」对话框（ImportConfirmDialog）
承接 ImportConfigDialog 的第二步：展示分组/冲突预览，收集 5 选项冲突决议，
确认后执行导入。导入成功后通过 ``result_config_params`` 把配置交回设置对话框
（ImportConfigDialog）转发 ``config_confirmed`` 信号给主窗口刷新。
"""
from __future__ import annotations
from typing import Any
from loguru import logger
from PySide6.QtCore import Qt, QTimer
from PySide6.QtWidgets import (
QApplication,
QComboBox,
QDialog,
QHeaderView,
QLabel,
QMessageBox,
QProgressBar,
QSizePolicy,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from certflow.config.settings import _get_user_config
from certflow.widgets import ActionToolbar, ConfirmDialog
from certflow.widgets.enhanced_table import EnhancedTable
[文档]
def run_import(session: Any, config_params: dict) -> dict:
"""执行导入（配置可含 conflict_resolutions），返回统计 dict。
供「0 冲突直接导入」与「确认对话框导入」共用，保证两条路径落库完全一致。
经 ImportController 统一调用面（蓝图 M5：导入链路补齐 Controller 层）。
"""
from certflow.controllers.import_controller import ImportController
with ImportController(db_session=session) as ctrl:
return ctrl.run_import(config_params)
[文档]
def should_skip_confirm_dialog(
db_conflicts: list,
intra_conflicts: list,
skip_enabled: bool = True,
) -> bool:
"""判定是否可跳过确认对话框（0 冲突时直接导入）。
Args:
db_conflicts: 跨库冲突列表（来自 _scan_conflicts）
intra_conflicts: 批内撞键冲突列表
skip_enabled: import.skip_confirm_when_no_conflict 开关
Returns:
bool: 开关开启且两类冲突均为空时为 True
"""
return bool(skip_enabled) and not db_conflicts and not intra_conflicts
[文档]
class ImportConfirmDialog(QDialog):
"""导入确认对话框：冲突预览 + 5 选项决议 + 执行导入。"""
def __init__(
self,
config_params: dict,
preview_records: list[dict],
db_conflicts: list[tuple[str, dict, str]],
intra_conflicts: list[str],
order_groups: int,
parent: Any = None,
db_manager: Any = None,
session: Any = None,
) -> None:
super().__init__(parent)
self._config_params = config_params
self._is_all_months = config_params.get("scope") == "all_months"
self._preview_records = preview_records
self._db_conflicts = db_conflicts
self._intra_conflicts = intra_conflicts
self._order_groups = order_groups
self.db_manager = db_manager
self._session = session
self.result_config_params: dict | None = None
self._conflict_resolutions: dict[str, str] = {}
self._importing = False
self.setWindowTitle("销售计划导入 · 确认")
self.setMinimumSize(680, 480)
self.setModal(True)
self.resize(820, 600)
self.setup_ui()
self._populate_conflicts()
[文档]
def setup_ui(self) -> None:
layout = QVBoxLayout(self)
# 概要
self.summary_label = QLabel()
self.summary_label.setStyleSheet("font-weight:bold; padding:4px;")
layout.addWidget(self.summary_label)
# 冲突表
self.conflict_table = EnhancedTable()
self.conflict_table.setColumnCount(5)
self.conflict_table.setHorizontalHeaderLabels(
["产品型号", "唯一键", "现有值摘要", "导入动作", "备注"]
)
for idx, field in enumerate(["product_model", "unique_key", "summary", "action", "note"]):
self.conflict_table.set_header_field(idx, field)
self.conflict_table.setMinimumHeight(200)
# 列宽策略：长文本列（现有值摘要/备注）拉伸占宽并换行，短列按内容自适应，
# 避免长中文被单行行高截断（EnhancedTable 默认 Interactive 不会撑开行高）。
header = self.conflict_table.horizontalHeader()
header.setSectionResizeMode(0, QHeaderView.ResizeMode.ResizeToContents) # 产品型号
header.setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents) # 唯一键
header.setSectionResizeMode(2, QHeaderView.ResizeMode.Stretch) # 现有值摘要（占宽+换行）
header.setSectionResizeMode(3, QHeaderView.ResizeMode.Fixed) # 导入动作（下拉固定宽）
header.resizeSection(3, 96)
self.conflict_table.setWordWrap(True)
layout.addWidget(self.conflict_table, stretch=1)
# 批量应用
batch_bar = ActionToolbar()
batch_bar.add_widget(QLabel("批量应用动作:"))
self.batch_combo = QComboBox()
self.batch_combo.addItems(["跳过", "覆盖", "合并", "隔离", "新建"])
batch_bar.add_widget(self.batch_combo)
self.batch_btn = batch_bar.add_button("应用到全部可决议行", self._batch_apply)
batch_bar.add_stretch()
layout.addWidget(batch_bar)
# 撞键处理策略（盲区3 修复：为批次内撞键行提供动作选项）
self.collision_bar = ActionToolbar()
self.collision_bar.add_widget(QLabel("撞键处理:"))
self.collision_policy_combo = QComboBox()
self.collision_policy_combo.addItems(["隔离(当前)", "取首行", "丢弃全部重复"])
self.collision_policy_combo.setCurrentText("隔离(当前)")
self.collision_bar.add_widget(self.collision_policy_combo)
self.collision_bar.add_widget(QLabel("（仅对批次内撞键行生效）"))
self.collision_bar.add_stretch()
self.collision_policy_combo.currentTextChanged.connect(self._update_isolation_hint)
layout.addWidget(self.collision_bar)
# 进度条
self.progress_bar = QProgressBar()
self.progress_bar.setVisible(False)
self.progress_bar.setMinimum(0)
self.progress_bar.setMaximum(0) # 不确定进度（动画模式）
self.progress_bar.setTextVisible(True)
self.progress_bar.setFormat("导入中...")
layout.addWidget(self.progress_bar)
# 撞键隔离提示（导入前让用户感知将自动隔离的撞键行）
self.isolation_hint_label = QLabel()
self.isolation_hint_label.setStyleSheet(
"color:#b8860b; font-weight:bold; padding:6px; "
"background-color:#fff8e1; border:1px solid #f0c36d; border-radius:4px;"
)
self.isolation_hint_label.setWordWrap(True)
self.isolation_hint_label.setVisible(False)
layout.addWidget(self.isolation_hint_label)
# 按钮
button_bar = ActionToolbar()
self.back_btn = button_bar.add_button("← 返回", self.reject)
button_bar.add_stretch()
self.confirm_btn = button_bar.add_button("确认导入", self._confirm_import, primary=True)
self.confirm_btn.setStyleSheet(
"background-color: #2c7be5; color: white; padding: 8px 20px;"
)
layout.addWidget(button_bar)
[文档]
def showEvent(self, event) -> None: # noqa: N802 Qt 覆写方法，驼峰命名由框架规定
"""首次显示后按内容撑开冲突表行高，并对过长行设上限。
行高由 ``resizeRowsToContents`` 按当前字体/DPI 推导，因此不同屏幕缩放下
表现一致（前提是应用启用了 ``AA_EnableHighDpiScaling``）。
不再强制所有行等高——旧实现取 ``max_h`` 统一行高，一旦某行摘要换行较多，
会把所有短行也拉高，视觉上「行高变高」。现改为：内容高 ≤ 上限时按内容，
超过上限则截断到上限（配合 ``setWordWrap`` 长文本仍完整换行显示）。
列 3（导入动作）的单元格放的是 QWidget 容器，内部 combo 用 ``minimumHeight``
占位并垂直居中，不依赖整行等高也能整齐对齐。
"""
super().showEvent(event)
def _fit_rows() -> None:
t = self.conflict_table
t.resizeRowsToContents()
n = t.rowCount()
if n == 0:
return
# 行高上限：约 3 行正文 + 上下边距，避免单行摘要过长把整表撑爆。
# 随字体度量派生，高 DPI 下自动放大，保证物理观感一致。
fm = t.fontMetrics()
line_h = fm.lineSpacing() or fm.height()
max_h = line_h * 3 + 12
for r in range(n):
h = t.rowHeight(r)
if h > max_h:
t.setRowHeight(r, max_h)
QTimer.singleShot(0, _fit_rows)
def _populate_conflicts(self) -> None:
# 全部月份表（大批量）：不做逐表冲突扫描，改为信息提示，由导入时默认冲突策略处理跨表重复
if self._is_all_months:
total = self._config_params.get("_all_months_total", len(self._preview_records))
self.summary_label.setText(
f"⚠ 全部月份表模式：将依次导入 1-12 月中存在的全部欠交表（约 {total} 行）。\n"
f"跨表重复行按「冲突处理默认策略」（建议在 Tab③ 设为「跳过」）自动排除结转重复。"
)
self.conflict_table.setRowCount(0)
self.isolation_hint_label.setVisible(False)
self.collision_bar.setVisible(False)
return
n = len(self._preview_records)
self.summary_label.setText(
f"待导入 {n} 行 → {self._order_groups} 个订单组 | "
f"批次内冲突 {len(self._intra_conflicts)} | "
f"数据库已存在冲突 {len(self._db_conflicts)}"
)
# 撞键行导入前提示：让用户感知撞键行将如何处理（隔离/取首行/丢弃全部）
self.collision_bar.setVisible(bool(self._intra_conflicts))
self._update_isolation_hint()
self.conflict_table.setRowCount(0)
self._conflict_resolutions = {}
for uk, rec, summary in self._db_conflicts:
self._add_conflict_row(uk, rec, summary, "数据库已存在，选择导入动作")
for uk in self._intra_conflicts:
self._add_conflict_row(
uk, {}, "(批次内多行同唯一键但内容不同)", "请先在源表合并/改值后重导", readonly=True
)
# 撞键处理策略：UI 中文 -> handler 英文（与 save_handler._apply_batch_collision 一致）
_COLLISION_POLICY_MAP = {
"隔离(当前)": "isolate",
"取首行": "first",
"丢弃全部重复": "discard",
}
def _update_isolation_hint(self) -> None:
"""根据当前撞键处理策略动态刷新提示文案（盲区3 修复）。"""
if not self._intra_conflicts:
self.isolation_hint_label.setVisible(False)
return
cnt = len(self._intra_conflicts)
policy_cn = self.collision_policy_combo.currentText()
if policy_cn == "取首行":
text = (
f"⚠ 本批含 {cnt} 个撞键唯一键：按「取首行」处理，重复行直接丢弃"
f"（首行保留入主表，不进待处理表）。请先回源合并/改值后重导，或确认后导入。"
)
elif policy_cn == "丢弃全部重复":
text = (
f"⚠ 本批含 {cnt} 个撞键唯一键：按「丢弃全部重复」处理，整组（含首行）"
f"直接丢弃，不进主表/待处理表。请先回源合并/改值后重导，或确认后导入。"
)
else: # 隔离(当前)
text = (
f"⚠ 本批含 {cnt} 个撞键唯一键：导入时将自动隔离到「待处理表」，"
f"主表不会重复写入。请先回源合并/改值后重导，或确认后导入。"
)
self.isolation_hint_label.setText(text)
self.isolation_hint_label.setVisible(True)
def _add_conflict_row(
self, uk: str, rec: dict, summary: str, note: str, readonly: bool = False
) -> None:
"""向冲突表追加一行，含动作下拉（5 选项）。"""
ridx = self.conflict_table.rowCount()
self.conflict_table.insertRow(ridx)
pm = str(rec.get("product_model", "") or "")[:30]
self.conflict_table.setItem(ridx, 0, QTableWidgetItem(pm))
self.conflict_table.setItem(ridx, 1, QTableWidgetItem(uk))
# 现有值摘要：完整保留（去掉原 [:40] 硬截断），配合列宽拉伸+换行+撑开行高显示
summary_item = QTableWidgetItem(str(summary))
summary_item.setTextAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
self.conflict_table.setItem(ridx, 2, summary_item)
combo = QComboBox()
combo.addItems(["跳过", "覆盖", "合并", "隔离", "新建"])
# 高度随字体度量派生（而非硬编码 22px），高 DPI / 大字体下自动放大，
# 避免因像素与 DPI 不一致而导致下拉框比同行文字偏矮、错位。
# 用 minimumHeight 占位：容器被表格拉伸填满整格，combo 在容器内垂直居中，
# 不再依赖「整行等高」即可形成对齐列（showEvent 仅对过长行封顶，短行不被拉高）。
combo_fm = combo.fontMetrics()
combo_h = combo_fm.lineSpacing() + 8
combo.setMinimumHeight(combo_h)
combo.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred)
if readonly:
combo.setEnabled(False)
combo.setCurrentText("跳过")
else:
combo.setCurrentText("跳过")
combo.currentTextChanged.connect(
lambda txt, k=uk: self._conflict_resolutions.update({k: self._action_en(txt)})
)
self._conflict_resolutions[uk] = "skip"
combo_container = QWidget()
cvl = QVBoxLayout(combo_container)
cvl.setContentsMargins(2, 2, 2, 2)
cvl.setSpacing(0)
# 垂直居中：无论行高被内容撑到多少，下拉框都居于单元格中央，列对齐更稳。
cvl.addWidget(combo, alignment=Qt.AlignmentFlag.AlignVCenter)
self.conflict_table.setCellWidget(ridx, 3, combo_container)
note_item = QTableWidgetItem(note)
note_item.setTextAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
self.conflict_table.setItem(ridx, 4, note_item)
def _batch_apply(self) -> None:
"""将批量动作应用到所有可决议（非只读）行。"""
text = self.batch_combo.currentText()
for r in range(self.conflict_table.rowCount()):
# 单元格放的是 QVBoxLayout 容器，需取出其中的 QComboBox
cell = self.conflict_table.cellWidget(r, 3)
combo = cell.findChild(QComboBox) if cell is not None else None
if combo is None or not combo.isEnabled():
continue
combo.setCurrentText(text)
@staticmethod
def _action_en(cn: str) -> str:
"""中文动作 -> 英文（与 save_handler._resolve_existing 一致）。"""
return {
"跳过": "skip",
"覆盖": "overwrite",
"合并": "merge",
"隔离": "isolate",
"新建": "new",
}.get(cn, "skip")
def _build_conflict_resolutions(self) -> dict[str, str]:
return dict(self._conflict_resolutions)
def _should_confirm_overwrite(self) -> bool:
"""检查用户偏好是否要求导入前确认。"""
return _get_user_config().get("preferences", {}).get("confirm_before_overwrite", True)
def _set_ui_enabled(self, enabled: bool) -> None:
self.back_btn.setEnabled(enabled)
self.batch_btn.setEnabled(enabled)
self.batch_combo.setEnabled(enabled)
self.conflict_table.setEnabled(enabled)
# confirm_btn 由调用方单独控制
def _confirm_import(self) -> None:
"""收集冲突决议并执行导入；成功后交回 config_params。"""
if self._session is None:
self._show_import_result("无法导入", "缺少数据库会话，无法执行导入。", is_success=False)
return
if self._should_confirm_overwrite() and self._db_conflicts:
confirmed = ConfirmDialog.danger(
self,
"确认导入",
"即将导入数据，可能覆盖已有记录。\n\n确定要继续吗？",
)
if not confirmed:
return
self._importing = True
self._set_ui_enabled(False)
self.progress_bar.setVisible(True)
self.confirm_btn.setText("导入中...")
QApplication.processEvents()
try:
config = dict(self._config_params)
resolutions = self._build_conflict_resolutions()
if resolutions:
config["conflict_resolutions"] = resolutions
# 盲区3 修复：批次内撞键处理策略透传给 service（覆盖 ON_COLLISION 配置）
config["collision_policy"] = self._COLLISION_POLICY_MAP[
self.collision_policy_combo.currentText()
]
result = run_import(self._session, config)
logger.info("导入完成")
msg = self._build_import_result_message(result)
self._show_import_result("导入成功", msg, is_success=True)
self.result_config_params = config
self.accept()
except Exception as e:
logger.error(f"导入失败: {e}")
self._show_import_result("导入失败", f"导入过程中发生错误:\n{e}", is_success=False)
finally:
self._importing = False
self._set_ui_enabled(True)
self.progress_bar.setVisible(False)
self.confirm_btn.setText("确认导入")
@staticmethod
def _build_import_result_message(result: dict) -> str:
"""构建导入结果消息，按隔离来源分别计数展示（门控/撞键/冲突决议）。"""
total = result.get("total", 0)
new_count = result.get("new_count", 0)
skipped = result.get(
"skipped_count", result.get("skipped", result.get("duplicate_count", 0))
)
changes = result.get("changes_count", result.get("changes", 0))
gate_q = result.get("gate_quarantined", 0)
collision_q = result.get("collision_quarantined", 0)
collision_dropped = result.get("collision_dropped", 0)
resolution_q = result.get("resolution_quarantined", 0)
change_categories = result.get("change_categories") or {}
parts = [f"总记录: {total}"]
if new_count:
parts.append(f"新增: {new_count}")
if skipped:
parts.append(f"跳过(已存在且无变更): {skipped}")
if gate_q:
parts.append(f"门控隔离(缺要货单号): {gate_q}")
if collision_q:
parts.append(f"撞键隔离(唯一键重复): {collision_q}")
if collision_dropped:
parts.append(f"撞键丢弃(取首行/全弃): {collision_dropped}")
if resolution_q:
parts.append(f"冲突决议隔离: {resolution_q}")
if changes:
parts.append(f"变更(已存在但内容变化): {changes}")
for cat, cnt in change_categories.items():
parts.append(f" · {cat}: {cnt}")
return "\n".join(parts)
def _show_import_result(self, title: str, message: str, is_success: bool = True) -> None:
"""显示导入结果消息。"""
if is_success:
QMessageBox.information(self, title, message)
else:
QMessageBox.critical(self, title, message)