引言

上一篇介紹完了觀察者模式的原理，本篇想就此再介紹一個小應用，雖然我也就玩了一下午，是當時看observer正好找到的，以及還有Django-observer，但Django很久沒用了，所以提下這個作為一個筆記。

watchdog介紹

Watchdog的中文的“看門狗”，有保護的意思。最早引入Watchdog是在單片機系統中，由于單片機的工作環境容易受到外界磁場的干擾，導致程序“跑飛”，造成整個系統無法正常工作，因此，引入了一個“看門狗”，對單片機的運行狀態進行實時監測，針對運行故障做一些保護處理，譬如讓系統重啟。這種Watchdog屬于硬件層面，必須有硬件電路的支持。

Linux也引入了Watchdog，在Linux內核下，當Watchdog啟動后，便設定了一個定時器，如果在超時時間內沒有對/dev/Watchdog進行寫操作，則會導致系統重啟。通過定時器實現的Watchdog屬于軟件層面。

嗯，這樣的嘛。好像上面這段話沒啥用，連成為談資都不行。我也是直接百度第一篇復制一段當做介紹，習慣使然。(手動狗頭)

在python中文件監控主要有兩個庫，一個是pyinotify ( https://github.com/seb-m/pyinotify/wiki )，一個是watchdog(http://pythonhosted.org/watchdog/)。pyinotify依賴于Linux平臺的inotify，后者則對不同平臺的的事件都進行了封裝。

watchdog使用

在python中可以直接通過pip安裝：

pip install watchdog -i https://pypi.tuna.tsinghua.edu.cn/simple

watchdog主要采用觀察者模型。主要有三個角色：observer，event_handler，被監控的文件夾。三者原本是獨立的，主要通過observer.schedule函數將三者串起來。

事件類(event)：

watchdog.events.FileSystemEvent(event_type,

src_path,

is_directory=False)

event_type為事件類型，為moved、deleted、created或modified的其中之一

src_path為觸發該事件的文件或目錄路徑

is_directory為該事件是否由一個目錄觸發

watchdog能實現在不同平臺下都能兼容，并監控相關事件，但是如果在Windows下，是有很多問題的，具體的會在后面提出，那懂了事件類，我們就可以看看事件處理方法：

那現在有了處

def on_created(event):

print(f"hey, {event.src_path} has been created!")

def on_deleted(event):

print(f"Someone deleted {event.src_path}!")

def on_modified(event):

print(f"hey buddy, {event.src_path} has been modified")

def on_moved(event):

print(f"ok ok ok, someone moved {event.src_path} to {event.dest_path}")

理事件的函數，就需要在主程序里創建一個監聽程序了：

path = "."

go_recursively = True

my_observer = Observer()

my_observer.schedule(my_event_handler, path, recursive=True)

observer.schedule(event_handler, path, recursive=False)相當于實例化監聽對象，監控指定路徑path，該路徑觸發任何事件都會調用event_handler來處理，如果path是目錄，則recursive=True則會遞歸監控該目錄的所有變化。每一次調用schedule()對一個路徑進行監控處理就叫做一個watch，schedule()方法會返回這個watch，接著可以對這個watch做其他操作，如為該watch增加多個event處理器等。

那了解到這里，就可以寫一個demo程序進行測試了：

from watchdog.observers import Observer

from watchdog.events import *

import time

class FileEventHandler(FileSystemEventHandler):

def __init__(self):

FileSystemEventHandler.__init__(self)

def on_moved(self, event):

if event.is_directory:

print("directory moved from {0} to {1}".format(event.src_path,event.dest_path))

else:

print("file moved from {0} to {1}".format(event.src_path,event.dest_path))

def on_created(self, event):

if event.is_directory:

print("directory created:{0}".format(event.src_path))

else:

print("file created:{0}".format(event.src_path))

def on_deleted(self, event):

if event.is_directory:

print("directory deleted:{0}".format(event.src_path))

else:

print("file deleted:{0}".format(event.src_path))

def on_modified(self, event):

if event.is_directory:

print("directory modified:{0}".format(event.src_path))

else:

print("file modified:{0}".format(event.src_path))

if __name__ == "__main__":

observer = Observer()

event_handler = FileEventHandler()

observer.schedule(event_handler,r"D:\code\dingshirenwu",True)

observer.start()

try:

while True:

time.sleep(1)

except KeyboardInterrupt:

observer.stop()

observer.join()

代碼參考自python中文件變化監控-watchdog

不過這里只是監控了單個，我們可以通過循環來監控多個文件夾：

dirs = [r'D:\code\dingshirenwu', r'D:\code\tuiliu']

for dir in dirs:

event_handler = FileEventHandler()

observer.schedule(event_handler, dir, True)

observer.start()

到此為止，基本上已經知道這個模塊到底怎么用了，但當我準備在事件里加一個上傳機制的時候，發現Windows下的一些問題。Windows下watchdog并沒有權限去監控文件是否完整。即我有一個大文件，2G的視頻即使是內部百M傳輸，也需要幾十秒的時間，但watchdog只能接收到文件創建的時間就立刻進行了文件上傳，而不是同Linux并使用的inotify，似乎沒有什么好的辦法，我也只是能上傳一些比較小的如圖片等秒傳秒下的文件，下面為我的代碼：

import logging

import queue

import threading

import time

import watchdog.observers as observers

import watchdog.events as events

from ftplib import FTP

logger = logging.getLogger(__name__)

SENTINEL = None

def upload(f, remote_path, local_path):

fp = open(local_path, "rb")

buf_size = 1024

f.storbinary("STOR {}".format(remote_path), fp, buf_size)

fp.close()

class MyEventHandler(events.FileSystemEventHandler):

def on_any_event(self, event):

super(MyEventHandler, self).on_any_event(event)

queue.put(event)

def __init__(self, queue):

self.queue = queue

def process(queue):

while True:

event = queue.get()

logger.info(event)

print(event.key) # tuple

('modified', 'C:\\Users\\admin\\Desktop\\公司文件\\test\\GitHub\\isadb\\.idea', True)

if (event.key)[0] == "created":

upload(ftp, remote_path, event.src_path)

if __name__ == '__main__':

logging.basicConfig(level=logging.DEBUG,

format='[%(asctime)s %(threadName)s] %(message)s',

datefmt='%H:%M:%S')

ftp = FTP()

ftp.connect("x.x.x.x", 21) # 第一個參數可以是ftp服務器的ip或者域名，第二個參數為ftp服務器的連接端口，默認為21

ftp.login(username, password) # 匿名登錄直接使用ftp.login()

queue = queue.Queue()

num_workers = 4

pool = [threading.Thread(target=process, args=(queue,)) for i in range(num_workers)]

for t in pool:

t.daemon = True

t.start()

event_handler = MyEventHandler(queue)

observer = observers.Observer()

observer.schedule(

event_handler,

path=r'C:\Users\admin\Desktop\公司文件\test\GitHub\isadb',

recursive=True)

observer.start()

try:

while True:

time.sleep(1)

except KeyboardInterrupt:

observer.stop()

observer.join()

建立了一個工作線程池，而不是累積文件系統事件，該線程從一個公共隊列中獲取任務。上傳文件我是寫了一個類調用，但那個文件找不到了。。所以改用了函數，這里會有問題是：IOError: [Errno 13] Permission denied: u'D:\pycharm\test.mp4'

然后再Stack Overflow找到了一個解決方案：當上傳一個大文件的時候，同時上傳一個空文本，記錄這個文件的大小，然后對這個文件進行輪詢，只有當該文件的大小不再發生變化時，我們認為這個文件已經生成成功，這時再考慮上傳，不過我也就寫個demo，太麻煩了。。。如果有人有更好的方式，可以評論或者私信我。

到此這篇關于python中watchdog文件監控與檢測上傳的文章就介紹到這了,更多相關python watchdog監控文件內容請搜索腳本之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持腳本之家！

總結

以上是生活随笔為你收集整理的python watchdog 同时检测到多个事件_python中watchdog文件监控与检测上传功能的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。