實戰(zhàn) | 如何用 Python 自動化監(jiān)控文件夾完成服務(wù)部署！

大家好，我是安果！

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

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

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

1. 準(zhǔn)備

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

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

  1. # 安裝依賴包
  2. pip3 install watchdog

項目地址：

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

2. 實戰(zhàn)一下

首先，我們需要創(chuàng)建一個監(jiān)聽器，用于監(jiān)聽文件夾目錄

  1. from watchdog.observers import Observer
  2.  
  3. ...
  4. # 創(chuàng)建一個監(jiān)聽器，用來監(jiān)聽文件夾目錄
  5. observer = Observer()
  6. ...

然后，創(chuàng)建 2 個事件處理對象

PS：該對象繼承于「 FileSystemEventHandler 」類

它們分別用于監(jiān)聽「 /tmp 」目錄、「 /home/project/frontend 」目錄，假設(shè)事件對象被命名為 obj1、obj2

obj1 負(fù)責(zé)監(jiān)聽 /tmp 目錄，重寫「 新建或修改 」事件方法，完成壓縮文件的移動操作

  1. from watchdog.events import *
  2. import ntpath
  3. import shutil
  4. import zipfile
  5.  
  6. def get_filename(filepath):
  7.     """
  8.     根據(jù)文件夾目錄，獲取文件名稱（待后綴）
  9.     :param filepath:
  10.     :return:
  11.     """
  12.     return ntpath.basename(filepath)
  13.  
  14. class FileMoveHandler(FileSystemEventHandler):
  15.     def __init__(self):
  16.         FileSystemEventHandler.__init__(self)
  17.  
  18.     ...
  19.  
  20.     # 文件新建
  21.     def on_created(self, event):
  22.         # 新建文件夾
  23.         if event.is_directory:
  24.             # print("directory created:{0}".format(event.src_path))
  25.             pass
  26.         # 新建文件
  27.         else:
  28.             # print("file created:{0}".format(event.src_path))
  29.             filename = get_filename(event.src_path)
  30.  
  31.             # 如果屬于前端的4個項目壓縮包，開始文件夾的操作
  32.             if filename in watch_tags:
  33.                 self.start(filename)
  34.    ...
  35.    def on_modified(self, event):
  36.         if event.is_directory:
  37.             # print("directory modified:{0}".format(event.src_path))
  38.             pass
  39.         else:
  40.             # print("file modified:{0}".format(event.src_path))
  41.             filename = get_filename(event.src_path)
  42.             if filename in watch_tags:
  43.                 self.start(filename)
  44.     ...
  45.         def start(self, filename):
  46.         """
  47.         文件處理邏輯
  48.         :param filename:
  49.         :return:
  50.         """
  51.         try:
  52.             # 文件名不帶后綴
  53.             filename_without_suffix = filename.split(".")[0]
  54.  
  55.             # 源文件路徑（壓縮包文件）
  56.             source_file_path = watch_folder + filename
  57.  
  58.             # 目標(biāo)文件路徑（壓縮包文件）
  59.             target_file_path = target_folder + filename
  60.  
  61.             # 目標(biāo)項目文件夾（目標(biāo)項目）
  62.             target_project_path = target_folder + filename_without_suffix
  63.  
  64.             # 1、復(fù)制文件到目標(biāo)文件夾
  65.             print(f"拷貝源目錄{source_file_path},目標(biāo)文件夾:{target_folder}")
  66.             # 刪除目標(biāo)文件夾下的壓縮文件
  67.             if os.path.exists(target_file_path):
  68.                 os.remove(target_file_path)
  69.             # 移動文件到目標(biāo)文件夾中
  70.             shutil.move(source_file_path, target_folder)
  71.  
  72.             # 2、清空目標(biāo)文件夾中內(nèi)的所有文件夾（如果存在）
  73.             # 如果不存在，新建一個文件夾
  74.             if os.path.exists(target_project_path):
  75.                 shutil.rmtree(target_project_path, ignore_errors=True)
  76.  
  77.             print(f"項目{filename_without_suffix}移動成功！")
  78.         except Exception as e:
  79.             print("部署失敗，錯誤原因:", str(e.args))





obj2 負(fù)責(zé)監(jiān)聽 /home/project/frontend 目錄，同樣重寫「 新建或修改 」事件方法，完成壓縮文件的解壓動作

  1. ...
  2.     def start(self, filename):
  3.         # 文件名不帶后綴
  4.         filename_without_suffix = filename.split(".")[0]
  5.         # 目標(biāo)文件路徑（壓縮包文件）
  6.         target_file_path = target_folder + filename
  7.  
  8.         # 目標(biāo)項目文件夾（目標(biāo)項目）
  9.         target_project_path = target_folder + filename_without_suffix
  10.         r = zipfile.is_zipfile(target_file_path)
  11.         if r:
  12.             fz = zipfile.ZipFile(target_file_path, 'r')
  13.             for file in fz.namelist():
  14.                 fz.extract(file, target_folder)
  15.         else:
  16.             print('這不是一個正常的zip壓縮包！')
  17. ...

接著，通過監(jiān)聽器啟動上面兩個事件的監(jiān)聽任務(wù)

  1. import time
  2. ...
  3. if __name__ == "__main__":
  4.     # 待監(jiān)聽的文件夾目錄
  5.     watch_folder = "/tmp/"
  6.  
  7.     # 項目目標(biāo)文件夾目錄
  8.     target_folder = "/home/project/frontend/"
  9.  
  10.     # 監(jiān)聽文件夾名稱，即：項目壓縮包名稱
  11.     watch_tags = ['proj1.zip', 'proj2.zip', 'proj3.zip', 'proj4.zip']
  12.  
  13.     # 創(chuàng)建一個監(jiān)聽器，用來監(jiān)聽文件夾目錄
  14.     observer = Observer()
  15.  
  16.     # 創(chuàng)建兩個事件處理對象
  17.     move_handler = FileMoveHandler()
  18.     unzip_handler = FileUnzipHandler()
  19.  
  20.     # 啟動監(jiān)控任務(wù)
  21.     # 參數(shù)分別是：觀察者、監(jiān)聽目錄、是否監(jiān)聽子目錄
  22.     observer.schedule(move_handler, watch_folder, True)
  23.     observer.schedule(unzip_handler, target_folder, True)
  24.     observer.start()
  25.     try:
  26.         while True:
  27.             time.sleep(1)
  28.     except KeyboardInterrupt:
  29.         observer.stop()
  30.     observer.join()
  31. ...

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

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

3. 總結(jié)

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


作者：星安果


歡迎關(guān)注微信公眾號 ：AirPython