快捷導航

基于Python Watchdog庫實現(xiàn)文件系統(tǒng)監(jiān)控

更新時間：2025年07月04日 08:47:37 作者：嘆一曲當時只道是尋常

Watchdog是一個優(yōu)秀的Python庫,用于監(jiān)控文件系統(tǒng)事件,本文將介紹Watchdog的基本用法,并通過一個實際案例展示如何用它來監(jiān)控下載目錄并自動轉(zhuǎn)換ICO文件為PNG格式

Watchdog是一個優(yōu)秀的Python庫，用于監(jiān)控文件系統(tǒng)事件。它可以檢測文件或目錄的創(chuàng)建、修改、刪除和移動等操作，并觸發(fā)相應(yīng)的回調(diào)函數(shù)。本文將介紹Watchdog的基本用法，并通過一個實際案例展示如何用它來監(jiān)控下載目錄并自動轉(zhuǎn)換ICO文件為PNG格式。

Watchdog簡介

Watchdog庫的主要組件包括：

Observer - 監(jiān)控文件系統(tǒng)變化的核心類
FileSystemEventHandler - 處理文件系統(tǒng)事件的基類
各種事件類 - 如FileCreatedEvent, FileModifiedEvent等

安裝Watchdog

pip install watchdog

實際案例：自動轉(zhuǎn)換ICO文件為PNG

下面是一個完整的示例，它監(jiān)控用戶的下載目錄，當檢測到新的ICO文件時，自動將其轉(zhuǎn)換為PNG格式（使用ImageMagick的convert工具）。

import os
import time
import subprocess
import shutil
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class ICOHandler(FileSystemEventHandler):
    def __init__(self, download_dir, convert_exe):
        self.download_dir = download_dir
        self.processed_files = set()
        self.convert_exe = convert_exe
        
        # 初始化時處理已存在的ico文件
        for filename in os.listdir(download_dir):
            if filename.lower().endswith('.ico'):
                self.processed_files.add(filename)
                self.convert_ico_to_png(filename)
    
    def on_created(self, event):
        if not event.is_directory and event.src_path.lower().endswith('.ico'):
            filename = os.path.basename(event.src_path)
            if filename not in self.processed_files:
                self.processed_files.add(filename)
                print(f"檢測到新的ICO文件: {filename}")
                self.convert_ico_to_png(filename)
    
    def convert_ico_to_png(self, ico_filename):
        ico_path = os.path.join(self.download_dir, ico_filename)
        base_name = os.path.splitext(ico_filename)[0]
        
        # 臨時輸出目錄
        temp_dir = os.path.join(self.download_dir, f"{base_name}_temp")
        os.makedirs(temp_dir, exist_ok=True)
        
        # 臨時輸出路徑模板
        temp_output = os.path.join(temp_dir, f"{base_name}_%d.png")
        
        try:
            # 使用ImageMagick的convert命令轉(zhuǎn)換ICO到PNG
            cmd = [
                self.convert_exe,
                ico_path,
                temp_output
            ]
            subprocess.run(cmd, check=True)
            print(f"已轉(zhuǎn)換: {ico_filename} -> 多個PNG文件")
            
            # 找出最大的PNG文件
            largest_file = self.find_largest_png(temp_dir)
            if largest_file:
                final_output = os.path.join(self.download_dir, f"{base_name}.png")
                os.rename(largest_file, final_output)
                print(f"已保存最大的PNG文件: {final_output}")
            
            # 清理臨時目錄
            self.cleanup_temp_dir(temp_dir)
            
        except subprocess.CalledProcessError as e:
            print(f"轉(zhuǎn)換失敗: {e}")
            self.cleanup_temp_dir(temp_dir)
        except Exception as e:
            print(f"發(fā)生錯誤: {e}")
            self.cleanup_temp_dir(temp_dir)
    
    def find_largest_png(self, directory):
        largest_size = 0
        largest_file = None
        
        for filename in os.listdir(directory):
            if filename.lower().endswith('.png'):
                filepath = os.path.join(directory, filename)
                file_size = os.path.getsize(filepath)
                if file_size > largest_size:
                    largest_size = file_size
                    largest_file = filepath
        
        return largest_file
    
    def cleanup_temp_dir(self, temp_dir):
        try:
            for filename in os.listdir(temp_dir):
                filepath = os.path.join(temp_dir, filename)
                os.remove(filepath)
            os.rmdir(temp_dir)
        except Exception as e:
            print(f"清理臨時目錄時出錯: {e}")

def main():
    # 獲取用戶下載目錄
    download_dir = os.path.expanduser('~/Downloads')

    # 在這里設(shè)置ImageMagick的convert.exe的完整路徑
    # 如果'convert'在您的系統(tǒng)PATH中，您可以直接寫'convert'
    # 否則，請?zhí)峁┩暾窂?，例? r'C:\Program Files\ImageMagick-7.1.1-Q16-HDRI\convert.exe'
    convert_exe = 'convert'
    
    # 查找并打印convert命令的完整路徑
    # 如果用戶沒有提供絕對路徑，則在PATH中搜索
    if convert_exe == 'convert' and not os.path.isabs(convert_exe):
        full_path = shutil.which(convert_exe)
        if full_path:
            print(f"查找到 'convert' 的完整路徑是: {full_path}")
            convert_exe = full_path  # 使用完整路徑以提高可靠性
        else:
            print(f"警告: 在系統(tǒng)PATH中找不到 '{convert_exe}'。程序可能會失敗。")

    # 確保ImageMagick已安裝
    try:
        subprocess.run([convert_exe, '--version'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    except (subprocess.CalledProcessError, FileNotFoundError):
        print(f"錯誤: 無法執(zhí)行 '{convert_exe}'。")
        print("請檢查路徑是否正確，或ImageMagick是否已安裝并添加到系統(tǒng)PATH中。")
        print("您可以修改腳本中 'convert_exe' 變量的值為 convert.exe 的完整路徑。")
        print("ImageMagick下載地址: https://imagemagick.org/script/download.php")
        return
    
    print(f"開始監(jiān)控目錄: {download_dir}")
    print("按Ctrl+C停止監(jiān)控")
    
    event_handler = ICOHandler(download_dir, convert_exe)
    observer = Observer()
    observer.schedule(event_handler, download_dir, recursive=False)
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == "__main__":
    main()