【Python 打造高效文件分类工具】

诗林  金牌会员 | 2025-2-18 20:15:12 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题

主题 578|帖子 578|积分 1734

该代码基于 Python 的 tkinter 库实现了 “文件分类工具 - 优化版” 图形化桌面应用。用户能通过界面选择文件夹和要分类的文件范例,支持全选或自定义部分文件扩展名进行分类。采用多线程技能执行文件整理任务,避免界面卡顿且可中途制止。利用 os.scandir 高效遍历文件并批量分组处置惩罚,对大文件(超 500MB)增加耽误处置惩罚并记录日志。通过消息队列记录整理过程,包含跳过文件、移动文件及错误信息等,支持将日志生存为文本文件。整理完成后表现总文件数、成功移动数、失败次数和跳过文件数等统计信息,还可直接打开整理后的文件夹查看结果。
一、代码团体结构

我们的文件分类工具基于 Python 的 Tkinter 库构建,Tkinter 是 Python 的尺度 GUI(Graphical User Interface,图形用户界面)库,它提供了丰富的组件来创建交互式应用步调。整个代码围绕一个名为FileOrganizerApp的类睁开,这个类负责管理界面元素、用户交互以及文件整理的核心逻辑。
二、关键代码剖析

(一)初始化部分

  1. class FileOrganizerApp:
  2.     def __init__(self, root):
  3.         self.root = root
  4.         self.root.title("文件分类工具-优化版")
  5.         self.root.geometry("800x600")
  6.         # 初始化统计变量
  7.         self.total_files = 0
  8.         self.moved_count = 0
  9.         self.error_count = 0
  10.         self.current_progress = 0
  11.         # 初始化文件类型选择相关变量
  12.         self.all_files_var = tk.BooleanVar()
  13.         self.checkbox_vars = {}
  14.         self.checkboxes = {}
  15.         self.preset_extensions = [
  16.             "txt", "doc", "docx", "pdf", "jpg",
  17.             "png", "gif", "mp3", "mp4", "xls",
  18.             "xlsx", "zip", "rar", "ppt", "pptx"
  19.         ]
  20.         # 创建消息队列
  21.         self.log_queue = queue.Queue()
  22.         self.create_widgets()
  23.         self.running = False
  24.         self.stop_event = threading.Event()
