Park Lam's 每日分享

Python实用工具：文件系统监控利器watchdog

作者：

Park Lam

在

1. Python在各领域的广泛性及重要性

Python作为一种高级编程语言，凭借其简洁易读的语法和强大的功能，已经广泛应用于多个领域。在Web开发中，Django、Flask等框架让开发者能够快速构建高效的Web应用；数据分析和数据科学领域，NumPy、Pandas等库提供了强大的数据处理和分析能力；机器学习和人工智能领域，TensorFlow、PyTorch等库推动了深度学习的发展；桌面自动化和爬虫脚本方面，Selenium、Requests等库让自动化操作和数据抓取变得简单；金融和量化交易领域，Python也发挥着重要作用；教育和研究领域，Python更是成为了首选的编程语言。

Python的广泛性和重要性得益于其丰富的库和工具。这些库和工具为开发者提供了便捷的方式来实现各种功能，大大提高了开发效率。本文将介绍其中一个实用的Python库——watchdog。

2. watchdog库概述

2.1 用途

watchdog是一个用于监控文件系统事件的Python库。它可以监控文件和目录的创建、修改、删除等事件，并在事件发生时执行相应的操作。这对于需要实时响应文件系统变化的应用程序非常有用，比如自动备份、实时编译、文件同步等。

2.2 工作原理

watchdog通过监听操作系统提供的文件系统通知机制来工作。不同的操作系统有不同的实现方式：

  • 在Linux系统上，使用inotify API；
  • 在Windows系统上，使用ReadDirectoryChangesW API；
  • 在macOS系统上，使用FSEvents API。

watchdog提供了一个统一的接口，让开发者可以在不同的操作系统上使用相同的代码来监控文件系统事件。

2.3 优缺点

优点

  • 跨平台支持： 可以在Linux、Windows和macOS等多种操作系统上使用。
  • 简单易用： 提供了简洁的API，让开发者可以快速上手。
  • 丰富的事件类型： 支持文件和目录的创建、修改、删除等多种事件类型。
  • 可扩展性： 可以自定义事件处理器，实现个性化的功能。

缺点

  • 性能开销： 长时间监控大量文件和目录可能会带来一定的性能开销。
  • 某些特殊情况处理不足： 在某些特殊情况下，可能会出现事件丢失或重复的问题。

2.4 License类型

watchdog库采用Apache License 2.0许可协议。这是一个非常宽松的开源许可证，允许用户自由使用、修改和分发该库，只需要保留原有的版权声明和许可证信息即可。

3. watchdog库的使用方式

3.1 安装

watchdog库可以通过pip包管理器进行安装，打开终端并执行以下命令：

pip install watchdog

如果需要安装特定版本的watchdog库，可以使用以下命令：

pip install watchdog==版本号

3.2 基本概念

在使用watchdog库之前，需要了解几个基本概念：

  • 事件(Event)： 表示文件系统发生的变化，如文件创建、修改、删除等。
  • 事件处理器(Event Handler)： 用于处理特定类型的事件，当特定事件发生时，相应的事件处理器会被调用。
  • 观察者(Observer)： 负责监控文件系统，并在检测到事件时通知相应的事件处理器。

3.3 简单示例

下面是一个简单的示例，展示了如何使用watchdog库监控指定目录下的文件变化：

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

# 自定义事件处理器
class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        print(f"文件 {event.src_path} 被修改了")

    def on_created(self, event):
        print(f"文件 {event.src_path} 被创建了")

    def on_deleted(self, event):
        print(f"文件 {event.src_path} 被删除了")

if __name__ == "__main__":
    # 创建事件处理器
    event_handler = MyHandler()

    # 创建观察者
    observer = Observer()

    # 监控指定目录，使用递归方式监控子目录
    path = "."  # 当前目录
    observer.schedule(event_handler, path, recursive=True)

    # 启动观察者
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 停止观察者
        observer.stop()

    # 等待观察者线程结束
    observer.join()

在这个示例中，我们创建了一个自定义的事件处理器MyHandler，它继承自FileSystemEventHandler类，并重写了on_modified、on_created和on_deleted方法。这些方法分别在文件被修改、创建和删除时被调用。

