太牛了,用Python實(shí)現(xiàn)服務(wù)部署自動(dòng)化!
最近在部署前端項(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 依賴(lài)庫(kù)「 watchdog 」
它可用于監(jiān)控某個(gè)文件目錄下的文件變化,包含:刪除、修改、新增等操作,每一個(gè)操作都會(huì)回調(diào)一個(gè)事件函數(shù),我們可以在內(nèi)部編寫(xiě)自定義的邏輯,以此滿(mǎn)足我們的需求。
# 安裝依賴(là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 」類(lèi)。
它們分別用于監(jiān)聽(tīng)「 /tmp 」目錄、「 /home/project/frontend 」目錄,假設(shè)事件對(duì)象被命名為 obj1、obj2
obj1 負(fù)責(zé)監(jiān)聽(tīng) /tmp 目錄,重寫(xiě)「 新建或修改 」事件方法,完成壓縮文件的移動(dòng)操作:
from watchdog.events import *
import ntpath
import shutil
import zipfile
def get_filename(filepath):
"""
根據(jù)文件夾目錄,獲取文件名稱(chēng)(待后綴)
: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 目錄,同樣重寫(xiě)「 新建或修改 」事件方法,完成壓縮文件的解壓動(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)文件夾名稱(chēng),即:項(xiàng)目壓縮包名稱(chē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)力!