python中watchdog文件监控与检测上传功能

脚本专栏 2024/12/23 佚名

3 2 1

引言

上一篇介绍完了观察者模式的原理，本篇想就此再介绍一个小应用，虽然我也就玩了一下午，是当时看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,检测上传