如何读取sFTP服务器上gzip压缩的zip文件并获取内部文件列表

• 未过滤ls返回的无效空行：split('\r\n')生成的列表包含空字符串，直接作为文件名拼接会导致后续命令执行失败
• 数据逻辑完全错误：远程执行zless返回的是gzip解压后的zip文件二进制内容，但代码将其作为文本拼接后直接传入gzip.open，gzip.open接收的是文件路径/类文件对象而非解压后的内容，此处逻辑完全颠倒
• 缩进错误：解压逻辑放在了ssh返回内容的逐行循环内，还没拿到完整的zip内容就触发解压，必然会解压失败
• 缺少错误捕获：ssh执行的错误输出完全没有读取，报错信息被丢弃，无法定位问题

两种方案可选，优先推荐第一种，不需要传输大量压缩包数据到本地，执行效率最高：

方案1：远程服务器直接执行命令获取结果（推荐） #

直接在远程服务器用原生命令完成解压、读取zip列表的操作，仅把最终结果返回到本地，代码如下：

import paramiko

# 假设conn是你已经初始化完成的paramiko SSH连接对象
client = "你的client参数值"
# 第一步：获取所有符合要求的gzip文件列表
_, stdout, stderr = conn.exec_command(f'ls /{client}/Data/*.ZIP.gz')
err = stderr.read().decode('utf-8')
if err:
    raise Exception(f"获取文件列表失败: {err}")
gziplist = [f.strip() for f in stdout.read().decode('utf-8').split('\n') if f.strip()]

all_zip_files = []
for gz_path in gziplist:
    # 远程执行命令：先用zcat解压gzip得到zip文件，再用zipinfo命令列出zip内的所有文件名
    _, stdout, stderr = conn.exec_command(f'zcat {gz_path} | zipinfo -1 /dev/stdin')
    err = stderr.read().decode('utf-8')
    if err:
        print(f"处理文件{gz_path}出错: {err}")
        continue
    # 收集当前zip内的文件名
    zip_files = [f.strip() for f in stdout.read().decode('utf-8').split('\n') if f.strip()]
    all_zip_files.extend(zip_files)
    # 打印当前zip的文件列表
    print(f"{gz_path} 内包含的文件：")
    for f in zip_files:
        print(f"- {f}")

# 打印全部文件的总列表
print("\n所有zip内的完整文件列表：")
for f in all_zip_files:
    print(f)

方案2：本地内存处理（适合远程没有zipinfo命令的场景） #

通过sftp将远程gzip文件读取到本地内存处理，不需要写入本地磁盘：

import paramiko
import gzip
import zipfile
import io

# 假设conn是你已经初始化完成的paramiko SSH连接对象
client = "你的client参数值"
# 打开sftp连接
sftp = conn.open_sftp()
# 列出目录下的文件
gziplist = [f for f in sftp.listdir(f'/{client}/Data/') if f.endswith('.ZIP.gz')]

all_zip_files = []
for gz_name in gziplist:
    gz_path = f'/{client}/Data/{gz_name}'
    try:
        # 读取远程gzip文件到内存
        with sftp.open(gz_path, 'rb') as remote_f:
            # 解压gzip得到zip内容
            with gzip.open(remote_f, 'rb') as gz_f:
                zip_content = gz_f.read()
        # 读取zip内的文件名
        with zipfile.ZipFile(io.BytesIO(zip_content)) as zip_f:
            zip_files = zip_f.namelist()
            all_zip_files.extend(zip_files)
            print(f"{gz_name} 内包含的文件：")
            for f in zip_files:
                print(f"- {f}")
    except Exception as e:
        print(f"处理文件{gz_name}出错: {str(e)}")

# 打印全部文件的总列表
print("\n所有zip内的完整文件列表：")
for f in all_zip_files:
    print(f)

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