最近在完善自己的轮子的时候,遇到了一个问题,需要监听文件变化,查了查资料,写下此篇以作记录。
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)
except KeyboardInterrupt:
observer.stop()
observer.join()
忽略掉其他细枝末节,关键在observer.schedule(event_handler, path, recursive=True)
这一行,它的含义是
在path这个路径下触发更改事件时,将事件传递给event_handler进行处理。
那么我们来自定义个event_handler。首先这个类需要从FileSystemEventHandler
这个事件处理基类继承,并重写事件处理方法。
事件处理方法如下:
-
self.dispatch(event)
接收到一个事件后,通过该方法来决定该event由下面哪个方法处理(一般情况下不予复写该方法) -
self.on_any_event(event)
任何事件发生都会首先执行该方法,该方法默认为空,dispatch()方法会先执行该方法,然后再把event分派给其他方法处理 -
self.on_moved(event)
处理DirMovedEvent和FileMovedEvent事件,子类需重写该方法 -
self.on_created(event)
处理DirCreatedEvent和FileCreatedEvent事件,子类需重写该方法 -
self.on_deleted(event)
处理DirDeletedEvent和FileDeletedEvent事件,子类需重写该方法 -
self.on_modified(event)
处理DirModifiedEvent和FileModifiedEvent事件,子类需重写该方法
建立如下的子类
from watchdog.events import FileSystemEventHandler
class MonitorFileEventHandler(FileSystemEventHandler):
def on_modified(self, event):
file = os.path.basename(event.src_path)
if file.split(".")[-1] == "html":
logging.info("Update article {}".format(file.replace(".html", "")))
update_blog(file.replace(".html", ""))
def on_deleted(self, event):
file = os.path.basename(event.src_path)
if file.split(".")[-1] == "html":
logging.info("Delete article {}".format(file.replace(".html", "")))
delete_blog(file.replace(".html", ""))
这个类只对更改和删除做出反应。你当然可以自己复写如上方法来做自己想做的事情。
event有如下的属性:
-
event.is_directory
该事件是否由一个目录触发 -
event.src_path
触发该事件的文件或目录路径 -
event.event_type
事件类型,为moved、deleted、created或modified的其中之一 -
event.key
返回元组(event_type, src_path, is_directory)
你可以在复写的方法里使用它们来做到一些有趣的事情。
然后我又编写了一个类来调用这个事件类
from watchdog.observers import Observer
class MonitorFile:
def __init__(self, path):
self.observer=Observer()
self.observer.schedule(MonitorFileEventHandler(), path, recursive=True)
self.observer.start()
def __del__(self):
self.observer.stop()
self.observer.join()
接下来只需要实例化一下这个类,然后保持自身不被kill就OK了。
类似于这样
main = MonitorFile("D:/")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
pass