Python 监控文件变化的 watchdog 模块

在本文中，我们将介绍如何使用 Python 的 watchdog 模块来监控文件的变化。监控文件变化在许多应用中是很有用的，比如文件同步、日志监控以及实时数据处理等。

什么是 watchdog ？

watchdog 是一个用于监控文件系统变化的 Python 库。它提供了一种简单而强大的方式来检测和处理文件或文件夹的变化。watchdog 可以检测到文件的创建、删除、重命名、修改等事件，从而帮助我们实时地处理这些变化。

安装 watchdog 模块

在使用 watchdog 模块之前，我们首先需要安装它。可以使用 pip 命令来安装 watchdog：

pip install watchdog

安装完成后，我们就可以在 Python 代码中引入 watchdog 模块来使用了。

使用 watchdog 监控文件的变化

下面我们来看一个简单的例子，展示了如何使用 watchdog 监控指定文件的变化，并在文件发生变化时触发相应的操作。

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 继承 FileSystemEventHandler 自定义文件事件处理类
class MyEventHandler(FileSystemEventHandler):
    def on_modified(self, event):
        # 文件被修改时的处理逻辑
        print("文件被修改了！")

# 创建 Observer 对象，并指定文件事件处理类
observer = Observer()
event_handler = MyEventHandler()
observer.schedule(event_handler, path='.', recursive=False)

# 启动 observer
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()

# 停止 observer
observer.join()

在这个例子中，我们首先导入了 watchdog 的相关模块，然后定义了一个继承自 FileSystemEventHandler 的自定义文件事件处理类 MyEventHandler。在这个类中，我们可以重写各种文件事件的处理方法，比如 on_created、on_deleted、on_modified 等。在这个例子中，我们只重写了 on_modified 方法，即文件被修改时要执行的逻辑。你可以根据自己的需求重写其他方法。

接下来，我们创建了一个 Observer 对象，并使用 MyEventHandler 类作为文件事件处理类。通过调用 Observer 对象的 schedule 方法，我们可以指定要监控的目录或文件路径。在这个例子中，我们监控的是当前目录下的所有文件，recursive=False 表示不会递归监控子目录。

最后，我们通过调用 start 方法启动了 Observer，并进入一个无限循环来监控文件的变化。按下 Ctrl + C 可以退出程序。

监控文件夹的变化

除了监控单个文件的变化，watchdog 还支持监控整个文件夹的变化。我们可以通过监听 on_created 和 on_deleted 这两个事件来实现对文件夹的监控。

以下是一个示例代码：

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 继承 FileSystemEventHandler 自定义文件事件处理类
class MyEventHandler(FileSystemEventHandler):
    def on_created(self, event):
        # 文件或文件夹被创建时的处理逻辑
        print("文件或文件夹被创建了！")

    def on_deleted(self, event):
        # 文件或文件夹被删除时的处理逻辑
        print("文件或文件夹被删除了！")

# 创建 Observer 对象，并指定文件事件处理类
observer = Observer()
event_handler = MyEventHandler()
observer.schedule(event_handler, path='.', recursive=True)

# 启动 observer
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()

# 停止 observer
observer.join()