Polars写入CSV时异常未被捕获，tkinter进度标签无法更新求助

架构老林

2026-8-10

问题

我用tkinter开发了一款工具，点击提交按钮后progress_label会显示“Loading . . .”。常规错误（比如列与yaml不匹配）能被try-except块捕获，并更新progress_label为错误信息；但部分异常（如Polars写入CSV时的异常）仅在后台（cmd/解释器）抛出，并未进入except块，导致progress_label一直显示加载状态。相关代码如下：

类实现代码

class IngestGenerator:
    def __init__(self, fn,fd,n3pl):
        self.filename = fn
        self.filedir = fd
        self.name3pl = n3pl
    
    def generate_result_csv(self):
        """To extend 3PL please refer to comment with <(extend this)> note
           Don't forget to extend the yaml when extending 3PL
        """
        
        start_time = time.time()

        # with open("columns 1a.yaml", 'r') as f:
        with open(os.path.join(os.path.dirname(__file__), 'columns 1a.yaml'), 'r') as f:
            data = yaml.load(f, Loader=SafeLoader)

        with tempfile.TemporaryDirectory() as tmpdirname:
            try :
                # create new list if there's new 3pl behavior (extend this)
                list_type_like = [data['behavior']['type_like'][i]['name3pl'] for i in range(0,len(data['behavior']['type_like']))] #collects names of 3pl which have categorical column to be divided based on
                
                status = False
                for i in range(0,len(data['behavior']['type_like'])):
                    if data['behavior']['type_like'][i]['name3pl']==self.name3pl: #get the name of category column and the values of categories (extend this whole if-statement)
                        list_types = data['behavior']['type_like'][i]['cats']
                        cat_column = data['behavior']['type_like'][i]['categorical_col']
                        status = True
                    else : pass
                if status == False: #for logging to check if its in the list of type-like 3pl (extend this whole if else statement)
                    print("3PL cannot be found on type-like 3PL list")
                else: print("3PL can be found on type-like 3PL list")

                try:
                    for cat in list_types: #dynamic list creation for each category (extend this line only)
                        globals()[f"{cat}_final_df"] = []
                except : print("3pl isn't split based on it's categories")

                xl = win32com.client.Dispatch("Excel.Application")
                print("Cast to CSV first (win32com)")
                wb = xl.Workbooks.Open(self.filename,ReadOnly=1)
                xl.DisplayAlerts = False
                xl.Visible = False
                xl.ScreenUpdating = False
                xl.EnableEvents = False
                sheet_names = [sheet.Name for sheet in wb.Sheets if sheet.Name.lower() != 'summary']
                print("Sheet names")
                print(sheet_names)

                for sheet_name in sheet_names:
                    print("Reading sheet "+sheet_name)
                    ws = wb.Worksheets(sheet_name)
                    ws.SaveAs(tmpdirname+"\\myfile_tmp_{}.csv".format(sheet_name), 24)

                    used_columns = data['sls_data'][f'{self.name3pl.lower()}_used_columns']
                    renamed_columns = data['sls_data'][f'{self.name3pl.lower()}_rename_columns']
                    rowskip = data['behavior']['row_skip'][f'{self.name3pl.lower()}']
                    list_dtypes = [str for u in used_columns]
                    print("CP 1")

                    scandf = pl.scan_csv(tmpdirname+"\\myfile_tmp_{}.csv".format(sheet_name),skip_rows= rowskip,n_rows=10) #scan csv to get column name
                    print(scandf.columns)
                    scanned_cols = scandf.columns.copy()
                    used_cols_inDF = [] #collects column names dynamically
                    for i in range(0,len(used_columns)):
                        if type(used_columns[i]) is list: #check for each scanned-columns which contained in yaml used_columns, append if the scanned columns exist in yaml
                            for sc in scanned_cols:
                                for uc in used_columns[i]:
                                    if sc == uc:
                                        print(f"Column match : {uc}")
                                        used_cols_inDF.append(uc)
                                    else:pass       
                        else:
                            for sc in scanned_cols: #purpose is same with the if statement
                                if sc == used_columns[i]:
                                    print(f"Column match : {used_columns[i]}")
                                    used_cols_inDF.append(used_columns[i])
                                else:pass
                    print(used_cols_inDF)
                    """
                    JNT files have everchanging column names. Some files only have Total Ongkir, some only have Total,
                    and some might have Total and Total Ongkir. If both exists then will use column Total Ongkir (extend this if necessary i.e for special cases of 3pl)
                    """
                    if self.name3pl == 'JNT':
                        if "Total" in used_cols_inDF and "Total Ongkir" in used_cols_inDF:
                            used_cols_inDF.remove("Total")
                        else:pass
                    else:pass 
                    
                    pldf_csv = pl.read_csv(tmpdirname+"\\myfile_tmp_{}.csv".format(sheet_name),
                            columns = used_cols_inDF,
                            new_columns = renamed_columns,
                            dtypes = list_dtypes,
                            skip_rows= rowskip
                        ).filter(~pl.fold(acc=True, f=lambda acc, s: acc & s.is_null(), exprs=pl.all(),)) #filter rows with all null values
                    print(pldf_csv)
                    print(pldf_csv.columns)
                    
                    for v in data['sls_data']['add_columns']: #create dynamic columns
                        if "3pl invoice distance (m) (optional)" in v.lower() or "3pl cod amount (optional)" in v.lower():
                            pldf_csv = pldf_csv.with_column(pl.Series(name="{}".format(v),values= np.zeros(shape=pldf_csv.shape[0])))
                        elif "3pl tn (mandatory)" in v.lower() or  "weight (kg) (optional)" in v.lower():
                            pass
                        elif "total fee (3pl) (optional)" in v.lower():
                            pldf_csv = pldf_csv.with_column(pl.col(v).str.replace_all(",","").str.strip().cast(pl.Float64,False).fill_null(0))
                        else : 
                            pldf_csv = pldf_csv.with_column(pl.lit(None).alias(v))
                    print(pldf_csv)
                    
                    endpath = self.filedir+"\\{}_{}_{}.csv".format(get_file_name(file_name_appear_label["text"]),sheet_name,"IngestResult").replace('/','\\')
                    if self.name3pl not in list_type_like: #(extend this line only)
                        if self.name3pl == 'JNT': #(extend this line and its statement if necessary i.e for special cases of 3pl)
                            pldf_csv = pldf_csv.with_column((pl.col("Total Fee (3PL) (Optional)")+pl.col("Biaya Asuransi").str.replace_all(",","").str.strip().cast(pl.Float64,False).fill_null(0)).alias("Total Fee (3PL) (Optional)"))
                        else: pass
                        print(pldf_csv)
                        pldf_csv[data['sls_data']['add_columns']].write_csv(endpath,sep='\t')
                    elif self.name3pl in list_type_like: #(extend this line only)
                        for cat in list_types:
                            globals()[f"{cat}_final_df"].append(pldf_csv.filter(pl.col(cat_column).str.contains(cat)))

                if self.name3pl not in list_type_like:
                    pass
                elif self.name3pl in list_type_like:
                    for cat in list_types:
                            globals()[f"{cat}_final_df"] = pl.concat(globals()[f"{cat}_final_df"])
                            print(globals()[f"{cat}_final_df"])
                            globals()[f"endpath_{cat}"] = self.filedir+"\\{}_{}_{}.csv".format(get_file_name(file_name_appear_label["text"]),cat,"IngestResult").replace('/','\\')
                            print("done creating paths")
                            globals()[f"{cat}_final_df"][data['sls_data']['add_columns']].write_csv(globals()[f"endpath_{cat}"],sep='\t')
                
                progress_label["text"] = "Successful!"
                submit_btn["state"] = "normal"
                browse_btn["state"] = "normal"
                print("Process finished")
            except Exception as e:
                print("ERROR with message")
                print(e)
                progress_label["text"] = "Failed due to {}".format(e)
            finally : 
                wb.Close(False)

        print("Total exec time : {}".format((time.time()-start_time)/60))

