太牛了，用Python实现服务部署自动化！

2023-02-28

文件文件夹目录

最近在部署前端项目的时候，需要先将前端项目压缩包通过堡垒机上传到应用服务器的/tmp目录下，然后进入应用服务器中，使用mv命令将压缩文件移动到Nginx项目设定目录，最后使用unzip命令解压文件，以此完成项目的部署。仔细分析，大部分操作都是重复性的动作，人工去完成这些操作会大大降低工作效率。本篇文

最近在部署前端项目的时候，需要先将前端项目压缩包通过堡垒机上传到应用服务器的 /tmp 目录下，然后进入应用服务器中，使用 mv 命令将压缩文件移动到 Nginx 项目设定目录，最后使用 unzip 命令解压文件，以此完成项目的部署。

仔细分析，大部分操作都是重复性的动作，人工去完成这些操作会大大降低工作效率。

本篇文章将介绍如何利用 Python 监控文件夹，以此辅助完成服务的部署动作。

1. 准备

这里要介绍一个 Python 依赖库「 watchdog 」

它可用于监控某个文件目录下的文件变化，包含：删除、修改、新增等操作，每一个操作都会回调一个事件函数，我们可以在内部编写自定义的逻辑，以此满足我们的需求。

# 安装依赖包
pip3 install watchdog1.
2.

项目地址：

https://pypi.org/project/watchdog/

2. 实战一下

首先，我们需要创建一个监听器，用于监听文件夹目录：

from watchdog.observers import Observer
...
# 创建一个监听器，用来监听文件夹目录
observer = Observer()
...1.
2.
3.
4.
5.

然后，创建 2 个事件处理对象。

PS：该对象继承于「 FileSystemEventHandler 」类。

它们分别用于监听「 /tmp 」目录、「 /home/project/frontend 」目录，假设事件对象被命名为 obj1、obj2

obj1 负责监听 /tmp 目录，重写「新建或修改」事件方法，完成压缩文件的移动操作：

from watchdog.events import *
import ntpath
import shutil
import zipfile
def get_filename(filepath):
   """
   根据文件夹目录，获取文件名称（待后缀）
   :param filepath:
   :return:
   """
   return ntpath.basename(filepath)
class FileMoveHandler(FileSystemEventHandler):
   def __init__(self):
       FileSystemEventHandler.__init__(self)
   ...
   # 文件新建
   def on_created(self, event):
       # 新建文件夹
       if event.is_directory:
           # print("directory created:{0}".format(event.src_path))
           pass
       # 新建文件
       else:
           # print("file created:{0}".format(event.src_path))
           filename = get_filename(event.src_path)
           # 如果属于前端的4个项目压缩包，开始文件夹的操作
           if filename in watch_tags:
               self.start(filename)
  ...
  def on_modified(self, event):
       if event.is_directory:
           # print("directory modified:{0}".format(event.src_path))
           pass
       else:
           # print("file modified:{0}".format(event.src_path))
           filename = get_filename(event.src_path)
           if filename in watch_tags:
               self.start(filename)
   ...
       def start(self, filename):
       """
       文件处理逻辑
       :param filename:
       :return:
       """
       try:
           # 文件名不带后缀
           filename_without_suffix = filename.split(".")[0]
           # 源文件路径（压缩包文件）
           source_file_path = watch_folder + filename
           # 目标文件路径（压缩包文件）
           target_file_path = target_folder + filename
           # 目标项目文件夹（目标项目）
           target_project_path = target_folder + filename_without_suffix
           # 1、复制文件到目标文件夹
           print(f"拷贝源目录{source_file_path},目标文件夹:{target_folder}")
           # 删除目标文件夹下的压缩文件
           if os.path.exists(target_file_path):
               os.remove(target_file_path)
           # 移动文件到目标文件夹中
           shutil.move(source_file_path, target_folder)
           # 2、清空目标文件夹中内的所有文件夹（如果存在）
           # 如果不存在，新建一个文件夹
           if os.path.exists(target_project_path):
               shutil.rmtree(target_project_path, ignore_errors=True)
           print(f"项目{filename_without_suffix}移动成功！")
       except Exception as e:
           print("部署失败，错误原因:", str(e.args))1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.

obj2 负责监听 /home/project/frontend 目录，同样重写「新建或修改」事件方法，完成压缩文件的解压动作：

...
   def start(self, filename):
       # 文件名不带后缀
       filename_without_suffix = filename.split(".")[0]
       # 目标文件路径（压缩包文件）
       target_file_path = target_folder + filename
       # 目标项目文件夹（目标项目）
       target_project_path = target_folder + filename_without_suffix
       r = zipfile.is_zipfile(target_file_path)
       if r:
           fz = zipfile.ZipFile(target_file_path, 'r')
           for file in fz.namelist():
               fz.extract(file, target_folder)
       else:
           print('这不是一个正常的zip压缩包！')
...1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.

接着，通过监听器启动上面两个事件的监听任务：

import time
...
if __name__ == "__main__":
   # 待监听的文件夹目录
   watch_folder = "/tmp/"
   # 项目目标文件夹目录
   target_folder = "/home/project/frontend/"
   # 监听文件夹名称，即：项目压缩包名称
   watch_tags = ['proj1.zip', 'proj2.zip', 'proj3.zip', 'proj4.zip']
   # 创建一个监听器，用来监听文件夹目录
   observer = Observer()
   # 创建两个事件处理对象
   move_handler = FileMoveHandler()
   unzip_handler = FileUnzipHandler()
   # 启动监控任务
   # 参数分别是：观察者、监听目录、是否监听子目录
   observer.schedule(move_handler, watch_folder, True)
   observer.schedule(unzip_handler, target_folder, True)
   observer.start()
   try:
       while True:
           time.sleep(1)
   except KeyboardInterrupt:
       observer.stop()
   observer.join()
...1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.

最后，我们在服务器上通过「 nohup 」命令，让文件监听程序在后台运行即可。

# 在后台运行  
# 项目文件：watch_folder.py
# 日志文件：watch_folder.log
nohup python3 -u watch_folder.py > watch_folder.log 2>&1 &
# 查看日志：
cat watch_folder.log1.
2.
3.
4.
5.
6.