Python使用Watchdog库监听文件变化

目录

• 一、简介
• 二、使用
• 1、监听单个目录的所有文件事件
• 2、监听特定类型文件（如.txt）
• 3、递归监控 – 监听目录及所有子目录
• 4、应用：监控到文件修改后自动执行脚本
• 5、监控多个目录 – 同时监听不同路径
• 6、动态启停监控 – 暂停 / 恢复文件监控

官方文档：https://python-watchdog.readthedocs.io/en/stable/index.html

watchdog 是 Python 生态中处理文件系统事件的主流库，跨平台（Windows/Linux/macOS）、效率高，支持监听文件 / 目录的创建、修改、删除、移动等事件，是工业级的解决方案。

# 安装
pip install watchdog

核心组件：
EventHandler 事件处理器：定义监控到文件事件后要执行的逻辑（核心，需自定义 / 继承）
Observer 观察者：负责启动监控线程，监听文件系统事件，并将事件分发给处理器
FileSystemEvent 事件对象：包含事件类型（创建 / 修改等）、文件路径、是否是目录等信息

watchdog基于系统原生的文件监控接口（如 Linux 的 inotify、Windows 的 ReadDirectoryChangesW），性能远高于手动轮询；
对于频繁修改的文件（如每秒多次修改），可能会触发多次on_modified事件，可通过加时间戳防抖处理；
跨平台使用时，路径建议用os.path.abspath()转为绝对路径，避免兼容问题。

运行代码前，先创建./test_dir目录；

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

# 自定义事件处理器：继承FileSystemEventHandler，重写对应事件方法
class MyFileHandler(FileSystemEventHandler):
    # 监听文件/目录创建事件
    def on_created(self, event):
        # event.is_directory 判断是否是目录，event.src_path 是触发事件的文件路径
        if event.is_directory:
            print(f"[创建目录] {event.src_path}")
        else:
            print(f"[创建文件] {event.src_path}")

    # 监听文件/目录修改事件
    def on_modified(self, event):
        if not event.is_directory:  # 只关注文件修改（排除目录修改）
            print(f"[修改文件] {event.src_path}")

    # 监听文件/目录删除事件
    def on_deleted(self, event):
        if event.is_directory:
            print(f"[删除目录] {event.src_path}")
        else:
            print(f"[删除文件] {event.src_path}")

    # 监听文件/目录移动事件
    def on_moved(self, event):
        # event.dest_path 是移动后的路径
        if event.is_directory:
            print(f"[移动目录] {event.src_path} -> {event.dest_path}")
        else:
            print(f"[移动文件] {event.src_path} -> {event.dest_path}")

if __name__ == "__main__":
    # 要监控的目录路径（绝对路径/相对路径都可）
    monitor_dir = "./test_dir"
    # 创建事件处理器实例
    event_handler = MyFileHandler()
    # 创建观察者实例
    observer = Observer()
    # 为观察者绑定处理器和监控目录
    # recursive=False 表示只监控当前目录，不包含子目录
    observer.schedule(event_handler, path=monitor_dir, recursive=False)
    
    # 启动观察者（后台线程，非阻塞）
    observer.start()
    print(f"开始监控目录：{monitor_dir}（按 Ctrl+C 停止）")
    
    try:
        # 保持主线程运行，避免程序退出
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 捕获 Ctrl+C，停止观察者
        observer.stop()
        print("n停止监控")
    # 等待观察者线程结束
    observer.join()

仅监控 ./test_dir下.txt文件的修改事件，忽略其他类型文件。

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

class TxtFileHandler(FileSystemEventHandler):
    # 重写修改事件，增加文件类型过滤
    def on_modified(self, event):
        # 排除目录，只处理文件
        if not event.is_directory:
            # 获取文件后缀名
            file_ext = os.path.splitext(event.src_path)[1]
            # 只处理 .txt 文件
            if file_ext == ".txt":
                print(f"[修改TXT文件] {event.src_path}")

if __name__ == "__main__":
    monitor_dir = "./test_dir"
    handler = TxtFileHandler()
    observer = Observer()
    observer.schedule(handler, monitor_dir, recursive=False)
    observer.start()
    
    print(f"开始监控 {monitor_dir} 下的 .txt 文件修改（按 Ctrl+C 停止）")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

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

class RecursiveHandler(FileSystemEventHandler):
    def on_created(self, event):
        print(f"[创建] {event.src_path} (目录: {event.is_directory})")

