Python引入threading.Thread()后错误处理失效问题求助

我是Python新手，原本已实现正常工作的错误处理逻辑。为给PySimpleGUI添加进度条，引入threading模块实现多线程后，错误处理失效。以下是添加进度条前后的代码（新版未包含错误处理），求解决办法。

旧代码（带错误处理） #

import PySimpleGUI as sg
import functions_new as fn
import sys
import os

sg.theme("Default1")
sg.set_options(font=("Segoe UI Variable", 10))

input1 = sg.Input()
archive_choose = sg.FolderBrowse("Select Source Directory",
                                 key="source",
                                 pad=(5, 5))

input2 = sg.Input()
BOM_choose = sg.FileBrowse("Select BOM",
                           key="excel",
                           pad=(5, 5))

input3 = sg.Input()
save_dir = sg.FolderBrowse("Select Save Location",
                           key="save",
                           pad=(5, 5))

exec_button = sg.Button("Execute",
                        pad=(5, 5),
                        button_color=('White', 'NavyBlue'))

window = sg.Window(title="Step Name Replace",
                   layout=[[input1, archive_choose],
                           [input2, BOM_choose],
                           [input3, save_dir],
                           [exec_button]])

while True:
    event, values = window.read()

    if event == sg.WIN_CLOSED or event == "Exit":
        break

    if event == "Execute":
        # calls func to search directory and store stepfiles in list
        filepaths = fn.stpFinder(values["source"])
        filepaths = [item.replace("\\", "/") for item in filepaths]

        try:
            print(f"Source Directory: {values['source']}")
            print(f"Source Directory: {values['excel']}")
            print(f"Source Directory: {values['save']}")

            # calls func to search directory and store stepfiles in list
            filepaths = fn.stpFinder(values["source"])
            filepaths = [item.replace("\\", "/") for item in filepaths]
            print(f"Found {len(filepaths)} .stp files")

            #gets df from excel
            df = fn.extractMREBOM(values["excel"])

            #runs name replace function
            for filepath in filepaths:
                new_name = os.path.join(values["save"], os.path.basename(filepath).replace(".stp", "_rename.stp"))

                print(f"Processing file: {filepath} -> {new_name}")
                fn.stepNameReplace(filepath, df, new_name)

            sg.popup("Process completed successfully!",
                     font=("Segoe UI Variable", 10))

        except PermissionError:
            sg.popup("Ensure that BOM and folder locations are closed before executing\n\n"
                     "Click OK to terminate",
                     font=("Segoe UI Variable", 10))
            sys.exit()

        except Exception as e:
            sg.popup(f"An error occurred: {e}",
                     font=("Segoe UI Variable", 10))

window.close()

新代码（带进度条但无错误处理） #

import PySimpleGUI as sg
import functions_new as fn
import sys
import os
import threading


def execute_func(window, values):
    # Create new directory within source directory
    output_dir = os.path.join(values["source"], "SNR_Output")
    os.makedirs(output_dir, exist_ok=True)

    # Calls func to search directory and store stepfiles in list
    files = fn.stpFinder(values["source"])
    files = [item.replace("\\", "/") for item in files]
    print(f"Found {len(files)} .stp files")

    # Extracts df from excel
    df = fn.extractMREBOM(values["excel"])

    total_files = len(files)

    for index, file in enumerate(files):

        # Creates name for output step file
        new_name = os.path.join(output_dir, os.path.basename(file))

        # "Update" assigned to be called by elif to update status bar through each iteration
        window.write_event_value("Update", (index + 1, total_files))

        # Processes the file
        print(f"Processing file: {file}")
        fn.stepNameReplace(file, df, new_name)

    # Once loop completes, "Done" value assigned to be called by elif
    window.write_event_value("Done", None)

sg.theme("Default1")
sg.set_options(font=("Segoe UI Variable", 10))

layout = [
    [sg.Input(), sg.FolderBrowse("Select Source Directory", key="source", pad=(5, 5))],
    [sg.Input(), sg.FileBrowse("Select BOM", key="excel", pad=(5, 5))],
    [sg.Button("Execute", button_color=("White", "Grey")),
     sg.ProgressBar(100, orientation='h', size=(20, 20), key='progressbar', bar_color=("Salmon", "LightGrey")),
     sg.Text("", key="info")]
]

window = sg.Window(title="Step Name Replace (SNR)", layout=layout)

while True:
    event, values = window.read(timeout=100)

    if event == sg.WIN_CLOSED:
        break

    elif event == "Execute":
        print(f"Source Directory: {values['source']}")
        print(f"Source Directory: {values['excel']}")

        threading.Thread(target=execute_func, args=(window, values), daemon=True).start()

    elif event == "Update":
        current_file, total_files = values[event]
        window["progressbar"].update(current_file * 100 / total_files)
        window["info"].update(f"Working file {current_file} of {total_files}.")

    elif event == "Done":
        sg.popup("Process completed successfully!",
                 font=("Segoe UI Variable", 10))
        sys.exit()

window.close()

functions_new.py #

from pandas import read_excel
import os
import sys
import PySimpleGUI as sg


