利用 Watchdog 和 rsync 實現實時文件同步

在現代軟件開發中，保持代碼庫和依賴文件的同步是至關重要的。尤其是在代碼庫和依賴文件分佈在不同工程時，自動化同步工具變得尤爲重要。本文將介紹如何使用 Python 的 Watchdog 庫以及 rsync 工具，構建一個實時文件同步腳本。

項目背景

在我們的項目中，我們需要在開發過程中實時同步兩個目錄，保證任何文件變動都能夠及時反映到目標目錄。通過這種方式，我們可以確保不同開發環境、不同開發工程之間的代碼和配置一致性。

項目實現

該項目的核心思想是利用 Watchdog 庫監聽源目錄中的文件變化事件，並通過 rsync 工具將變動同步到目標目錄。以下是詳細的實現步驟：

1. 引入必要的庫

import os
import time
import fnmatch
import subprocess
from queue import Queue
from threading import Thread
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import argparse

2. 定義打印日誌的輔助函數

爲了方便調試和監控，我們定義了一個簡單的日誌打印函數：

def printMSG(msg):  
    """打印日誌信息"""  
    print(f"[lib_sync.py] {msg}")

3. 定義自定義的文件系統事件處理程序

我們通過繼承FileSystemEventHandler類，定義了自己的事件處理程序SyncHandler：

# 自定義文件系統事件處理程序類  
class SyncHandler(FileSystemEventHandler):  
    def __init__(self, src_dir, dest_dir, ignore_patterns, event_queue):  
        """  
        初始化SyncHandler  
        :param src_dir: 源目錄  
        :param dest_dir: 目標目錄  
        :param ignore_patterns: 忽略模式列表  
        :param event_queue: 事件隊列  
        """  
        self.src_dir = src_dir  
        self.dest_dir = dest_dir  
        self.ignore_patterns = ignore_patterns  
        self.event_queue = event_queue  
  
    def should_ignore(self, path):  
        """  
        判斷路徑是否應被忽略  
        :param path: 文件路徑  
        :return: 是否應忽略  
        """  
        for pattern in self.ignore_patterns:  
            if fnmatch.fnmatch(path, pattern):  
                return True  
        return False  
  
    def on_any_event(self, event):  
        """  
        處理任何文件系統事件  
        :param event: 文件系統事件  
        """  
        # 如果事件路徑不在忽略列表中，則將其放入事件隊列  
        if not self.should_ignore(event.src_path):  
            self.event_queue.put(event.src_path)

4. 定義同步函數

我們使用rsync工具來同步源目錄和目標目錄：

def sync_directories(src_dir, dest_dir, ignore_patterns):  
    """  
    使用 rsync 同步目錄  
    :param src_dir: 源目錄  
    :param dest_dir: 目標目錄  
    :param ignore_patterns: 忽略模式列表  
    """  
    rsync_command = [  
        "rsync", "-avz", "--delete"  
    ] + [  
        "--exclude=" + pattern for pattern in ignore_patterns  
    ] + [src_dir, dest_dir]  
      
    # 運行 rsync 命令進行同步  
    subprocess.run(rsync_command)

5. 定義事件處理器

事件處理器從事件隊列中獲取文件變動信息，並調用同步函數進行同步：

def event_processor(event_queue, src_dir, dest_dir, ignore_patterns):  
    """  
    事件處理器，從隊列中獲取事件並進行同步  
    :param event_queue: 事件隊列  
    :param src_dir: 源目錄  
    :param dest_dir: 目標目錄  
    :param ignore_patterns: 忽略模式列表  
    """  
    while True:  
        time.sleep(2)  # 每2秒檢查一次事件隊列，可以根據需要調整  
        paths_to_sync = set()  
          
        # 獲取隊列中的所有事件路徑  
        while not event_queue.empty():  
            paths_to_sync.add(event_queue.get())  
          
        if paths_to_sync:  
            # 如果有事件需要處理，則進行同步  
            printMSG("有文件改動，正在同步...")  
            sync_directories(src_dir, dest_dir, ignore_patterns)

6. 主函數

主函數負責解析命令行參數，初始化事件處理器和觀察者，並啓動同步過程：

def main():  
    """  
    主函數，配置並啓動文件同步程序  
    """  
    # 設置命令行參數解析器  
    parser = argparse.ArgumentParser(description='依賴同步工具.')  
    parser.add_argument('src', type=str, help='Source folder path')  
    parser.add_argument('dest', type=str, help='Destination folder path')  
    parser.add_argument('--ignore', type=str, nargs='*', default=['*/build/*','*/node_modules/*','*/oh_modules/*','*/.idea/*','*/.hvigor/*','*/local.properties','*/.git/*'], help='Patterns to ignore')  
  
    # 解析命令行參數  
    args = parser.parse_args()  
  
    # 獲取絕對路徑  
    src_dir = os.path.abspath(args.src)  
    dest_dir = os.path.abspath(args.dest)  
    ignore_patterns = args.ignore  
    printMSG(f"src_dir : {src_dir}")  
    printMSG(f"dest_dir : {dest_dir}")  
    printMSG(f"ignore_patterns : {ignore_patterns}")  
  
    # 創建事件隊列  
    event_queue = Queue()  
      
    # 創建文件系統事件處理器  
    event_handler = SyncHandler(src_dir, dest_dir, ignore_patterns, event_queue)  
      
    # 創建並配置觀察者  
    observer = Observer()  
    observer.schedule(event_handler, path=src_dir, recursive=True)  
  
    # 創建並啓動事件處理線程  
    event_processor_thread = Thread(target=event_processor, args=(event_queue, src_dir, dest_dir, ignore_patterns))  
    event_processor_thread.daemon = True  # 設置爲守護線程  
    event_processor_thread.start()  
  
    try:  
        # 啓動觀察者  
        observer.start()  
        while True:  
            time.sleep(1)  
    except KeyboardInterrupt:  
        # 停止觀察者  
        observer.stop()  
    observer.join()  
  
if __name__ == "__main__":  
    main()

總結

通過本文，我們展示瞭如何利用 Python 和 Watchdog 庫實現一個實時文件同步工具。這個工具通過監聽源目錄中的文件變化，並使用 rsync 工具進行同步，確保源目錄和目標目錄的內容保持一致。這個方案簡單高效，適用於各種需要實時文件同步的場景。希望這篇文章對你有所幫助！

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/IfCO8L3DyO56RWsa7aumSg