然后，我们创建了一个观察者对象，并将事件处理器和要监控的目录传递给它。最后，启动观察者并让它持续运行，直到用户按下Ctrl+C停止程序。

3.4 监控特定类型的文件

如果你只需要监控特定类型的文件，可以在事件处理器中添加过滤逻辑。以下是一个示例，只监控.py文件的变化：

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

class PythonFileHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith('.py'):
            print(f"Python文件 {event.src_path} 被修改了")

    def on_created(self, event):
        if event.src_path.endswith('.py'):
            print(f"Python文件 {event.src_path} 被创建了")

    def on_deleted(self, event):
        if event.src_path.endswith('.py'):
            print(f"Python文件 {event.src_path} 被删除了")

if __name__ == "__main__":
    event_handler = PythonFileHandler()
    observer = Observer()
    path = "."
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

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

    observer.join()

3.5 使用模式匹配事件处理器

watchdog库提供了一个PatternMatchingEventHandler类，可以更方便地监控特定类型的文件。以下是一个使用PatternMatchingEventHandler的示例：

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

if __name__ == "__main__":
    # 定义要监控的文件模式
    patterns = ["*.py", "*.txt"]
    # 定义不需要监控的文件模式
    ignore_patterns = None
    # 是否忽略目录事件
    ignore_directories = True
    # 是否区分大小写
    case_sensitive = True

    # 创建模式匹配事件处理器
    event_handler = PatternMatchingEventHandler(patterns, ignore_patterns, ignore_directories, case_sensitive)

    # 定义事件处理方法
    def on_modified(event):
        print(f"文件 {event.src_path} 被修改了")

    def on_created(event):
        print(f"文件 {event.src_path} 被创建了")

    def on_deleted(event):
        print(f"文件 {event.src_path} 被删除了")

    # 绑定事件处理方法
    event_handler.on_modified = on_modified
    event_handler.on_created = on_created
    event_handler.on_deleted = on_deleted

    # 创建观察者并启动监控
    observer = Observer()
    path = "."
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

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

    observer.join()

3.6 异步监控

上面的示例都是同步监控，会阻塞主线程。如果需要在不阻塞主线程的情况下监控文件系统，可以使用异步方式。以下是一个异步监控的示例：

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

class AsyncHandler(FileSystemEventHandler):
    def __init__(self, loop):
        self.loop = loop

    def on_modified(self, event):
        # 在事件循环中执行异步任务
        asyncio.run_coroutine_threadsafe(self.handle_event(event), self.loop)

    async def handle_event(self, event):
        # 模拟一个异步操作
        await asyncio.sleep(0.1)
        print(f"异步处理文件 {event.src_path} 的修改事件")

async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # 创建事件处理器
    event_handler = AsyncHandler(loop)

    # 创建观察者
    observer = Observer()
    path = "."
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

    try:
        # 保持主线程运行
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

if __name__ == "__main__":
    asyncio.run(main())

3.7 高级用法：自定义事件和事件处理器

除了使用内置的事件处理器，还可以自定义事件和事件处理器。以下是一个自定义事件和事件处理器的示例：

import time
from watchdog.observers import Observer
from watchdog.events import Event, FileSystemEventHandler, FileSystemEvent

# 定义自定义事件类
class CustomEvent(FileSystemEvent):
    event_type = "custom"

    def __init__(self, src_path):
        super().__init__(src_path)
        self.is_directory = False

# 定义自定义事件处理器
class CustomEventHandler(FileSystemEventHandler):
    def on_custom(self, event):
        print(f"自定义事件发生在 {event.src_path}")

# 创建自定义事件处理器实例
event_handler = CustomEventHandler()

# 创建观察者
observer = Observer()
path = "."
observer.schedule(event_handler, path, recursive=True)
observer.start()

try:
    # 模拟触发自定义事件
    time.sleep(2)
    custom_event = CustomEvent("./test.txt")

    # 手动调用事件处理器的方法
    event_handler.dispatch(custom_event)

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

observer.join()

4. 实际案例

4.1 自动备份文件

下面是一个使用watchdog库实现自动备份文件的实际案例：

