作者：星安果

來源：AirPython

最近在部署前端項目的時候，需要先將前端項目壓縮包通過堡壘機上傳到應用服務器的 /tmp 目錄下，然后進入應用服務器中，使用 mv 命令將壓縮文件移動到 Nginx 項目設定目錄，最后使用 unzip 命令解壓文件，以此完成項目的部署

仔細分析，大部分操作都是重復性的動作，人工去完成這些操作會大大降低工作效率

本篇文章將介紹如何利用 Python 監控文件夾，以此輔助完成服務的部署動作

1. 準備

這里要介紹一個 Python 依賴庫「 watchdog 」

它可用于監控某個文件目錄下的文件變化，包含：刪除、修改、新增等操，每一個操作都會回調一個事件函數，我們可以在內部編寫自定義的邏輯，以此滿足我們的需求

# 安裝依賴包
pip3 install watchdog

項目地址：

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

2. 實戰一下

首先，我們需要創建一個監聽器，用于監聽文件夾目錄

from watchdog.observers import Observer
...
# 創建一個監聽器，用來監聽文件夾目錄
observer = Observer()
...

然后，創建 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))

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壓縮包！')
...

接著，通過監聽器啟動上面兩個事件的監聽任務

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()
...

最后，我們在服務器上通過「 nohup 」命令，讓文件監聽程序在后臺運行即可

# 在后臺運行
# 項目文件：watch_folder.py
# 日志文件：watch_folder.log
nohup python3 -u watch_folder.py > watch_folder.log 2>&1 &
# 查看日志：
cat watch_folder.log

3. 總結

通過上面的操作，每次我通過堡壘機將前端 zip 壓縮項目文件上傳到應用服務器的 /tmp 目錄下，程序會自動進行后面的操作，自動完成應用部署