Python基于PyQt5和openpyxl实现Excel单元格合并工具

import sys
from pyqt5.qtwidgets import (qapplication, qmainwindow, qwidget, qvboxlayout, qhboxlayout,
                             qlabel, qlineedit, qpushbutton, qfiledialog, qcombobox,
                             qcheckbox, qmessagebox, qspinbox, qprogressbar)
from pyqt5.qtcore import qthread, pyqtsignal
from openpyxl import load_workbook
from openpyxl.utils import get_column_letter
import os

class mergethread(qthread):
    progress_signal = pyqtsignal(int)
    finished_signal = pyqtsignal(str)
    error_signal = pyqtsignal(str)

    def __init__(self, file_path, sheet_name, start_row, columns, merge_same_content, output_path, parent=none):
        super().__init__(parent)
        self.file_path = file_path
        self.sheet_name = sheet_name
        self.start_row = start_row
        self.columns = columns 
        self.merge_same_content = merge_same_content
        self.output_path = output_path

    def run(self):
        try:
            wb = load_workbook(self.file_path)
            ws = wb[self.sheet_name]
            total_steps = len(self.columns)
            current_step = 0
            
            for col_info in self.columns:
                col = col_info['col']
                target_list = []
                
                for row in range(self.start_row, ws.max_row + 1):
                    cell_value = ws[f"{col}{row}"].value
                    target_list.append(cell_value if cell_value is not none else "")
                    
                self._merge_cells(ws, target_list, self.start_row, col)
                current_step += 1
                progress = int(current_step / total_steps * 100)
                self.progress_signal.emit(progress)
                
            wb.save(self.output_path)
            self.finished_signal.emit(self.output_path)
            
        except exception as e:
            self.error_signal.emit(str(e))

    def _merge_cells(self, ws, target_list, start_row, col):
        start = 0
        end = 0
        reference = target_list[0]

        for i in range(len(target_list)):
            if not self.merge_same_content and i > 0:
                if i > 0 and start != i - 1:
                    ws.merge_cells(f'{col}{start + start_row}:{col}{i - 1 + start_row}')
                start = i
            else:
                if i < len(target_list) - 1 and target_list[i] == target_list[i + 1]:
                    end = i
                    continue

                if start == i: 
                    pass
                else:
                    if target_list[start] == target_list[i]:
                        ws.merge_cells(f'{col}{start + start_row}:{col}{i + start_row}')

                start = i + 1