import os
import time
import shutil
from datetime import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class BackupHandler(FileSystemEventHandler):
    def __init__(self, backup_dir):
        self.backup_dir = backup_dir
        # 如果备份目录不存在，则创建
        if not os.path.exists(backup_dir):
            os.makedirs(backup_dir)

    def on_modified(self, event):
        if not event.is_directory:
            src_path = event.src_path
            file_name = os.path.basename(src_path)

            # 创建带时间戳的备份文件名
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            backup_file_name = f"{os.path.splitext(file_name)[0]}_{timestamp}{os.path.splitext(file_name)[1]}"
            backup_path = os.path.join(self.backup_dir, backup_file_name)

            try:
                # 备份文件
                shutil.copy2(src_path, backup_path)
                print(f"已备份文件 {src_path} 到 {backup_path}")
            except Exception as e:
                print(f"备份文件 {src_path} 失败: {e}")

    def on_created(self, event):
        if not event.is_directory:
            src_path = event.src_path
            file_name = os.path.basename(src_path)

            # 等待一小段时间，确保文件写入完成
            time.sleep(0.1)

            # 创建带时间戳的备份文件名
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            backup_file_name = f"{os.path.splitext(file_name)[0]}_{timestamp}{os.path.splitext(file_name)[1]}"
            backup_path = os.path.join(self.backup_dir, backup_file_name)

            try:
                # 备份文件
                shutil.copy2(src_path, backup_path)
                print(f"已备份新创建的文件 {src_path} 到 {backup_path}")
            except Exception as e:
                print(f"备份新创建的文件 {src_path} 失败: {e}")

if __name__ == "__main__":
    # 要监控的目录
    monitored_dir = "."
    # 备份目录
    backup_dir = "./backups"

    # 创建事件处理器
    event_handler = BackupHandler(backup_dir)

    # 创建观察者
    observer = Observer()
    observer.schedule(event_handler, monitored_dir, recursive=True)

    # 启动观察者
    observer.start()

    try:
        print(f"开始监控目录 {monitored_dir}，备份目录为 {backup_dir}")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

这个脚本会监控指定目录下的文件变化，当文件被创建或修改时，会自动备份到指定的备份目录，并在备份文件名中添加时间戳。

4.2 实时编译Sass文件

下面是一个使用watchdog库实现实时编译Sass文件的实际案例：

import os
import time
import subprocess
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

class SassHandler(PatternMatchingEventHandler):
    def __init__(self):
        # 只监控.scss和.sass文件
        patterns = ["*.scss", "*.sass"]
        super().__init__(patterns=patterns)

    def on_modified(self, event):
        src_path = event.src_path
        print(f"检测到Sass文件 {src_path} 被修改")

        # 获取输出CSS文件的路径
        base_dir = os.path.dirname(src_path)
        file_name = os.path.basename(src_path)
        css_file_name = os.path.splitext(file_name)[0] + ".css"
        css_path = os.path.join(base_dir, css_file_name)

        # 编译Sass文件
        try:
            # 使用sass命令编译文件
            # 注意：需要先安装sass命令行工具
            result = subprocess.run(
                ["sass", src_path, css_path],
                capture_output=True,
                text=True
            )

            if result.returncode == 0:
                print(f"成功编译 {src_path} 到 {css_path}")
            else:
                print(f"编译失败: {result.stderr}")
        except Exception as e:
            print(f"编译过程中发生错误: {e}")

if __name__ == "__main__":
    # 要监控的目录
    monitored_dir = "./sass"

    # 创建事件处理器
    event_handler = SassHandler()

    # 创建观察者
    observer = Observer()
    observer.schedule(event_handler, monitored_dir, recursive=True)

    # 启动观察者
    observer.start()

    try:
        print(f"开始监控Sass目录 {monitored_dir}")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

这个脚本会监控指定目录下的Sass文件变化，当Sass文件被修改时，会自动调用sass命令将其编译为CSS文件。

4.3 文件同步工具

下面是一个使用watchdog库实现简单文件同步工具的实际案例：

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

