# src/certflow/views/import_config_dialog.py
"""销售计划导入配置对话框
提供Excel导入前的参数配置界面，包括：
- 文件选择
- 工作表选择
- 表头行设置
- 数据起始行设置
- 列映射预览
- 跳过行数设置
"""
from __future__ import annotations
import calendar
import os
import re
from datetime import date, datetime
from pathlib import Path
from typing import Any
import pandas as pd
from loguru import logger
from PySide6.QtCore import QDate, Qt, Signal
from PySide6.QtGui import QColor
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QDateEdit,
QDialog,
QFileDialog,
QFormLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QMessageBox,
QProgressBar,
QPushButton,
QScrollArea,
QSpinBox,
QTableWidget,
QTableWidgetItem,
QTabWidget,
QTextEdit,
QVBoxLayout,
QWidget,
)
from certflow.config.paths_override import cfg
from certflow.config.settings import (
COLUMN_ALIASES,
REQUIRED_FIELDS,
SALE_PLAN_COLUMNS,
SKIP_CONFIRM_WHEN_NO_CONFLICT,
_get_user_config,
_save_user_config,
add_recent_import_file,
clear_recent_files,
get_import_default_strategy,
get_import_required_fields,
get_import_views,
get_recent_import_files,
merge_user_config,
)
from certflow.controllers.import_controller import ImportController
from certflow.utils.date_utils import normalize_plan_date
from certflow.views.common.qdate_utils import (
QT_MIN_VALID_DATE,
parse_qdate,
safe_qdate,
)
from certflow.widgets import FlowLayout, SectionCard
from certflow.widgets.quick_search import QuickSearch
def _labeled(label_text: str, widget: QWidget) -> QWidget:
"""将「标签 + 控件」包成水平控件，用于放进 FlowLayout 自动换行。
避免 QFormLayout.addRow 让每个字段独占一整行、在窄配置区浪费纵向空间。
"""
box = QWidget()
h = QHBoxLayout(box)
h.setContentsMargins(0, 0, 0, 0)
h.setSpacing(6)
h.addWidget(QLabel(label_text))
h.addWidget(widget)
return box
[文档]
class ImportConfigDialog(QDialog):
"""导入配置对话框
提供Excel文件导入前的参数配置界面，支持:
- 文件选择（新增）
- 工作表选择
- 表头行和数据起始行设置
- 列映射预览和手动配置
- 跳过列管理
- 配置记忆(自动保存下次导入时恢复)
Signals:
config_confirmed: 配置确认信号，参数为配置参数字典
Attributes:
file_path: Excel文件路径
df_raw: 原始DataFrame(未处理表头)
available_sheets: 可用工作表列表
_skipped_columns: 跳过的列名集合
_column_mapped_status: 每列映射状态缓存
sheet_configs: 各工作表的保存配置
"""
config_confirmed = Signal(dict)
# ================================================================
# 1. 生命周期方法
# ================================================================
def __init__(
self,
file_path: str = "",
view_key: str = "sales_plan",
parent: Any = None,
db_manager: Any = None,
session: Any = None,
) -> None:
"""初始化导入配置对话框
Args:
file_path: Excel文件路径，为空时从最近文件列表加载
view_key: 视图键
parent: 父组件，默认为None
db_manager: 数据库管理器实例
session: SQLAlchemy 会话对象（用于执行导入）
"""
super().__init__(parent)
self.db_manager = db_manager
self._session = session
# ========== 一次加载，全部透传（和 QueryView 一样）==========
self.view_key = view_key
self._views = get_import_views() # 视图配置
self._view_template = self._views.get(view_key, {})
self._required_fields = get_import_required_fields(view_key)
self._default_strategy = get_import_default_strategy()
# 直接读 YAML 配置
self._header_default = cfg("sales_plan.excel_config.header_row", 1)
self._data_start_default = cfg("sales_plan.data_start_row", 2)
# 用户配置（按需读写）
self._user_config = _get_user_config()
# 初始化状态
self.file_path = None
self.df_raw = None
self.current_sheet = ""
self.available_sheets = []
self._skipped_columns: set = set()
self._column_mapped_status: dict = {}
self._updating_checkboxes = False
self._loading_config = False
self._populating_recent = False # 标记：正在填充下拉列表，跳过事件
self._importing = False # 标记：正在执行导入，禁用按钮
# ========== B0-8 导入范围 / 冲突决议状态 ==========
self.range_mode = "all" # all(整表) / order(按订单筛选) / rows(勾选行)
# 导入范围（按月份筛选）三层层次：
# all_months — 全部月份表（1-12 月存在的全部欠交表，大批量，禁用快速筛选）
# single_month — 指定单个月份表（可选按计划日期范围筛选）
# current_month — 默认：当月欠交整表（含历史积压，可选按计划日期范围筛选）
self.scope = "current_month"
self.scope_month: int | None = None # 指定月份时的月份(1-12)
self.date_range_enabled = False
self.date_range_start: str | None = None # YYYY-MM-DD
self.date_range_end: str | None = None # YYYY-MM-DD
self._all_months_total: int = 0 # 全部月份表预览用：真实总行数
self.order_filter: dict[str, str] = {} # 4 字段筛选值（英文键）
self.selected_rows: list[int] = [] # 勾选行模式：0-based 数据行序号
self._conflict_resolutions: dict[str, str] = {} # unique_key -> 动作
self._preview_records: list[dict] = [] # 预览用（已清洗）记录
self._all_clean_records: list[dict] = [] # 级联筛选用：去重前的全部清洗记录
self._order_tuples: list[dict] = [] # 去重后的唯一订单（4 字段元组），供级联下拉
self.filter_combos: dict[str, Any] = {} # en -> QComboBox（级联筛选）
# 4 字段筛选：中文标签 -> 英文键（对齐 B0-6 分组键）
self._ORDER_FIELDS = [
("计划日期", "plan_date"),
("订货单位", "customer"),
("项目名称", "project_name"),
("要货计划单号", "plan_no"),
]
self.setWindowTitle(self._view_template.get("name", "导入"))
self.setMinimumSize(1000, 700)
self.setModal(True)
self.setup_ui()
# 加载最近文件列表
self._populate_recent_files()
# 如果有明确文件路径，直接加载
if file_path and file_path != ".":
p = Path(file_path)
if p.exists():
self._set_file_path(p)
return
# 自动加载最近文件列表中第一个存在的文件
self._load_first_recent_file()
[文档]
def setup_ui(self) -> None:
"""初始化UI（三页 Tab：① 源文件与导入设置 / ② 列映射 / ③ 导入配置项）"""
layout = QVBoxLayout(self)
# ============ 三页 Tab 容器 ============
self._tabs = QTabWidget()
self.tab_source = QWidget()
self.tab_source_layout = QVBoxLayout(self.tab_source)
self.tab_source_layout.setContentsMargins(12, 12, 12, 12)
self._tabs.addTab(self.tab_source, "① 源文件与导入设置")
self.tab_mapping = QWidget()
self.tab_mapping_layout = QVBoxLayout(self.tab_mapping)
self.tab_mapping_layout.setContentsMargins(0, 0, 0, 0)
self._tabs.addTab(self.tab_mapping, "② 列映射")
self.tab_config = QWidget()
self.tab_config_layout = QVBoxLayout(self.tab_config)
self.tab_config_layout.setContentsMargins(0, 0, 0, 0)
self._tabs.addTab(self.tab_config, "③ 导入配置项")
# ============ 三个页分别构建 ============
self._build_source_tab()
self._build_mapping_tab()
self._build_config_tab()
layout.addWidget(self._tabs, stretch=1)
# 限制对话框默认尺寸，超出部分在各 Tab 内滚动（底部进度条+按钮栏始终可见）
self.setMinimumWidth(780)
self.resize(1060, 780)
# ========== 进度条 ==========
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)
# ========== 按钮区域 ==========
button_layout = QHBoxLayout()
self.auto_detect_btn = QPushButton("🔍 自动检测")
self.auto_detect_btn.clicked.connect(self.auto_detect)
button_layout.addWidget(self.auto_detect_btn)
self.preview_btn = QPushButton("👁️ 预览数据")
self.preview_btn.clicked.connect(self.preview_data)
button_layout.addWidget(self.preview_btn)
button_layout.addStretch()
self.clear_memory_btn = QPushButton("🗑️ 清除记忆")
self.clear_memory_btn.setToolTip("清除当前文件保存的跳过列和配置")
self.clear_memory_btn.clicked.connect(self.clear_saved_config)
button_layout.addWidget(self.clear_memory_btn)
self.cancel_btn = QPushButton("关闭")
self.cancel_btn.clicked.connect(self.reject)
button_layout.addWidget(self.cancel_btn)
self.confirm_btn = QPushButton("下一步：查看冲突 →")
self.confirm_btn.setToolTip(
"打开冲突预览与确认对话框：展示批次内/数据库冲突，逐行决议后执行导入"
)
self.confirm_btn.setStyleSheet(
"background-color: #2c7be5; color: white; padding: 8px 20px;"
)
self.confirm_btn.clicked.connect(self._goto_confirm)
button_layout.addWidget(self.confirm_btn)
layout.addLayout(button_layout)
# ================================================================
# 1.1 Tab①：源文件与导入设置（文件选择 + 基本配置 + 导入时间段 + 导入范围）
# ================================================================
def _build_source_tab(self) -> None:
"""Tab①：源文件与导入设置。
按用户意图将「选择文件 + 基本配置 + 导入方式」归为一类：
- 源文件与基本配置（文件选择 / 文件大小·工作表数·当前表 写在一起 / 工作表·表头·数据起始·跳过行）
- 导入时间段与范围（按月份筛选 + B0-8 导入范围）
"""
root = self.tab_source_layout
# ============ 源文件与基本配置 ============
src_card = SectionCard("📁 源文件与基本配置")
src = src_card.content
# 文件选择
file_layout = QHBoxLayout()
self.recent_file_combo = QComboBox()
self.recent_file_combo.setMinimumWidth(300)
self.recent_file_combo.setToolTip("最近使用的文件（点击切换）")
self.recent_file_combo.currentIndexChanged.connect(self._on_recent_file_changed)
file_layout.addWidget(self.recent_file_combo, stretch=1)
self.clear_history_btn = QPushButton("🗑️")
self.clear_history_btn.setToolTip("清除最近文件历史")
self.clear_history_btn.setFixedWidth(30)
self.clear_history_btn.clicked.connect(self._clear_recent_history)
file_layout.addWidget(self.clear_history_btn)
self.browse_btn = QPushButton("浏览...")
self.browse_btn.clicked.connect(self.browse_file)
file_layout.addWidget(self.browse_btn)
src.addLayout(file_layout)
# 文件信息（文件大小 / 工作表数 / 当前表 写在一起，紧凑流式）
self.file_size_label = QLabel("未选择文件")
self.file_sheets_label = QLabel("未知")
self.file_current_sheet_label = QLabel("—")
info_flow = FlowLayout()
info_flow.addWidget(_labeled("文件大小:", self.file_size_label))
info_flow.addWidget(_labeled("工作表数:", self.file_sheets_label))
info_flow.addWidget(_labeled("当前表:", self.file_current_sheet_label))
src.addLayout(info_flow)
# 基本配置（FlowLayout 紧凑排版，根治整行单控件）
# 注：数据源表（工作表）选择已交由上方「导入范围」统一驱动，此处不再重复放置工作表下拉
self.header_row_spin = QSpinBox()
self.header_row_spin.setRange(0, 50)
self.header_row_spin.setValue(1)
self.header_row_spin.valueChanged.connect(self.on_preview_changed)
self.data_start_spin = QSpinBox()
self.data_start_spin.setRange(0, 100)
self.data_start_spin.setValue(2)
self.data_start_spin.valueChanged.connect(self.on_preview_changed)
self.skip_rows_spin = QSpinBox()
self.skip_rows_spin.setRange(0, 100)
self.skip_rows_spin.setValue(0)
self.skip_rows_spin.valueChanged.connect(self.on_preview_changed)
config_flow = FlowLayout()
config_flow.addWidget(_labeled("表头行号 (0=第一行):", self.header_row_spin))
config_flow.addWidget(_labeled("数据起始行 (0=第一行):", self.data_start_spin))
config_flow.addWidget(_labeled("跳过顶部行数:", self.skip_rows_spin))
src.addLayout(config_flow)
root.addWidget(src_card)
# ============ 导入时间段与范围 ============
scope_card = SectionCard("📅 导入时间段与范围")
scope = scope_card.content
# 导入范围（导入方式，新三层层次）—— 由 ui.yaml import_page.import_scope 配置驱动
scope_cfg = cfg("import_page.import_scope", {}) or {}
scope_options = scope_cfg.get("options", ["all_months", "single_month", "current_month"])
scope_labels = scope_cfg.get("labels", {}) or {}
self._scope_default_labels = {
"all_months": scope_labels.get("all_months", "全部月份表（1-12月存在的全部表）"),
"single_month": scope_labels.get("single_month", "指定月份表"),
"current_month": scope_labels.get("current_month", "当月欠交整表（默认）"),
}
self._scope_value_to_index: dict[str, int] = {}
scope_layout = QHBoxLayout()
scope_layout.addWidget(QLabel("导入范围:"))
self.scope_combo = QComboBox()
self.scope_combo.setToolTip(
"全部月份表：依次导入 1-12 月中存在的全部欠交表（大批量，禁用快速筛选；"
"跨表重复按默认冲突策略处理，建议设「跳过」排除结转重复）\n"
"指定月份表：仅导入选定单个月份的欠交表（可再按计划日期范围筛选）\n"
"当月欠交整表：默认，导入当月欠交表全量（含历史积压），可再按计划日期范围筛选"
)
for i, opt in enumerate(scope_options):
self.scope_combo.addItem(self._scope_default_labels.get(opt, opt))
self._scope_value_to_index[opt] = i
self.scope_combo.currentIndexChanged.connect(self._on_scope_changed)
scope_layout.addWidget(self.scope_combo, stretch=1)
self.scope_month_combo = QComboBox()
self.scope_month_combo.setMinimumWidth(80)
self.scope_month_combo.addItems([f"{m}月" for m in range(1, 13)])
self.scope_month_combo.setVisible(False)
self.scope_month_combo.currentIndexChanged.connect(self._on_scope_month_changed)
scope_layout.addWidget(self.scope_month_combo)
scope.addLayout(scope_layout)
# 计划日期范围筛选（仅 current_month / single_month 可用，用于剔除历史积压）
# 是否显示由 ui.yaml import_scope.show_date_range 控制
self._scope_show_date_range = bool(scope_cfg.get("show_date_range", True))
if self._scope_show_date_range:
self.date_range_row = QWidget()
dr_layout = QHBoxLayout()
dr_layout.setContentsMargins(0, 0, 0, 0)
self.date_range_chk = QCheckBox("按计划日期范围筛选（剔除历史积压）")
self.date_range_chk.toggled.connect(self._on_date_range_toggled)
dr_layout.addWidget(self.date_range_chk)
dr_layout.addWidget(QLabel("起:"))
self.date_start = QDateEdit()
self.date_start.setCalendarPopup(True)
self.date_start.setDate(QDate.currentDate().addMonths(-2))
self.date_start.setEnabled(False)
self.date_start.dateChanged.connect(self._on_date_range_changed)
dr_layout.addWidget(self.date_start)
dr_layout.addWidget(QLabel("止:"))
self.date_end = QDateEdit()
self.date_end.setCalendarPopup(True)
self.date_end.setDate(QDate.currentDate())
self.date_end.setEnabled(False)
self.date_end.dateChanged.connect(self._on_date_range_changed)
dr_layout.addWidget(self.date_end)
dr_layout.addStretch()
self.date_range_row.setLayout(dr_layout)
self.date_range_row.setVisible(False)
scope.addWidget(self.date_range_row)
# 全部月份表（大批量）提示
self.scope_bulk_note = QLabel(
"⚠ 全部月份表（大批量）：将依次导入 1-12 月中存在的全部欠交表，已禁用上方快速筛选；"
"跨表重复行按 Tab③「冲突处理默认策略」处理（建议设为「跳过」以排除结转重复）。"
)
self.scope_bulk_note.setWordWrap(True)
self.scope_bulk_note.setStyleSheet(
"color:#b8860b; padding:6px; background-color:#fff8e1; "
"border:1px solid #f0c36d; border-radius:4px;"
)
self.scope_bulk_note.setVisible(False)
scope.addWidget(self.scope_bulk_note)
# 导入范围（B0-8 现有逻辑）
range_group = QGroupBox("🎯 导入范围·快速筛选（B0-8）")
self.range_group = range_group # 全部月份表模式时禁用
range_layout = QVBoxLayout()
self.range_combo = QComboBox()
self.range_combo.addItems(["整表导入", "按订单筛选", "勾选行"])
self.range_combo.setToolTip(
"整表：导入全部数据行；按订单筛选：填4字段只导该订单组；勾选行：勾选指定数据行"
)
self.range_combo.currentIndexChanged.connect(self._on_range_mode_changed)
range_layout.addWidget(self.range_combo)
# 快查框：从已加载记录去重出的订单中模糊匹配，选中即自动设 4 级联下拉
# 复用 §1.7 统一 QuickSearch 组件（editable + completer 模糊匹配）。
self.order_quick_search = QuickSearch(
None,
placeholder="🔍 输入关键字快查订单（日期/单位/项目/单号）",
min_width=260,
)
self.order_quick_search.setToolTip(
"从去重订单中模糊匹配，选中即自动设 4 级联下拉定位该订单"
)
# activated(str) 仅在用户激活(选中/确认)条目时触发，直接给文本，
# 避开 QComboBox.activated 的 (int) 默认重载导致收到索引的坑
self.order_quick_search.activated.connect(self._on_quick_search_activated)
range_layout.addWidget(self.order_quick_search)
# -- 按订单筛选（级联下拉，对齐 VBA：去重唯一订单 → 逐字段级联）--
self.filter_widget = QWidget()
filter_form = QFormLayout()
self.filter_combos = {}
for level, (cn, en) in enumerate(self._ORDER_FIELDS):
cb = QComboBox()
cb.setMinimumWidth(240)
cb.addItem("（全部）")
cb.setToolTip(f"选择{cn}；下游字段仅列出与上游匹配的取值（级联）")
cb.currentIndexChanged.connect(
lambda _idx, lvl=level: self._on_filter_combo_changed(lvl)
)
self.filter_combos[en] = cb
filter_form.addRow(cn + ":", cb)
self.filter_widget.setLayout(filter_form)
range_layout.addWidget(self.filter_widget)
# 级联筛选结果指示（对齐 VBA：汇总表每行4数据保证级联成功）
self.order_status_label = QLabel(
"当前筛选: 计划日期=全部 / 订货单位=全部 / 项目名称=全部 / 要货计划单号=全部 → 匹配 0 行"
)
self.order_status_label.setStyleSheet("color:#888;")
range_layout.addWidget(self.order_status_label)
# 下一步行动提示：引导用户从「按订单筛选」进入冲突预览/确认
self.next_step_hint = QLabel()
self.next_step_hint.setStyleSheet("color:#2c7be5; font-weight:bold; padding-top:2px;")
range_layout.addWidget(self.next_step_hint)
# -- 勾选行 --
self.rows_widget = QWidget()
rows_layout = QVBoxLayout()
self.row_table = QTableWidget()
self.row_table.setColumnCount(6)
self.row_table.setHorizontalHeaderLabels(
["导入", "#", "计划日期", "订货单位", "项目名称", "产品型号"]
)
self.row_table.horizontalHeader().setStretchLastSection(True)
self.row_table.setMaximumHeight(220)
rows_layout.addWidget(self.row_table)
rows_layout.addWidget(QLabel("切换至此模式时自动载入数据行（默认全选），取消勾选=排除该行"))
self.rows_widget.setLayout(rows_layout)
range_layout.addWidget(self.rows_widget)
range_group.setLayout(range_layout)
scope.addWidget(range_group)
root.addWidget(scope_card)
# 导入范围初始选中（配置驱动 default）。延迟到此处初始化，避免 setup 早期
# scope_month_combo/date_range_row 尚未创建时触发 _on_scope_changed 报错。
default_scope = scope_cfg.get("default", "current_month")
self.scope = (
default_scope if default_scope in self._scope_value_to_index else "current_month"
)
default_idx = self._scope_value_to_index.get(
self.scope, next(iter(self._scope_value_to_index.values()), 0)
)
self.scope_combo.blockSignals(True)
self.scope_combo.setCurrentIndex(default_idx)
self.scope_combo.blockSignals(False)
self._on_scope_changed() # 统一初始化显隐 + 自动定位工作表
self._on_range_mode_changed() # 初始隐藏非整表子控件
# ================================================================
# 1.2 Tab②：列映射（批量操作 + 列映射预览 + 数据预览，独立成页避免拥挤）
# ================================================================
def _build_mapping_tab(self) -> None:
"""Tab②：列映射。独立成页，避免与源配置挤在一起。"""
scroll = QScrollArea()
scroll.setWidgetResizable(True)
inner = QWidget()
inner_layout = QVBoxLayout(inner)
# ========== 批量操作按钮栏 ==========
batch_layout = QHBoxLayout()
self.skip_unmapped_btn = QPushButton("⏭️ 跳过未配置列")
self.skip_unmapped_btn.clicked.connect(self.skip_unmapped_columns)
batch_layout.addWidget(self.skip_unmapped_btn)
self.reset_skip_btn = QPushButton("🔄 重置跳过")
self.reset_skip_btn.clicked.connect(self.reset_skipped_columns)
batch_layout.addWidget(self.reset_skip_btn)
batch_layout.addStretch()
self.select_all_btn = QPushButton("全选")
self.select_all_btn.clicked.connect(self.select_all)
batch_layout.addWidget(self.select_all_btn)
self.deselect_all_btn = QPushButton("取消全选")
self.deselect_all_btn.clicked.connect(self.deselect_all)
batch_layout.addWidget(self.deselect_all_btn)
batch_layout.addStretch()
self.apply_all_months_btn = QPushButton("📅 应用到所有月份表")
self.apply_all_months_btn.setToolTip(
"将当前工作表的解析配置(表头行/数据起始行/跳过列/列映射)复制到本工作簿所有其他工作表"
)
self.apply_all_months_btn.clicked.connect(self._apply_config_to_all_months)
batch_layout.addWidget(self.apply_all_months_btn)
inner_layout.addLayout(batch_layout)
# ========== 列映射预览表格 ==========
preview_group = QGroupBox("列映射预览(勾选☑️表示导入，未勾选表示跳过)")
preview_layout = QVBoxLayout()
self.preview_table = QTableWidget()
self.preview_table.setColumnCount(5)
self.preview_table.setHorizontalHeaderLabels(
["导入", "Excel列名", "目标字段", "配置状态", "示例数据"]
)
self.preview_table.horizontalHeader().setStretchLastSection(True)
self.preview_table.setColumnWidth(0, 50)
preview_layout.addWidget(self.preview_table)
# ========== 数据预览 ==========
self.data_preview = QTextEdit()
self.data_preview.setPlaceholderText("数据预览将显示在这里...")
self.data_preview.setMaximumHeight(200)
self.data_preview.setReadOnly(True)
preview_layout.addWidget(self.data_preview)
preview_group.setLayout(preview_layout)
inner_layout.addWidget(preview_group)
scroll.setWidget(inner)
self.tab_mapping_layout.addWidget(scroll)
# ================================================================
# 1.3 Tab③：导入配置项（黄标标记规则 / 冲突处理默认策略 / 自动编号，多子页）
# ================================================================
def _build_config_tab(self) -> None:
"""Tab③：导入配置项。多子页 Tab，对应 correction_queue / import_confirm 的配置驱动思路。"""
scroll = QScrollArea()
scroll.setWidgetResizable(True)
inner = QWidget()
inner_layout = QVBoxLayout(inner)
sub = QTabWidget()
sub.addTab(self._build_yellow_flag_page(), "🟡 黄标标记规则")
sub.addTab(self._build_conflict_page(), "⚔️ 冲突处理默认策略")
sub.addTab(self._build_auto_number_page(), "🔢 自动编号")
inner_layout.addWidget(sub)
scroll.setWidget(inner)
self.tab_config_layout.addWidget(scroll)
def _build_yellow_flag_page(self) -> QWidget:
"""🟡 黄标标记规则（导入后进入校正队列的触发条件）。"""
page = QWidget()
layout = QVBoxLayout(page)
prefs = self._user_config.setdefault("preferences", {})
yf = prefs.setdefault("yellow_flag", {})
self.yf_on_dict_miss = QCheckBox("字典未命中时自动黄标（进入校正队列）")
self.yf_on_dict_miss.setChecked(yf.get("on_dict_miss", True))
self.yf_on_dict_miss.stateChanged.connect(self._on_yellow_flag_changed)
layout.addWidget(self.yf_on_dict_miss)
self.yf_on_missing_param = QCheckBox("口径/型号参数缺失时也黄标")
self.yf_on_missing_param.setChecked(yf.get("on_missing_param", False))
self.yf_on_missing_param.stateChanged.connect(self._on_yellow_flag_changed)
layout.addWidget(self.yf_on_missing_param)
layout.addWidget(
QLabel(
"提示：黄标行导入后进入「校正队列 → 黄标校正」，可逐步补字段；"
"补任意字段即触发对应字典自学习并清除黄标。"
)
)
layout.addStretch()
return page
def _build_conflict_page(self) -> QWidget:
"""⚔️ 冲突处理默认策略（无人工决议时的默认动作）。"""
page = QWidget()
layout = QVBoxLayout(page)
prefs = self._user_config.setdefault("preferences", {})
cf = prefs.setdefault("conflict", {})
layout.addWidget(QLabel("数据库冲突默认动作（无人工决议时采用）："))
self.conflict_db_combo = QComboBox()
self.conflict_db_combo.addItems(["跳过", "覆盖", "合并", "隔离", "新建"])
self.conflict_db_combo.setCurrentText(cf.get("default_db_action", "跳过"))
self.conflict_db_combo.currentTextChanged.connect(self._on_conflict_changed)
layout.addWidget(self.conflict_db_combo)
layout.addWidget(QLabel("批次内冲突默认动作："))
self.conflict_intra_combo = QComboBox()
self.conflict_intra_combo.addItems(["跳过", "覆盖", "合并", "隔离", "新建"])
self.conflict_intra_combo.setCurrentText(cf.get("default_intra_action", "跳过"))
self.conflict_intra_combo.currentTextChanged.connect(self._on_conflict_changed)
layout.addWidget(self.conflict_intra_combo)
layout.addStretch()
return page
def _build_auto_number_page(self) -> QWidget:
"""🔢 自动编号开关。"""
page = QWidget()
layout = QVBoxLayout(page)
prefs = self._user_config.setdefault("preferences", {})
an = prefs.setdefault("auto_number", {})
self.auto_number_enabled = QCheckBox("导入时启用自动编号")
self.auto_number_enabled.setChecked(an.get("enabled", False))
self.auto_number_enabled.stateChanged.connect(self._on_auto_number_changed)
layout.addWidget(self.auto_number_enabled)
layout.addStretch()
return page
def _on_yellow_flag_changed(self) -> None:
yf = self._user_config.setdefault("preferences", {}).setdefault("yellow_flag", {})
yf["on_dict_miss"] = self.yf_on_dict_miss.isChecked()
yf["on_missing_param"] = self.yf_on_missing_param.isChecked()
merge_user_config({"preferences": {"yellow_flag": dict(yf)}})
def _on_conflict_changed(self) -> None:
cf = self._user_config.setdefault("preferences", {}).setdefault("conflict", {})
cf["default_db_action"] = self.conflict_db_combo.currentText()
cf["default_intra_action"] = self.conflict_intra_combo.currentText()
merge_user_config({"preferences": {"conflict": dict(cf)}})
def _on_auto_number_changed(self) -> None:
an = self._user_config.setdefault("preferences", {}).setdefault("auto_number", {})
an["enabled"] = self.auto_number_enabled.isChecked()
merge_user_config({"preferences": {"auto_number": dict(an)}})
# ================================================================
# 1.4 导入时间段（按月份筛选）
# ================================================================
def _on_scope_changed(self) -> None:
"""导入范围切换：更新模式、显隐月份/日期范围控件、禁用/启用快速筛选、自动定位工作表。"""
idx = self.scope_combo.currentIndex()
# 反向映射下拉 index → scope 值（与 ui.yaml import_scope.options 顺序一致）
self.scope = next(
(opt for opt, i in self._scope_value_to_index.items() if i == idx),
"current_month",
)
is_single = self.scope == "single_month"
is_bulk = self.scope == "all_months"
self.scope_month_combo.setVisible(is_single)
if self._scope_show_date_range:
self.date_range_row.setVisible(not is_bulk)
self.scope_bulk_note.setVisible(is_bulk)
self._set_quick_filter_enabled(not is_bulk) # 大批量导入禁用快速筛选
if is_single:
self.scope_month = self.scope_month_combo.currentIndex() + 1
self._auto_select_month_sheet(self.scope_month)
elif is_bulk:
# 全部月份表：选首个月份表作列映射预览（真实导入读取全部月份表）
self._apply_scope_sheet_selection()
else: # current_month
# 默认当月欠交整表：自动定位当月表
self._auto_select_month_sheet(date.today().month)
# 单月/当月范围：若已启用日期范围，按所选月份自动填充 [1号, 月末]
self._auto_fill_month_date_range()
self.preview_data() # 重新预览以反映范围
def _on_scope_month_changed(self) -> None:
"""指定月份表的月份下拉变化：记录月份并自动定位对应工作表。"""
self.scope_month = self.scope_month_combo.currentIndex() + 1
self._auto_select_month_sheet(self.scope_month)
# 已启用日期范围时，按所选月份自动填充 [1号, 月末]
self._auto_fill_month_date_range()
self.preview_data()
def _on_date_range_toggled(self, checked: bool) -> None:
"""计划日期范围筛选开关：启用时解锁起止日期编辑，并按所选月份自动填充当月区间。"""
self.date_range_enabled = bool(checked)
self.date_start.setEnabled(checked)
self.date_end.setEnabled(checked)
if checked:
# 按当前所选月份自动填充 [1号, 月末] 闭区间；all_months 下控件隐藏不会触发
self._auto_fill_month_date_range()
else:
self.date_range_start = None
self.date_range_end = None
self.preview_data()
def _on_date_range_changed(self, _date) -> None:
"""起/止日期变化：同步字符串并刷新预览（仅在启用时）。"""
if self.date_range_enabled:
self.date_range_start = self.date_start.date().toString("yyyy-MM-dd")
self.date_range_end = self.date_end.date().toString("yyyy-MM-dd")
self.preview_data()
def _resolve_workbook_year(self) -> int | None:
"""从工作簿文件名提取完整年份（如 2026年销售计划.xlsx → 2026），失败回退 None。
注意：Sorter._extract_year_from_filename 返回的是年份后两位(0-99)，
而日期区间构造需要完整 4 位年份。若直接把后两位当完整年份使用，会被
`_auto_fill_month_date_range` 的下界校验(year<=1752)判定为非法年份，
导致每次都报「工作簿年份非法（低于可表示下限）: 26」且日期区间填不上。
提取合法区间限定在 2000-2100（见 sorter），故后两位 +2000 还原安全。
"""
if not self.file_path:
return None
try:
yy = ImportController.extract_year_from_filename(self.file_path)
if yy is None:
return None
# 后两位 (0-99) → 完整年份 (2000-2099)；2100 边界亦兼容
return 2000 + yy if yy < 100 else yy
except Exception: # noqa: BLE001
return None
def _auto_fill_month_date_range(self) -> None:
"""按所选月份自动填充计划日期范围：[当月1号, 当月最后一天] 闭区间（闰年安全）。
- 触发条件：scope 为 single_month / current_month 且已启用日期范围。
all_months 下日期范围控件已隐藏，不触发。
- 月份：single_month 取 self.scope_month；current_month 取当前月。
- 年份：从工作簿文件名提取（如 2026年销售计划.xlsx → 2026），
提取失败回退当前年。
- 结束取当月真实最后一天（calendar.monthrange 处理闰年，2024-02 → 29）。
与查询页 between [start, end]（同为闭区间）语义对齐，临界无错位。
仅写控件值与起止字符串，不自行刷新预览（由调用方统一 preview_data）。
"""
if self.scope == "all_months" or not self.date_range_enabled:
return
month = self.scope_month if self.scope == "single_month" else date.today().month
if not month or not (1 <= month <= 12):
return
year = self._resolve_workbook_year() or date.today().year
# 年份下界校验：QDateEdit 默认最小可表示日期为 1752-09-14，
# 早于该下限的年份（含 1752 全年，其 1~9 月会被控件静默钳制回退到
# 零日期 1752-09-14）即使能被 QDate 构造为“逻辑有效”日期，setDate
# 时也会被控件钳制回退。因此年份必须 > Qt 最小有效日期所在年份。
if year <= QT_MIN_VALID_DATE[0]:
logger.warning(f"工作簿年份非法（低于可表示下限）: {year}")
QMessageBox.warning(
self,
"日期错误",
f"无法构造有效日期区间：年份 {year} 低于可表示下限 {QT_MIN_VALID_DATE[0]}",
)
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}"
# 安全构造 QDate：解析/构造失败时不会把 QDateEdit 静默回退到
# Qt 零日期 1752-09-14，而是保留原值并提示。
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.blockSignals(True)
self.date_end.blockSignals(True)
self.date_start.setDate(start_qd)
self.date_end.setDate(end_qd)
self.date_start.blockSignals(False)
self.date_end.blockSignals(False)
self.date_range_start = start_str
self.date_range_end = end_str
logger.debug(
f"[_auto_fill_month_date_range] 自动填充月份范围: {start_str} ~ {end_str} (scope={self.scope})"
)
def _set_quick_filter_enabled(self, enabled: bool) -> None:
"""启用/禁用 B0-8 快速筛选组（全部月份表大批量时禁用）。"""
self.range_group.setEnabled(enabled)
def _apply_scope_sheet_selection(self) -> None:
"""根据当前导入范围（scope）自动选择数据源表，替代原工作表下拉的手动选择。
- single_month : 定位指定月份表
- current_month: 定位当月表
- all_months : 选首个月份表用于「列映射预览」（真实导入仍读取全部月份表）
"""
if not self.available_sheets:
return
if self.scope == "single_month" and self.scope_month:
self._auto_select_month_sheet(self.scope_month)
elif self.scope == "current_month":
self._auto_select_month_sheet(date.today().month)
else: # all_months：取首个月份表作列映射预览
month_sheets = self._resolve_month_sheets()
first = month_sheets[0] if month_sheets else self.available_sheets[0]
self._select_sheet(first)
def _select_sheet(self, sheet_name: str) -> None:
"""加载指定工作表（替代原 on_sheet_changed 的 combo 依赖，由导入范围/月份自动驱动）。"""
if not sheet_name or sheet_name not in (self.available_sheets or []):
return
logger.info(f"[_select_sheet] 切换到工作表: {sheet_name}")
self.current_sheet = sheet_name
self._load_config_for_sheet(sheet_name)
self.load_sheet_data(sheet_name)
self._update_file_status() # 刷新文件信息
# 大批量模式：当前表统一显示「全部月份表」（实际导入读取全部月份表）
if self.scope == "all_months":
self.file_current_sheet_label.setText("全部月份表")
def _auto_select_month_sheet(self, month: int) -> None:
"""在可用工作表中匹配指定月份（支持 '7月'/'07'/'2026-07' 等命名），命中即加载。"""
if not self.available_sheets:
return
for name in self.available_sheets:
s = str(name)
if re.search(rf"{month}\s*月", s) or re.search(rf"[-\\/]0?{month}(?![0-9])", s):
self._select_sheet(s)
return
# 未命中：回退到首个工作表，避免空选
logger.warning(
f"[_auto_select_month_sheet] 未找到 {month} 月对应表，回退到 {self.available_sheets[0]}"
)
self._select_sheet(self.available_sheets[0])
def _resolve_month_sheets(self) -> list[str]:
"""从可用工作表中匹配全部 1-12 月欠交表（按名称含月份号），按顺序返回。"""
out: list[str] = []
seen: set[str] = set()
for m in range(1, 13):
for name in self.available_sheets or []:
s = str(name)
if s in seen:
continue
if re.search(rf"{m}\s*月", s) or re.search(rf"[-\\/]0?{m}(?![0-9])", s):
out.append(s)
seen.add(s)
break
return out
def _apply_time_scope_filter(self, records: list[dict]) -> list[dict]:
"""按「导入范围」对清洗后记录做行级筛选。
- all_months：多表导入由服务层逐表处理，视图层不在此过滤（预览仅代表当前表）。
- current_month / single_month 且启用日期范围：仅保留 计划日期 落在 [起,止] 的行。
- 未启用日期范围：返回全量（即该月欠交整表，含历史积压）。
"""
if self.scope == "all_months" or not self.date_range_enabled:
return records
start, end = self.date_range_start, self.date_range_end
if not start or not end:
return records
kept = []
for r in records:
pd = normalize_plan_date(r.get("plan_date"))
if pd and start <= pd <= end:
kept.append(r)
return kept
def _build_all_months_preview_records(self) -> list[dict]:
"""全部月份表模式：依次读取 1-12 月中存在的欠交表，拼接预览记录（限预览上限）。
仅用于预览/冲突概览，不落库；真实导入由服务层逐表执行。
"""
if not self.file_path:
return []
sheets = self._resolve_month_sheets()
combined: list[dict] = []
total = 0
cap = 800
header_row = self.header_row_spin.value()
skip_rows = self.skip_rows_spin.value()
mapping = self.get_current_mapping()
for sh in sheets:
try:
recs = ImportController.read_and_clean_sheet(
file_path=str(self.file_path),
sheet_name=sh,
header_row=header_row,
skip_rows=skip_rows,
column_mapping=mapping,
)
total += len(recs)
remaining = cap - len(combined)
if remaining > 0:
combined.extend(recs[:remaining])
if len(combined) >= cap:
break
except Exception as e: # noqa: BLE001
logger.warning(f"全部月份表预览读取 {sh} 失败(跳过): {e}")
self._all_months_total = total
return combined
# ================================================================
# 2. 配置加载/保存
# ================================================================
[文档]
def load_user_config(self) -> Any:
"""加载用户保存的配置（从 file_sheet_configs 读取）"""
file_sheet_configs = self._user_config.get("file_sheet_configs", {})
file_key = str(self.file_path) if self.file_path else ""
file_config = file_sheet_configs.get(file_key, {})
result = file_config.get(self.current_sheet, {})
logger.debug(
f"[load_user_config] file_path={file_key}, "
f"current_sheet={self.current_sheet}, "
f"file_keys={list(file_config.keys())}, "
f"result_empty={not bool(result)}, "
f"result_keys={list(result.keys()) if result else []}"
)
return result
[文档]
def save_user_config(self, config: dict) -> None:
"""保存用户配置（到 file_sheet_configs）"""
file_sheet_configs = self._user_config.setdefault("file_sheet_configs", {})
file_key = str(self.file_path) if self.file_path else ""
file_config = file_sheet_configs.setdefault(file_key, {})
file_config[self.current_sheet] = config
file_config["_last_sheet"] = self.current_sheet
logger.debug(
f"[save_user_config] file_key={file_key}, "
f"sheet={self.current_sheet}, "
f"config_keys={list(config.keys())}"
)
_save_user_config(self._user_config)
def _should_confirm_overwrite(self) -> bool:
"""检查用户偏好是否要求导入前确认"""
return self._user_config.get("preferences", {}).get("confirm_before_overwrite", True)
def _load_config_for_sheet(self, sheet_name: str) -> bool:
"""加载指定工作表的配置，返回是否加载成功"""
logger.debug(f"[_load_config_for_sheet] 开始加载 sheet={sheet_name}")
self.current_sheet = sheet_name
user_config = self.load_user_config()
if user_config:
hr = user_config.get("header_row", self._header_default)
dsr = user_config.get("data_start_row", self._data_start_default)
sr = user_config.get("skip_rows", 0)
sc = user_config.get("skipped_columns", [])
self.header_row_spin.setValue(hr)
self.data_start_spin.setValue(dsr)
self.skip_rows_spin.setValue(sr)
self._skipped_columns = set(sc)
logger.info(
f"[_load_config_for_sheet] ✅ 加载用户配置: sheet={sheet_name}, "
f"header_row={hr}, data_start_row={dsr}, "
f"skip_rows={sr}, skipped_columns={sc}"
)
return True
self.header_row_spin.setValue(self._header_default)
self.data_start_spin.setValue(self._data_start_default)
self.skip_rows_spin.setValue(0)
self._skipped_columns.clear()
logger.info(f"[_load_config_for_sheet] ⚠️ 使用默认配置: sheet={sheet_name}")
return False
# ================================================================
# 3. 文件操作
# ================================================================
def _populate_recent_files(self) -> None:
"""填充最近文件下拉列表"""
self._populating_recent = True
self.recent_file_combo.clear()
recent_files = get_recent_import_files(max_count=10)
if recent_files:
self.recent_file_combo.addItem("-- 选择历史文件 --")
for rf in recent_files:
path_str = rf.get("path", "")
name = rf.get("name", Path(path_str).name if path_str else "")
display = f"{name} ({path_str})" if path_str else name
self.recent_file_combo.addItem(display, path_str)
else:
self.recent_file_combo.addItem("-- 没有历史文件，请浏览 --")
self._populating_recent = False
logger.info(f"[_populate_recent_files] 加载了 {len(recent_files)} 个最近文件")
def _clear_recent_history(self) -> None:
"""清除最近文件历史"""
reply = QMessageBox.question(
self,
"确认清除",
"确定要清除所有最近文件历史吗？",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply == QMessageBox.StandardButton.Yes:
clear_recent_files()
self._populate_recent_files()
self.recent_file_combo.setCurrentIndex(0)
logger.info("已清除最近文件历史")
def _load_first_recent_file(self) -> None:
"""自动加载最近文件列表中第一个存在的文件"""
for i in range(1, self.recent_file_combo.count()):
path_str = self.recent_file_combo.itemData(i)
if path_str and Path(path_str).exists():
logger.info(f"[_load_first_recent_file] 自动加载: {path_str}")
self.recent_file_combo.setCurrentIndex(i)
# setCurrentIndex 会触发 _on_recent_file_changed
return
logger.info("[_load_first_recent_file] 没有找到可用的最近文件")
def _on_recent_file_changed(self, index: int) -> None:
"""最近文件下拉切换事件"""
if self._populating_recent or index < 0:
return
# index=0 是占位文字
if index == 0:
return
path_str = self.recent_file_combo.itemData(index)
if not path_str:
return
p = Path(path_str)
if not p.exists():
QMessageBox.warning(
self, "文件不存在", f"文件已不存在:\n{path_str}\n\n请浏览选择其他文件。"
)
self._populate_recent_files()
return
self._set_file_path(p)
def _set_file_path(self, path: Path) -> None:
"""设置文件路径并加载（统一入口）"""
logger.info(f"[_set_file_path] 设置文件: {path}")
self.file_path = path
self._update_file_status()
self.load_excel_info()
# 更新最近文件列表（把当前文件置顶）
self._populate_recent_files()
# 选中当前文件项
for i in range(1, self.recent_file_combo.count()):
if self.recent_file_combo.itemData(i) == str(path):
self._populating_recent = True
self.recent_file_combo.setCurrentIndex(i)
self._populating_recent = False
break
def _update_file_status(self) -> None:
"""更新文件信息显示"""
if self.file_path and self.file_path.exists():
size = self.file_path.stat().st_size
if size > 1024 * 1024:
size_str = f"{size / (1024 * 1024):.2f} MB"
else:
size_str = f"{size / 1024:.2f} KB"
self.file_size_label.setText(size_str)
self.file_sheets_label.setText(f"{len(self.available_sheets)} 个工作表")
self.file_current_sheet_label.setText(self.current_sheet or "—")
self.setWindowTitle(f"销售计划导入 - {self.file_path.name}")
else:
self.file_size_label.setText("未选择文件")
self.file_sheets_label.setText("未知")
self.file_current_sheet_label.setText("—")
[文档]
def browse_file(self) -> None:
"""浏览选择Excel文件"""
# 默认目录：优先「上次文件所在目录」；首次/无历史文件时回退销售计划配置目录。
# paths.sales_plan.default_dir（可由 paths.local.yaml 覆盖）缺失时，
# 回退 vba_import.sales_plan_files.primary 所在目录（见讨论稿 §12/§13）。
default_dir = cfg("paths.sales_plan.default_dir") or os.path.dirname(
cfg("vba_import.sales_plan_files.primary", "")
)
start_dir = default_dir
if self.file_path and self.file_path.parent.exists():
start_dir = str(self.file_path.parent)
file_path, _ = QFileDialog.getOpenFileName(
self,
"选择销售计划Excel文件",
start_dir,
"Excel文件 (*.xlsx *.xls);;所有文件 (*.*)",
)
if not file_path:
return
self._set_file_path(Path(file_path))
logger.info(f"浏览选择文件: {file_path}")
[文档]
def load_excel_info(self) -> None:
"""加载Excel文件信息（仅读取工作表清单；实际数据源表由「导入范围」驱动选择）。"""
if not self.file_path or not self.file_path.exists():
logger.debug(f"[load_excel_info] 跳过: file_path={self.file_path}, exists=False")
return
logger.info(f"[load_excel_info] 开始加载: {self.file_path}")
self._loading_config = True
try:
excel_file = pd.ExcelFile(self.file_path)
self.available_sheets = excel_file.sheet_names
logger.debug(f"[load_excel_info] 工作表列表: {self.available_sheets}")
self._update_file_status()
# 数据源表由「导入范围」(scope) 统一驱动：current/single 自动定位月份表，
# all_months 取首个月份表用于列映射预览。不再依赖独立的工作表下拉。
self._apply_scope_sheet_selection()
self._loading_config = False
except Exception as e:
self._loading_config = False
logger.error(f"加载Excel信息失败: {e}")
QMessageBox.warning(self, "错误", f"无法读取Excel文件: {e}")
[文档]
def load_sheet_data(self, sheet_name: str) -> None:
"""加载指定工作表的数据"""
if not self.file_path or not sheet_name:
return
try:
self.df_raw = pd.read_excel(
self.file_path, sheet_name=sheet_name, header=None, dtype=str
)
logger.debug(
f"[load_sheet_data] 加载完成: sheet={sheet_name}, shape={self.df_raw.shape}"
)
self.update_preview()
if self.range_mode == "order":
# 数据已变，重建去重订单 + 级联 + 快查 + 匹配计数
self._order_tuples = self._build_order_tuples()
for lvl in range(len(self._ORDER_FIELDS)):
self._repopulate_filter_combo(lvl)
self._populate_quick_search()
self._refresh_order_status()
except Exception as e:
logger.error(f"加载工作表失败: {e}")
QMessageBox.warning(self, "错误", f"加载工作表失败: {e}")
# ================================================================
# 4. 预览表格
# ================================================================
[文档]
def update_preview(self) -> None:
"""更新列映射预览
根据当前配置刷新预览表格的显示内容。
"""
if self.df_raw is None:
return
self._updating_checkboxes = True
header_row = self.header_row_spin.value()
data_start_row = self.data_start_spin.value()
skip_rows = self.skip_rows_spin.value()
effective_data_start = data_start_row + skip_rows
logger.debug(
f"[update_preview] header_row={header_row}, "
f"data_start_row={data_start_row}, skip_rows={skip_rows}, "
f"skipped_columns_count={len(self._skipped_columns)}"
)
header_data = self._get_header_data(header_row)
sample_data = self._get_sample_data(effective_data_start)
# 清空表格
self.preview_table.clearContents()
self.preview_table.setRowCount(len(header_data))
# 清理旧的 cell widget(避免重复)
for row in range(self.preview_table.rowCount()):
for col in range(self.preview_table.columnCount()):
old_widget = self.preview_table.cellWidget(row, col)
if old_widget:
old_widget.deleteLater()
for col_idx, col_name in enumerate(header_data):
self._fill_preview_row(col_idx, col_name, sample_data)
self.preview_table.resizeColumnsToContents()
self._updating_checkboxes = False
def _get_header_data(self, header_row: int) -> list[str]:
"""获取表头数据"""
if header_row < len(self.df_raw):
return self.df_raw.iloc[header_row].fillna("").astype(str).tolist()
return [f"列{i}" for i in range(len(self.df_raw.columns))]
def _get_sample_data(self, effective_data_start: int) -> list[list[str]]:
"""获取示例数据行"""
sample_data = []
for i in range(effective_data_start, min(effective_data_start + 3, len(self.df_raw))):
if i < len(self.df_raw):
sample_data.append(self.df_raw.iloc[i].fillna("").astype(str).tolist())
return sample_data
def _fill_preview_row(self, col_idx: int, col_name: str, sample_data: list[list[str]]) -> None:
"""填充预览表格的一行"""
if not col_name or col_name == "nan":
col_name = f"列{col_idx + 1}"
# 解析映射状态并缓存
target_field, is_mapped = self._resolve_column_mapping(col_name)
self._column_mapped_status[col_name] = is_mapped
# 检查是否被跳过
is_skipped = col_name in self._skipped_columns
if is_skipped:
target_field = "-"
status = "⏭️ 跳过"
elif is_mapped:
status = "✅ 已映射"
else:
target_field = "-"
status = "未配置"
sample_value = ""
if sample_data and col_idx < len(sample_data[0]):
sample_value = sample_data[0][col_idx] if sample_data else ""
self._add_preview_widgets(col_idx, col_name, target_field, status, sample_value, is_skipped)
def _add_preview_widgets(
self,
col_idx: int,
col_name: str,
target_field: str,
status: str,
sample_value: str,
is_skipped: bool,
) -> None:
"""添加预览表格的控件"""
# 复选框(是否导入)
checkbox = QCheckBox()
checkbox.setChecked(not is_skipped)
checkbox.stateChanged.connect(
lambda state, row=col_idx, col=col_name: self.on_checkbox_changed(row, col, state)
)
self.preview_table.setCellWidget(col_idx, 0, checkbox)
# Excel列名
self.preview_table.setItem(col_idx, 1, QTableWidgetItem(col_name))
# 目标字段(只读)
target_item = QTableWidgetItem(target_field or "-")
target_item.setFlags(target_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
self.preview_table.setItem(col_idx, 2, target_item)
# 状态(带颜色)
status_item = QTableWidgetItem(status)
status_item.setFlags(status_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
self._set_status_color(status_item, status)
self.preview_table.setItem(col_idx, 3, status_item)
# 示例数据
self.preview_table.setItem(col_idx, 4, QTableWidgetItem(sample_value[:50]))
@staticmethod
def _set_status_color(item: QTableWidgetItem, status: str) -> None:
"""根据状态设置背景颜色"""
if "✅" in status:
item.setBackground(QColor(200, 230, 200))
elif "⏭️" in status:
item.setBackground(QColor(255, 200, 200))
else:
item.setBackground(QColor(255, 255, 200))
# ================================================================
# 5. 列映射
# ================================================================
def _resolve_column_mapping(self, col_name: str) -> tuple:
"""解析列的映射状态
Args:
col_name: Excel列名
Returns:
tuple: (目标字段名或None, 是否已映射)
"""
direct_mapping = SALE_PLAN_COLUMNS
alias_mapping = COLUMN_ALIASES
if col_name in direct_mapping:
return direct_mapping[col_name], True
for target, aliases in alias_mapping.items():
if col_name in aliases:
return target, True
return None, False
[文档]
def get_current_mapping(self) -> dict[str, str]:
"""获取当前列映射(只包含选中的列)
Returns:
Dict[str, str]: Excel列名到目标字段的映射字典
"""
mapping = {}
for row in range(self.preview_table.rowCount()):
checkbox = self.preview_table.cellWidget(row, 0)
if checkbox and checkbox.isChecked():
excel_col = self.preview_table.item(row, 1).text()
target_field = self.preview_table.item(row, 2).text()
if target_field and target_field != "-" and excel_col:
mapping[excel_col] = target_field
return mapping
# ================================================================
# 6. 跳过列操作
# ================================================================
def _save_skipped_columns(self) -> None:
"""保存跳过列到用户配置（使用新的 save_user_config）"""
# 收集当前配置
config = {
"header_row": self.header_row_spin.value(),
"data_start_row": self.data_start_spin.value(),
"skip_rows": self.skip_rows_spin.value(),
"skipped_columns": list(self._skipped_columns),
"column_mapping": self.get_current_mapping(),
"last_used": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
self.save_user_config(config)
logger.debug(f"已保存配置: {self.current_sheet}")
def _apply_config_to_all_months(self) -> None:
"""将当前工作表的解析配置复制到本工作簿所有其他工作表
解决同一工作簿多个月表列结构相同、但跳过列/列映射按 sheet 独立存储
导致"2月漏列"的问题: 配好 1月 后一键复制到 2-12 月, 预览标记也一致。
"""
if not self.file_path:
QMessageBox.information(self, "提示", "请先选择并加载 Excel 文件。")
return
if not self.available_sheets:
QMessageBox.information(self, "提示", "尚未读取到工作表列表。")
return
# 取当前 UI 实时状态(即使尚未"保存"当前表也生效)
current = {
"header_row": self.header_row_spin.value(),
"data_start_row": self.data_start_spin.value(),
"skip_rows": self.skip_rows_spin.value(),
"skipped_columns": sorted(self._skipped_columns),
"column_mapping": dict(self.get_current_mapping()),
}
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
file_key = str(self.file_path)
file_sheet_configs = self._user_config.setdefault("file_sheet_configs", {})
file_config = file_sheet_configs.setdefault(file_key, {})
# 仅匹配"数字+月"的月份表(如 1月/2月/12月), 排除"2026年不计价"等非月份表
month_pattern = re.compile(r"^\d+月$")
copied: list[str] = []
for sheet in self.available_sheets:
if sheet == self.current_sheet or not month_pattern.match(sheet):
continue
entry = file_config.get(sheet, {})
entry.update(
{
"header_row": current["header_row"],
"data_start_row": current["data_start_row"],
"skip_rows": current["skip_rows"],
"skipped_columns": list(current["skipped_columns"]),
"column_mapping": dict(current["column_mapping"]),
"last_used": now,
}
)
file_config[sheet] = entry
copied.append(sheet)
_save_user_config(self._user_config)
QMessageBox.information(
self,
"已应用",
f"已将「{self.current_sheet}」的解析配置复制到 {len(copied)} 个工作表:\n"
+ "、".join(copied),
)
logger.info(f"[_apply_config_to_all_months] 已复制配置到 {len(copied)} 个表: {copied}")
[文档]
def skip_unmapped_columns(self) -> None:
"""跳过所有未配置的列"""
self._updating_checkboxes = True
for row in range(self.preview_table.rowCount()):
status_item = self.preview_table.item(row, 3)
if status_item and "未配置" in status_item.text():
col_name = self.preview_table.item(row, 1).text()
self._skipped_columns.add(col_name)
self._updating_checkboxes = False
self.update_preview()
self._save_skipped_columns()
logger.info("已跳过所有未配置列并保存配置")
[文档]
def reset_skipped_columns(self) -> None:
"""重置所有跳过列"""
self._skipped_columns.clear()
self.update_preview()
self._save_skipped_columns()
logger.info("已重置所有跳过列并保存配置")
[文档]
def select_all(self) -> None:
"""全选所有列
清空跳过列，恢复所有列的原始映射状态。
"""
self._updating_checkboxes = True
self._skipped_columns.clear()
self._updating_checkboxes = False
self.update_preview()
[文档]
def deselect_all(self) -> None:
"""取消全选
跳过所有列，将所有列标记为不导入。
"""
self._updating_checkboxes = True
for row in range(self.preview_table.rowCount()):
col_name_item = self.preview_table.item(row, 1)
if col_name_item:
self._skipped_columns.add(col_name_item.text())
self._updating_checkboxes = False
self.update_preview()
[文档]
def clear_saved_config(self) -> None:
"""清除当前文件保存的配置
清除跳过的列配置，恢复为默认状态。
"""
reply = QMessageBox.question(
self,
"确认清除",
f"确定要清除文件 '{self.file_path.name}' 的保存配置吗?\n下次导入时将使用默认配置。",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply == QMessageBox.StandardButton.Yes:
self._skipped_columns.clear()
self.update_preview()
QMessageBox.information(self, "完成", "已清除保存的配置")
# ================================================================
# 7. 复选框事件
# ================================================================
[文档]
def on_checkbox_changed(self, row: int, col_name: str, state: int) -> None:
"""复选框状态改变处理
三种状态的处理逻辑：
- 已配置(✅): 列有目标映射 + 复选框选中 → 导入时使用该映射
- 未配置(⚠️): 列无目标映射 + 复选框选中 → 导入但不映射(保留原列名)
- 跳过(⏭️): 复选框未选中 → 导入时忽略该列
转换规则：
- 从未选中→选中: 从跳过恢复为原始状态(已配置或未配置)
- 从选中→未选中: 标记为跳过
Args:
row: 行索引
col_name: 列名
state: 复选框状态
"""
if self._updating_checkboxes:
return
logger.debug(
f"复选框状态改变: row={row}, col={col_name}, state={state}, "
f"当前跳过列={self._skipped_columns}"
)
is_checked = state == Qt.CheckState.Checked.value
if is_checked:
# 选中 = 取消跳过，恢复原始映射状态
if col_name in self._skipped_columns:
self._skipped_columns.discard(col_name)
logger.debug(f"从跳过列中移除: {col_name}, 剩余跳过列={self._skipped_columns}")
self._restore_column_mapping(row, col_name)
else:
logger.debug(f"列 {col_name} 不在跳过列中，无需恢复")
else:
# 未选中 = 跳过此列
if col_name not in self._skipped_columns:
self._skipped_columns.add(col_name)
logger.debug(f"添加到跳过列: {col_name}, 跳过列={self._skipped_columns}")
self._mark_as_skipped(row, col_name)
# 自动保存到配置
self._save_skipped_columns()
logger.debug(f"保存配置后跳过列: {list(self._skipped_columns)}")
def _mark_as_skipped(self, row: int, _col_name: str) -> None:
"""将列标记为跳过状态
Args:
row: 行索引
_col_name: 列名(用于调试)
"""
# 目标字段
target_item = QTableWidgetItem("-")
target_item.setFlags(target_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
self.preview_table.setItem(row, 2, target_item)
# 状态
status_item = QTableWidgetItem("⏭️ 跳过")
status_item.setFlags(status_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
status_item.setBackground(QColor(255, 200, 200))
status_item.setForeground(Qt.GlobalColor.black)
self.preview_table.setItem(row, 3, status_item)
def _restore_column_mapping(self, row: int, col_name: str) -> None:
"""根据缓存的映射状态恢复列显示
Args:
row: 行索引
col_name: 列名
"""
is_mapped = self._column_mapped_status.get(col_name, False)
target_field, _ = self._resolve_column_mapping(col_name)
# 目标字段
target_item = QTableWidgetItem(target_field or "-")
target_item.setFlags(target_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
self.preview_table.setItem(row, 2, target_item)
# 状态
if is_mapped:
status = "✅ 已映射"
bg_color = QColor(200, 230, 200)
else:
status = "未配置"
bg_color = QColor(255, 255, 200)
status_item = QTableWidgetItem(status)
status_item.setFlags(status_item.flags() & ~Qt.ItemFlag.ItemIsEditable)
status_item.setBackground(bg_color)
status_item.setForeground(Qt.GlobalColor.black)
self.preview_table.setItem(row, 3, status_item)
logger.debug(f"恢复列映射: {col_name} -> {target_field}, status={status}")
def _rematch_column(self, row: int, col_name: str) -> None:
"""重新匹配列(保留用于兼容)
Args:
row: 行索引
col_name: 列名
"""
logger.debug(f"重新匹配列: row={row}, col_name={col_name}")
self._restore_column_mapping(row, col_name)
# ================================================================
# 8. 工作表/配置变更事件
# ================================================================
[文档]
def on_preview_changed(self) -> None:
"""配置变更事件
当表头行、数据起始行或跳过行数改变时更新预览。
"""
self.update_preview()
# ================================================================
# 9. 预览/自动检测
# ================================================================
[文档]
def preview_data(self) -> None:
"""预览数据
显示当前配置下即将导入的数据预览。
"""
if self.df_raw is None:
return
header_row = self.header_row_spin.value()
data_start_row = self.data_start_spin.value()
skip_rows = self.skip_rows_spin.value()
effective_data_start = data_start_row + skip_rows
preview_text = []
preview_text.append("=" * 80)
preview_text.append("配置参数:")
preview_text.append(f" 工作表: {self.current_sheet or '—'}")
preview_text.append(f" 表头行: {header_row} (第{header_row + 1}行)")
preview_text.append(
f" 数据起始行: {effective_data_start} (第{effective_data_start + 1}行)"
)
preview_text.append(f" 跳过顶部行数: {skip_rows}")
preview_text.append("=" * 80)
preview_text.append("")
if header_row < len(self.df_raw):
headers = self.df_raw.iloc[header_row].fillna("").astype(str).tolist()
headers = [h for h in headers if h and h != "nan"]
preview_text.append("检测到的表头列:")
preview_text.append(f" {', '.join(headers[:15])}")
if len(headers) > 15:
preview_text.append(f" ... 还有 {len(headers) - 15} 列")
preview_text.append("")
preview_text.append("数据预览 (前10行):")
preview_text.append("-" * 80)
for i in range(effective_data_start, min(effective_data_start + 10, len(self.df_raw))):
if i < len(self.df_raw):
row_data = self.df_raw.iloc[i].fillna("").astype(str).tolist()
non_empty = [d for d in row_data[:8] if d and d != "nan"]
preview_text.append(f" 第{i + 1}行: {' | '.join(non_empty[:5])}...")
self.data_preview.setText("\n".join(preview_text))
[文档]
def auto_detect(self) -> None:
"""自动检测配置
根据关键词自动检测表头行位置。
"""
if self.df_raw is None:
return
keywords = ["订货单位", "产品名称", "型号", "数量", "规格"]
detected_row = ImportController.detect_header_row(self.df_raw, keywords, max_rows=20)
if detected_row is not None:
self.header_row_spin.setValue(detected_row)
self.data_start_spin.setValue(detected_row + 1)
QMessageBox.information(
self,
"检测完成",
f"自动检测到表头在第 {detected_row + 1} 行\n数据起始行建议设为 {detected_row + 2} 行",
)
else:
QMessageBox.information(self, "提示", "未能自动检测表头行，请手动设置")
self.update_preview()
self.preview_data()
# ================================================================
# 10. 验证/确认
# ================================================================
[文档]
def validate_config(self, config_params: dict[str, Any]) -> bool:
"""验证配置参数
Args:
config_params: 配置参数字典
Returns:
bool: 配置是否有效
"""
required_fields = REQUIRED_FIELDS
mapped_fields = set(config_params["column_mapping"].values())
missing_fields = [f for f in required_fields if f not in mapped_fields]
if missing_fields:
field_names = {
"product_model": "产品型号",
"order_no": "订单号",
"quantity": "数量",
"customer": "客户",
"specification": "规格",
"material": "材质",
"sales_order_no": "销售订单号",
"production_order_no": "生产令号",
}
missing_names = [field_names.get(f, f) for f in missing_fields]
QMessageBox.warning(
self,
"配置不完整",
f"缺少必需的列映射:\n{', '.join(missing_names)}\n\n请确保这些列已勾选且正确映射。",
)
return False
return True
def _goto_confirm(self) -> None:
"""校验配置 + 计算预览/冲突，打开「确认」对话框（ImportConfirmDialog）。
导入执行与冲突决议均移至 ImportConfirmDialog；导入成功后将
config_params 经 config_confirmed 信号转发给主窗口刷新。
"""
if not self.file_path or not self.file_path.exists():
QMessageBox.warning(self, "提示", "请先选择有效的Excel文件")
return
sheet_name = self.current_sheet or ""
if not sheet_name:
QMessageBox.warning(self, "提示", "请选择要导入的工作表")
return
config_params = self._build_config_params(sheet_name)
if config_params is None:
return
# 计算预览记录 + 冲突（与真实导入同清洗管线）
if self.scope == "all_months":
# 大批量：拼接各欠交表预览；跨表重复由导入时默认冲突策略处理，不做逐表冲突扫描
records = self._build_all_months_preview_records()
if not records:
QMessageBox.warning(self, "提示", "未找到任何月份欠交表，请检查工作簿结构")
return
groups: dict[str, int] = {}
intra_conflicts: list[str] = []
db_conflicts: list[tuple[str, dict, str]] = []
config_params["_all_months_total"] = self._all_months_total
else:
records = self._build_preview_records()
if not records:
QMessageBox.warning(self, "提示", "当前范围无可导出的数据行，请调整导入范围")
return
groups = self._compute_groups(records)
intra_conflicts, db_conflicts = self._scan_conflicts(records)
from certflow.views.import_confirm_dialog import (
ImportConfirmDialog,
should_skip_confirm_dialog,
)
# 0 冲突直接导入（配置驱动，蓝图 §4.1.3）：冲突扫描均为空时跳过确认框
if should_skip_confirm_dialog(db_conflicts, intra_conflicts, SKIP_CONFIRM_WHEN_NO_CONFLICT):
self._run_direct_import(config_params)
return
dlg = ImportConfirmDialog(
config_params=config_params,
preview_records=records,
db_conflicts=db_conflicts,
intra_conflicts=intra_conflicts,
order_groups=len(groups),
parent=self,
db_manager=self.db_manager,
session=self._session,
)
dlg.exec()
if dlg.result_config_params is not None:
self._save_import_config(config_params)
self.config_confirmed.emit(dlg.result_config_params)
dlg.deleteLater()
def _run_direct_import(self, config_params: dict) -> None:
"""0 冲突时跳过确认对话框，直接执行导入并弹一次完成提示。
复用 import_confirm_dialog.run_import，与确认路径落库完全一致；
完成后保存导入配置并转发 config_confirmed 信号给主窗口刷新。
"""
from certflow.views.import_confirm_dialog import (
ImportConfirmDialog,
run_import,
)
try:
result = run_import(self._session, config_params)
except Exception as e:
logger.error(f"直接导入失败: {e}")
QMessageBox.critical(self, "导入失败", f"导入过程中发生错误:\n{e}")
return
msg = ImportConfirmDialog._build_import_result_message(result)
QMessageBox.information(self, "导入完成", msg)
self._save_import_config(config_params)
self.config_confirmed.emit(config_params)
def _build_config_params(self, sheet_name: str) -> dict | None:
"""构建并验证配置参数（不含 conflict_resolutions，由确认对话框补充）。"""
config_params = {
"file_path": str(self.file_path),
"sheet_name": sheet_name,
"header_row": self.header_row_spin.value(),
"data_start_row": self.data_start_spin.value(),
"skip_rows": self.skip_rows_spin.value(),
"column_mapping": self.get_current_mapping(),
}
# B0-8：导入范围（冲突决议在确认对话框内收集）
# 全部月份表（大批量）→ 禁用快速筛选，不附加 range_filter/selected_rows
if self.scope != "all_months":
if self.range_mode == "order":
config_params["range_filter"] = self._collect_order_filter()
elif self.range_mode == "rows":
config_params["selected_rows"] = self._collect_selected_rows()
# 导入范围（按月份的三层层次）
config_params["scope"] = self.scope
if self.scope == "single_month":
config_params["scope_month"] = self.scope_month
if self.scope in ("current_month", "single_month") and self.date_range_enabled:
config_params["date_range"] = {
"enabled": True,
"start": self.date_range_start,
"end": self.date_range_end,
}
if not self.validate_config(config_params):
return None
return config_params
def _save_import_config(self, config_params: dict) -> None:
"""保存导入配置到用户设置"""
sheet_config = {
"header_row": config_params["header_row"],
"data_start_row": config_params["data_start_row"],
"skip_rows": config_params["skip_rows"],
"skipped_columns": list(self._skipped_columns),
"column_mapping": config_params["column_mapping"],
"last_used": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
self.save_user_config(sheet_config)
add_recent_import_file(str(self.file_path), config_params)
# 导入执行已移至 ImportConfirmDialog（确认对话框）；本对话框仅负责「设置」
def _set_ui_enabled(self, enabled: bool) -> None:
"""设置导入期间 UI 控件状态"""
self.recent_file_combo.setEnabled(enabled)
self.browse_btn.setEnabled(enabled)
self.clear_history_btn.setEnabled(enabled)
self.scope_combo.setEnabled(enabled)
self.scope_month_combo.setEnabled(enabled)
self.header_row_spin.setEnabled(enabled)
self.data_start_spin.setEnabled(enabled)
self.skip_rows_spin.setEnabled(enabled)
self.auto_detect_btn.setEnabled(enabled)
self.preview_btn.setEnabled(enabled)
self.clear_memory_btn.setEnabled(enabled)
self.skip_unmapped_btn.setEnabled(enabled)
self.reset_skip_btn.setEnabled(enabled)
self.select_all_btn.setEnabled(enabled)
self.deselect_all_btn.setEnabled(enabled)
self.cancel_btn.setEnabled(enabled)
self.range_combo.setEnabled(enabled)
self.filter_widget.setEnabled(enabled)
self.rows_widget.setEnabled(enabled)
# confirm_btn 由调用方单独控制
# ================================================================
# 11. B0-8 导入范围 / 冲突决议
# ================================================================
def _on_range_mode_changed(self) -> None:
"""导入范围模式切换：显示/隐藏对应子控件，并按需构建级联下拉。"""
idx = self.range_combo.currentIndex()
self.range_mode = ["all", "order", "rows"][idx] if 0 <= idx < 3 else "all"
self.filter_widget.setVisible(self.range_mode == "order")
self.order_quick_search.setVisible(self.range_mode == "order")
self.next_step_hint.setVisible(self.range_mode == "order")
self.rows_widget.setVisible(self.range_mode == "rows")
if self.range_mode == "order":
# 构建去重唯一订单（对齐 VBA：去重 4 字段 → 唯一订单），初始化级联下拉
self._order_tuples = self._build_order_tuples()
for lvl in range(len(self._ORDER_FIELDS)):
self._repopulate_filter_combo(lvl)
self._populate_quick_search()
self._refresh_order_status()
elif self.range_mode == "rows":
self._populate_row_table()
def _collect_order_filter(self) -> dict[str, str]:
"""读取「按订单筛选」级联下拉的 4 字段值（仅非「全部」）。"""
flt: dict[str, str] = {}
for _cn, en in self._ORDER_FIELDS:
v = self.filter_combos[en].currentText().strip()
if v and v != "（全部）":
flt[en] = v
return flt
def _build_order_tuples(self) -> list[dict]:
"""从已清洗记录抽取 4 字段去重元组（对齐 VBA：去重唯一订单）。
值取自与真实导入同一清洗管线，保证下拉取值与 apply_range_filter 精确匹配。
"""
records = self._build_preview_records(apply_range=False)
self._all_clean_records = records
tuples: list[dict] = []
seen: set[tuple] = set()
for rec in records:
t = tuple(
str(rec.get(f, "") or "").strip()
for f in ("plan_date", "customer", "project_name", "plan_no")
)
if t in seen or not any(t):
continue
seen.add(t)
tuples.append(
{"plan_date": t[0], "customer": t[1], "project_name": t[2], "plan_no": t[3]}
)
return tuples
def _rebuild_clean_records(self) -> None:
"""仅重建清洗记录缓存（不触碰级联下拉），用于加载/切表后刷新匹配计数。"""
self._all_clean_records = self._build_preview_records(apply_range=False)
def _populate_quick_search(self) -> None:
"""用去重订单填充快查框候选项（复用 QuickSearch.set_items：editable + 模糊匹配）。"""
items = [
f"{t['plan_date']} | {t['customer']} | {t['project_name']} | {t['plan_no']}"
for t in self._order_tuples
]
self.order_quick_search.set_items(items)
def _on_quick_search_activated(self, text) -> None:
"""选中快查项：自动设 4 级联下拉定位该订单（级联保证下游可选项存在）。"""
text = str(text) if text is not None else ""
if not text or " | " not in text:
return
parts = [p.strip() for p in text.split(" | ")]
if len(parts) != 4:
return
for level, val in enumerate(parts):
en = self._ORDER_FIELDS[level][1]
combo = self.filter_combos[en]
idx = combo.findText(val) if val else -1
if idx >= 0:
combo.setCurrentIndex(idx) # 触发 _on_filter_combo_changed 级联
else:
combo.setCurrentText("（全部）")
self._refresh_order_status()
def _filter_order_tuples(self, upto: int) -> list[dict]:
"""返回满足前 upto 个字段已选约束的元组子集（upto=0 表示无约束）。"""
conds: dict[str, str] = {}
for i in range(upto):
en = self._ORDER_FIELDS[i][1]
val = self.filter_combos[en].currentText()
if val and val != "（全部）":
conds[en] = val
if not conds:
return self._order_tuples
return [t for t in self._order_tuples if all(t[en] == v for en, v in conds.items())]
def _repopulate_filter_combo(self, level: int) -> None:
"""按上游已选值，刷新第 level 个级联下拉的可选项。"""
en = self._ORDER_FIELDS[level][1]
combo = self.filter_combos[en]
vals = sorted({t[en] for t in self._filter_order_tuples(level)})
combo.blockSignals(True)
combo.clear()
combo.addItem("（全部）")
combo.addItems(vals)
combo.blockSignals(False)
def _on_filter_combo_changed(self, level: int) -> None:
"""某级联下拉变化时：刷新其下游下拉 + 更新结果指示。"""
for lvl in range(level + 1, len(self._ORDER_FIELDS)):
self._repopulate_filter_combo(lvl)
self._refresh_order_status()
def _count_matched_rows(self) -> int:
"""统计当前级联筛选命中的实际数据行数（按清洗记录精确过滤）。"""
if not self._all_clean_records:
if self.df_raw is not None:
self._rebuild_clean_records()
else:
return 0
flt = self._collect_order_filter()
return len(ImportController.apply_range_filter(self._all_clean_records, flt, None))
def _refresh_order_status(self) -> None:
"""更新级联筛选结果指示（对齐 VBA：汇总表每行4数据保证级联成功）。"""
sel = [self.filter_combos[en].currentText() for _cn, en in self._ORDER_FIELDS]
matched = self._count_matched_rows()
parts = [
f"{cn}={v if v != '（全部）' else '全部'}"
for (cn, _en), v in zip(self._ORDER_FIELDS, sel, strict=True)
]
self.order_status_label.setText("当前筛选: " + " / ".join(parts) + f" → 匹配 {matched} 行")
# 下一步提示：明确「看冲突」的入口就是底部按钮
if matched > 0:
self.next_step_hint.setText(
f"✅ 已匹配 {matched} 行 → 点击底部「下一步：查看冲突 →」查看冲突并确认导入"
)
else:
self.next_step_hint.setText(
"⚠ 未匹配到数据行；可调整筛选，或仍点「下一步：查看冲突 →」查看整表冲突"
)
def _collect_selected_rows(self) -> list[int]:
"""读取「勾选行」模式下被勾选的数据行序号（0-based）。"""
rows: list[int] = []
for r in range(self.row_table.rowCount()):
cb = self.row_table.cellWidget(r, 0)
if cb and cb.isChecked():
rows.append(r)
return rows
def _populate_row_table(self) -> None:
"""载入数据行到勾选表（默认全选）。"""
if self.df_raw is None:
return
header_row = self.header_row_spin.value()
data_start = self.data_start_spin.value() + self.skip_rows_spin.value()
try:
headers = self._get_header_data(header_row)
except Exception:
headers = []
self.row_table.setRowCount(0)
max_rows = min(data_start + 500, len(self.df_raw))
for i in range(data_start, max_rows):
if i >= len(self.df_raw):
break
row_data = self.df_raw.iloc[i].fillna("").astype(str).tolist()
self.row_table.insertRow(self.row_table.rowCount())
ridx = self.row_table.rowCount() - 1
cb = QCheckBox()
cb.setChecked(True)
self.row_table.setCellWidget(ridx, 0, cb)
self.row_table.setItem(ridx, 1, QTableWidgetItem(str(i - data_start)))
# 计划日期/订货单位/项目名称/产品型号 列索引（按表头匹配英文/中文）
for col, en in (
(2, "plan_date"),
(3, "customer"),
(4, "project_name"),
(5, "product_model"),
):
val = self._cell_by_field(row_data, headers, en)
self.row_table.setItem(ridx, col, QTableWidgetItem(val[:30]))
self.row_table.resizeColumnsToContents()
@staticmethod
def _cell_by_field(row_data: list[str], headers: list[str], en_field: str) -> str:
"""按英文/中文字段名（含别名）从一行数据中取值。"""
cn_name = {
"plan_date": "计划日期",
"customer": "订货单位",
"project_name": "项目名称",
"product_model": "产品型号",
}.get(en_field, en_field)
raw_aliases = COLUMN_ALIASES.get(en_field)
if isinstance(raw_aliases, (list, tuple, set)):
alias_set = set(raw_aliases)
elif raw_aliases:
alias_set = {raw_aliases}
else:
alias_set = set()
for idx, h in enumerate(headers):
if idx < len(row_data) and (h in (en_field, cn_name) or h in alias_set):
return row_data[idx]
return ""
def _build_preview_records(self, apply_range: bool = True) -> list[dict]:
"""构建预览用的已清洗记录（复用无格式导入管线，确保 unique_key 与真实导入一致）。
仅用于分组/冲突预览，不落库。读取路径与 `SalePlanService.import_from_excel_with_config`
完全一致：经 `ExcelHandler.read_excel(header_row=, skiprows=)` 读取——pandas 在
`header=` 下会自动对重名列加 `.1`/`.2` 后缀去重，从而避免 `df[col]` 因重名列返回
DataFrame、进而 `clean_sale_plan` 内 `.astype(str).str` 报
'DataFrame' object has no attribute 'str'。
全部月份表模式：拼接各欠交表预览记录（见 `_build_all_months_preview_records`）。
"""
if self.scope == "all_months":
return self._build_all_months_preview_records()
if self.df_raw is None or not self.file_path or not self.current_sheet:
return []
header_row = self.header_row_spin.value()
skip_rows = self.skip_rows_spin.value()
try:
# 与真实导入同款读取（自动去重重名列），不再手动拼表头
records = ImportController.read_and_clean_sheet(
file_path=str(self.file_path),
sheet_name=self.current_sheet,
header_row=header_row,
skip_rows=skip_rows,
column_mapping=self.get_current_mapping(),
)
# 应用当前范围筛选（与真实导入一致）
order_filter = (
self._collect_order_filter()
if (apply_range and self.range_mode == "order")
else None
)
selected = (
self._collect_selected_rows()
if (apply_range and self.range_mode == "rows")
else None
)
records = ImportController.apply_range_filter(records, order_filter, selected)
# 导入时间段筛选（按月份）：与真实导入同源语义
return self._apply_time_scope_filter(records)
except Exception as e:
logger.warning(f"预览记录构建失败(降级为空): {e}")
return []
@staticmethod
def _compute_groups(records: list[dict]) -> dict[str, int]:
"""按 B0-6 的 4 字段分组，返回 组键->行数。"""
group_fields = ["plan_date", "customer", "project_name", "plan_no"]
groups: dict[str, int] = {}
for rec in records:
key = "::".join(str(rec.get(f, "") or "").strip() for f in group_fields)
groups[key] = groups.get(key, 0) + 1
return groups
@staticmethod
def _normalize_record_for_key(rec: dict, session: Any) -> dict:
"""预览/扫描阶段与入库同款清洗：口径/压力规范化后返回新 dict（不污染入参）。
与 ``save_handler`` 入库前 ``normalize_spec`` / ``normalize_pn`` 保持一致，
确保预览算出的 ``unique_key`` 与入库时一致（否则带 DN/φ 前缀的口径会导致
「数据库已存在」漏报）。
"""
return ImportController.normalize_record_for_key(dict(rec), session)
def _scan_conflicts(self, records: list[dict]) -> tuple[list[str], list[tuple[str, dict, str]]]:
"""检测批次内冲突（同 unique_key 内容不同）与数据库已存在冲突。
预览阶段与真实导入保持同一清洗管线：先对记录做口径/压力规范化
（``_normalize_record_for_key``），再生成 ``unique_key``。否则预览算出的 uk
用原始 ``product_spec``（如 ``"DN150"``），与入库时用的 ``product_spec_norm``
（如 ``"150"``）不一致，导致「数据库已存在」严重漏报（如重导同表时只报出
134 而非应接近 816）。
"""
seen: dict[str, str] = {}
intra_conflicts: list[str] = []
db_conflicts: list[tuple[str, dict, str]] = []
for raw in records:
try:
rec = self._normalize_record_for_key(raw, self._session)
uk = ImportController.generate_unique_key(rec)
except Exception:
continue
sig = ImportController.generate_change_signature(rec)
if uk in seen and seen[uk] != sig:
if uk not in intra_conflicts:
intra_conflicts.append(uk)
else:
seen[uk] = sig
if self._session is not None:
existing = ImportController.find_existing_by_unique_key(self._session, uk)
if existing is not None:
db_conflicts.append((uk, rec, f"{existing.product_model}/{existing.customer}"))
return intra_conflicts, db_conflicts
# 冲突表渲染与导入执行见 ImportConfirmDialog（确认对话框）