以下为个人学习笔记整理

# 基于 watchdog 实现自动化更新

# 背景：

每次修改 Python 后为了不重启服务器进行调试，需要手动执行一个指令来指定需要热更的文件，有时候会比较麻烦。为此写了一个自动识别文件修改的工具，可以实时监听文件的修改并自动执行热更操作。

# 原理：

基于 watchdog 监听某个文件下的文件
发现文件修改后触发相应的 Event。
根据 Event 内容生成热更指令，写入热更文件。
热更程序定时读取文件，执行热更指令，并清空文件。

# 核心代码：

	# -- coding: utf8 --
	# DATE: 2020/10/16 Fri
	import sys
	import time
	import logging
	from watchdog.observers import Observer
	from watchdog.events import LoggingEventHandler

	class EventHandler(LoggingEventHandler):
	def on_modified(self, event):
	print(f"{event.src_path}")
	...

	if __name__ == "__main__":
	event_handler = EventHandler()
	observer = Observer()
	observer.schedule(event_handler, path="./", recursive=True)
	observer.start()
	try:
	while True:
	time.sleep(1)
	except KeyboardInterrupt:
	observer.stop()
	observer.join()

# 代码实现：

	# -- coding: utf8 --
	# DATE: 2020/10/16 Fri
	import os
	import time
	import enum
	import json
	from watchdog.observers import Observer
	from watchdog.events import FileSystemEventHandler


	class EventType(enum.IntEnum):
	UPDATE = 1


	class FilterPath:
	def __init__(self):
	self._paths = WatchDog.get_json_info("listen_path")

	def __call__(self, path: str):
	for _path in self._paths:
	if path.startswith(_path):
	return True
	return False


	class FilterFile:
	def __init__(self):
	self._files = WatchDog.get_json_info("listen_file_ext")

	def __call__(self, path: str):
	_, ext = os.path.splitext(path)
	if ext not in self._files:
	return False
	return True


	class CmdConverter:
	def __init__(self):
	self.root_path = WatchDog.get_json_info("converter_root")
	self._replaces = WatchDog.get_json_info("replace_char")
	self._cmd_format = WatchDog.get_json_info("cmd_format")

	def __call__(self, event_type: EventType, path: str):
	if not path.startswith(self.root_path):
	return ""
	path = path[len(self.root_path):]
	for _replace in self._replaces:
	path = path.replace(_replace, ".")

	if event_type == EventType.UPDATE:
	return self._cmd_format.format("update", path)
	return ""


	class FileWriter:
	def __init__(self):
	self.write_path = WatchDog.get_json_info("update_path")

	def write_to_file(self, content):
	try:
	with open(self.write_path, "a") as f:
	f.writelines(content)
	except Exception as e:
	print(e)


	class AutoUpdateHandler(FileSystemEventHandler):
	def __init__(self):
	self._filters = [FilterPath(), FilterFile()]
	self._converter = CmdConverter()
	self._writer = FileWriter()

	def on_modified(self, event):
	for _filter in self._filters:
	if not _filter(event.src_path):
	return
	cmd = self._converter(EventType.UPDATE, event.src_path)
	self._writer.write_to_file(cmd)
	print(f'cmd:{cmd}', end='')


	class WatchDog:

	json_info = {}
	valid_init = False

	@classmethod
	def init(cls):
	if cls.valid_init:
	return
	try:
	with open("./config.json", "r") as f:
	cls.json_info = json.load(f)
	cls.valid_init = True
	except Exception:
	pass

	@classmethod
	def get_json_info(cls, key):
	if not cls.valid_init:
	cls.init()
	return cls.json_info.get(key)

	@classmethod
	def listen_change(cls):
	observer = Observer()
	observer.schedule(AutoUpdateHandler(), path=cls.get_json_info("root_path"), recursive=True)
	observer.start()
	try:
	while True:
	time.sleep(1)
	except KeyboardInterrupt:
	observer.stop()
	observer.join()


	if __name__ == "__main__":
	WatchDog.init()
	WatchDog.listen_change()

部分路径参数抽离到了 config.json 文件内：

	{
	"root_path":"/root",
	"listen_path":[
	"/root/xxxx"
	// 需要监听的文件目录，有些文件可以忽略修改，在这里进行定制
	],
	"listen_file_ext":[".py"], // 监听文件后缀
	"update_path":"需要写入的 update 文件，该文件会被热更程序定时读取",
	"converter_root":"用于转换更新指令的路径例如监听路径是 /root/ 修改文件是 /root/code/Python/test.py 而热更指令可能是 update Python.test.py, 所以这里应该填 /root/code/ 用于进行路径转换",
	"cmd_format":"{0} {1}\n", // 热更指令，例如：update Python.test.py
	"replace_char":[
	"/",
	"\\"
	]
	}

参考内容：

watchdog github 源码

Python watchdog