复制代码
在__init__ 方法中,我们首先设置了应用步调窗口的标题和初始巨细。接着,初始化了一系列统计变量,用于记录整理过程中的文件总数、成功移动的文件数、出错的文件数以及当前的进度。
为了让用户能够选择要分类的文件范例,我们定义了all_files_var布尔变量用于全选功能,checkbox_vars字典来存储每个文件扩展名对应的复选框状态,checkboxes字典用于存储复选框组件,preset_extensions列表则列出了我们预设支持分类的文件扩展名。
消息队列log_queue用于在不同线程间传递日志信息,确保界面的流畅更新。末了,调用create_widgets方法创建界面元素,并初始化一些控制变量。
(二)界面创建部分

  1. def create_widgets(self):
  2.     # 文件夹选择部分
  3.     frame_top = ttk.Frame(self.root, padding=10)
  4.     frame_top.pack(fill=tk.X)
  5.     self.btn_select = ttk.Button(frame_top, text="选择文件夹", command=self.select_directory)
  6.     self.btn_select.pack(side=tk.LEFT, padx=5)
  7.     self.path_var = tk.StringVar()
  8.     self.entry_path = ttk.Entry(frame_top, textvariable=self.path_var, width=60)
  9.     self.entry_path.pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True)
  10.     # 文件类型选择部分
  11.     type_frame = ttk.LabelFrame(self.root, text="选择要分类的文件类型", padding=10)
  12.     type_frame.pack(fill=tk.BOTH, expand=True, pady=5)
  13.     # 全选复选框
  14.     ttk.Checkbutton(
  15.         type_frame,
  16.         text="全部分类",
  17.         variable=self.all_files_var,
  18.         command=self.toggle_checkboxes
  19.     ).grid(row=0, column=0, sticky=tk.W, padx=5)
  20.     # 创建扩展名复选框
  21.     for idx, ext in enumerate(self.preset_extensions):
  22.         self.checkbox_vars[ext] = tk.BooleanVar()
  23.         cb = ttk.Checkbutton(
  24.             type_frame,
  25.             text=ext.upper(),
  26.             variable=self.checkbox_vars[ext]
  27.         )
  28.         cb.grid(row=idx // 5 + 1, column=idx % 5, sticky=tk.W, padx=5)
  29.         self.checkboxes[ext] = cb
  30.     # 控制按钮部分
  31.     frame_controls = ttk.Frame(self.root, padding=10)
  32.     frame_controls.pack(fill=tk.X)
  33.     self.btn_start = ttk.Button(frame_controls, text="开始整理", command=self.start_organize)
  34.     self.btn_start.pack(side=tk.LEFT, padx=5)
  35.     self.btn_stop = ttk.Button(frame_controls, text="停止", command=self.stop_organize, state=tk.DISABLED)
  36.     self.btn_stop.pack(side=tk.LEFT, padx=5)
  37.     # 保存日志按钮
  38.     self.btn_save_log = ttk.Button(frame_controls, text="保存日志", command=self.save_log, state=tk.DISABLED)
  39.     self.btn_save_log.pack(side=tk.LEFT, padx=5)
  40.     # 查看文件按钮,初始状态为禁用
  41.     self.btn_view_files = ttk.Button(frame_controls, text="查看文件", command=self.open_folder, state=tk.DISABLED)
  42.     self.btn_view_files.pack(side=tk.LEFT, padx=5)
  43.     # 日志显示部分
  44.     frame_log = ttk.Frame(self.root, padding=10)
  45.     frame_log.pack(fill=tk.BOTH, expand=True)
  46.     self.log_area = scrolledtext.ScrolledText(frame_log, wrap=tk.WORD)
  47.     self.log_area.pack(fill=tk.BOTH, expand=True)
  48.     # 进度条
  49.     self.progress = ttk.Progressbar(self.root, orient=tk.HORIZONTAL, mode='determinate')
  50.     self.progress.pack(fill=tk.X, padx=10, pady=5)
复制代码
create_widgets方法负责构建应用步调的整个界面。它分为几个重要部分:

  • 文件夹选择部分:包含一个按钮btn_select用于打开文件选择对话框,以及一个输入框entry_path用于表现用户选择的文件夹路径。
  • 文件范例选择部分:通过一个标签框架type_frame包含一个全选复选框和多个文件扩展名复选框。全选复选框的command参数绑定到toggle_checkboxes方法,用于切换全部扩展名复选框的状态。
  • 控制按钮部分:包罗开始整理按钮btn_start、制止按钮btn_stop(初始状态为禁用)、生存日志按钮btn_save_log(初始状态为禁用)和查看文件按钮btn_view_files(初始状态为禁用)。这些按钮分别绑定到对应的功能方法。
  • 日志表现部分:使用scrolledtext.ScrolledText组件创建一个可滚动的文本区域log_area,用于表现文件整理过程中的日志信息。
  • 进度条部分:通过ttk.Progressbar创建一个水平进度条progress,用于直观展示文件整理的进度。
(三)核心功能部分

  1. def start_organize(self):
  2.     if not self.path_var.get():
  3.         messagebox.showwarning("警告", "请先选择要整理的文件夹")
  4.         return
  5.     if self.running:
  6.         return
  7.     self.running = True
  8.     self.stop_event.clear()
  9.     self.btn_start['state'] = tk.DISABLED
  10.     self.btn_stop['state'] = tk.NORMAL
  11.     self.btn_save_log['state'] = tk.DISABLED
  12.     self.btn_view_files['state'] = tk.DISABLED
  13.     self.progress['value'] = 0
  14.     self.log_area.delete(1.0, tk.END)
  15.     # 启动队列处理
  16.     self.process_log_queue()
  17.     self.update_progress()
  18.     worker = threading.Thread(target=self.organize_files, daemon=True)
  19.     worker.start()
复制代码
start_organize方法是启动文件整理流程的入口。首先,它检查用户是否选择了要整理的文件夹路径,如果没有则弹出告诫框提示用户。然后,检查当前是否已经在运行整理任务,如果是则直接返回。
接着,设置运行状态变量running为True,清除制止变乱stop_event,并根据任务状态更新界面按钮的状态,清空进度条和日志区域。
为了确保日志能够及时更新,调用process_log_queue方法启动日志队列处置惩罚,同时调用update_progress方法开始更新进度条。末了,创建一个新的线程来执行核心的文件整理逻辑organize_files,并将线程设置为守护线程,如许当主线程结束时,该线程也会自动结束。
  1. def organize_files(self):
  2.     target_dir = self.path_var.get()
  3.     try:
  4.         # 使用更高效的scandir遍历文件
  5.         files = []
  6.         with os.scandir(target_dir) as entries:
  7.             for entry in entries:
  8.                 if entry.is_file():
  9.                     files.append(entry.name)
  10.         self.total_files = len(files)
  11.         processed_files = 0
  12.         # 批量分组处理文件
  13.         selected_exts = None
  14.         if not self.all_files_var.get():
  15.             selected_exts = {ext for ext, var in self.checkbox_vars.items() if var.get()}
  16.         file_groups = defaultdict(list)
  17.         skipped_count = 0
  18.         for filename in files:
  19.             if self.stop_event.is_set():
  20.                 break
  21.             _, ext = os.path.splitext(filename)
  22.             ext = ext.lower().lstrip('.') if ext else 'no_extension'
  23.             if selected_exts is None or ext in selected_exts:
  24.                 file_groups[ext].append(filename)
  25.             else:
  26.                 skipped_count += 1
  27.         # 处理跳过的文件
  28.         self.log_queue.put(f"已跳过 {skipped_count} 个未选类型的文件")
  29.         # 批量移动文件
  30.         total_to_process = sum(len(v) for v in file_groups.values())
  31.         processed = 0
  32.         for ext, filenames in file_groups.items():
  33.             if self.stop_event.is_set():
  34.                 break
  35.             dest_dir = os.path.join(target_dir, ext)
  36.             os.makedirs(dest_dir, exist_ok=True)
  37.             for filename in filenames:
  38.                 if self.stop_event.is_set():
  39.                     break
  40.                 file_path = os.path.join(target_dir, filename)
  41.                 try:
  42.                     # 大文件处理(超过500MB时增加延迟)
  43.                     file_size = os.path.getsize(file_path)
  44.                     if file_size > 500 * 1024 * 1024:
  45.                         self.log_queue.put(f"正在处理大文件: {filename} ({file_size // 1024 // 1024}MB)")
  46.                         time.sleep(0.5)
  47.                     shutil.move(file_path, os.path.join(dest_dir, filename))
  48.                     self.moved_count += 1
  49.                     # 每处理50个文件更新一次日志
  50.                     if self.moved_count % 50 == 0:
  51.                         self.log_queue.put(f"已移动 {self.moved_count} 个文件")
  52.                 except Exception as e:
  53.                     self.error_count += 1
  54.                     self.log_queue.put(f"错误: {filename} ({str(e)})")
  55.                 processed += 1
  56.                 self.current_progress = (processed / total_to_process) * 100
  57.         self.root.after(10, self.organize_complete)
  58.     except Exception as e:
  59.         self.log_queue.put(f"系统错误: {str(e)}")
  60.         self.root.after(10, self.organize_complete)
复制代码
organize_files方法是文件分类的核心逻辑所在。它首先获取用户选择的目标文件夹路径target_dir。然后,使用os.scandir函数更高效地遍历目标文件夹中的全部文件,将文件列表存储在files变量中,并记录文件总数total_files。
接下来,根据用户在界面上选择的文件范例进行分组处置惩罚。如果用户选择了全部分类(all_files_var为True),则selected_exts为None,表示处置惩罚全部文件;否则,通过列表推导式获取用户勾选的文件扩展名集合selected_exts。
使用defaultdict创建file_groups字典,将文件按扩展名分组。在遍历文件过程中,如果文件范例不在用户选择的范围内,则跳过该文件并记录跳过的文件数。
在批量移动文件阶段,首先计算必要处置惩罚的文件总数total_to_process。对于每个文件组,创建对应的目标文件夹(如果不存在),然后逐个移动文件。在移动文件时,增加了对大文件(超过 500MB)的处置惩罚逻辑,当碰到大文件时,在日志中记录并增加 0.5 秒的耽误,以避免在处置惩罚大文件时导致系统卡顿。同时,每成功移动 50 个文件,在日志中记录已移动的文件数。如果移动过程中出现错误,记录错误信息并增加错误计数。
在整个过程中,根据已处置惩罚文件数和总文件数实时更新当进步度current_progress。末了,通过root.after方法在 10 毫秒后调用organize_complete方法,用于处置惩罚整理完成后的后续操作。
(四)其他辅助功能部分

  1. def stop_organize(self):
  2.     self.stop_event.set()
  3.     self.log_queue.put("正在停止整理进程...")
复制代码
stop_organize方法用于制止文件整理任务。它通过设置stop_event变乱,关照正在执行文件整理的线程制止操作,并在日志队列中添加制止进程的提示信息。
  1. def process_log_queue(self):
  2.     try:
  3.         while True:
  4.             msg = self.log_queue.get_nowait()
  5.             self.log_area.insert(tk.END, msg + "\n")
  6.             self.log_area.see(tk.END)
  7.     except queue.Empty:
  8.         pass
  9.     self.root.after(100, self.process_log_queue)
复制代码
process_log_queue方法负责从日志队列log_queue中获取日志信息,并将其表现在日志区域log_area中。它通过一个循环不断实验从队列中获取消息,使用get_nowait方法避免阻塞。如果队列为空,捕捉queue.Empty非常并跳过。每次获取并表现消息后,通过root.after方法设置 100 毫秒后再次调用自身,以实现实时更新日志。
  1. def update_progress(self):
  2.     self.progress['value'] = self.current_progress
  3.     self.root.after(100, self.update_progress)
复制代码
update_progress方法用于更新进度条的表现。它将当进步度current_progress的值设置到进度条progress上,并通过root.after方法设置 100 毫秒后再次调用自身,以实现进度条的实时更新。
  1. def organize_complete(self):
  2.     self.running = False
  3.     self.btn_start['state'] = tk.NORMAL
  4.     self.btn_stop['state'] = tk.DISABLED
  5.     self.btn_save_log['state'] = tk.NORMAL
  6.     # 整理完成后启用查看文件按钮
  7.     self.btn_view_files['state'] = tk.NORMAL
  8.     self.current_progress = 100
  9.     stats_message = (
  10.         f"\n整理完成:\n"
  11.         f"总文件数: {self.total_files}\n"
  12.         f"成功移动: {self.moved_count}\n"
  13.         f"失败次数: {self.error_count}\n"
  14.         f"跳过文件: {self.total_files - self.moved_count - self.error_count}"
  15.     )
  16.     self.log_queue.put(stats_message)
  17.     messagebox.showinfo("整理统计", stats_message)
复制代码
organize_complete方法在文件整理完成后被调用。它首先将运行状态变量running设置为False,并规复界面按钮的初始状态,启用开始按钮、禁用制止按钮、启用生存日志按钮和查看文件按钮。同时,将进度条设置为 100%。
然后,生成整理统计信息字符串stats_message,包含总文件数、成功移动的文件数、失败次数和跳过的文件数。将统计信息添加到日志队列中,并通过messagebox.showinfo弹出对话框表现整理统计结果。
  1. def save_log(self):
  2.     log_text = self.log_area.get(1.0, tk.END)
  3.     file_path = filedialog.asksaveasfilename(defaultextension=".txt", filetypes=[("Text files", "*.txt")])
  4.     if file_path:
  5.         try:
  6.             with open(file_path, 'w', encoding='utf-8') as f:
  7.                 f.write(log_text)
  8.             messagebox.showinfo("保存成功", "日志已成功保存。")
  9.         except Exception as e:
  10.             messagebox.showerror("保存失败", f"保存日志时出现错误:{str(e)}")
复制代码
save_log方法用于将日志区域中的内容生存为文本文件。它首先获取日志区域的全部文本内容log_text,然后通过filedialog.asksaveasfilename打开文件生存对话框,让用户选择生存路径和文件名。如果用户选择了生存路径,实验将日志内容写入文件。如果生存成功,弹出提示框告知用户;如果生存过程中出现错误,捕捉非常并弹出错误提示框。
  1. def open_folder(self):
  2.     folder = self.path_var.get()
  3.     if folder:
  4.         try:
  5.             if os.name == 'nt':  # Windows 系统
  6.                 os.startfile(folder)
  7.             elif os.name == 'posix':  # Linux 或 macOS 系统
  8.                 webbrowser.open(folder)
  9.         except Exception as e:
  10.             messagebox.showerror("错误", f"打开文件夹时出错: {str(e)}")
复制代码
open_folder方法用于在文件整理完成后,打开用户选择的目标文件夹。它
首先获取用户在界面上选择的文件夹路径folder。如果路径存在,根据当前操作系统范例进行不同的操作:对于 Windows 系统,使用os.startfile函数打开文件夹;对于 Linux 或 macOS 系统,使用webbrowser.open函数打开文件夹。如果在打开过程中出现非常,捕捉非常并通过messagebox.showerror弹出错误提示框,告知用户打开文件夹时出现的错误信息。
三、运行与使用

当你运行这段代码时,会弹出一个图形化界面的文件分类工具窗口。在窗口中,你可以通过点击 “选择文件夹” 按钮选择必要整理文件的文件夹路径,该路径会表现在输入框中。
在 “选择要分类的文件范例” 区域,你可以选择 “全部分类”,也可以单独勾选必要分类的文件扩展名范例,如 “TXT”“JPG”“PDF” 等。完成选择后,点击 “开始整理” 按钮,工具将开始扫描文件夹内的文件,并按照你选择的范例进行分类整理。
在整理过程中,进度条会实时表现整理进度,日志区域会记录整理过程中的详细信息,包罗每个文件的处置惩罚情况、错误信息等。如果在整理过程中你想制止操作,可以点击 “制止” 按钮。整理完成后,你可以点击 “生存日志” 按钮将整理过程中的日志信息生存为文本文件,也可以点击 “查看文件” 按钮直接打开整理后的文件夹查看分类结果。
四、示图


五、作者有话说

代码功能已颠末严格测试,确认无误。然而,值得注意的是,测试集的巨细仅为3个G,相对而言规模较小,因此无法百分之百包管在全部情况下均无任何题目。鉴于此,猛烈建议在使用之前做好数据备份,以防万一出现数据丢失的情况。如果您在使用过程中发现任何bug或题目,欢迎随时留言反馈,作者将及时相应并进行相应的修改。

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看,没有账号?立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

诗林

金牌会员
这个人很懒什么都没写!

标签云

快速回复 返回顶部 返回列表