1. 寫在前面

在軟件開發(fā)中，有時(shí)候需要通過 Python 去監(jiān)聽指定區(qū)域文件或目錄的創(chuàng)建、修改，或者刪除，從而引發(fā)特定的事件處理。本篇博客為你介紹第三方模塊 Watchdog 實(shí)現(xiàn)對(duì)文件事件的監(jiān)控。

公眾號(hào)： 滑翔的紙飛機(jī)

2. Watchdog

2.1 什么是 Watchdog?

用于監(jiān)視文件系統(tǒng)事件的 Python API 和 shell 實(shí)用程序。

**項(xiàng)目地址：**https://pypi.org/project/watchdog/
**最新版本：**Watchdog 3.0.0 適用于 Python 3.7+
**安裝：**需要運(yùn)行以下命令進(jìn)行安裝（確保使用的是 Python 3.7+）：

pip install watchdog

2.2 官方快速入門示例

以下示例程序：將以遞歸方式監(jiān)視當(dāng)前目錄文件系統(tǒng)變更，并簡單地將它們輸出到控制臺(tái)；

import sys
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

if __name__ == "__main__":
    # 設(shè)置日志信息格式
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    # 要監(jiān)控的目錄路徑
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    # 創(chuàng)建一個(gè)日志事件處理程序
    event_handler = LoggingEventHandler()
    # 創(chuàng)建一個(gè)觀察者對(duì)象
    observer = Observer()
    # 聲明一個(gè)定時(shí)任務(wù)
    observer.schedule(event_handler, path, recursive=True)
    # 啟動(dòng)定時(shí)任務(wù)
    observer.start()
    try:
        while observer.is_alive():
            observer.join(1)
    finally:
        observer.stop()
        observer.join()

輸出: 跟蹤目錄變更事件，通過日志輸出變更記錄。

例如: 創(chuàng)建 test > 1.txt 控制臺(tái)輸出：

2023-10-19 00:56:18 - Created directory: /Users/demo/2023/10/watchdog/test
2023-10-19 00:56:18 - Modified directory: /Users/demo/2023/10/watchdog
2023-10-19 00:56:27 - Created file: /Users/demo/2023/10/watchdog/test/1.txt
2023-10-19 00:56:27 - Modified directory: /Users/demo/2023/10/watchdog/test

2.3 Event Handler 和 Observer

Watchdog 的主要實(shí)現(xiàn)或者可以說 Watchdog 的構(gòu)件是基于以下類：

• Observer：觀察者，用于監(jiān)視目錄并將調(diào)用分派給事件處理程序；
• Event handler: 文件系統(tǒng)事件和事件處理程序；

說白了，Observer 監(jiān)控目錄，觸發(fā) Event handler 針對(duì)事件做出響應(yīng)；

導(dǎo)入方式：

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler 

2.3.1 Event handler

以下是Watchdog中默認(rèn)提供的4個(gè)事件處理類：

• FileSystemEventHandler：文件，事件處理器的基類，用于處理事件；
• PatternMatchingEventHandler：模式匹配文件；
• RegexMatchingEventHandler：正則匹配文件；
• LoggingEventHandler：記錄日志。

有關(guān)處理程序的更多詳情，請(qǐng)參閱此鏈接

通過擴(kuò)展 Watchdog 提供的默認(rèn)事件處理程序類，實(shí)現(xiàn)自定義的函數(shù)來處理修改、創(chuàng)建、刪除和移動(dòng)事件。還可以覆蓋 FileSystemEventHandler 中的函數(shù)（以下函數(shù)），因?yàn)槠渌录幚眍惗祭^承自該類。

**on_any_event(event)：**捕獲所有事件處理程序；
**on_created(event)：**在創(chuàng)建文件或目錄時(shí)調(diào)用；
**on_deleted(event)：**刪除文件或目錄時(shí)調(diào)用；
**on_modified(event)：**當(dāng)文件或目錄被修改時(shí)調(diào)用；
**on_moved(event)：**在移動(dòng)或重命名文件或目錄時(shí)調(diào)用；
**on_closed(event)：**文件已關(guān)閉時(shí)調(diào)用；
**on_opened(event)：**打開文件時(shí)調(diào)用；

