Python中识别S3中新上传文件的最简方法（存在旧文件覆盖场景）

这问题我做FTP到S3同步的时候也踩过坑，之前傻乎乎全量下载覆盖，既费带宽又慢。结合你的场景，给你两个最省心的方案，按最简程度排序：

方案1：直接对比S3已存文件列表（零额外存储，首选） #

既然你是把文件上传到S3的同一路径，那最直接的办法就是先拉取S3桶里已有的所有文件名，再和FTP上的文件列表做差集，只处理S3里没有的文件。完全不需要额外存任何记录，利用S3本身的对象列表就行。

代码示例（用boto3实现） #

import boto3
from ftplib import FTP

# 初始化S3客户端（确保本地配置了AWS凭证，或者通过环境变量传入）
s3 = boto3.client('s3')
BUCKET_NAME = 'your-target-bucket'

def get_existing_s3_files():
    """获取S3桶中已有的所有文件Key"""
    existing_keys = set()
    # 处理S3分页（如果文件多的话必须用分页）
    paginator = s3.get_paginator('list_objects_v2')
    for page in paginator.paginate(Bucket=BUCKET_NAME):
        if 'Contents' in page:
            for obj in page['Contents']:
                existing_keys.add(obj['Key'])
    return existing_keys

def get_sorted_ftp_files(ftp_host, ftp_user, ftp_pass, remote_dir):
    """从FTP获取按时间戳升序排列的文件列表"""
    ftp = FTP(ftp_host)
    ftp.login(ftp_user, ftp_pass)
    ftp.cwd(remote_dir)
    
    # 假设文件名包含时间戳（比如file_202405201430），按时间戳排序
    # 如果你的文件名本身就是按时间戳命名的，直接sorted()就行
    ftp_files = ftp.nlst()
    sorted_files = sorted(ftp_files, key=lambda x: x.split('_')[-1])
    
    ftp.quit()
    return sorted_files

# 主同步逻辑
if __name__ == "__main__":
    existing_files = get_existing_s3_files()
    ftp_files = get_sorted_ftp_files('your-ftp-host', 'ftp-user', 'ftp-pass', '/remote/files')
    
    # 只处理S3中没有的新增文件
    for file in ftp_files:
        if file not in existing_files:
            print(f"发现新增文件，开始处理：{file}")
            # 这里填入你的FTP下载代码，比如：
            # download_ftp_file(file)
            # 然后上传到S3：
            # s3.upload_file(f'/local/tmp/{file}', BUCKET_NAME, file)

方案2：用S3存储已处理文件记录（适合文件名可能重复的场景） #

如果你的FTP文件名可能存在重复（比如后续会出现同名但内容不同的文件），或者担心S3文件被误删导致重复处理，那可以在S3里存一个简单的元数据文件（比如processed_files.txt），记录所有已经处理过的文件名。每次运行前下载这个文件，对比FTP列表，处理完新文件后更新记录再上传回去。

代码示例 #

def get_processed_files():
    """从S3获取已处理的文件列表"""
    try:
        # 下载记录文件到本地临时目录
        s3.download_file(BUCKET_NAME, 'processed_files.txt', '/tmp/processed.txt')
        with open('/tmp/processed.txt', 'r') as f:
            return set(f.read().splitlines())
    except s3.exceptions.NoSuchKey:
        # 如果记录文件不存在，返回空集合
        return set()

def update_processed_files(processed_files):
    """更新S3中的已处理文件记录"""
    with open('/tmp/processed.txt', 'w') as f:
        f.write('\n'.join(processed_files))
    s3.upload_file('/tmp/processed.txt', BUCKET_NAME, 'processed_files.txt')

# 主逻辑
if __name__ == "__main__":
    processed_files = get_processed_files()
    ftp_files = get_sorted_ftp_files('your-ftp-host', 'ftp-user', 'ftp-pass', '/remote/files')
    
    new_files = [f for f in ftp_files if f not in processed_files]
    for file in new_files:
        print(f"处理新增文件：{file}")
        # 下载+上传代码...
        processed_files.add(file)
    
    # 更新已处理记录
    update_processed_files(processed_files)

额外小提示 #

如果你的FTP服务器支持MDTM命令（可以获取文件的修改时间），也可以通过对比文件修改时间来识别新增/更新的文件，但这个需要记录上次同步的时间，复杂度比前两个方案高一些。除非你的文件名里的时间戳不可靠，否则优先用前两个方案。

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