Python监控目录文件夹，并使用SFTP上传目录及文件到linux服务器

Python 扫描监控本地文件夹并进行超大文件上传

方案1：WebUploader大文件分块多线程并发上传
方案2：watchdog目录文件夹监控，paramiko STFP上传服务器
方案3：优化2，压缩后上传，服务器解压
- （1）监控本地文件夹
- （2）paramiko利用sftp连接远程服务器，并上传目录或者文件；
- （3）源码
参考

Python监控本地目录文件夹，并分块等方式上传文件到服务器

有几种思路：大文件分块上传，或者先压缩后上传在解压；

目录监控方案： Java WatchEvent（可以监控到目录/文件的新建、修改、移动、删除）
Python watchdog（同Java WatchEvent）
大文件上传方案：WebUploader（h5+js）大文件分块，上传后在合并
压缩后上传，在解压
可以使用FTP、SFTP等连接服务器上传

方案1：WebUploader大文件分块多线程并发上传

WebUploader 是由Baidu WebFE(FEX)团队开发的一个简单的以HTML5为主，FLASH为辅的现代文件上传组件。

集成了文件分块，异步多线程上传；

原理：可设置是否分块、分块大小、是否多线程并发上传；
H5页面触发，js里可设置是否分块、分块大小、线程并发数；上传文件时，同一个文件会带有一个唯一的taskId，及当前分块数；上传结束后根据taskId调用回调函数合并文件；

分片与并发结合，将一个大文件分割成多块，并发上传，极大地提高大文件的上传速度。当网络问题导致传输错误时，只需要重传出错分片，而不是整个文件。另外分片传输能够更加实时的跟踪上传进度。

优点：支持分块并发上传，速度快;
缺点：需要h5页面触发（不适用于自动化）

方案2：watchdog目录文件夹监控，paramiko STFP上传服务器

watchdog 目录文件夹监控可以监控到目录/文件的新建、修改、移动、删除；如下图；

监控到目录文件新增或者修改，可能有一个问题；就是某个文件很大，需要1分钟完全复制过来，但在第2秒已经监测到其新增，监测到其更新，一直持续到复制完；
可以通过捕获报错并sleep(10)在进行上传操作解决；

paramiko SFTP连接服务器直接上传文件

方案3：优化2，压缩后上传，服务器解压

watchdog目录文件夹监控
zip、unzip压缩上传后解压
paramiko SFTP连接服务器上传文件
该方案对方案2进行优化，增加了zip压缩文件，到服务器端在解压；

（1）监控本地文件夹

可以监听到文件创建、更新、删除、重命名；

Python下用 win32file 或者 watchdog
Java下用 WatchEvent

无论是win32file、watchdog，监控到文件有更新时，大文件很可能还在更新状态，同时操作这个文件会报error；

如下图，对于大文件，当第一次监控到文件是新增或者修改，文件本身还未操作结束，检测到有变更立即处理就会有问题；

（2）paramiko利用sftp连接远程服务器，并上传目录或者文件；

运用python：paramkio模块；
安装：pin install paramkio

paramkio实现了ssh，用来操作文件（上传、删除）;

paramkio 可以完成以下操作：

判断目录存在与否；
删除目录；
新建目录；
上传、下载文件；

（3）源码

监控本地文件夹，并上传其变更项（新增、修改及删除均进行同步）

有一个问题需要注意：当有新增时，会先检测到新增、然后检测到更新，如果文件比较大，会持续监控到一个文件的多个更新；操作时会报错，可以采用sleep(10)秒在操作；避免持续报错；