按钮点击事件代码

def file_submit_btn_click():
    if (file_name_appear_label["text"]==""):
        progress_label["text"] = "Please input your file"
    elif option_menu.get() == '':
        progress_label["text"] = "Please select 3pl name"
    else:
        try :
            submit_btn["state"] = "disabled"
            browse_btn["state"] = "disabled"
            progress_label["text"] = "Loading . . ."
            name3pl = option_menu.get()
            ingest = IngestGenerator(file_name,file_path,name3pl)
            print(get_file_name(file_name_appear_label["text"]))
            threading.Thread(target=ingest.generate_result_csv).start()
        except Exception as e:
            print(e)

请问我遗漏了什么内容？为什么这类异常无法被except语句捕获？有没有办法实现全局捕获所有类型的异常？

分析与解决方案

异常未被捕获的核心原因

线程隔离导致异常无法冒泡
你把generate_result_csv放到新线程执行，Python的线程异常是隔离的——子线程抛出的异常不会传递到启动它的主线程，主线程的try-except无法捕获子线程的异常。所以Polars写入CSV时的异常只会在子线程的控制台输出，不会触发主线程里的UI更新逻辑。
内部try-except吞掉异常
代码里有一段嵌套的try-except仅打印错误信息但不重新抛出异常，会导致部分异常被隐藏，无法被外层的try-except捕获处理：
```
try:
    for cat in list_types:
        globals()[f"{cat}_final_df"] = []
except : print("3pl isn't split based on it's categories")
```
UI操作线程不安全
就算你在子线程的except块里直接修改progress_label，这也不符合tkinter规则——tkinter的UI组件只能在主线程中修改，子线程直接操作可能导致崩溃或不生效。

