在程序工具的开发过程中，监控文件的变化对于许多应用程序来说至关重要。

无论是自动化构建系统、文件同步工具还是相关的自动化程序，都需要对系统中文件的变化（创建、修改、删除等）做出及时响应。

而 Python Watchdog 库 就是一个强大的工具，可以帮助我们实现这一目标。

Watchdog介绍

Watchdog 是一个用于监视文件系统变化的库，它提供了一种简单而灵活的方式来监视文件或目录的创建、修改、删除和移动等操作。

无论是在本地文件系统还是网络文件系统中，Watchdog都能够高效地捕获这些变化，并触发相应的事件。

Watchdog库具有以下特色功能：

• 跨平台性：可以在Windows、Linux和macOS等主流操作系统上运行。
• 灵活性：可以通过自定义处理程序来响应不同类型的文件系统事件。
• 高效性：采用异步事件处理机制，能够高效地处理大量文件系统事件。
• 易用性：提供简洁清晰的API，便于开发人员快速上手。

安装与使用

安装 Watchdog 库

首先，我们需要使用 pip 安装 Watchdog 库：

pip install watchdog

安装完成以后，就可以在Python程序开发时使用了。

Watchdog 四大API方法（事件处理）

1、创建事件（文件或目录被创建）

要监视文件或目录的创建事件，可以在自定义的处理程序类中重写 on_created 方法，并在其中处理相应的逻辑。

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        print(f'文件 {event.src_path} 被创建')

2、修改事件（文件或目录被修改）

要监视文件或目录的修改事件，可以在自定义的处理程序类中重写 on_modified 方法，并在其中处理相应的逻辑。

class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        print(f'文件 {event.src_path} 被修改')

3、移动事件（文件或目录被移动）

要监视文件或目录的移动事件，可以在自定义的处理程序类中重写 on_moved 方法，并在其中处理相应的逻辑。

class MyHandler(FileSystemEventHandler):
    def on_moved(self, event):
        if event.is_directory:
            print(f'目录 {event.src_path} 被移动到 {event.dest_path}')
        else:
            print(f'文件 {event.src_path} 被移动到 {event.dest_path}')

4、删除事件（文件或目录被删除）

要监视文件或目录的删除事件，可以在自定义的处理程序类中重写 on_deleted 方法，并在其中处理相应的逻辑。

class MyHandler(FileSystemEventHandler):
    def on_deleted(self, event):
        if event.is_directory:
            print(f'目录 {event.src_path} 被删除')
        else:
            print(f'文件 {event.src_path} 被删除')

当然，小编为帮助大家更好地理解 Watchdog 库 API 的使用，为大家展示一个修改的示例，已更深入的理解它！

编写一个监控文件修改的示例

1、导入所需的模块

# 导入监听器类（观察者）
from watchdog.observers import Observer
# 导入文件系统事件类
from watchdog.events import FileSystemEventHandler

2、创建一个自定义的处理程序类，继承自 FileSystemEventHandler，并重写相应的方法来处理文件系统事件：

class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        print(f'文件 {event.src_path} 被修改')

3、创建一个观察者对象并将处理程序类与要监视的路径关联起来：

observer = Observer()
observer.schedule(MyHandler(), path='要监视的路径', recursive=True)
observer.start()

4、在需要的时候停止观察

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

5、完整代码示例，演示如何使用 Watchdog 监视文件系统中目录（包括其子目录及文件）的修改事件：

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

class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        print(f'文件 {event.src_path} 被修改')

if __name__ == "__main__":
    path = '.'  # 监视当前目录
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

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

上面为大家介绍了目录的监控方式，如果只想监控某一个文件的变化，只需要将 recursive 改为 False 即可。

...
if __name__ == "__main__":
    path = 'e:\src\event_202401281535.log'  # 监视当前文件
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=False)
    observer.start()
    
    ...

Watchdog可以使用在哪些业务场景下？

Watchdog可应用于诸多场景，包括但不限于：

• 自动化构建系统：监控源代码文件的变化，自动触发构建和部署操作。

class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        print(f'检测到文件 {event.src_path} 的修改，开始自动化构建...')
        # 在这里添加触发构建任务的代码

• 文件同步工具：实时监控文件变化，保持多个目录之间的同步。

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        print(f'检测到文件 {event.src_path} 的创建，开始同步到目标目录...')
        # 在这里添加文件同步逻辑的代码

    def on_modified(self, event):
        if event.is_directory:
            return
        print(f'检测到文件 {event.src_path} 的修改，开始同步到目标目录...')
        # 在这里添加文件同步逻辑的代码

    def on_deleted(self, event):
        if event.is_directory:
            return
        print(f'检测到文件 {event.src_path} 的删除，开始同步到目标目录...')
        # 在这里添加文件同步逻辑的代码

