多线程处理 Excel 文件及单元格翻译

问题：

先总结以下以下代码的作用，和要点，然后修改以下代码，不止用多线程处理多个xlsx文件，处理某个xlsx的单元格的时候，也采用多线程，同时还要保证效率和消息的正常更新。 import os import ujson import tkinter as tk from tkinter import filedialog, messagebox import pandas as pd from concurrent.futures import ThreadPoolExecutor import queue import threading from tqdm import tqdm

def load_and_sort_json(file_path): """加载 JSON 文件并按键的长度排序（最长的键排在前面）""" with open(file_path, 'r', encoding='utf-8') as f: data = ujson.load(f) # 按键的长度从大到小排序 sorted_data = dict(sorted(data.items(), key=lambda item: len(item[0]), reverse=True)) return sorted_data

def chunk_sorted_data(sorted_data, chunk_size=1000): """将排序后的数据分块""" chunks = [] chunk = {} for key, value in sorted_data.items(): chunk[key] = value if len(chunk) >= chunk_size: chunks.append(chunk) chunk = {} # 添加剩余的块 if chunk: chunks.append(chunk) return chunks

def replace_with_dict(text, translations): """使用字典查找替换文本""" for original, translated in translations.items(): text = text.replace(original, translated) return text

def process_excel_file_with_pandas(file_path, translations_chunks, progress_queue): """使用 pandas 处理单个 Excel 文件""" try: # 读取 Excel 文件的第二列 df = pd.read_excel(file_path, usecols=[1]) total_cells = len(df) progress_queue.put(("sheet", "Sheet1", total_cells)) # 发送工作表进度信息

    # 遍历第二列的每个单元格
    for idx, cell_value in enumerate(df.iloc[:, 0]):
        if pd.notna(cell_value):  # 检查单元格是否为空
            # 按从长到短的顺序依次替换
            for chunk in translations_chunks:
                cell_value = replace_with_dict(str(cell_value), chunk)
            df.iloc[idx, 0] = cell_value  # 更新单元格值
        progress_queue.put(("cell", 1))  # 发送单元格进度信息

    # 保存修改后的 Excel 文件
    df.to_excel(file_path, index=False, header=False)
    progress_queue.put(("done",))  # 发送完成信息
    return True
except Exception as e:
    print(f"处理文件 {file_path} 时出错: {e}")
    progress_queue.put(("error", str(e)))  # 发送错误信息
    return False

def process_folder_with_threads(folder_path, translations_chunks, progress_queue): """使用多线程处理文件夹中的所有 Excel 文件""" success_count = 0 total_files = len([f for f in os.listdir(folder_path) if f.endswith('.xlsx')]) if total_files == 0: messagebox.showwarning("警告", "选择的文件夹中没有 Excel 文件（.xlsx）") return

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = []
    for file_name in os.listdir(folder_path):
        if file_name.endswith('.xlsx'):
            file_path = os.path.join(folder_path, file_name)
            future = executor.submit(process_excel_file_with_pandas, file_path, translations_chunks, progress_queue)
            futures.append(future)

    for future in futures:
        if future.result():
            success_count += 1
        progress_queue.put(("file", 1))  # 发送文件进度信息

progress_queue.put(("finish", success_count, total_files))  # 发送完成信息

def update_progress_bar(progress_queue, file_pbar, sheet_pbar, cell_pbar): """更新进度条""" while True: msg = progress_queue.get() if msg[0] == "file": file_pbar.update(msg[1]) # 更新文件进度条 elif msg[0] == "sheet": sheet_pbar.reset(total=msg[2]) # 重置工作表进度条 sheet_pbar.set_description(f"处理工作表: {msg[1]}") elif msg[0] == "cell": sheet_pbar.update(msg[1]) # 更新单元格进度条 elif msg[0] == "done": file_pbar.update(1) # 更新文件进度条 elif msg[0] == "finish": messagebox.showinfo("完成", f"处理完成！成功处理 {msg[1]}/{msg[2]} 个文件。") break elif msg[0] == "error": messagebox.showerror("错误", f"处理文件时出错: {msg[1]}")

def select_files_and_process(): """选择 JSON 文件和 Excel 文件夹并开始处理""" # 选择 JSON 文件 json_path = filedialog.askopenfilename( title="选择 JSON 文件", filetypes=[("JSON 文件", "*.json")] ) if not json_path: messagebox.showwarning("警告", "未选择 JSON 文件") return

# 加载并排序 JSON 文件
try:
    translations = load_and_sort_json(json_path)
except Exception as e:
    messagebox.showerror("错误", f"加载 JSON 文件时出错: {e}")
    return