# argparse 命令行参数构建和解析
# watchdog 监控文件的变化
# sftp 上传文件到服务器
import argparse
import os
import time
from stat import S_ISDIRimport paramiko
from watchdog.events import *
from watchdog.observers import Observerimport log_exec_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))rm(event.src_path)else:print("file deleted:{0}".format(event.src_path))rm(event.src_path)def on_modified(self, event):if event.is_directory:print("directory modified:{0}".format(event.src_path))# 只处理修改、删除的事情upload(event.src_path, remote_dir)else:print("file modified:{0}".format(event.src_path))uploadFile(event.src_path)# hostname = 'xxx.xxx.xxx.xxx'
# username = 'root'
# password = 'ffff'
# port = 22
# local_dir = 'D:/gifCreate/'
# remote_dir = '/data/home/mdir/'ap = argparse.ArgumentParser()
ap.add_argument("-host", "--hostname", required=False, default='xxx.xxx.xxx.xxx',help="the server host")
ap.add_argument("-username", "--username", required=False, default='root',help="ther server username")
ap.add_argument("-pwd", "--password", required=False, default='fffff',help="ther server password")
ap.add_argument("-port", "--port", required=False, default=22,help="ther server port")
ap.add_argument("-localdir", "--localdir", required=False, default='D:/gifCreate/',help="path of the localdir")
ap.add_argument("-remotedir", "--remotedir", required=False, default='/data/home/mdir/',help="ther server remotedir")
args = vars(ap.parse_args())hostname = args['hostname']
username = args['username']
password = args['password']
port = args['port']
local_dir = args['localdir']
remote_dir = args['remotedir']def isdir(path):try:t = paramiko.Transport((hostname, port))t.connect(username=username, password=password)sftp = paramiko.SFTPClient.from_transport(t)return S_ISDIR(sftp.stat(path).st_mode)except IOError as e:return Falsedef rm(path):try:t = paramiko.Transport((hostname, port))t.connect(username=username, password=password)sftp = paramiko.SFTPClient.from_transport(t)path = path.replace('\\', '/').replace(local_dir, remote_dir).replace('\\', '/')if not isdir(path):sftp.remove(path)print('\t%s deleted.' % path)returnprint(path)files = sftp.listdir(path=path)for f in files:filepath = os.path.join(path, f)if isdir(filepath):rm(filepath)else:sftp.remove(filepath)print('\t%s deleted.' % path)sftp.rmdir(path)print('\t%s dir deleted.' % path)except Exception as e:print(e)@log_exec_time.log_execution_time
def upload(local_dir, remote_dir):try:t = paramiko.Transport((hostname, port))t.connect(username=username, password=password)sftp = paramiko.SFTPClient.from_transport(t)for root, dirs, files in os.walk(local_dir):for name in dirs:# print(name)# 过滤掉图片文件夹if (str(name) == 'Img'):continuelocal_path = os.path.join(root, name)a = local_path.replace(local_dir, '').replace('\\', '/').replace('\\', '/')remote_path = remote_dir + a# print('dir: %s,dirname: %s' % (a, name))try:sftp.mkdir(remote_path)print("mkdir path %s" % remote_path)except Exception as e:print("dir path is exists: %s" % remote_path)for filepath in files:# 过滤掉图片if (filepath.endswith('.jpg') or filepath.endswith('.JPG') or filepath.endswith('.JPEG')):continuelocal_file = os.path.join(root, filepath)a = local_file.replace(local_dir, '').replace('\\', '/').replace('\\', '/')remote_file = remote_dir + a# print('\tlocal_file: %s,  filepath: %s' % (local_file, filepath))try:sftp.put(local_file, remote_file)except Exception as e:sftp.mkdir(os.path.split(remote_file)[0])sftp.put(local_file, remote_file)# print("upload %s to remote %s" % (local_file, remote_file))# print('upload file success %s ' % datetime.datetime.now())t.close()except Exception as e:print(e)@log_exec_time.log_execution_time
def uploadFile(filepath):remote_file = filepath.replace('\\', '/').replace(local_dir, remote_dir).replace('\\', '/')# print(remote_file)try:if not os.path.exists(filepath):print('error: %s not exists' % filepath)returnt = paramiko.Transport((hostname, port))t.connect(username=username, password=password)sftp = paramiko.SFTPClient.from_transport(t)try:# print(filepath)sftp.put(filepath, remote_file)except Exception as e:sftp.mkdir(os.path.split(remote_file)[0])sftp.put(filepath, remote_file)# print(e)if (e == '[Errno 2] No such file'):print('error: %s' % e)# print(e)print('uploadFileSuccess: %s' % filepath)t.close()except Exception as e:time.sleep(10)uploadFile(filepath)# print('e1: %s' % e)if __name__ == '__main__':# 1.88G  630s  10 mins 30 s# 18G  2.1hprint(args)print("Paramaters:")print("hostname: " + hostname)print("username: " + username)print("password: " + password)print("port: " + str(port))print("local_dir: " + local_dir)print("remote_dir: " + remote_dir)# 上传原始目录upload(local_dir, remote_dir)# 监控目录，有文件变化时及时处理observer = Observer()event_handler = FileEventHandler()observer.schedule(event_handler, local_dir, True)observer.start()try:while True:time.sleep(1)except KeyboardInterrupt:observer.stop()observer.join()

参考

Python 监控本地文件夹
Java 监控本地文件夹
Java 监控某个目录下的文件
Python paramkio 利用ssh连接远程服务器，并上传目录或者文件；
Python 使用SFTP删除远程目录
Python WebUploader超大文件并发分块上传