自动化文件监控与分类压缩：实现高效文件管理

引言

在现代数据处理和文件管理中，如何高效地管理和归档大量文件是一个常见的挑战。特别是在需要根据文件类型进行分类并定期归档的场景下，手动操作不仅耗时且容易出错。为此，我们开发了一款基于Python的自动化工具，能够实时监控指定目录中的文件，并根据文件类型自动进行压缩归档。本文将详细介绍该工具的功能、实现原理及其应用场景。

功能概述

该工具的主要功能包括：

实时监控：持续监控指定目录中的文件变化。
按类型分组：根据文件扩展名对文件进行分类。
自动压缩：当某一类型的文件数量超过设定阈值（默认为5个）时，自动生成压缩包并将其移动到指定的输出目录。
删除原文件：为节省空间，压缩后的原文件会被删除。
用户交互：提供简单的用户交互界面，允许用户随时停止监控。

实现原理

1. 目录创建与初始化

首先，确保输出目录存在。如果不存在，则自动创建该目录。这一步骤保证了后续压缩包有地方存放。

if not os.path.exists(output_directory):os.makedirs(output_directory)print(f"Output directory '{output_directory}' created.")

2. 文件监控与分类

通过os.listdir()获取指定目录下的所有文件，并使用os.path.splitext()分离文件名和扩展名。然后，将相同扩展名的文件归为一组，存储在一个字典中，以便后续处理。

files = [f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))]
file_groups = {}
for file in files:_, ext = os.path.splitext(file)ext = ext.lower()if ext in file_groups:file_groups[ext].append(file)else:file_groups[ext] = [file]

3. 压缩归档

对于每种类型的文件，当其数量超过5个时，生成一个以当前时间戳命名的压缩包，并将这些文件添加到压缩包中。同时，删除已压缩的原文件以释放空间。

for ext, group in file_groups.items():if len(group) > 5:archive_date_num = datetime.datetime.now().strftime("%Y%m%d_%H:%M:%S")archive_prefix = ext.strip('.')archive_name = f"{archive_prefix}_{archive_date_num}.zip"zip_path = os.path.join(output_directory, archive_name)with zipfile.ZipFile(zip_path, 'w') as zipf:for file in group:file_path = os.path.join(directory, file)try:zipf.write(file_path, arcname=file)os.remove(file_path)print(f"Deleted {file_path}")except Exception as e:print(f"Failed to delete {file_path}: {e}")print(f"Created archive {zip_path} with {len(group)} files.")

4. 用户交互与循环监控

程序会每隔5秒检查一次文件变化，并询问用户是否继续监控。用户可以输入q来退出程序。

user_input = input("Press 'q' to quit, any other key to continue: ")
if user_input.lower() == 'q':print("Program stopped by user.")break
time.sleep(5)

应用场景

日志文件管理：对于生成大量日志文件的应用，可以定期压缩旧日志，防止磁盘空间被占用。
图片或文档库：在图片或文档库中，可以根据文件类型自动整理和归档，方便日后查找和管理。
临时文件清理：对于频繁生成临时文件的环境，可以自动清理不再需要的文件，保持系统整洁。

完整源码

import os
import datetime
import zipfile
import timedef monitor_and_zip_by_type(directory, output_directory):# 确保输出目录存在if not os.path.exists(output_directory):os.makedirs(output_directory)  # 创建输出目录print(f"Output directory '{output_directory}' created.")while True:try:# 获取指定目录下所有文件files = [f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))]file_groups = {}  # 用于存放不同类型文件的字典# 将文件按扩展名分组for file in files:_, ext = os.path.splitext(file)  # 分离文件扩展名ext = ext.lower()  # 转换为小写以统一处理if ext in file_groups:file_groups[ext].append(file)  # 添加到对应扩展名的列表中else:file_groups[ext] = [file]  # 创建新的扩展名列表# 遍历每种文件类型及其文件列表for ext, group in file_groups.items():if len(group) > 5:  # 如果同类型文件数量超过5个# 获取当前时间作为压缩包的日期标识archive_date_num = datetime.datetime.now().strftime("%Y%m%d_%H:%M:%S")archive_prefix = ext.strip('.')  # 使用文件类型作为压缩包前缀archive_name = f"{archive_prefix}_{archive_date_num}.zip"  # 生成压缩包名称zip_path = os.path.join(output_directory, archive_name)  # 压缩包完整路径# 创建压缩包并写入文件with zipfile.ZipFile(zip_path, 'w') as zipf:for file in group:file_path = os.path.join(directory, file)  # 获取文件完整路径try:zipf.write(file_path, arcname=file)  # 写入文件到压缩包os.remove(file_path)  # 删除已压缩的原文件print(f"Deleted {file_path}")  # 输出删除文件的信息except Exception as e:print(f"Failed to delete {file_path}: {e}")  # 捕获删除失败的异常print(f"Created archive {zip_path} with {len(group)} files.")  # 输出压缩结果信息except Exception as e:print(f"An error occurred: {e}")  # 捕获并输出其他异常# 用户输入决定是否继续监控user_input = input("Press 'q' to quit, any other key to continue: ")if user_input.lower() == 'q':  # 如果用户输入'q'则退出print("Program stopped by user.")breaktime.sleep(5)  # 暂停5秒后继续监控# 获取用户输入的监控目录和输出目录
monitor_directory = input('请输入监控的目录：').strip()
if not monitor_directory:  # 如果未输入，则使用当前工作目录monitor_directory = os.getcwd()output_directory = input('请输入压缩文件的目录：').strip()
if not output_directory:  # 如果未输入，则使用当前工作目录output_directory = os.getcwd()# 启动监控和压缩功能
monitor_and_zip_by_type(monitor_directory, output_directory)