class FileSyncHandler(FileSystemEventHandler):
    def __init__(self, source_dir, target_dir):
        self.source_dir = source_dir
        self.target_dir = target_dir

        # 如果目标目录不存在，则创建
        if not os.path.exists(target_dir):
            os.makedirs(target_dir)

    def on_modified(self, event):
        src_path = event.src_path
        relative_path = os.path.relpath(src_path, self.source_dir)
        target_path = os.path.join(self.target_dir, relative_path)

        if event.is_directory:
            # 如果是目录被修改，创建对应的目标目录
            if not os.path.exists(target_path):
                os.makedirs(target_path)
                print(f"创建目录 {target_path}")
        else:
            # 如果是文件被修改，复制文件到目标位置
            try:
                # 确保目标目录存在
                target_dir = os.path.dirname(target_path)
                if not os.path.exists(target_dir):
                    os.makedirs(target_dir)

                # 复制文件
                shutil.copy2(src_path, target_path)
                print(f"同步文件 {src_path} 到 {target_path}")
            except Exception as e:
                print(f"同步文件 {src_path} 失败: {e}")

    def on_created(self, event):
        src_path = event.src_path
        relative_path = os.path.relpath(src_path, self.source_dir)
        target_path = os.path.join(self.target_dir, relative_path)

        if event.is_directory:
            # 如果是新创建的目录，创建对应的目标目录
            os.makedirs(target_path)
            print(f"创建目录 {target_path}")
        else:
            # 如果是新创建的文件，复制文件到目标位置
            try:
                # 确保目标目录存在
                target_dir = os.path.dirname(target_path)
                if not os.path.exists(target_dir):
                    os.makedirs(target_dir)

                # 复制文件
                shutil.copy2(src_path, target_path)
                print(f"同步新创建的文件 {src_path} 到 {target_path}")
            except Exception as e:
                print(f"同步新创建的文件 {src_path} 失败: {e}")

    def on_deleted(self, event):
        src_path = event.src_path
        relative_path = os.path.relpath(src_path, self.source_dir)
        target_path = os.path.join(self.target_dir, relative_path)

        # 删除目标位置对应的文件或目录
        try:
            if os.path.exists(target_path):
                if os.path.isfile(target_path):
                    os.remove(target_path)
                    print(f"删除文件 {target_path}")
                else:
                    shutil.rmtree(target_path)
                    print(f"删除目录 {target_path}")
        except Exception as e:
            print(f"删除目标文件/目录 {target_path} 失败: {e}")

if __name__ == "__main__":
    # 源目录
    source_dir = "./source"
    # 目标目录
    target_dir = "./target"

    # 创建事件处理器
    event_handler = FileSyncHandler(source_dir, target_dir)

    # 创建观察者
    observer = Observer()
    observer.schedule(event_handler, source_dir, recursive=True)

    # 启动观察者
    observer.start()

    try:
        print(f"开始同步目录 {source_dir} 到 {target_dir}")

        # 初始同步 - 将源目录中的所有文件复制到目标目录
        print("执行初始同步...")
        for root, dirs, files in os.walk(source_dir):
            for dir_name in dirs:
                src_dir_path = os.path.join(root, dir_name)
                rel_path = os.path.relpath(src_dir_path, source_dir)
                target_dir_path = os.path.join(target_dir, rel_path)

                if not os.path.exists(target_dir_path):
                    os.makedirs(target_dir_path)
                    print(f"初始同步：创建目录 {target_dir_path}")

            for file_name in files:
                src_file_path = os.path.join(root, file_name)
                rel_path = os.path.relpath(src_file_path, source_dir)
                target_file_path = os.path.join(target_dir, rel_path)

                target_dir_path = os.path.dirname(target_file_path)
                if not os.path.exists(target_dir_path):
                    os.makedirs(target_dir_path)

                shutil.copy2(src_file_path, target_file_path)
                print(f"初始同步：复制文件 {src_file_path} 到 {target_file_path}")

        print("初始同步完成")

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

    observer.join()

这个脚本会监控源目录的变化，并实时将这些变化同步到目标目录。包括文件和目录的创建、修改和删除操作。

5. 相关资源

  • Pypi地址： https://pypi.org/project/watchdog/
  • Github地址： https://github.com/gorakhargosh/watchdog
  • 官方文档地址： https://python-watchdog.readthedocs.io/en/stable/

关注我，每天分享一个实用的Python自动化工具。

更多文章