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

在实际的开发过程中，有时候需要通过 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 LoggingEventHandlerif __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 FileSystemEventHandlerclass 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 FileSystemEventHandlerclass 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, LoggingEventHandlerclass 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 handlerobserver.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, LoggingEventHandlerlogging.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_pathif event.is_directory:passelse:logging.info(path + "文件新增")class MyHandler(FileSystemEventHandler):def on_created(self, event):path = event.src_pathfile_name = os.path.basename(path)if file_name.endswith("mp4") or file_name.endswith("avi") or file_name.endswith("flv"):print("文件格式正确")else:passprint(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 年，不断更！）