utils/operate_s3.py

import boto3
import yaml


# 上传PNG图片到S3
def save_S3(file_name,
            s3_name='fashion-imgs'):
    # 设置暂缓存图片地址
    file_path = '../data/temp/temp.png'

    # 创建S3资源对象
    s3 = config_S3()

    # 获取桶对象
    bucket = s3.Bucket(s3_name)

    # 设置元数据格式（避免获取URL时变成下载）
    metadata = {'Content-Type': 'image/png'}

    # 上传文件
    bucket.upload_file(file_path, file_name, ExtraArgs={'ContentType': 'image/png'})


# 从S3取出图片URL
def take_url_S3(img_name='person1.png',
                s3_name='fashion-imgs'):
    # 创建S3资源对象
    s3 = config_S3()

    # 获取对象的URL
    img_url = s3.meta.client.generate_presigned_url(
        'get_object',
        Params={
            'Bucket': s3_name,
            'Key': img_name
        }
    )

    return img_url


# 从S3下载图片
def take_img_S3(s3_file_key='20240722-191721-5.jpeg',
                local_file_path='../data/temp/temp.jpeg',
                bucket_name='fashion-guangzhou-dataset'):
    # 创建S3资源对象
    s3 = config_S3()

    # 从指定桶下载指定图片
    s3.meta.client.download_file(bucket_name, s3_file_key, local_file_path)


# 配置S3
def config_S3(yaml_path='../config/S3/S3.yaml'):
    # 读取S3的配置
    f = open(yaml_path, 'r', encoding='utf-8')
    res = yaml.load(f, Loader=yaml.FullLoader)

    # 配置S3的访问信息
    access_key = res['access_key']
    secret_key = res['secret_key']

    # 创建S3资源对象
    s3 = boto3.resource('s3', aws_access_key_id=access_key, aws_secret_access_key=secret_key)

    return s3


# 检查S3存储桶中是否存在指定名称的图片（忽略文件扩展名）
def check_image_name_in_s3(image_name,
                           bucket_name='fashion-guangzhou-dataset'):
    # 创建S3资源对象
    s3 = config_S3()

    # 获取桶对象
    bucket = s3.Bucket(bucket_name)

    # 列出Bucket中的所有对象
    objects = bucket.objects.all()

    # 检查每个对象的键（key），看是否与image_name匹配（忽略后缀）
    for obj in objects:
        # 去除键中的文件扩展名后比较
        if obj.key.split('.')[:-1] == image_name.split('.'):
            return True

    return False