if __name__ == "__main__":
    monitor_dir = "./test_dir"
    handler = RecursiveHandler()
    observer = Observer()
    # 关键：recursive=True 开启递归监控
    observer.schedule(handler, monitor_dir, recursive=True)
    observer.start()
    
    print(f"递归监控 {monitor_dir} 及其所有子目录（按 Ctrl+C 停止）")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

监控配置文件 ./config.ini，当文件修改后，自动重新加载配置并执行指定脚本（如重启服务、刷新缓存）。

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

class ConfigChangeHandler(FileSystemEventHandler):
    def on_modified(self, event):
        # 只监控 config.ini 文件
        if event.src_path.endswith("config.ini") and not event.is_directory:
            print(f"n[配置文件修改] {event.src_path}")
            print("开始重新加载配置并执行脚本...")
            
            # 自定义动作1：打印配置内容（模拟加载配置）
            with open(event.src_path, "r", encoding="utf-8") as f:
                config_content = f.read()
                print(f"最新配置内容：n{config_content}")
            
            # 自定义动作2：执行外部脚本（如 restart_service.py）
            # 注意：需提前创建 restart_service.py，或替换为你的脚本路径
            try:
                result = subprocess.run(
                    ["python", "restart_service.py"],
                    capture_output=True,
                    text=True,
                    timeout=10
                )
                print(f"脚本执行结果：nstdout: {result.stdout}nstderr: {result.stderr}")
            except Exception as e:
                print(f"脚本执行失败：{e}")

if __name__ == "__main__":
    monitor_dir = "./"  # 监控当前目录
    handler = ConfigChangeHandler()
    observer = Observer()
    observer.schedule(handler, monitor_dir, recursive=False)
    observer.start()
    
    print(f"监控配置文件 config.ini 修改（按 Ctrl+C 停止）")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

同时监控./log_dir和./config_dir两个目录的文件变化

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

class MultiDirHandler(FileSystemEventHandler):
    def on_created(self, event):
        # 区分不同目录的事件
        if "log_dir" in event.src_path:
            print(f"[日志目录] 新增文件/目录：{event.src_path}")
        elif "config_dir" in event.src_path:
            print(f"[配置目录] 新增文件/目录：{event.src_path}")

if __name__ == "__main__":
    # 定义多个监控目录
    dir1 = "./log_dir"
    dir2 = "./config_dir"
    
    handler = MultiDirHandler()
    observer = Observer()
    
    # 为观察者添加多个监控任务
    observer.schedule(handler, dir1, recursive=False)
    observer.schedule(handler, dir2, recursive=False)
    
    observer.start()
    print(f"同时监控 {dir1} 和 {dir2}（按 Ctrl+C 停止）")
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

支持手动触发暂停监控，一段时间后恢复（比如临时维护目录时暂停监控）。

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

class SimpleHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            print(f"[监控中] 修改文件：{event.src_path}")

def control_observer(observer):
    """单独线程处理监控的启停控制"""
    while True:
        cmd = input("n输入命令（pause/continue/exit）：").strip().lower()
        if cmd == "pause":
            observer.stop()  # 暂停监控
            print("监控已暂停")
        elif cmd == "continue":
            observer.start()  # 恢复监控
            print("监控已恢复")
        elif cmd == "exit":
            observer.stop()
            print("退出程序")
            break
        else:
            print("无效命令，请输入 pause/continue/exit")

if __name__ == "__main__":
    monitor_dir = "./test_dir"
    handler = SimpleHandler()
    observer = Observer()
    observer.schedule(handler, monitor_dir, recursive=False)
    observer.start()
    
    print(f"开始监控 {monitor_dir}（可输入命令控制）")
    # 启动控制线程，避免阻塞主线程
    control_thread = threading.Thread(target=control_observer, args=(observer,))
    control_thread.daemon = True  # 主线程退出时，控制线程也退出
    control_thread.start()
    
    # 等待控制线程结束
    control_thread.join()
    observer.join()

到此这篇关于Python使用Watchdog库监听文件变化的文章就介绍到这了,更多相关Python Watchdog监听文件变化内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• python使用watchdog实现文件资源监控
• Python watchdog灵活监控文件和目录的变化
• Python神器之使用watchdog监控文件变化
• Python Watchdog实现实时监控文件系统
• Python利用watchdog模块监控文件变化