# searches selected directory and extracts stepfiles into list
def stpFinder(source_dir):
    stp_files = []
    extensions = (".stp", ".step", ".STP", ".STEP")
    for root, dirs, files in os.walk(source_dir):
        for file in files:
            if file.endswith(extensions):
                stp_files.append(os.path.join(root, file))
    return stp_files


# Reads BOM excel and returns dataframe
def extractMREBOM(path):
    df = read_excel(path, sheet_name="BOM", dtype="str")

    des_count = df["DESCRIPTION"].count()
    num_count = df["DOC NUMBER"].count()
    type_count = df["DOC TYPE"].count()
    part_count = df["DOC PART"].count()

    # Checks if BOM is formatted correctly before returning df
    if des_count == num_count and type_count == part_count:
        return df

    else:
        sg.popup("BOM formatted incorrectly (Inconsistent number of rows)\n \n"
                 "Click OK to Terminate",
                 font=("Segoe UI Variable", 10))
        sys.exit()


# Uses returned dataframe to replace string names with BOM Description, then rename
def stepNameReplace(filepath_arg, df_arg, newName):

    output = open(newName, 'w+')
    file = open(filepath_arg, 'rb')

    for row in file:
        tmp_str = row.decode("utf-8").replace("\r\n", "\r")

        # Perform all replacements for the current line
        for i in range(len(df_arg["DESCRIPTION"])):
            tmp_str = tmp_str.replace(
                f"{df_arg['DOC NUMBER'][i]}_{df_arg['DOC TYPE'][i].upper()}_{df_arg['DOC PART'][i]}",
                df_arg['DESCRIPTION'][i])  # case Upper

            tmp_str = tmp_str.replace(
                f"{df_arg['DOC NUMBER'][i]}_{df_arg['DOC TYPE'][i].lower()}_{df_arg['DOC PART'][i]}",
                df_arg['DESCRIPTION'][i])  # case Lower

        # After all replacements, write the modified line to the output file
        output.write(tmp_str)
    output.close()
    return

多线程环境下，子线程的异常不会自动传递到主线程，需手动捕获子线程异常并通知主线程处理，同时修改工具函数中直接终止程序的逻辑。

1. 修改execute_func函数，添加异常捕获 #

将函数代码包裹在try-except块中，捕获异常后通过window.write_event_value发送错误事件给主线程：

def execute_func(window, values):
    try:
        # Create new directory within source directory
        output_dir = os.path.join(values["source"], "SNR_Output")
        os.makedirs(output_dir, exist_ok=True)

        # Calls func to search directory and store stepfiles in list
        files = fn.stpFinder(values["source"])
        files = [item.replace("\\", "/") for item in files]
        print(f"Found {len(files)} .stp files")

        # Extracts df from excel
        df = fn.extractMREBOM(values["excel"])

        total_files = len(files)

        for index, file in enumerate(files):
            # Creates name for output step file
            new_name = os.path.join(output_dir, os.path.basename(file))
            # Update progress bar
            window.write_event_value("Update", (index + 1, total_files))
            # Processes the file
            print(f"Processing file: {file}")
            fn.stepNameReplace(file, df, new_name)

        # Notify process completion
        window.write_event_value("Done", None)
    except PermissionError as e:
        window.write_event_value("Error", ("PermissionError", str(e)))
    except Exception as e:
        window.write_event_value("Error", ("GeneralError", str(e)))

2. 修改主线程事件循环，添加错误处理分支 #

在原有事件判断中增加"Error"事件的处理逻辑：

while True:
    event, values = window.read(timeout=100)

    if event == sg.WIN_CLOSED:
        break

    elif event == "Execute":
        print(f"Source Directory: {values['source']}")
        print(f"Source Directory: {values['excel']}")
        threading.Thread(target=execute_func, args=(window, values), daemon=True).start()

    elif event == "Update":
        current_file, total_files = values[event]
        window["progressbar"].update(current_file * 100 / total_files)
        window["info"].update(f"Working file {current_file} of {total_files}.")

    elif event == "Done":
        sg.popup("Process completed successfully!",
                 font=("Segoe UI Variable", 10))
        window.close()

    elif event == "Error":
        error_type, error_msg = values[event]
        if error_type == "PermissionError":
            sg.popup("确保BOM文件和文件夹都已关闭后再执行\n\n点击OK退出",
                     font=("Segoe UI Variable", 10))
        else:
            sg.popup(f"发生错误: {error_msg}",
                     font=("Segoe UI Variable", 10))
        window.close()

window.close()

3. 修改functions_new.py中的extractMREBOM函数 #

去掉sys.exit()，改为抛出异常让上层处理：

# Reads BOM excel and returns dataframe
def extractMREBOM(path):
    df = read_excel(path, sheet_name="BOM", dtype="str")

    des_count = df["DESCRIPTION"].count()
    num_count = df["DOC NUMBER"].count()
    type_count = df["DOC TYPE"].count()
    part_count = df["DOC PART"].count()

    # Checks if BOM is formatted correctly before returning df
    if des_count == num_count and type_count == part_count:
        return df

    else:
        raise ValueError("BOM格式错误（行数不一致）")

修改后，子线程中所有异常都会被捕获并通知主线程弹出对应提示，同时保持进度条功能正常运行。

内容的提问来源于stack exchange，提问作者Dominick Cole