解决办法

1. 子线程内完整捕获异常，通过主线程更新UI

修改generate_result_csv的异常处理逻辑，用tkinter的after方法把UI更新任务提交到主线程（需确保你有tkinter根窗口对象root）：

except Exception as e:
    print("ERROR with message")
    print(e)
    # 用after把UI更新任务放到主线程执行
    root.after(0, lambda: progress_label.config(text=f"Failed due to {e}"))
    root.after(0, lambda: submit_btn.config(state="normal"))
    root.after(0, lambda: browse_btn.config(state="normal"))

2. 修复吞异常的嵌套try-except

把只打印不抛出的try-except改成处理后重新抛出，让外层逻辑捕获：

try:
    for cat in list_types:
        globals()[f"{cat}_final_df"] = []
except Exception as e:
    print(f"3pl isn't split based on it's categories: {e}")
    raise  # 重新抛出异常，让外层try-except捕获处理

3. 全局捕获子线程异常（可选）

给所有子线程设置全局异常钩子，统一处理未捕获的异常：

import sys
import threading

def thread_exception_hook(args):
    print(f"子线程异常: {args.exc_value}")
    # 用after更新UI
    root.after(0, lambda: progress_label.config(text=f"Failed due to {args.exc_value}"))
    root.after(0, lambda: submit_btn.config(state="normal"))
    root.after(0, lambda: browse_btn.config(state="normal"))

# 程序初始化时设置
threading.excepthook = thread_exception_hook

4. 线程函数包装器（可选）

给线程函数加包装器，确保所有异常都被捕获并转发到主线程：

def thread_wrapper(func):
    def wrapper(*args, **kwargs):
        try:
            func(*args, **kwargs)
        except Exception as e:
            print(f"线程执行异常: {e}")
            root.after(0, lambda: progress_label.config(text=f"Failed due to {e}"))
            root.after(0, lambda: submit_btn.config(state="normal"))
            root.after(0, lambda: browse_btn.config(state="normal"))
    return wrapper

# 启动线程时使用包装器
threading.Thread(target=thread_wrapper(ingest.generate_result_csv)).start()

关键注意点

tkinter的所有UI操作必须在主线程执行，子线程只能通过after方法提交UI更新任务。
避免直接在子线程中修改UI组件，否则会导致不可预测的问题。
不要吞掉异常，确保所有异常都能被捕获并处理，同时更新UI状态。

内容的提问来源于stack exchange，提问作者random student

Polars写入CSV时异常未被捕获，tkinter进度标签无法更新求助

类实现代码 #

按钮点击事件代码 #

异常未被捕获的核心原因 #

解决办法 #

1. 子线程内完整捕获异常，通过主线程更新UI #

2. 修复吞异常的嵌套try-except #

3. 全局捕获子线程异常（可选） #

4. 线程函数包装器（可选） #

关键注意点 #