class excelmergetool(qmainwindow):
    def __init__(self):
        super().__init__()
        self.setwindowtitle("excel 单元格合并工具")
        self.setgeometry(100, 100, 600, 450)  
        self.file_path = ""
        self.sheet_names = []
        self.columns = []
        self.init_ui()

    def init_ui(self):
        main_widget = qwidget()
        main_layout = qvboxlayout()
        file_layout = qhboxlayout()
        file_layout.addwidget(qlabel("excel文件:"))
        self.file_edit = qlineedit()
        self.file_edit.setreadonly(true)
        file_layout.addwidget(self.file_edit)
        browse_btn = qpushbutton("浏览...")
        browse_btn.clicked.connect(self.browse_file)
        file_layout.addwidget(browse_btn)
        main_layout.addlayout(file_layout)
        sheet_layout = qhboxlayout()
        sheet_layout.addwidget(qlabel("工作表:"))
        self.sheet_combo = qcombobox()
        self.sheet_combo.setenabled(false)
        sheet_layout.addwidget(self.sheet_combo)
        main_layout.addlayout(sheet_layout)
        row_layout = qhboxlayout()
        row_layout.addwidget(qlabel("起始行:"))
        self.start_row_spin = qspinbox()
        self.start_row_spin.setminimum(1)
        self.start_row_spin.setmaximum(9999)
        self.start_row_spin.setvalue(1)
        row_layout.addwidget(self.start_row_spin)
        main_layout.addlayout(row_layout)
        col_layout = qvboxlayout()
        col_layout.addwidget(qlabel("需要合并的列:"))
        self.col_list_widget = qwidget()
        self.col_list_layout = qvboxlayout()
        self.col_list_widget.setlayout(self.col_list_layout)
        col_layout.addwidget(self.col_list_widget)
        add_col_btn = qpushbutton("添加列")
        add_col_btn.clicked.connect(self.add_column_setting)
        col_layout.addwidget(add_col_btn)
        main_layout.addlayout(col_layout)
        option_layout = qhboxlayout()
        self.merge_same_check = qcheckbox("仅合并相同内容")
        self.merge_same_check.setchecked(true)
        option_layout.addwidget(self.merge_same_check)
        main_layout.addlayout(option_layout)
        output_layout = qhboxlayout()
        output_layout.addwidget(qlabel("输出文件名:"))
        self.output_edit = qlineedit()
        self.output_edit.setplaceholdertext("自动在原文件名后添加'_wenqingzhou'")
        output_layout.addwidget(self.output_edit)
        main_layout.addlayout(output_layout)
        self.progress_bar = qprogressbar()
        self.progress_bar.setvalue(0)
        main_layout.addwidget(self.progress_bar)
        execute_btn = qpushbutton("执行合并")
        execute_btn.clicked.connect(self.execute_merge)
        main_layout.addwidget(execute_btn)
        main_widget.setlayout(main_layout)
        self.setcentralwidget(main_widget)

    def browse_file(self):
        file_path, _ = qfiledialog.getopenfilename(
            self, "选择excel文件", "", "excel文件 (*.xlsx *.xls)"
        )

        if file_path:
            self.file_path = file_path
            self.file_edit.settext(file_path)

            try:
                wb = load_workbook(file_path, read_only=true)
                self.sheet_names = wb.sheetnames
                self.sheet_combo.clear()
                self.sheet_combo.additems(self.sheet_names)
                self.sheet_combo.setenabled(true)
                base, ext = os.path.splitext(file_path)
                default_output = f"{base}_wenqingzhou{ext}"
                self.output_edit.settext(default_output)

                for i in reversed(range(self.col_list_layout.count())):
                    widget = self.col_list_layout.itemat(i).widget()
                    if widget is not none:
                        widget.deletelater()

                self.columns = []

            except exception as e:
                qmessagebox.critical(self, "错误", f"无法读取excel文件:\n{str(e)}")

    def add_column_setting(self):
        if not self.sheet_names:
            qmessagebox.warning(self, "警告", "请先选择excel文件")
            return

        col_widget = qwidget()
        col_layout = qhboxlayout()
        col_layout.addwidget(qlabel("列:"))
        col_combo = qcombobox()
        col_combo.additems([get_column_letter(i) for i in range(1, 100)])  # a-cz
        col_layout.addwidget(col_combo)
        del_btn = qpushbutton("删除")
        del_btn.clicked.connect(lambda: self.remove_column_setting(col_widget))
        col_layout.addwidget(del_btn)
        col_widget.setlayout(col_layout)
        self.col_list_layout.addwidget(col_widget)
        self.columns.append({
            'widget': col_widget,
            'col_combo': col_combo,
        })

    def remove_column_setting(self, widget):
        self.col_list_layout.removewidget(widget)
        widget.deletelater()
        self.columns = [col for col in self.columns if col['widget'] != widget]

    def execute_merge(self):
        if not self.file_path:
            qmessagebox.warning(self, "警告", "请先选择excel文件")
            return

        if not self.columns:
            qmessagebox.warning(self, "警告", "请至少添加一列需要合并的列")
            return

        sheet_name = self.sheet_combo.currenttext()
        start_row = self.start_row_spin.value()
        merge_same_content = self.merge_same_check.ischecked()

        columns_info = []
        for col in self.columns:
            col_letter = col['col_combo'].currenttext()
            columns_info.append({'col': col_letter, 'name': ''})  

        output_path = self.output_edit.text().strip()
        if not output_path:
            base, ext = os.path.splitext(self.file_path)
            output_path = f"{base}_wenqingzhou{ext}"

        self.thread = mergethread(
            file_path=self.file_path,
            sheet_name=sheet_name,
            start_row=start_row,
            columns=columns_info,
            merge_same_content=merge_same_content,
            output_path=output_path
        )

        self.thread.progress_signal.connect(self.update_progress)
        self.thread.finished_signal.connect(self.merge_completed)
        self.thread.error_signal.connect(self.merge_failed)
        self.thread.start()

    def update_progress(self, value):
        self.progress_bar.setvalue(value)

    def merge_completed(self, output_path):
        qmessagebox.information(
            self,
            "完成",
            f"单元格合并完成!\n输出文件已保存为:\n{output_path}"
        )
        self.progress_bar.setvalue(0)

    def merge_failed(self, error_msg):
        qmessagebox.critical(
            self,
            "错误",
            f"合并过程中发生错误:\n{error_msg}"
        )
        self.progress_bar.setvalue(0)

