引言

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

watchdog介紹

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

Linux也引入了Watchdog，在Linux內(nèi)核下，當(dāng)Watchdog啟動后，便設(shè)定了一個定時器，如果在超時時間內(nèi)沒有對/dev/Watchdog進(jìn)行寫操作，則會導(dǎo)致系統(tǒng)重啟。通過定時器實(shí)現(xiàn)的Watchdog屬于軟件層面。

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

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

watchdog使用

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

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

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

事件類(event)：

watchdog.events.FileSystemEvent(event_type,

src_path,

is_directory=False)

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

src_path為觸發(fā)該事件的文件或目錄路徑

is_directory為該事件是否由一個目錄觸發(fā)

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

那現(xiàn)在有了處

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}")

理事件的函數(shù)，就需要在主程序里創(chuàng)建一個監(jiān)聽程序了：

path = "."

go_recursively = True

my_observer = Observer()

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

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

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

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中文件變化監(jiān)控-watchdog

不過這里只是監(jiān)控了單個，我們可以通過循環(huán)來監(jiān)控多個文件夾：

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

for dir in dirs:

event_handler = FileEventHandler()

observer.schedule(event_handler, dir, True)

observer.start()

到此為止，基本上已經(jīng)知道這個模塊到底怎么用了，但當(dāng)我準(zhǔn)備在事件里加一個上傳機(jī)制的時候，發(fā)現(xiàn)Windows下的一些問題。Windows下watchdog并沒有權(quán)限去監(jiān)控文件是否完整。即我有一個大文件，2G的視頻即使是內(nèi)部百M(fèi)傳輸，也需要幾十秒的時間，但watchdog只能接收到文件創(chuàng)建的時間就立刻進(jìn)行了文件上傳，而不是同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) # 第一個參數(shù)可以是ftp服務(wù)器的ip或者域名，第二個參數(shù)為ftp服務(wù)器的連接端口，默認(rèn)為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()

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

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

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。