杰瑞科技汇

Python Watchdog示例如何监控文件变化?

环境准备

你需要安装 watchdog 库,打开你的终端或命令行工具,运行:

Python Watchdog示例如何监控文件变化?-图1
(图片来源网络,侵删)
pip install watchdog

示例 1:基础监控器(打印所有事件)

这是最简单的示例,它会监控指定目录下的所有文件系统事件,并在控制台打印出来。

代码 (basic_watcher.py)

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# 1. 定义一个事件处理类
class MyHandler(FileSystemEventHandler):
    """
    自定义事件处理类,继承自 FileSystemEventHandler。
    我们可以重写它的方法来响应特定的事件。
    """
    def on_any_event(self, event):
        """
        当任何文件系统事件发生时被调用。
        event 对象包含了事件的详细信息。
        """
        print(f"事件类型: {event.event_type} | 路径: {event.src_path}")
if __name__ == "__main__":
    # 2. 指定要监控的目录
    path_to_watch = "."  # 监控当前目录
    # 3. 创建事件处理器实例
    event_handler = MyHandler()
    # 4. 创建观察者实例
    observer = Observer()
    # 5. 将观察者与事件处理器和要监控的路径关联起来
    observer.schedule(event_handler, path_to_watch, recursive=True)
    # 6. 启动观察者(在后台线程中运行)
    print(f"开始监控目录: {path_to_watch}")
    observer.start()
    try:
        # 7. 保持主线程运行,以便观察者可以持续工作
        # 当按下 Ctrl+C 时,会捕获 KeyboardInterrupt 异常并停止观察者
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    # 8. 等待观察者线程完全停止
    observer.join()
    print("监控已停止。")

如何运行

  1. 将上面的代码保存为 basic_watcher.py

  2. 在同一个目录下创建一些文件或文件夹,或者修改、删除现有文件。

  3. 在终端中运行脚本:

    Python Watchdog示例如何监控文件变化?-图2
    (图片来源网络,侵删)
    python basic_watcher.py

预期输出

你会在终端看到类似下面的输出(取决于你的操作):

开始监控目录: .
事件类型: created | 路径: ./test.txt
事件类型: modified | 路径: ./test.txt
事件类型: deleted | 路径: ./test.txt

示例 2:特定事件处理器(只响应文件修改和创建)

在实际应用中,我们通常只关心特定的事件,这个示例只处理文件(非目录)的创建和修改事件。

代码 (specific_events_handler.py)

import time
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class SpecificEventHandler(FileSystemEventHandler):
    def on_created(self, event):
        # 只处理文件创建事件,忽略目录
        if not event.is_directory:
            print(f"检测到文件创建: {event.src_path}")
            # 在这里可以添加你的逻辑,例如读取文件内容等
    def on_modified(self, event):
        # 只处理文件修改事件,忽略目录
        if not event.is_directory:
            print(f"检测到文件修改: {event.src_path}")
            # 在这里可以添加你的逻辑,例如重新加载配置等
if __name__ == "__main__":
    # 监控当前目录
    path = "."
    event_handler = SpecificEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    print(f"开始监控目录: {path} (仅响应文件创建和修改)")
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()
    print("监控已停止。")

如何运行

与示例1类似,运行脚本并尝试创建或修改文件,你会发现删除或移动文件不会产生任何输出。


示例 3:实用场景 - 自动同步文件

这个示例更具实用性,它会监控一个“源”目录,每当有文件被修改或创建时,就自动将其复制到另一个“目标”目录。

代码 (auto_sync.py)

import time
import shutil
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class SyncHandler(FileSystemEventHandler):
    def __init__(self, source_dir, dest_dir):
        self.source_dir = source_dir
        self.dest_dir = dest_dir
        # 确保目标目录存在
        if not os.path.exists(self.dest_dir):
            os.makedirs(self.dest_dir)
    def on_any_event(self, event):
        # 我们只关心文件的创建和修改
        if event.is_directory:
            return
        # 获取源文件相对于源目录的相对路径
        relative_path = os.path.relpath(event.src_path, self.source_dir)
        dest_path = os.path.join(self.dest_dir, relative_path)
        print(f"事件: {event.event_type} | 文件: {event.src_path}")
        if event.event_type == 'created':
            # 如果是创建,直接复制
            print(f"正在复制新文件: {event.src_path} -> {dest_path}")
            shutil.copy2(event.src_path, dest_path)
        elif event.event_type == 'modified':
            # 如果是修改,也进行复制(覆盖)
            print(f"正在更新文件: {event.src_path} -> {dest_path}")
            shutil.copy2(event.src_path, dest_path)
if __name__ == "__main__":
    # 设置源目录和目标目录
    source_directory = "./source_folder"
    destination_directory = "./destination_folder"
    # 确保源目录存在
    if not os.path.exists(source_directory):
        os.makedirs(source_directory)
        print(f"已创建源目录: {source_directory}")
    print(f"开始监控源目录: {source_directory}")
    print(f"所有更改将同步到目标目录: {destination_directory}")
    event_handler = SyncHandler(source_directory, destination_directory)
    observer = Observer()
    observer.schedule(event_handler, source_directory, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()
    print("同步已停止。")

如何运行

  1. 创建两个文件夹:source_folderdestination_folder

  2. 将上面的代码保存为 auto_sync.py

  3. 在终端运行脚本:

    python auto_sync.py
  4. source_folder 中创建、修改或删除文件,你会看到它们被自动同步到 destination_folder(删除操作需要额外代码处理,此处未展示)。


watchdog 核心概念总结

  1. Observer (观察者):

    • 这是核心类,它会在一个单独的线程中运行,持续监控指定的文件系统。
    • 你需要调用 observer.start() 来启动监控,observer.stop()observer.join() 来停止它。
  2. FileSystemEventHandler (事件处理基类):

    • 这是一个基类,你可以继承它并重写其方法来定义你自己的事件处理逻辑。
    • 常用的事件方法:
      • on_any_event(self, event): 所有事件的通用入口。
      • on_created(self, event): 文件或目录被创建时触发。
      • on_deleted(self, event): 文件或目录被删除时触发。
      • on_modified(self, event): 文件或目录被修改时触发。
      • on_moved(self, event): 文件或目录被移动或重命名时触发。event 对象包含 src_path (旧路径) 和 dest_path (新路径)。
  3. event (事件对象):

    • 当事件发生时,watchdog 会创建一个事件对象并将其传递给你的事件处理方法。
    • 它包含以下重要属性:
      • event_type: 事件的类型,如 'created', 'modified', 'deleted', 'moved'
      • src_path: 触发事件的文件或目录的路径。
      • is_directory: 一个布尔值,如果事件是关于目录的则为 True
      • dest_path: 仅在 'moved' 事件中存在,表示文件/目录的新路径。

希望这些示例能帮助你快速上手 watchdog!你可以根据这些基础,构建更复杂的文件监控和自动化工具。

分享:
扫描分享到社交APP
上一篇
下一篇