if __name__ == "__main__":
    app = qapplication(sys.argv)
    window = excelmergetool()
    window.show()
    sys.exit(app.exec_())

class mergethread(qthread):
    
    """后台处理excel合并的线程类，避免界面卡顿"""
    
    # 定义线程信号
    progress_signal = pyqtsignal(int)  # 进度信号
    finished_signal = pyqtsignal(str)  # 完成信号（带输出路径）
    error_signal = pyqtsignal(str)     # 错误信号

    def __init__(self, file_path, sheet_name, start_row, columns, merge_same_content, output_path, parent=none):
        super().__init__(parent)
        # 初始化参数
        self.file_path = file_path
        self.sheet_name = sheet_name
        self.start_row = start_row
        self.columns = columns  # 格式：[{'col': 'a'}, {'col': 'b'}]
        self.merge_same_content = merge_same_content
        self.output_path = output_path

    def run(self):
        """线程执行的主逻辑"""
        try:
            # 1. 加载excel文件
            wb = load_workbook(self.file_path)
            ws = wb[self.sheet_name]
            
            # 2. 计算总进度步数（基于列数）
            total_steps = len(self.columns)
            current_step = 0
            
            # 3. 遍历处理每一列
            for col_info in self.columns:
                col = col_info['col']
                target_list = []
                
                # 3.1 收集该列所有单元格数据
                for row in range(self.start_row, ws.max_row + 1):
                    cell_value = ws[f"{col}{row}"].value
                    target_list.append(cell_value if cell_value is not none else "")
                
                # 3.2 执行单元格合并
                self._merge_cells(ws, target_list, self.start_row, col)
                
                # 3.3 更新进度
                current_step += 1
                progress = int(current_step / total_steps * 100)
                self.progress_signal.emit(progress)
            
            # 4. 保存结果
            wb.save(self.output_path)
            self.finished_signal.emit(self.output_path)
            
        except exception as e:
            self.error_signal.emit(str(e))

    def _merge_cells(self, ws, target_list, start_row, col):
        """
        实际执行单元格合并的算法:
        :param ws: 工作表对象
        :param target_list: 该列所有单元格值列表
        :param start_row: 起始行号
        :param col: 列字母
        """
        start = 0
        reference = target_list[0]

        for i in range(len(target_list)):
            if not self.merge_same_content and i > 0:
                # 模式1：强制合并（无论内容是否相同）
                if i > 0 and start != i - 1:
                    ws.merge_cells(f'{col}{start + start_row}:{col}{i - 1 + start_row}')
                start = i
            else:
                # 模式2：仅合并相同内容
                if i < len(target_list) - 1 and target_list[i] == target_list[i + 1]:
                    continue  # 相同内容则继续扩展合并范围
                
                if start == i: 
                    pass  # 单个单元格无需合并
                else:
                    if target_list[start] == target_list[i]:
                        ws.merge_cells(f'{col}{start + start_row}:{col}{i + start_row}')
                start = i + 1