每個(gè)函數(shù)都有一個(gè)名為 event 的輸入?yún)?shù)，其中包含以下變量：

• event_type：字符串形式的事件類型（“moved”、“deleted”、“created”、“modified”、“closed”、“opened”），默認(rèn)為 “無”；
• is_directory：True：表示事件針對(duì)目錄；
• src_path：觸發(fā)此事件的文件系統(tǒng)對(duì)象的源路徑；

2.3.2 Observer

如果大家熟悉設(shè)計(jì)模式，那么 Watchdog 就遵循觀察設(shè)計(jì)模式。因此，每個(gè)觀察者都會(huì)有事件，如果文件或目錄有任何變化，它就會(huì)查看并顯示變化。

Observer，觀察目錄，針對(duì)事件調(diào)用處理程序，也可以直接導(dǎo)入特定平臺(tái)的類，并用它代替 Observer

2.3.3 回顧上文簡單示例

通過上述介紹，對(duì) Event handler 和 Observer有一個(gè)簡單的理解，現(xiàn)在我們回過頭繼續(xù)來看官方示例：

import sys 
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

if __name__ == "__main__":
    # 設(shè)置日志信息格式
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    # 要監(jiān)控的目錄路徑
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    # 創(chuàng)建一個(gè)日志事件處理程序
    event_handler = LoggingEventHandler()
    # 創(chuàng)建一個(gè)觀察者對(duì)象
    observer = Observer()
    # 聲明一個(gè)定時(shí)任務(wù)
    observer.schedule(event_handler, path, recursive=True)
    # 啟動(dòng)定時(shí)任務(wù)
    observer.start()
    try:
        while observer.is_alive():
            observer.join(1)
    finally:
        observer.stop()
        observer.join()

（1）event_handler = LoggingEventHandler()： 創(chuàng)建一個(gè)日志事件處理程序；
（2）observer = Observer()：創(chuàng)建一個(gè)觀察者對(duì)象；
（3）observer.schedule(event_handler, path, recursive=True)：聲明一個(gè)定時(shí)任務(wù)，傳入事件處理程序、監(jiān)控路徑、以及是否遞歸子目錄；
（4）observer.start()：啟動(dòng)定時(shí)任務(wù)；

進(jìn)一步分析下：

schedule(self, event_handler, path, recursive=False):該方法用于監(jiān)視 path 路徑，并調(diào)用給定的事情 event_handler 。

參數(shù) recursive 表示是否遞歸子目錄，即監(jiān)聽子目錄，默認(rèn)為 False。

start():啟動(dòng)線程，這里開啟了新的守護(hù)線程，主程序如果結(jié)束， 該線程也會(huì)停止。
每個(gè)線程對(duì)象只能調(diào)用1次，它安排對(duì)象的 run() 方法在單獨(dú)的控制線程中調(diào)用，如果在同一線程對(duì)象上多次調(diào)用此方法將引發(fā) RuntimeError。

2.4. 理解和使用

基于上述關(guān)鍵概念介紹以及官方示例，自己實(shí)現(xiàn)一個(gè)文件事件監(jiān)聽；

在本例中，使用 FileSystemEventHandler 事件類。對(duì)一個(gè)文件夾設(shè)置監(jiān)視，并在有文件產(chǎn)生時(shí)觸發(fā)另一個(gè)函數(shù)。處理完成后，將把文件移到另一個(gè)文件夾。

（1）首先，你需要?jiǎng)?chuàng)建一個(gè)繼承自 FileSystemEventHandler 事件處理類，并創(chuàng)建一個(gè)觀察者和自定義的事件處理程序?qū)嵗?/p>

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class MyHandler(FileSystemEventHandler):
  pass

