python 监目录文件变动，然后在做些其它的操作，watchdog 详细解答

Posted 2022-02-05 梦想橡皮擦

在实际的开发过程中，有时候需要通过 Python 去监听某文件夹的变动，从而实现针对文件变化的操作。
Python 中有2个不错的库实现了该功能，分别是 pyinotify 和 watchdog 本篇博客为你介绍第三方模块 watchdog 实现对文件夹的监控。

文章目录

• • watchdog 安装与准备
  • • 官方提供最简单的入门案例
  • 重写事件
  • 监听指定内容
  • 添加多个事件
  • 监听特定文件夹，特定后缀的文件
  • 记录时间

watchdog 安装与准备

安装就比较简单了

pip install watchdog

项目地址是：https://pypi.org/project/watchdog/#description
文档参考地址：https://python-watchdog.readthedocs.io/en/stable/

watchdog 是一个实时监控库，其原理是通过操作系统的时间触发，需要循环等待。

官方提供最简单的入门案例

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    event_handler = LoggingEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    finally:
        observer.stop()
	observer.join()

基于上述源码，说明一下 watchdog 的相关用法。

observer = Observer()
创建一个观察者对象。

observer.schedule()
声明一个定时任务。

observer.start()
启动定时任务。

observer.schedule() 的函数原型如下

schedule(event_handler, path, recursive=False)

该方法用于监视 path 路径，并调用给定的事情 event_handler 。

最后一个参数 recursive 表示是否递归子目录，即监听子目录，默认为 False。

start()
启动线程，这里开启了新的守护线程，主程序如果结束， 该线程也会停止。

每个线程对象只能调用1次，它安排对象的 run() 方法在单独的控制线程中调用，如果在同一线程对象上多次调用此方法将引发RuntimeError。

重写事件

接下来我们使用自定义的时间，实现对文件操作的监控

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

class MyEventHandler(FileSystemEventHandler):
    # 文件移动
    def on_moved(self, event):
        print("文件移动触发")
        print(event)

    def on_created(self, event):
        print("文件创建触发")
        print(event)

    def on_deleted(self, event):
        print("文件删除触发")
        print(event)

    def on_modified(self, event):
        print("文件编辑触发")
        print(event)

if __name__ == '__main__':

    observer = Observer()  # 创建观察者对象
    file_handler = MyEventHandler()  # 创建事件处理对象
    observer.schedule(file_handler, "./images", False)  # 向观察者对象绑定事件和目录
    observer.start() # 启动
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

这里会出现一个问题，就是当复制进去一个文件时，会同时触发 创建/编辑 事件，如下所示。

先说原因：
文件创建这个操作会触发多种事件，包括 FileCreatedEvent ， FileModifiedEvent 事件，对应到代码中就是 on_created 和 on_modified 方法被调用，其原因在于 f = open("file.txt", "w") 这样文件创建动作会触发 FileCreatedEvent 事件，执行 on_created 函数，往文件写数据的 f.flush() 和 f.close() 操作，会触发 FileModifiedEvent 事件，执行 on_modified 函数，所以触发了 3次。

FileSystemEvent 文件类派生出来的子类包括如下内容

• watchdog.events.FileCreatedEvent() ：文件被创建时触发该事件；
• watchdog.events.DirCreatedEvent() ：目录被创建时触发该事件；
• watchdog.events.FileDeletedEvent() ：文件被删除时触发该事件；
• watchdog.events.DirDeletedEvent() ：目录被删除时触发该事件；
• watchdog.events.FileModifiedEvent() ：文件被修改时触发该事件；
• watchdog.events.DirModifiedEvent() ：目录被修改时触发该事件；
• watchdog.events.FileMovedEvent() ：文件被移动或重命名时触发该事件（event.src_path 表示原路径，还有 event.dest_path ）；
• watchdog.events.DirMovedEvent() ：目录被移动或重命名时触发该事件（event.src_path 表示原路径，还有 event.dest_path ）；

watchdog 默认提供的一些事件处理类

