基于ftplib的TLS加密FTP嵌套目录断点续传上传方案咨询

实现逻辑 #

• 用显式的进度栈替代递归隐式栈，存储每个待处理目录的本地绝对路径、远程目录相对路径、已处理文件索引，异常中断时进度栈会保留最后一次正常处理的位置
• 上传文件前先查询远程服务器对应文件的大小，本地文件指针跳转到对应偏移量后，调用storbinary时传入rest参数实现断点续传
• 创建远程目录前先校验目录是否存在，避免重复创建抛出异常
• 重连后直接读取进度栈恢复到中断的目录位置，不需要重新遍历所有已处理的文件和目录

完整修改后代码 #

import ftplib
import os
import ssl
import time

class ReusedSslSocket(ssl.SSLSocket):
    def unwrap(self):
        pass

class MyFTP_TLS(ftplib.FTP_TLS):
    """Explicit FTPS, with shared TLS session"""
    def ntransfercmd(self, cmd, rest=None):
        conn, size = ftplib.FTP.ntransfercmd(self, cmd, rest)
        if self._prot_p:
            conn = self.context.wrap_socket(conn,
                                            server_hostname=self.host,
                                            session=self.sock.session)  # 复用TLS会话         
            conn.__class__ = ReusedSslSocket  # 传输完成不重复关闭TLS连接
        return conn, size

# 配置参数（替换为实际值）
server = "你的FTP服务器地址"
username = "用户名"
password = "密码"
root_local_path = "要上传的本地根目录绝对路径"
root_remote_path = "/"  # 上传到FTP的根路径

# 全局进度栈：每个元素为 (本地目录路径, 远程目录路径, 已处理文件索引)
upload_stack = []
# 初始化根目录任务
upload_stack.append((root_local_path, root_remote_path, 0))

def is_remote_dir_exists(ftp_session, remote_dir):
    """判断远程目录是否存在"""
    try:
        original_pwd = ftp_session.pwd()
        ftp_session.cwd(remote_dir)
        ftp_session.cwd(original_pwd)
        return True
    except ftplib.error_perm:
        return False

def get_remote_file_size(ftp_session, remote_file):
    """获取远程文件大小，不存在返回0"""
    try:
        return ftp_session.size(remote_file)
    except ftplib.error_perm:
        return 0

def upload_task():
    global upload_stack
    session = MyFTP_TLS(server, username, password, timeout=30)
    session.prot_p()
    
    while upload_stack:
        local_dir, remote_dir, processed_idx = upload_stack.pop()
        # 切换到对应远程目录
        if not is_remote_dir_exists(session, remote_dir):
            session.mkd(remote_dir)
        session.cwd(remote_dir)
        # 获取本地目录下所有文件/文件夹
        local_entries = sorted(os.listdir(local_dir))
        # 从已处理的索引位置继续遍历
        for i in range(processed_idx, len(local_entries)):
            entry = local_entries[i]
            local_entry_path = os.path.join(local_dir, entry)
            if os.path.isfile(local_entry_path):
                # 处理文件：断点续传
                remote_file_size = get_remote_file_size(session, entry)
                local_file_size = os.path.getsize(local_entry_path)
                # 已上传完成则跳过
                if remote_file_size >= local_file_size:
                    continue
                # 断点续传
                with open(local_entry_path, 'rb') as f:
                    f.seek(remote_file_size)
                    session.storbinary(f'STOR {entry}', f, rest=remote_file_size)
            else:
                # 处理文件夹：先把当前目录的进度压回栈，下次恢复直接从下一个索引开始
                upload_stack.append((local_dir, remote_dir, i+1))
                # 新目录任务压入栈，优先处理子目录
                new_remote_dir = os.path.join(remote_dir, entry).replace('\\', '/')  # FTP路径统一用正斜杠
                upload_stack.append((local_entry_path, new_remote_dir, 0))
                # 跳出当前循环，下次迭代处理子目录
                break
    # 全部处理完成退出
    session.quit()

def reset_connection():
    print("FTP连接断开，尝试重连")
    time.sleep(2)
    # 重连后直接执行上传任务，进度栈已保留中断位置
    run_upload()

def run_upload():
    try:
        upload_task()
    except (ConnectionResetError, WindowsError, OSError, ftplib.error_temp, ftplib.error_perm) as e:
        print(f"上传异常: {e}")
        reset_connection()

if __name__ == "__main__":
    run_upload()

优化说明 #

• 用迭代+栈的逻辑替代原生递归，避免深层目录递归层级溢出问题
• 路径统一处理，兼容Windows和Linux系统的路径格式差异
• 新增的目录存在判断、文件大小校验逻辑，不会重复上传已完成的内容
• 进度栈常驻内存，重连后直接恢复，不需要二次遍历已上传目录，适合大数量级文件上传场景

内容的提问来源于stack exchange，提问作者ionosphere