observer = Observer()
event_handler = MyHandler()

（2）創(chuàng)建一個(gè)定時(shí)任務(wù)，傳入以下參數(shù)

• event_handler：剛創(chuàng)建的處理程序?qū)ο螅?/li>
• path：要跟蹤的文件夾路徑；
• recursive：是否遞歸子目錄；

observer.schedule(event_handler, path='./input_files', recursive=True)

(3) observer.start() - 啟動(dòng)任務(wù)，等待目錄產(chǎn)生事件，觸發(fā)事件處理程序中的代碼。

(4) observer.stop() - 該函數(shù)將清理資源。

(5) 最后用 observer.join() 結(jié)束，因?yàn)槲覀冊(cè)谶@里使用的是多線程概念。join() 將連接多個(gè)線程，直到調(diào)用 join 方法的線程終止。

observer.start()
try:
    while True:
        time.sleep(300)
except KeyboardInterrupt:
    observer.stop()
observer.join()

接下去，自定義事件處理類：MyHandler

在這個(gè)示例中，我將檢查是否有文件上傳到所跟蹤的文件夾中。為此，我可以使用 on_created(event)：

def create_directory(file_path=None):
    # 以'年-月-日'的格式獲取當(dāng)前日期
    current_date = datetime.now().strftime('%Y-%m-%d')

    # 創(chuàng)建一個(gè)包含當(dāng)前日期的文件夾
    folder_path = f'{file_path}/{current_date}'
    if not os.path.exists(folder_path):
        os.makedirs(folder_path)
        return folder_path
    else:
        return folder_path


class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        dir_path = event.src_path.split('/input_files')
        processed_files = f'{dir_path[0]}/processed_files'

        child_processed_dir = create_directory(file_path=processed_files)

        if event:
            print("file created:{}".format(event.src_path))
            # 這里調(diào)用其他處理函數(shù)
            main(file_name=event.src_path)

            file_name = event.src_path.split('/')[-1]
            destination_path = f'{child_processed_dir}/{file_name}'

            # 將文件移動(dòng)到其他目錄
            shutil.move(event.src_path, destination_path)
            print("file moved:{} to {}".format(event.src_path, destination_path))

在上面的示例中，我使用函數(shù) create_directory() 來檢查目標(biāo)路徑中是否有當(dāng)前日期的文件夾，否則就創(chuàng)建相同的文件夾。

然后，在其他 python 腳本函數(shù) main() 中做了一些處理后，使用相同的路徑作為目標(biāo)路徑來移動(dòng)文件

下面是最終代碼：my_event_handler.py

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import shutil
import time
import os
from datetime import datetime
from watchdog_fileobserver import main


def create_directory(file_path=None):
    # 以'年-月-日'的格式獲取當(dāng)前日期
    current_date = datetime.now().strftime('%Y-%m-%d')

    # 創(chuàng)建一個(gè)包含當(dāng)前日期的文件夾
    folder_path = f'{file_path}/{current_date}'
    if not os.path.exists(folder_path):
        os.makedirs(folder_path)
        return folder_path
    else:
        return folder_path


class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        dir_path = event.src_path.split('/input_files')
        processed_files = f'{dir_path[0]}/processed_files'

        child_processed_dir = create_directory(file_path=processed_files)

        if event:
            print("file created:{}".format(event.src_path))
            # 這里調(diào)用其他處理函數(shù)
            main(file_name=event.src_path)

            file_name = event.src_path.split('/')[-1]
            destination_path = f'{child_processed_dir}/{file_name}'

            shutil.move(event.src_path, destination_path)
            print("file moved:{} to {}".format(event.src_path, destination_path))


