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 paramkioparamkio 可以完成以下操作： 判断目录存在与否；删除目录；新建目录；上传、下载文件；paramkio实现了ssh，用来操作文件（上传、删除）;
（3）源码
监控本地文件夹，并上传其变更项（新增、修改及删除均进行同步）
有一个问题需要注意：当有新增时，会先检测到新增、然后检测到更新，如果文件比较大，会持续监控到一个文件的多个更新；操作时会报错，可以采用sleep(10)秒在操作；避免持续报错；
# argparse 命令行参数构建和解析# watchdog 监控文件的变化# sftp 上传文件到服务器import argparseimport osimport timefrom stat import S_ISDIRimport paramikofrom 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_timedef 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_timedef 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()
