现在有一个应用场景,需要对文件系统进行监控,发生变化时产生日志,对新增的文件做一些相应的操作。比如说应用到我们之前的高潮提取器:若当前文件夹下增加了一个音乐文件,监控器就调用高潮提取器提取该音乐文件的高潮部分。
这样的监控器写起来也不难,但是很花时间,有许多情况要考虑。不过幸好我们是写Python的,有许多轮子可以使用。
1.安装”看门狗”
“看门狗”模块就是用于监控文件事件变化的一个Python”轮子”,代码架构优秀,可以注册许多事件处理器,方便用户做自定义操作。如果你还没有安装Python,请看这篇文章《Python详细安装指南》,在终端输入以下命令即可安装看门狗:
pip install watchdog
2.基本使用
看门狗的使用其实不复杂,请认真看以下代码和注释:
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()
可以看到代码中有几个关键步骤,1.配置各项信息;2.生成事件处理器、监控器;3.注册事件处理器、配置目录、递归执行(即同时监控子文件夹);4.启动。
其实,看门狗的observer是基于 threading.Thread 对象的,所以observer很多属性都继承了 threading.Thread 的属性。
如果你不带参数地运行该脚本,就是要监控脚本文件所在的文件夹,如果要监控其他文件夹,记得运行时带文件夹的路径参数,如:
python obserber.py /data/home/ckend/
我们来试着运行看看:
可以看到,我在当前文件夹下做的所有操作都被记录下来了。接下来我们就试试怎么自定义一些操作。
3.监控提取音乐高潮
如果你不知道怎么提取音乐文件的高潮部分,请看这篇文章:《Python自动提取音乐文件高潮》。
要实现这样的功能,我们有几种方法,一个是在原来log的处理器上做一些新增修改,比如多增一个函数调用音乐高潮提取器。第二个是重新继承 FileSystemEventHandler 类,并做相应的修改。这里我们还是要保留log的样式,只是在log的时候顺便提取音乐高潮,因此采用第一个方法。
看看LoggingEventHandler源代码中的on_created,这就是当文件创建时监控器的操作:
class LoggingEventHandler(FileSystemEventHandler): """Logs all the events captured.""" # ...省略其他源代码... def on_created(self, event): super(LoggingEventHandler, self).on_created(event) what = 'directory' if event.is_directory else 'file' logging.info("Created %s: %s", what, event.src_path)
我们仅需要继承这个类并对on_created进行修改,就能完成我们想要的功能:
# Python实用宝典 # 2019/12/29 import sys import time import logging from watchdog.observers import Observer from watchdog.events import LoggingEventHandler from pychorus import find_and_output_chorus class extractor(LoggingEventHandler): def on_created(self, event): super(LoggingEventHandler, self).on_created(event) what = 'directory' if event.is_directory else 'file' logging.info("Created %s: %s", what, event.src_path) NameExt = event.src_path.split('.') if NameExt[-1] == 'mp3': logging.info("mp3文件, 提取音乐高潮中...") output_path = "."+"".join(NameExt[:-1])+'_high.wav' find_and_output_chorus(event.src_path, output_path, 30) 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 = extractor() # 生成监控器对象 observer = Observer() # 注册事件处理器 observer.schedule(event_handler, path, recursive=True) # 监控器启动——创建线程 observer.start() # 以下代码是为了保持主线程运行 try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() # 主线程任务结束之后,进入阻塞状态,一直等待其他的子线程执行结束之后,主线程再终止 observer.join()
首先声明一个类,继承LoggingEventHandler,然后重载on_created函数,在这个函数中不仅记录文件事件变化,还要对mp3文件做一次音乐高潮提取。最后别忘了,生成事件处理器时要用我们新的类名。
看看效果,将小永远.mp3复制过来:
成功监控文件变化并提取到音乐高潮,生成高潮文件。
我们的文章到此就结束啦,如果你希望我们今天的Python 教程,请持续关注我们,如果对你有帮助,麻烦在下面点一个赞/在看哦
Python实用宝典 (pythondict.com)
不只是一个宝典
欢迎关注公众号:Python实用宝典