# 将排序后的数据分块
translations_chunks = chunk_sorted_data(translations)

# 选择 Excel 文件夹
folder_path = filedialog.askdirectory(
    title="选择包含 Excel 文件的文件夹"
)
if not folder_path:
    messagebox.showwarning("警告", "未选择文件夹")
    return

# 创建进度条和消息队列
progress_queue = queue.Queue()
file_pbar = tqdm(total=len([f for f in os.listdir(folder_path) if f.endswith('.xlsx')]), desc="处理文件")
sheet_pbar = tqdm(total=0, desc="处理工作表", leave=False)
cell_pbar = tqdm(total=0, desc="处理单元格", leave=False)

# 启动进度条更新线程
progress_thread = threading.Thread(target=update_progress_bar, args=(progress_queue, file_pbar, sheet_pbar, cell_pbar))
progress_thread.start()

# 启动文件处理线程
process_thread = threading.Thread(target=process_folder_with_threads, args=(folder_path, translations_chunks, progress_queue))
process_thread.start()

def main(): """主函数""" root = tk.Tk() root.title("Excel 翻译工具") root.geometry("400x200")

# 创建按钮
btn_select = tk.Button(root, text="选择文件并处理", command=select_files_and_process, width=20, height=3)
btn_select.pack(pady=40)

# 运行主循环
root.mainloop()

if name == "main": main()

答案：

代码作用总结

这段代码是一个用于处理 Excel 文件的翻译工具，主要功能包括：

加载和排序 JSON 文件：从指定的 JSON 文件中加载翻译数据，并按键的长度进行排序。
分块处理数据：将排序后的翻译数据分块，以便于逐块替换 Excel 文件中的内容。
处理 Excel 文件：使用 pandas 库读取 Excel 文件的第二列，逐个单元格进行翻译替换，并更新单元格的值。
多线程处理：使用多线程来并行处理文件夹中的多个 Excel 文件，提高处理效率。
进度更新：通过进度条和消息队列实时更新处理进度，提供用户反馈。
图形用户界面：使用 tkinter 创建简单的 GUI，允许用户选择 JSON 文件和包含 Excel 文件的文件夹。

代码要点

使用 ThreadPoolExecutor 来实现多线程处理多个 Excel 文件。
使用 queue.Queue 来传递进度信息。
使用 tqdm 库来显示进度条。
处理 Excel 文件时，逐个单元格进行翻译替换。

修改代码以实现更细粒度的多线程处理

为了在处理某个 Excel 文件的单元格时也采用多线程，我们可以在 process_excel_file_with_pandas 函数中引入多线程处理单元格的逻辑。以下是修改后的代码：

import os
import ujson
import tkinter as tk
from tkinter import filedialog, messagebox
import pandas as pd
from concurrent.futures import ThreadPoolExecutor
import queue
import threading
from tqdm import tqdm


def load_and_sort_json(file_path):
    """加载 JSON 文件并按键的长度排序（最长的键排在前面）"""
    with open(file_path, 'r', encoding='utf-8') as f:
        data = ujson.load(f)
    sorted_data = dict(sorted(data.items(), key=lambda item: len(item[0]), reverse=True))
    return sorted_data


def chunk_sorted_data(sorted_data, chunk_size=1000):
    """将排序后的数据分块"""
    chunks = []
    chunk = {}
    for key, value in sorted_data.items():
        chunk[key] = value
        if len(chunk) >= chunk_size:
            chunks.append(chunk)
            chunk = {}
    if chunk:
        chunks.append(chunk)
    return chunks


def replace_with_dict(text, translations):
    """使用字典查找替换文本"""
    for original, translated in translations.items():
        text = text.replace(original, translated)
    return text


def process_cell(cell_value, translations_chunks, progress_queue):
    """处理单个单元格的翻译"""
    if pd.notna(cell_value):
        for chunk in translations_chunks:
            cell_value = replace_with_dict(str(cell_value), chunk)
    progress_queue.put(("cell", 1))  # 发送单元格进度信息
    return cell_value


def process_excel_file_with_pandas(file_path, translations_chunks, progress_queue):
    """使用 pandas 处理单个 Excel 文件"""
    try:
        df = pd.read_excel(file_path, usecols=[1])
        total_cells = len(df)
        progress_queue.put(("sheet", "Sheet1", total_cells))  # 发送工作表进度信息

        with ThreadPoolExecutor(max_workers=4) as executor:
            futures = []
            for idx, cell_value in enumerate(df.iloc[:, 0]):
                futures.append(executor.submit(process_cell, cell_value, translations_chunks, progress_queue))

            # 更新单元格值
            for idx, future in enumerate(futures):
                df.iloc[idx, 0] = future.result()

        df.to_excel(file_path, index=False, header=False)
        progress_queue.put(("done",))  # 发送完成信息
        return True
    except Exception as e:
        print(f"处理文件 {file_path} 时出错: {e}")
        progress_queue.put(("error", str(e)))  # 发送错误信息
        return False