if __name__ == "__main__":
    observer = Observer()
    event_handler = MyHandler()
    observer.schedule(event_handler, path='./input_files', recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(300)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

watchdog_fileobserver.py:

import csv

def read_csv_file(file_name):
    try:
        with open(f"{file_name}", 'r') as file:
          csvreader = csv.DictReader(file)
          for row in csvreader:
            print(row)
        return csvreader
    except Exception as e:
        pass

def main(file_name=None):
    if file_name:
        dict_data = read_csv_file(file_name)
        print("Process completed")
    else:
        print("Invalid file path")

在這種情況下，需要等待文件上傳，然后執(zhí)行所需的操作。為此，你可以在事件函數(shù)中添加以下代碼：

def on_created(self, event):
  file_size = -1
  while file_size != os.path.getsize(event.src_path):
      file_size = os.path.getsize(event.src_path)
      print(file_size)
      time.sleep(1)

###    OR   ###

def on_created(self, event):
  file = None
  while file is None:
    try:
        file = open(event.src_path)
    except OSError:
        logger.info('Waiting for file transfer....')
        time.sleep(1)
        continue

驗(yàn)證：

腳本所在路徑下，創(chuàng)建用于監(jiān)聽目錄 input_files， 在該目錄下創(chuàng)建一個(gè)文件，控制臺(tái)輸出：

file created:/Users/demo/2023/10/watchdog/input_files/text.txt
Process completed
file moved:/Users/demo/2023/10/watchdog/input_files/text.txt to /Users/demo/2023/10/watchdog/processed_files/2023-10-20/text.txt

目錄如下：

.
├── input_files
├── my_event_handler.py
├── processed_files
│   └── 2023-10-20
│       └── text.txt
└── watchdog_fileobserver.py

2.5. Watchdog 使用案例

2.5.1 忽略子目錄或只包含模式匹配文件的情況

如果要忽略某個(gè)目錄中的某些文件，可以使用最簡單的方法之一，即使用 PatternMatchingEventHandler

在文件 my_event_handler.py 中，修改 MyHandler 中的繼承類（PatternMatchingEventHandler），如下所示：

class MyHandler(PatternMatchingEventHandler):
    ....
    ....

if __name__ == "__main__":
    event_handler = MyHandler(patterns=["*.csv", "*.pdf"],
                              ignore_patterns=[],
                              ignore_directories=True
                              )
     ....
     ....

2.5.2 使用 Celery 來啟動(dòng)/停止 Watchdog

可以使用下面的示例來實(shí)現(xiàn) Watchdog。不過，這個(gè)示例只是一個(gè)關(guān)于如何將 celery 集成到Watchdog 中的想法。

from celery import Celery
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler
import os
import time

app = Celery('celery_ex.celery_apptask_ex', broker='redis://localhost:6379/0')


@app.task
def process_file(file_path):
    # do something with the file
    with open(file_path, 'r') as f:
        print(f.read())


class MyHandler(PatternMatchingEventHandler):
    def on_created(self, event):
        file_size = -1
        while file_size != os.path.getsize(event.src_path):
            file_size = os.path.getsize(event.src_path)
            print(file_size)
            time.sleep(1)

        if event:
            print("file created:{}".format(event.src_path))
            # call function here
            process_file.apply_async(args=(event.src_path,))


if __name__ == "__main__":
    observer = Observer()
    event_handler = MyHandler(patterns=["*.csv", "*.pdf"],
                              ignore_patterns=[],
                              ignore_directories=True
                              )
    observer.schedule(event_handler, path='./input_files', recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

此示例需要 redis、celery 支持；

2.5.3 監(jiān)控目錄變化

觀察者（observer）可以設(shè)置指定目錄及其所有子目錄，在文件或目錄創(chuàng)建、刪除或修改時(shí)調(diào)用相應(yīng)的方法（on_created、on_deleted 或 on_modified），觀察者以無限循環(huán)的方式運(yùn)行，可以被鍵盤中斷打斷。

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class EventHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            print("Directory created:", event.src_path)
        else:
            print("File created:", event.src_path)

    def on_deleted(self, event):
        if event.is_directory:
            print("Directory deleted:", event.src_path)
        else:
            print("File deleted:", event.src_path)

    def on_modified(self, event):
        if event.is_directory:
            print("Directory modified:", event.src_path)
        else:
            print("File modified:", event.src_path)

event_handler = EventHandler()
observer = Observer()
observer.schedule(event_handler, "/path/to/dir", recursive=True)
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

2.5.4 使用線程和多進(jìn)程執(zhí)行 Watchdog 來啟動(dòng)獨(dú)立進(jìn)程

可以運(yùn)行 Watchdog，使用線程和多進(jìn)程并行處理多個(gè)文件。下面是一個(gè)相同的示例:

from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler
import os
import ntpath
import time
import optparse
import multiprocessing
import threading
from collections import OrderedDict

lock = threading.RLock()


def process_function(get_event, event_dict):
    print(f"Process started for event: {get_event}")
    dir_path = ntpath.abspath(get_event)
    file_name = ntpath.basename(get_event)

    if len(get_event) > 0:
        your_own_function()
        do something....

class Handler(PatternMatchingEventHandler):
    def __init__(self, queue):
        PatternMatchingEventHandler.__init__(self, patterns=['*.csv'],
                                             ignore_patterns=[],
                                             ignore_directories=True)
        self.queue = queue

    def on_created(self, event):
        # logger.info(f"Wait while the transfer of the file is finished before processing it")
        # file_size = -1
        # while file_size != os.path.getsize(event.src_path):
        #     file_size = os.path. getsize(event.src_path)
        #     time.sleep(1)

        file = None
        while file is None:
            try:
                file = open(event.src_path)
            except OSError:
                logger.info('Waiting for file transfer')
                time.sleep(5)
                continue

        self.queue.put(event.src_path)

    def on_modified(self, event):
        pass


def start_watchdog(watchdog_queue, dir_path):
    logger.info(f"Starting Watchdog Observer\n")
    event_handler = Handler(watchdog_queue)
    observer = Observer()
    observer.schedule(event_handler, dir_path, recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except Exception as error:
        observer.stop()
        logger.error(f"Error: {str(error)}")
    observer.join()


if __name__ == '__main__':
    dir_path = r'//file_path/'

    watchdog_queue = Queue()

    logger.info(f"Starting Worker Thread")
    worker = threading.Thread(target=start_watchdog, name="Watchdog",
                              args=(watchdog_queue, dir_path), daemon=True)
    worker.start()

    mp = Manager()
    event_dict = mp.dict()

    while True:
        if not watchdog_queue.empty():
            logger.info(f"Is Queue empty: {watchdog_queue.empty()}")
            pool = Pool()
            pool.apply_async(process_function, (watchdog_queue.get(), event_dict))
        else:
            time.sleep(1)

2.5.5 在 Watchdog 中進(jìn)行日志記錄

要記錄事件，可以創(chuàng)建一個(gè)繼承自 FileSystemEventHandler 類的自定義事件處理程序類，并重寫與要記錄的事件相對(duì)應(yīng)的方法。

下面舉例說明如何使用 Watchdog 庫記錄文件創(chuàng)建和修改事件：

import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class LogEventHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory:
            logging.info(f"File created: {event.src_path}")

    def on_modified(self, event):
        if not event.is_directory:
            logging.info(f"File modified: {event.src_path}")

logging.basicConfig(filename='watchdog.log', level=logging.INFO, format='%(asctime)s - %(message)s')
event_handler = LogEventHandler()
observer = Observer()
observer.schedule(event_handler, "/path/to/", recursive=True)
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

3. 最后

無論你是在一個(gè)需要跟蹤多個(gè)文件的大型項(xiàng)目中工作，還是只想關(guān)注單個(gè)文件的任務(wù)，Watchdog 庫都能滿足你的需求。文章來源地址http://www.zghlxwxcb.cn/news/detail-782589.html

感謝您花時(shí)間閱讀文章

關(guān)注公眾號(hào)不迷路

到了這里，關(guān)于Python Watchdog：高效的文件系統(tǒng)監(jiān)控的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！