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

引言

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

功能概述

该工具的主要功能包括：

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

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

if not os.path.exists(output_directory):
    os.makedirs(output_directory)
    print(f"Output directory '{output_directory}' created.")
  • 1.
  • 2.
  • 3.
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]
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
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.")
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
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)
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
应用场景
  1. 日志文件管理：对于生成大量日志文件的应用，可以定期压缩旧日志，防止磁盘空间被占用。
  2. 图片或文档库：在图片或文档库中，可以根据文件类型自动整理和归档，方便日后查找和管理。
  3. 临时文件清理：对于频繁生成临时文件的环境，可以自动清理不再需要的文件，保持系统整洁。
完整源码
import os
import datetime
import zipfile
import time


def 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.")
            break
        time.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)
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.
  • 35.
  • 36.
  • 37.
  • 38.
  • 39.
  • 40.
  • 41.
  • 42.
  • 43.
  • 44.
  • 45.
  • 46.
  • 47.
  • 48.
  • 49.
  • 50.
  • 51.
  • 52.
  • 53.
  • 54.
  • 55.
  • 56.
  • 57.
  • 58.
  • 59.
  • 60.
  • 61.
  • 62.
  • 63.
  • 64.
  • 65.
  • 66.
  • 67.
  • 68.
  • 69.
  • 70.
  • 71.
结论

通过上述工具，我们可以轻松实现文件的自动化监控与分类压缩，极大地提高了文件管理的效率和准确性。无论是个人用户还是企业级应用，都可以从中受益。未来，还可以进一步优化此工具，例如增加更多的配置选项、支持多线程处理等，以满足更多复杂的需求。

希望这篇文章能帮助你更好地理解这个工具的功能和应用场景。如果有任何问题或改进建议，欢迎随时交流！