def process_folder_with_threads(folder_path, translations_chunks, progress_queue):
    """使用多线程处理文件夹中的所有 Excel 文件"""
    success_count = 0
    total_files = len([f for f in os.listdir(folder_path) if f.endswith('.xlsx')])
    if total_files == 0:
        messagebox.showwarning("警告", "选择的文件夹中没有 Excel 文件（.xlsx）")
        return

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []
        for file_name in os.listdir(folder_path):
            if file_name.endswith('.xlsx'):
                file_path = os.path.join(folder_path, file_name)
                future = executor.submit(process_excel_file_with_pandas, file_path, translations_chunks, progress_queue)
                futures.append(future)

        for future in futures:
            if future.result():
                success_count += 1
            progress_queue.put(("file", 1))  # 发送文件进度信息

    progress_queue.put(("finish", success_count, total_files))  # 发送完成信息


def update_progress_bar(progress_queue, file_pbar, sheet_pbar, cell_pbar):
    """更新进度条"""
    while True:
        msg = progress_queue.get()
        if msg[0] == "file":
            file_pbar.update(msg[1])  # 更新文件进度条
        elif msg[0] == "sheet":
            sheet_pbar.reset(total=msg[2])  # 重置工作表进度条
            sheet_pbar.set_description(f"处理工作表: {msg[1]}")
        elif msg[0] == "cell":
            sheet_pbar.update(msg[1])  # 更新单元格进度条
        elif msg[0] == "done":
            file_pbar.update(1)  # 更新文件进度条
        elif msg[0] == "finish":
            messagebox.showinfo("完成", f"处理完成！成功处理 {msg[1]}/{msg[2]} 个文件。")
            break
        elif msg[0] == "error":
            messagebox.showerror("错误", f"处理文件时出错: {msg[1]}")


def select_files_and_process():
    """选择 JSON 文件和 Excel 文件夹并开始处理"""
    json_path = filedialog.askopenfilename(
        title="选择 JSON 文件",
        filetypes=[("JSON 文件", "*.json")]
    )
    if not json_path:
        messagebox.showwarning("警告", "未选择 JSON 文件")
        return

    try:
        translations = load_and_sort_json(json_path)
    except Exception as e:
        messagebox.showerror("错误", f"加载 JSON 文件时出错: {e}")
        return

    translations_chunks = chunk_sorted_data(translations)

    folder_path = filedialog.askdirectory(
        title="选择包含 Excel 文件的文件夹"
    )
    if not folder_path:
        messagebox.showwarning("警告", "未选择文件夹")
        return

    progress_queue = queue.Queue()
    file_pbar = tqdm(total=len([f for f in os.listdir(folder_path) if f.endswith('.xlsx')]), desc="处理文件")
    sheet_pbar = tqdm(total=0, desc="处理工作表", leave=False)
    cell_pbar = tqdm(total=0, desc="处理单元格", leave=False)

    progress_thread = threading.Thread(target=update_progress_bar, args=(progress_queue, file_pbar, sheet_pbar, cell_pbar))
    progress_thread.start()

    process_thread = threading.Thread(target=process_folder_with_threads, args=(folder_path, translations_chunks, progress_queue))
    process_thread.start()


def main():
    """主函数"""
    root = tk.Tk()
    root.title("Excel 翻译工具")
    root.geometry("400x200")

    btn_select = tk.Button(root, text="选择文件并处理", command=select_files_and_process, width=20, height=3)
    btn_select.pack(pady=40)

    root.mainloop()


if __name__ == "__main__":
    main()

主要修改点

增加了 process_cell 函数：该函数负责处理单个单元格的翻译，允许在多线程中并行处理单元格。
在 process_excel_file_with_pandas 中使用 ThreadPoolExecutor：在处理单元格时，使用线程池来并行处理每个单元格的翻译。
更新进度信息：在处理单元格时，仍然通过 progress_queue 发送进度信息，以便更新进度条。

这样修改后，代码能够在处理多个 Excel 文件的同时，也能在处理单个 Excel 文件的单元格时实现多线程处理，从而提高整体效率。

多线程处理 Excel 文件及单元格翻译

使用 OpenAI o1 mini 由 OpenAI o1 回答

代码作用总结

代码要点

修改代码以实现更细粒度的多线程处理

主要修改点