TechOverflow Logo
TechOverflow

如何使用 boto3 和 Python 按文件名中的时间戳对 S3 上的文件排序

English Deutsch 简体中文

假设我们在 S3 目录中有备份对象，如：

s3_backup_list.txt
production-backup-2022-03-29_14-40-16.xz
production-backup-2022-03-29_14-50-16.xz
production-backup-2022-03-29_15-00-03.xz
production-backup-2022-03-29_15-10-04.xz
production-backup-2022-03-29_15-20-06.xz
production-backup-2022-03-29_15-30-06.xz
production-backup-2022-03-29_15-40-00.xz
production-backup-2022-03-29_15-50-07.xz
production-backup-2022-03-29_16-00-06.xz
production-backup-2022-03-29_16-10-12.xz
production-backup-2022-03-29_16-20-18.xz
production-backup-2022-03-29_16-30-18.xz
production-backup-2022-03-29_16-40-00.xz
production-backup-2022-03-29_16-50-09.xz
production-backup-2022-03-29_17-00-18.xz
production-backup-2022-03-29_17-10-13.xz
production-backup-2022-03-29_17-20-18.xz
production-backup-2022-03-29_17-30-18.xz
production-backup-2022-03-29_17-40-06.xz
production-backup-2022-03-29_17-50-21.xz
production-backup-2022-03-29_18-00-06.xz

我们想识别最新的一个。在这些情况下，你通常不能真正依赖修改时间戳，因为这些时间戳在同步旧文件或更改文件夹结构或名称时可能会改变。

因此最好的方式是依赖文件名中的时间戳作为参考点。我们这里使用的日期时间戳基于我们的文章如何在命令行上生成包含日期和时间的文件名；如果你使用不同的对象键格式，你可能需要相应地调整 date_regex。

以下示例脚本迭代特定 S3 文件夹中的所有对象，按文件名中的时间戳排序并选择最新的一个，从 S3 下载到本地文件系统。

此脚本基于我们之前的一些文章，包括：

s3_sort_by_timestamp.py
#!/usr/bin/env python3
import boto3
import re
import os.path
from collections import namedtuple
from datetime import datetime

# 创建到 Wasabi / S3 的连接
s3 = boto3.resource('s3',
    endpoint_url = 'https://minio.mydomain.com',
    aws_access_key_id = 'my-access-key',
    aws_secret_access_key = 'my-password'
)

# 获取存储桶对象
backups = s3.Bucket('mfwh-backup')

date_regex = re.compile(r"(?P<year>\d{4})-(?P<month>\d{2})-(?P<day>\d{2})_(?P<hour>\d{2})-(?P<minute>\d{2})-(?P<second>\d{2})")

DatedObject =  namedtuple("DatedObject", ["Date", "Object"])
entries = []
# 迭代存储桶中的对象
for obj in backups.objects.filter(Prefix="production/"):
    date_match = date_regex.search(obj.key)
    # 忽略其他文件（没有日期戳的），如果有的话
    if date_match is None:
        continue
    dt = datetime(year=int(date_match.group("year")), month=int(date_match.group("month")),
        day=int(date_match.group("day")), hour=int(date_match.group("hour")), minute=int(date_match.group("minute")),
        second=int(date_match.group("second")))
    entries.append(DatedObject(dt, obj))
# 按日期排序条目
entries.sort(key=lambda entry: entry.Date)

newest_date, newest_obj = entries[-1]
#print(f"Downloading {newest_obj.key} from {newest_date.isoformat()}")
filename = os.path.basename(newest_obj.key)

with open(filename, "wb") as outfile:
    backups.download_fileobj(newest_obj.key, outfile)

# 出于自动化目的打印文件名
print(filename)
Check out similar posts by category: S3