Pertempuran praktikal: Cara menggunakan Python untuk memantau folder secara automatik untuk melengkapkan penggunaan perkhidmatan

Helo semua, saya Peng Tao!

Apabila menggunakan projek bahagian hadapan baru-baru ini, anda perlu memuat naik pakej termampat projek bahagian hadapan ke direktori /tmp pelayan aplikasi melalui hos benteng, kemudian masukkan pelayan aplikasi dan gunakan arahan mv untuk alihkan fail yang dimampatkan ke projek Nginx Tetapkan direktori, dan akhirnya gunakan arahan unzip untuk unzip fail untuk menyelesaikan penggunaan projek.

Analisis yang teliti menunjukkan bahawa kebanyakan operasi adalah tindakan berulang, dan melengkapkan operasi ini secara manual akan mengurangkan kecekapan kerja dengan ketara.

Artikel ini akan memperkenalkan cara menggunakan Python untuk memantau folder untuk membantu dalam penggunaan perkhidmatan.

1. Persediaan

Di sini kami akan memperkenalkan pustaka pergantungan Python " watchdog "

Ia boleh digunakan untuk memantau perubahan fail dalam direktori fail tertentu, termasuk: pemadaman, pengubahsuaian, Menambah operasi menunggu baharu, setiap operasi akan memanggil semula fungsi acara, dan kami boleh menulis logik tersuai secara dalaman untuk memenuhi keperluan kami.

# 安装依赖包
pip3 install watchdog

Alamat projek:

​​https://www.php.cn/link/690e6de4e3e2c0916b6160d9959d156b​​​

><🎜 ia dalam amalan

Pertama, kita perlu mencipta pendengar untuk mendengar direktori folder

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

Kemudian, cipta 2 objek pengendalian acara.

PS: Objek ini diwarisi daripada kelas "FileSystemEventHandler".

Mereka digunakan untuk memantau direktori "/tmp" dan direktori "/home/project/frontend" masing-masing Anggap objek acara dinamakan obj1 dan obj2

obj1 bertanggungjawab untuk. memantau direktori /tmp. Tulis semula kaedah acara "Baharu atau Diubahsuai" untuk melengkapkan operasi pemindahan fail termampat:

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 bertanggungjawab untuk memantau direktori /home/project/frontend, dan juga menulis semula kaedah acara "Baharu atau Diubah suai" untuk melengkapkan penyahmampatan fail yang dimampatkan:

...
 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压缩包！')
...

Kemudian, mulakan tugas pemantauan dua peristiwa di atas melalui pendengar

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

Akhir sekali, kita boleh menggunakan arahan "nohup" pada pelayan untuk membiarkan program pemantauan fail berjalan di latar belakang

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

3 Ringkasan

Melalui operasi di atas, setiap kali saya memuat naik bahagian hadapan-. zip tamatkan fail projek dimampatkan ke direktori /tmp pelayan aplikasi melalui mesin benteng, program secara automatik akan melaksanakan operasi berikut , secara automatik menyelesaikan penggunaan aplikasi.

Atas ialah kandungan terperinci Pertempuran praktikal: Cara menggunakan Python untuk memantau folder secara automatik untuk melengkapkan penggunaan perkhidmatan. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!