class excelmergetool(qmainwindow):
    
    """主窗口类，负责ui展示和用户交互"""
    def __init__(self):
        super().__init__()
        self.setwindowtitle("excel 单元格合并工具")
        self.setgeometry(100, 100, 600, 450)  
        self.file_path = ""
        self.sheet_names = []
        self.columns = []  # 存储用户添加的列设置
        self.init_ui()

    def init_ui(self):
        
        """初始化用户界面"""
        main_widget = qwidget()
        main_layout = qvboxlayout()
        
        # ------- 文件选择区域 -------
        file_layout = qhboxlayout()
        file_layout.addwidget(qlabel("excel文件:"))
        self.file_edit = qlineedit()
        self.file_edit.setreadonly(true)
        file_layout.addwidget(self.file_edit)
        
        browse_btn = qpushbutton("浏览...")
        browse_btn.clicked.connect(self.browse_file)
        file_layout.addwidget(browse_btn)
        main_layout.addlayout(file_layout)
        
        # ------- 工作表选择区域 -------
        sheet_layout = qhboxlayout()
        sheet_layout.addwidget(qlabel("工作表:"))
        self.sheet_combo = qcombobox()
        self.sheet_combo.setenabled(false)  # 初始禁用，直到选择文件
        sheet_layout.addwidget(self.sheet_combo)
        main_layout.addlayout(sheet_layout)
        
        # ------- 起始行设置 -------
        row_layout = qhboxlayout()
        row_layout.addwidget(qlabel("起始行:"))
        self.start_row_spin = qspinbox()
        self.start_row_spin.setminimum(1)
        self.start_row_spin.setmaximum(9999)
        self.start_row_spin.setvalue(1)
        row_layout.addwidget(self.start_row_spin)
        main_layout.addlayout(row_layout)
        
        # ------- 列设置区域 -------
        col_layout = qvboxlayout()
        col_layout.addwidget(qlabel("需要合并的列:"))
        
        # 动态列设置容器
        self.col_list_widget = qwidget()
        self.col_list_layout = qvboxlayout()
        self.col_list_widget.setlayout(self.col_list_layout)
        col_layout.addwidget(self.col_list_widget)
        
        add_col_btn = qpushbutton("添加列")
        add_col_btn.clicked.connect(self.add_column_setting)
        col_layout.addwidget(add_col_btn)
        main_layout.addlayout(col_layout)
        
        # ------- 选项设置 -------
        option_layout = qhboxlayout()
        self.merge_same_check = qcheckbox("仅合并相同内容")
        self.merge_same_check.setchecked(true)
        option_layout.addwidget(self.merge_same_check)
        main_layout.addlayout(option_layout)
        
        # ------- 输出设置 -------
        output_layout = qhboxlayout()
        output_layout.addwidget(qlabel("输出文件名:"))
        self.output_edit = qlineedit()
        self.output_edit.setplaceholdertext("自动在原文件名后添加'_wenqingzhou'")
        output_layout.addwidget(self.output_edit)
        main_layout.addlayout(output_layout)
        
        # ------- 进度条 -------
        self.progress_bar = qprogressbar()
        self.progress_bar.setvalue(0)
        main_layout.addwidget(self.progress_bar)
        
        # ------- 执行按钮 -------
        execute_btn = qpushbutton("执行合并")
        execute_btn.clicked.connect(self.execute_merge)
        main_layout.addwidget(execute_btn)
        
        main_widget.setlayout(main_layout)
        self.setcentralwidget(main_widget)

    def browse_file(self):
        
        """打开文件对话框选择excel文件"""
        file_path, _ = qfiledialog.getopenfilename(
            self, "选择excel文件", "", "excel文件 (*.xlsx *.xls)"
        )

        if file_path:
            self.file_path = file_path
            self.file_edit.settext(file_path)

            try:
                # 读取工作表列表
                wb = load_workbook(file_path, read_only=true)
                self.sheet_names = wb.sheetnames
                self.sheet_combo.clear()
                self.sheet_combo.additems(self.sheet_names)
                self.sheet_combo.setenabled(true)
                
                # 设置默认输出文件名
                base, ext = os.path.splitext(file_path)
                default_output = f"{base}_wenqingzhou{ext}"
                self.output_edit.settext(default_output)

                # 清空已有列设置
                for i in reversed(range(self.col_list_layout.count())):
                    widget = self.col_list_layout.itemat(i).widget()
                    if widget is not none:
                        widget.deletelater()
                self.columns = []

            except exception as e:
                qmessagebox.critical(self, "错误", f"无法读取excel文件:\n{str(e)}")

    def add_column_setting(self):
        
        """添加一列合并设置"""
        if not self.sheet_names:
            qmessagebox.warning(self, "警告", "请先选择excel文件")
            return

        # 创建列设置控件组
        col_widget = qwidget()
        col_layout = qhboxlayout()
        col_layout.addwidget(qlabel("列:"))
        
        # 列选择下拉框（a-cz）
        col_combo = qcombobox()
        col_combo.additems([get_column_letter(i) for i in range(1, 100)])
        col_layout.addwidget(col_combo)
        
        # 删除按钮
        del_btn = qpushbutton("删除")
        del_btn.clicked.connect(lambda: self.remove_column_setting(col_widget))
        col_layout.addwidget(del_btn)
        
        col_widget.setlayout(col_layout)
        self.col_list_layout.addwidget(col_widget)
        
        # 记录到列设置列表
        self.columns.append({
            'widget': col_widget,
            'col_combo': col_combo,
        })

    def remove_column_setting(self, widget):
        
        """删除指定的列设置"""
        self.col_list_layout.removewidget(widget)
        widget.deletelater()
        self.columns = [col for col in self.columns if col['widget'] != widget]

    def execute_merge(self):
        
        """执行合并操作"""
        # 参数验证
        if not self.file_path:
            qmessagebox.warning(self, "警告", "请先选择excel文件")
            return

        if not self.columns:
            qmessagebox.warning(self, "警告", "请至少添加一列需要合并的列")
            return

        # 收集参数
        sheet_name = self.sheet_combo.currenttext()
        start_row = self.start_row_spin.value()
        merge_same_content = self.merge_same_check.ischecked()

        # 准备列信息
        columns_info = []
        for col in self.columns:
            col_letter = col['col_combo'].currenttext()
            columns_info.append({'col': col_letter, 'name': ''})  # name保留字段

        # 处理输出路径
        output_path = self.output_edit.text().strip()
        if not output_path:
            base, ext = os.path.splitext(self.file_path)
            output_path = f"{base}_wenqingzhou{ext}"

        # 创建并启动后台线程
        self.thread = mergethread(
            file_path=self.file_path,
            sheet_name=sheet_name,
            start_row=start_row,
            columns=columns_info,
            merge_same_content=merge_same_content,
            output_path=output_path
        )

        # 连接信号槽
        self.thread.progress_signal.connect(self.update_progress)
        self.thread.finished_signal.connect(self.merge_completed)
        self.thread.error_signal.connect(self.merge_failed)
        self.thread.start()

    def update_progress(self, value):
        
        """更新进度条"""
        self.progress_bar.setvalue(value)

    def merge_completed(self, output_path):
        
        """合并完成处理"""
        qmessagebox.information(
            self,
            "完成",
            f"单元格合并完成!\n输出文件已保存为:\n{output_path}"
        )
        self.progress_bar.setvalue(0)

    def merge_failed(self, error_msg):
        
        """合并失败处理"""
        qmessagebox.critical(
            self,
            "错误",
            f"合并过程中发生错误:\n{error_msg}"
        )
        self.progress_bar.setvalue(0)