Python Boto3筛选S3中LastModified≤15天对象失败问题咨询

我们可以通过以下AWS CLI命令筛选最后修改时间≤15天前的S3对象，能得到预期结果：

lastModifiedMax=`date -d "15 days ago" '+%Y-%m-%d'`
aws s3api list-objects --bucket $bucket  --query "Contents[?LastModified<='${lastModifiedMax}'][].{object: Key}"

但改用Python Boto3实现时，修改代码后未得到预期结果，返回了所有S3对象。修改前后的代码片段如下：

原参考代码片段 #

while True:
    # The S3 API response is a large blob of metadata.
    # 'Contents' contains information about the listed objects.
    resp = s3.list_objects_v2(**kwargs)
    for content in resp.get('Contents', []):
        last_modified_date = content['LastModified']
        if (
            content['Key'].endswith(suffixes) and
            last_modified_rule(last_modified_min, last_modified_date, last_modified_max)
        ):
            yield content

修改后的错误代码片段 #

end_date = date.today() - timedelta(days=15)

for prefix in prefixes:
    kwargs['Prefix'] = prefix
    while True:
        # The S3 API response is a large blob of metadata.
        # 'Contents' contains information about the listed objects.
        resp = s3.list_objects_v2(**kwargs)
        for content in resp.get('Contents', ['?LastModified<='+str(end_date)]):
                yield content

问题原因分析 #

1. 错误使用resp.get()的默认值：你把resp.get('Contents', [])的默认值改成了['?LastModified<='+str(end_date)]，这个默认值仅当API返回结果中没有Contents字段时才会生效，正常情况下API会返回所有对象的Contents列表，这行代码完全没有起到筛选作用。

2. 丢失了核心过滤逻辑：原参考代码通过if条件判断（调用last_modified_rule）来筛选符合时间要求的对象，但你修改后直接yield所有content，没有任何过滤，自然会返回全部对象。

3. 日期类型不匹配：date.today()返回的是不带时区的date对象，而S3返回的content['LastModified']是带UTC时区的datetime对象，直接比较会导致逻辑错误，甚至可能出现所有对象都被判定为符合条件的情况。

正确的代码修改方案 #

from datetime import datetime, timedelta, timezone

# 生成15天前的UTC时区时间（匹配S3的LastModified时区）
end_date = datetime.now(timezone.utc) - timedelta(days=15)

for prefix in prefixes:
    kwargs['Prefix'] = prefix
    while True:
        resp = s3.list_objects_v2(**kwargs)
        # 遍历所有对象，添加时间过滤条件
        for content in resp.get('Contents', []):
            last_modified = content['LastModified']
            # 比较带时区的时间对象
            if last_modified <= end_date:
                yield content
        # 处理S3分页，获取下一页数据
        continuation_token = resp.get('NextContinuationToken')
        if not continuation_token:
            break
        kwargs['ContinuationToken'] = continuation_token

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