實(shí)戰(zhàn) | 如何用 Python 自動(dòng)化監(jiān)控文件夾完成服務(wù)部署！
大家好，我是安果！
最近在部署前端項(xiàng)目的時(shí)候，需要先將前端項(xiàng)目壓縮包通過(guò)堡壘機(jī)上傳到應(yīng)用服務(wù)器的 /tmp 目錄下，然后進(jìn)入應(yīng)用服務(wù)器中，使用 mv 命令將壓縮文件移動(dòng)到 Nginx 項(xiàng)目設(shè)定目錄，最后使用 unzip 命令解壓文件，以此完成項(xiàng)目的部署
仔細(xì)分析，大部分操作都是重復(fù)性的動(dòng)作，人工去完成這些操作會(huì)大大降低工作效率
本篇文章將介紹如何利用 Python 監(jiān)控文件夾，以此輔助完成服務(wù)的部署動(dòng)作
1. 準(zhǔn)備
這里要介紹一個(gè) Python 依賴庫(kù)「 watchdog 」
它可用于監(jiān)控某個(gè)文件目錄下的文件變化，包含：刪除、修改、新增等操，每一個(gè)操作都會(huì)回調(diào)一個(gè)事件函數(shù)，我們可以在內(nèi)部編寫自定義的邏輯，以此滿足我們的需求
# 安裝依賴包
pip3 install watchdog項(xiàng)目地址：
https://pypi.org/project/watchdog/
2. 實(shí)戰(zhàn)一下
首先，我們需要?jiǎng)?chuàng)建一個(gè)監(jiān)聽(tīng)器，用于監(jiān)聽(tīng)文件夾目錄
from watchdog.observers import Observer
...
# 創(chuàng)建一個(gè)監(jiān)聽(tīng)器，用來(lái)監(jiān)聽(tīng)文件夾目錄
observer = Observer()
...然后，創(chuàng)建 2 個(gè)事件處理對(duì)象
PS：該對(duì)象繼承于「 FileSystemEventHandler 」類
它們分別用于監(jiān)聽(tīng)「 /tmp 」目錄、「 /home/project/frontend 」目錄，假設(shè)事件對(duì)象被命名為 obj1、obj2
obj1 負(fù)責(zé)監(jiān)聽(tīng) /tmp 目錄，重寫「 新建或修改 」事件方法，完成壓縮文件的移動(dòng)操作
from watchdog.events import *
import ntpath
import shutil
import zipfile
def get_filename(filepath):
"""
根據(jù)文件夾目錄，獲取文件名稱（待后綴）
: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個(gè)項(xiàng)目壓縮包，開(kāi)始文件夾的操作
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
# 目標(biāo)文件路徑（壓縮包文件）
target_file_path = target_folder + filename
# 目標(biāo)項(xiàng)目文件夾（目標(biāo)項(xiàng)目）
target_project_path = target_folder + filename_without_suffix
# 1、復(fù)制文件到目標(biāo)文件夾
print(f"拷貝源目錄{source_file_path},目標(biāo)文件夾:{target_folder}")
# 刪除目標(biāo)文件夾下的壓縮文件
if os.path.exists(target_file_path):
os.remove(target_file_path)
# 移動(dòng)文件到目標(biāo)文件夾中
shutil.move(source_file_path, target_folder)
# 2、清空目標(biāo)文件夾中內(nèi)的所有文件夾（如果存在）
# 如果不存在，新建一個(gè)文件夾
if os.path.exists(target_project_path):
shutil.rmtree(target_project_path, ignore_errors=True)
print(f"項(xiàng)目{filename_without_suffix}移動(dòng)成功！")
except Exception as e:
print("部署失敗，錯(cuò)誤原因:", str(e.args))
obj2 負(fù)責(zé)監(jiān)聽(tīng) /home/project/frontend 目錄，同樣重寫「 新建或修改 」事件方法，完成壓縮文件的解壓動(dòng)作
...
def start(self, filename):
# 文件名不帶后綴
filename_without_suffix = filename.split(".")[0]
# 目標(biāo)文件路徑（壓縮包文件）
target_file_path = target_folder + filename
# 目標(biāo)項(xiàng)目文件夾（目標(biāo)項(xiàng)目）
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('這不是一個(gè)正常的zip壓縮包！')
...接著，通過(guò)監(jiān)聽(tīng)器啟動(dòng)上面兩個(gè)事件的監(jiān)聽(tīng)任務(wù)
import time
...
if __name__ == "__main__":
# 待監(jiān)聽(tīng)的文件夾目錄
watch_folder = "/tmp/"
# 項(xiàng)目目標(biāo)文件夾目錄
target_folder = "/home/project/frontend/"
# 監(jiān)聽(tīng)文件夾名稱，即：項(xiàng)目壓縮包名稱
watch_tags = ['proj1.zip', 'proj2.zip', 'proj3.zip', 'proj4.zip']
# 創(chuàng)建一個(gè)監(jiān)聽(tīng)器，用來(lái)監(jiān)聽(tīng)文件夾目錄
observer = Observer()
# 創(chuàng)建兩個(gè)事件處理對(duì)象
move_handler = FileMoveHandler()
unzip_handler = FileUnzipHandler()
# 啟動(dòng)監(jiān)控任務(wù)
# 參數(shù)分別是：觀察者、監(jiān)聽(tīng)目錄、是否監(jiān)聽(tīng)子目錄
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()
...
最后，我們?cè)诜?wù)器上通過(guò)「 nohup 」命令，讓文件監(jiān)聽(tīng)程序在后臺(tái)運(yùn)行即可
# 在后臺(tái)運(yùn)行
# 項(xiàng)目文件：watch_folder.py
# 日志文件：watch_folder.log
nohup python3 -u watch_folder.py > watch_folder.log 2>&1 &
# 查看日志：
cat watch_folder.log
3. 總結(jié)
通過(guò)上面的操作，每次我通過(guò)堡壘機(jī)將前端 zip 壓縮項(xiàng)目文件上傳到應(yīng)用服務(wù)器的 /tmp 目錄下，程序會(huì)自動(dòng)進(jìn)行后面的操作，自動(dòng)完成應(yīng)用部署
我已經(jīng)將文中源碼上傳到后臺(tái)，回復(fù)關(guān)鍵字「 watchdog 」獲取完整的源碼
如果你覺(jué)得文章還不錯(cuò)，請(qǐng)大家 點(diǎn)贊、分享、留言 下，因?yàn)檫@將是我持續(xù)輸出更多優(yōu)質(zhì)文章的最強(qiáng)動(dòng)力！
END