简介
日常工作生活中经常遇到一个文件当它被创建和修改之后需要被处理，而那些处理都是非常重复性的手动操作，比如说移动到相应的目录或者是解压打包备份之类等等。那有没有一种可以对文件手动处理标准化成自动化从而使人从枯燥乏味的操作中解脱出来的方法呢？python的watchdog是一个实时监控文件变化并且触发相应的事件的一个第三方模块。
watchdog is an open-source python API library that is a cross-platform API to monitor file system events.
你可以指定一个文件或目录作为watchdog observer的参数对象，它会不断监视该文件夹的任何更改，如文件的创建、修改、删除或文件从一个文件夹移动到另一个文件夹。当观察者记录或观察到事件时，事件处理程序会执行指定的事件操作。
简单使用
首先安装watchdog:
pip install watchdog
主要分为三个部分：
- watchdog.events.FileSystemEventHandler：文件系统事件基类，使用者需要继承该类，并在子类中重写对应子类方法来实现触发后的相应动作
- watchdog.observers.Observer：定义一个观察者对象来监控文件系统事件
- event.src_path: 指定的被监控的目录
为了处理不同的文件系统事件，首先需要创建watchdog.events.FileSystemEventHandler的子类并重写对应实例方法。子类主要处理以下的方法：
on_created() : 当一个文件或目录被创建
on_modified()：当一个文件或目录被修改
on_deleted()：当一个文件或目录被删除
on_moved() ：当一个文件或目录被移动
on_closed(): 当一个文件被写完
on_any_event() ：所有的事件
class FileEventHandler(FileSystemEventHandler):
def __init__(self):
def on_moved(self, event):
print(f"file moved from {event.src_path} to {event.dest_path}.")
def on_created(self, event):
print(f"file created:{event.src_path}.")
def on_deleted(self, event):
print(f"file deleted:{event.src_path}.")
def on_modified(self, event):
print(f"file modified:{event.src_path}.")
接下来我们需要创建Observer的实例去监控指定的目录去捕获相应的事件。
默认情况下，watchdog.observers.Observer不会监控子文件夹。如果要监控子文件夹，则必须将递归标志recursive设置为true。
要开始监控文件夹，启动observer线程并等待目录事件的触发，并且不能阻塞我们的主线程。
下面这个例子传入参数指定目录后可以监控该目录，缺省时是当前目录。
if __name__ == "__main__":
src_path = sys.argv[1] if len(sys.argv) > 1 else '.'
event_handler = FileEventHandler()
observer = Observer()
observer.schedule(event_handler, path=src_path, recursive=True)
print("Monitoring started")
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
自动分类下载后的文件
根据上面的例子，我们可以编写一个脚本来自动归类下载后的文件。比如文件下载后默认存放的目录一般都是Downloads目录。但是我们希望下载之后能够根据文件的后缀名来自动归类到相应的文件夹来存放。这时候我们可以通过watchdog来处理类似的需求。
下面代码会监控Downloads目录，当有文件被创建之后会自动触发move_file这个方法并把文件移动到对应的文件夹。
import os
import shutil
import time
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
# 建立新的目录
def make_new_dir(dir, type_dir):
for td in type_dir:
new_td = os.path.join(dir, td)
if not os.path.isdir(new_td):
os.makedirs(new_td)
# 定义子类继承FileSystemEventHandler并重写on_any_event方法
class FileEventHandler(FileSystemEventHandler):
def on_any_event(self, event):
if not event.is_directory:
if event.event_type == 'created':
print(f"file created:{event.src_path}.")
self.move_file(event.src_path)
# 移动文件至相应的目录
@staticmethod
def move_file(src_path: str):
time.sleep(5)
for file_type, file_ext in file_type_mapping.items():
if src_path.endswith(file_ext):
shutil.move(src_path, os.path.join(source_dir, file_type))
print(f"file {src_path} has been successfully moved to {os.path.join(source_dir, file_type)}")
# 定义Watcher类包含observer和event_handler
class Watcher:
def __init__(self, event_handler: FileSystemEventHandler, path: str):
self.event_handler = event_handler
self.observer = Observer()
self.path = path
def run(self):
self.observer.schedule(self.event_handler, self.path, recursive=True)
self.observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
self.observer.stop()
print(f'Error')
self.observer.join()
if __name__ == "__main__":
source_dir = "/Users/tony/Downloads"
# 定义文件类型和它的扩展名
file_type_mapping = {
"music": ("mp3", "wav"),
"movie": ("mp4", "rmvb", "rm", "avi"),
"execute": ("exe", "bat"),
"sheet": ("csv", "xlsx")
}
# 建立新的文件夹
make_new_dir(source_dir, file_type_mapping)
# 创建Watcher实例并传入监控目录和FileEventHandler
w = Watcher(path=source_dir, event_handler=FileEventHandler())
w.run()
如果需要在后台不间断运行的话，可以用nohup python watchdog_auto_classify >output.log 2>&1 &。
监控多个目录
如果需要用不同event_handler监控多个文件目录，可以重新封装Watcher类，构造handler_dict存放event_hander和对应的被监控目录。然后增加add_handler方法来添加handler和目录。
import logging
import time
from watchdog.events import FileSystemEventHandler, LoggingEventHandler
from watchdog.observers import Observer
from watchdog_example import FileEventHandler
class Watcher:
def __init__(self):
self.handler_dict = dict()
self.observer = Observer()
def add_handler(self, handler: FileSystemEventHandler, path: str):
self.handler_dict[handler] = path
def run(self):
if self.handler_dict is not None:
for handler, path in self.handler_dict.items():
self.observer.schedule(handler, path, recursive=True)
self.observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
self.observer.stop()
print(f'Error')
self.observer.join()
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
event_handler1 = LoggingEventHandler()
event_handler2 = FileEventHandler()
src_path1 = '/Users/tony/Downloads/'
src_path2 = '/Users/tony/Documents/'
w = Watcher()
w.add_handler(event_handler1, src_path1)
w.add_handler(event_handler2, src_path2)
w.run()
总结
watchdog是一个非常容易上手的python监控开源框架。主要采用观察者模式，通过observer.schedule将observer, event_handler，被监控的文件夹串联起来，observer不断观察系统文件的任何事件，当发生改变时会通知event_handler去处理。
参考资料
https://pythonhosted.org/watchdog/quickstart.html#quickstart
https://blog.csdn.net/chdhust/article/details/50514391
