python中watchdog文件监控与检测上传功能
引言
上一篇介绍完了观察者模式的原理，本篇想就此再介绍一个小应用，虽然我也就玩了一下午，是当时看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 多线程中join()的作用