python watchdog 详细讲解

在实际的开发过程中，有时候需要通过 Python 去监听某文件夹的变动，从而实现针对文件变化的操作。

Python 中有2个不错的库实现了该功能，分别是 pyinotify 和 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()

Python

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。