如何使用pysftp/Paramiko读取远程SFTP服务器文件内容与属性

使用pysftp跨节点拉取文件做比对时，调用mysftp.listdir()仅返回纯文件名列表，无法获取文件大小、权限、内容等信息，无法支撑后续元数据校验、跨节点文件内容比对需求。原有测试代码还存在异常捕获变量缺失、内置类型错误赋值、SFTP连接未释放、连接地址写死不生效等问题。

• 替换listdir()为listdir_attr()方法：该方法直接返回SFTPAttributes对象列表，每个对象自带文件大小、修改时间、权限、文件名等完整元数据，不需要额外单独调用查询接口
• 远程文件内容比对无需下载全量文件到本地：通过SFTP文件句柄分块读取文件内容计算哈希值，固定内存占用即可完成内容一致性校验，避免大文件占满本地磁盘
• 修复原有代码逻辑bug：补全异常捕获的异常对象、修正列表赋值错误、增加连接释放逻辑、修正连接地址写死问题，避免连接泄漏和逻辑错误

修正后的可运行代码如下：

import pysftp
import hashlib
from multiprocessing import Pool

# 分块计算远程文件MD5，默认1MB块大小，内存占用固定
def get_remote_file_md5(sftp_conn, remote_path, chunk_size=1024*1024):
    md5_obj = hashlib.md5()
    with sftp_conn.open(remote_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            md5_obj.update(chunk)
    return md5_obj.hexdigest()

def readFileNode(host):
    file_info_map = {}
    mysftp = None
    try:
        cnopts = pysftp.CnOpts()
        cnopts.hostkeys = None
        mysftp = pysftp.Connection(
            host=host,
            username="ubuntu",
            private_key="/Users/test.pem",
            cnopts=cnopts,
        )
        work_dir = "/home/ubuntu/demo"
        mysftp.cwd(work_dir)
        # 获取带完整属性的文件列表
        attr_list = mysftp.listdir_attr()
        for attr in attr_list:
            # 跳过目录、软链等非普通文件
            if not attr.st_mode or not pysftp.st_regular(attr.st_mode):
                continue
            file_full_path = f"{work_dir}/{attr.filename}"
            file_info_map[attr.filename] = {
                "filename": attr.filename,
                "size": attr.st_size,
                "mtime": attr.st_mtime,
                "mode": attr.st_mode,
                "md5": get_remote_file_md5(mysftp, file_full_path)
            }
    except Exception as e:
        print(f"连接节点{host}出错: {e}")
    finally:
        if mysftp:
            mysftp.close()
    return file_info_map

if __name__ == '__main__':
    p = Pool(2)
    node_hosts = ['3.93.XX.XX', '3.93.XX.XX']
    result = p.map(readFileNode, node_hosts)
    node1_files = result[0]
    node2_files = result[1]

    # 输出两边共有的文件比对结果
    common_files = set(node1_files.keys()) & set(node2_files.keys())
    for filename in common_files:
        f1 = node1_files[filename]
        f2 = node2_files[filename]
        print(f"===== 比对文件: {filename} =====")
        print(f"节点1: 大小{f1['size']}B, 修改时间戳{f1['mtime']}, MD5值{f1['md5']}")
        print(f"节点2: 大小{f2['size']}B, 修改时间戳{f2['mtime']}, MD5值{f2['md5']}")
        # 先比大小快速筛除不一致文件，大小一致再比MD5
        if f1['size'] != f2['size']:
            print("*比对结果：文件大小不一致，内容不同*")
        elif f1['md5'] != f2['md5']:
            print("*比对结果：文件大小一致但哈希值不同，内容不同*")
        else:
            print("*比对结果：文件元数据、内容完全一致*")
    
    # 输出仅在单节点存在的文件
    only_node1 = set(node1_files.keys()) - set(node2_files.keys())
    only_node2 = set(node2_files.keys()) - set(node1_files.keys())
    if only_node1:
        print(f"仅在节点1存在的文件: {list(only_node1)}")
    if only_node2:
        print(f"仅在节点2存在的文件: {list(only_node2)}")

• 比对逻辑可以先校验文件大小，大小不一致直接判定内容不同，大小一致再计算哈希校验，能减少不必要的文件读取传输，提升比对速度
• 如果对一致性要求不是极高，也可以用st_mtime（修改时间）+文件大小做快速校验，速度比计算哈希快很多
• listdir_attr()返回的属性对象还支持st_uid、st_gid等属性，需要校验文件属主、用户组的话可以直接取用

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