• 数据备份应用：监控数据文件的修改，及时备份重要数据。

class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        print(f'检测到文件 {event.src_path} 的修改，开始备份数据...')
        # 在这里添加触发数据备份操作的代码

    def on_created(self, event):
        if event.is_directory:
            return
        print(f'检测到文件 {event.src_path} 的创建，开始备份数据...')
        # 在这里添加触发数据备份操作的代码

• 日志分析：通过日志文件的变化，记录、定位系统发生问题。

class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        print(f'检测到日志文件 {event.src_path} 的修改，开始实时分析日志...')
        # 在这里添加实时日志分析的代码

Watchdog现有的弊端：

文件操作引发的事件比我们想象的只多不少，而且难以在事件函数中做出针对性处理。

在 windows 平台上文件修改可能引发两次 modified 事件？在 windows 平台上，由于 watchdog 封装的是 windows 系统的 FileSystemWatcher Events，处理文件的过程中执行了多次文件系统操作，无法避免地触发了多次事件。

改进方案：

如果对监视文件的实时性要求不高，又懒得处理一大堆事件，那么，比较事件前后的文件夹快照就是一个值得尝试的改进方案。实现这个思路，有三个前提条件：

• 快速获取文件夹快照。幸运的是，watchdog 模块为我们提供了 DirectorySnapshot 功能
• 可以接受200毫秒的延时。文件操作引发的一大堆事件会集中在一个较短的时间内，一般情况下，在文件操作之后200毫秒获取文件夹快照，是一个不错的间隔
• 快速比较文件夹快照。这也不是问题，watchdog 模块有 DirectorySnapshotDiff 子模块

改进思路：设置一个定时器， 200毫秒后抓取快照，并与上一张快照比较。每当有事件发生时，检查定时器是否已经启动。如果未启动，则直接启动定时器；否则，说明该事件距离上次事件不足200毫秒，可视为是同一组事件，此时终止定时器，再次重启。

解决代码示例：

import time
import os, threading
from watchdog.observers import Observer
from watchdog.events import *
from watchdog.utils.dirsnapshot import DirectorySnapshot, DirectorySnapshotDiff

class FileEventHandler(FileSystemEventHandler):
    def __init__(self, aim_path):
        FileSystemEventHandler.__init__(self)
        self.aim_path = aim_path
        self.timer = None
        self.snapshot = DirectorySnapshot(self.aim_path)
    
    def on_any_event(self, event):
        if self.timer:
            self.timer.cancel()
        
        self.timer = threading.Timer(0.2, self.checkSnapshot)
        self.timer.start()
    
    def checkSnapshot(self):
        snapshot = DirectorySnapshot(self.aim_path)
        diff = DirectorySnapshotDiff(self.snapshot, snapshot)
        self.snapshot = snapshot
        self.timer = None
        
        print("files_created:", diff.files_created)
        print("files_deleted:", diff.files_deleted)
        print("files_modified:", diff.files_modified)
        print("files_moved:", diff.files_moved)
        print("dirs_modified:", diff.dirs_modified)
        print("dirs_moved:", diff.dirs_moved)
        print("dirs_deleted:", diff.dirs_deleted)
        print("dirs_created:", diff.dirs_created)

class DirMonitor(object):
    """文件夹监视类"""
    
    def __init__(self, aim_path):
        """构造函数"""
        self.aim_path= aim_path
        self.observer = Observer()
    
    def start(self):
        """启动"""
        event_handler = FileEventHandler(self.aim_path)
        self.observer.schedule(event_handler, self.aim_path, True)
        self.observer.start()
    
    def stop(self):
        """停止"""
        self.observer.stop()
    
if __name__ == "__main__": 
    monitor = DirMonitor(r"./monitor_folder_1") 
    monitor.start() 
    try: 
        while True: 
            time.sleep(1) 
    except KeyboardInterrupt: 
        monitor.stop() 

总结

Watchdog 是一个功能强大且易于使用的文件系统监控API工具（Python），它为开发人员提供了一种简单而灵活的方式来响应文件系统变化。

无论是在个人项目中还是企业级应用中，Watchdog都能够发挥重要作用，帮助我们实现对文件系统变化的实时监控和响应。

希望通过本文的介绍，让大家对Python Watchdog库有了更深入的了解，并能够在实际项目中灵活运用它来解决各种文件系统监控的需求。