• FileSystemEventHandler：文件，事件处理器的基类，用于处理事件；
• PatternMatchingEventHandler：模式匹配文件；
• RegexMatchingEventHandler：正则匹配文件；
• LoggingEventHandler：记录日志。

PatternMatchingEventHandler 函数原型如下

watchdog.events.PatternMatchingEventHandler(patterns=None,ignore_patterns=None,ignore_directories=False, case_sensitive=False)

该类会检查触发事件的 src_path 和 dest_path ，是否与 patterns 指定的模式匹配；

• ignore_patterns ：需要排除不处理的模式，如果路径匹配该模式则不处理
• ignore_directories：为 True 表示不处理由目录引起的事件；
• case_sensitive：为 True 则表示路径不区分大小写。

RegexMatchingEventHandler 函数原型如下

watchdog.events.RegexMatchingEventHandler(regexes=[r".*"], ignore_regexes=[], ignore_directories=False,  case_sensitive=False)            

监听指定内容

继承监听事件函数，然后监听特定文件。

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

class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        print(dir(event))
        print(event.src_path)
        path = os.path.abspath(event.src_path)
        print(path)
        if path == r"F:\\xxx\\监控文件变动\\imgs\\1.log":  # 监控指定文件
            print("文件 %s 有编辑" % event.src_path)

if __name__ == "__main__":
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path='./imgs', recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

其中 event 对象的属性如下所示：

• event.is_directory：该事件是否由一个目录触发；
• event.src_path：触发该事件的文件或目录路径；
• event.event_type：事件类型，例如 moved，deleted，created，modified；
• event.key：元组格式返回 (event_type, src_path, is_directory)。

observer.schedule(event_handler, path, recursive=False) 的详细说明
每一次调用 schedule() 对一个路径（ path ）进行监控处理叫做 watch ， schedule() 方法会返回这个 watch ，我们可以对 watch 增加多个 event 事件处理器。

• observer.add_handler_for_watch(event_handler, watch)：添加1个新的事件处理器到 ；
• observer.remove_handler_for_watch(event_handler, watch)：从 watch 移除1个事件处理器；
• observer.unschedule(watch)：移除1个 watch 及其所有事件处理器；
• observer.unschedule_all()：移除所有 watch 及关联的事件处理器；
• observer.on_thread_stop()：等同于 observer.stop()

添加多个事件

import time
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, LoggingEventHandler

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        print(event)

if __name__ == "__main__":
    event_handler1 = MyHandler()
    observer = Observer()
    watch = observer.schedule(event_handler1, path='.', recursive=True)

    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')

    event_handler2 = LoggingEventHandler()
    observer.add_handler_for_watch(event_handler2, watch)  # 添加event handler
    observer.start()

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

监听特定文件夹，特定后缀的文件

import time
import logging
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, LoggingEventHandler

logging.basicConfig(level=logging.INFO, filename='./video_log.log', filemode='a+', format='%(asctime)s - %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S', encoding="utf-8")

class LogHandler(LoggingEventHandler):
    def on_created(self, event):
        path = event.src_path
        if event.is_directory:
            pass
        else:
            logging.info(path + "文件新增")

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        path = event.src_path
        file_name = os.path.basename(path)
        if file_name.endswith("mp4") or file_name.endswith("avi") or file_name.endswith("flv"):
            print("文件格式正确")
        else:
            pass

        print(event)

if __name__ == "__main__":
    event_handler = MyHandler()
    observer = Observer()
    watch = observer.schedule(event_handler, path='./videos', recursive=False)

    log_handler = LogHandler()
    observer.add_handler_for_watch(log_handler, watch)  # 写入日志
    observer.start()

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

记录时间

今天是持续写作的第 283 / 365 天。
可以关注我，点赞我、评论我、收藏我啦。

更多精彩

• Python 爬虫 100 例教程导航帖（连续 3 年，不断更！）

---

👇👇👇扫码加入【78技术人】~ Python 事业部👇👇👇