亲宝软件园·资讯

展开

Python watchdog模块

Ckend 人气:2

假设现在有一个应用场景,需要对文件系统进行监控,发生变化时产生日志,对新增的文件做一些相应的操作。

比如说应用到我们之前的音乐高潮提取器:若当前文件夹下增加了一个音乐文件,监控器就调用音乐高潮提取器,自动提取该音乐文件的高潮部分。

这样的监控器写起来也不难,但是很花时间,有许多情况要考虑。不过幸好我们是写Python的,有许多轮子可以使用,本文介绍的就是一个名为 watchdog 的模块,它能帮助我们实现上述功能。

1.准备

开始之前,你要确保Python和pip已经成功安装在电脑上,如果没有,可以访问这篇文章:超详细Python安装指南 进行安装。

如果你用Python的目的是数据分析,可以直接安装Anaconda,它内置了Python和pip.

Windows环境下打开Cmd(开始—运行—CMD),苹果系统环境下请打开Terminal(command+空格输入Terminal),准备开始输入命令安装依赖。

当然,我更推荐大家用VSCode编辑器,把本文代码Copy下来,在编辑器下方的终端运行命令安装依赖模块,多舒服的一件事啊

在终端输入以下命令安装我们所需要的依赖模块:

pip install watchdog

看到 Successfully installed xxx 则说明安装成功。

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进程不关闭,它就会一直监控这个文件夹,一旦有音乐文件进入,就会自动提取它的音乐高潮,在linux系统下,可以搭配supervisor使用,非常好用。

加载全部内容

相关教程
猜你喜欢
用户评论