在前一篇博客文章 《使用 Python 编写脚本并发布》 中,我介绍了如何使用 Python 进行脚本编程,说实话这是我在尝试 Python 进行网站和网络编程之后首次使用 Python 进行脚本编程,前面也说过之前虽然使用 Bash 构建过一些脚本,但是由于我对 Bash 不熟练,对它的使用都仅限于最基础的命令行操作,仅仅是比 alias 别名操作稍微简单一点。上次介绍的脚本是如何添加命令行参数以及将现有的操作流程用一个脚本简单化,这一次介绍的脚本是一个非常实用而且经过优化的文件变动事件监视脚本。
P1 Python 脚本:文件变动检测在廖雪峰 Python 教程实战部分的 Day 13 - 提升开发效率 中,他给我们介绍了一种用于提升开发效率的方法:
首先执行我们需要的命令
监听当前目录,并判断变动文件的后缀名,若后缀名为 .py,则触发回调函数
回调函数触发后,自动重新启动命令
流程很清楚,实现起来也很简答,廖雪峰利用 Python 的 subprocess 和第三方库 watchdog 分别实现了重启命令和监听当前目录的文件变动情况。大概 70 余行代码就能完成这样一个简单且实用的脚本。
在我编程的过程中,经常需要用到这样一个监控文件变动并自动重新执行预设命令的操作,比如我在编写 SSPYMGR 这个网站程序时经常要用到文件改动后自动重启服务器的操作,或者我经常需要在改动某些文件后自动上传到虚拟机上。当我有这些需求时,我之前的做法就是将上面廖雪峰介绍的脚本复制到我要监视的文件夹中,然后直接修改脚本里面的命令参数,这样做很直接,但是很繁琐。
我要做的是:将上面的简单脚本进行优化,使得可以通过命令行参数对脚本的行为进行设置。主要的优化目标有:
可以预设命令,并且该命令可以带参数
可以设置监听的目录,并且设置是否递归监听子目录
可以设置监听的文件后缀名,并设置可以排除在监听范围内的文件名
增加保存参数功能,并且能够读取保存的配置文件
P2 优化脚本为了实现上面这些目标,就像我们在上一篇博客那样,用 argparse 库来对复杂的命令行参数进行解析,这一次我们换一种代码的组织方式,将命令行参数的解析和配置文件封装到类中,然后通过实例化类对象解析参数,然后将配置写入到字典中,程序执行流程以指定的配置文件为主:
若指定了要读取的配置文件,则将配置文件中的内容作为配置,忽略掉其他选项。指定配置文件主要可以简化命令行的参数输入过程。若没有指定读取的配置文件,则以命令行中其他的选项为配置。
在 monitor.py 这个脚本中我将配置和命令行参数读取封装到类 Configuration 中:
class Configuration(object): _DEFAULT_LOC = _CONFIG_DIR / "monitor_default.json" def __init__(self): self.config = {} self._addArgs() def readConfig(self, file: Path): pass def _addArgs(self): pass def parseArgs(self): pass 监听目录接下来就要用到第三方库 watchdog 来监听指定的目录及指定事件触发时的操作了。事件处理器要用到 watchdog.events.FileSystemEventHandler,我们用继承的方式处理事件:
from watchdog.events import FileSystemEventHandler class MyFileSystemEventHander(FileSystemEventHandler): def __init__(self, fn, config: Configuration): super(MyFileSystemEventHander, self).__init__() self.restart = fn self.config = config self.last = time.time()构造事件处理器时需要传入回调函数和配置对象,接下来定义事件处理函数,这里会监听目标文件夹中所有的文件事件 on_any_event,但是该事件会在保存文件时触发两次,因此需要对它做一个防抖处理,防抖处理就是判断两次事件触发的时间间隔是否超过预设值,若两次事件时间间隔过短,则忽略第二次事件。
以下时事件处理的代码:
class MyFileSystemEventHander(FileSystemEventHandler): def on_any_event(self, event): # for debounce cur = time.time() if cur - self.last < 0.25: return self.last = cur ext_able = False src = Path(event.src_path) if src.name not in self.config["exclude"]: for ext in self.config["mon_ext"]: if src.suffix == ext: ext_able = True break if ext_able: logger.info('File changed: {}'.format(src)) self.restart()上面的防抖处理时以第一次事件为准,忽略掉之后一段时间内的其它事件,这样做更方便。
还有另一种复杂但更合理的处理方式,即事件触发时不立即调用处理函数,延迟一段时间,在该段时间内若有其他事件发生,则以新事件为准,重新计算延迟时间,超过时间后再执行事件处理的代码。
第二种处理方式更合理。打个比方,我在很短的时间内先后保存了两个不同的文件 A 和 B,用第一种方式,程序重启后只会重新加载 A 文件而 B 文件的改动很可能被忽略掉了;而用第二种方式 A 文件改动后程序并不会立即重新加载,而 B 文件的改动会被监听到,最终就是在延迟一段时间后程序会重新加载 A 和 B 这两个文件。
自动重启程序