utils/operate_tx.py

import yaml
from datetime import datetime
from qcloud_cos import CosConfig, CosS3Client


# 配置腾讯云客户端
def config_tx(yaml_path='./config/txcloud/tx.yaml'):
    # 读取腾讯云的配置
    f = open(yaml_path, 'r', encoding='utf-8')
    res = yaml.load(f, Loader=yaml.FullLoader)

    # 配置腾讯云的访问信息
    secret_id = res['SecretId']
    secret_key = res['SecretKey']
    region = res['region']

    # 创建COS配置对象
    config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key)

    # 创建COS客户端对象
    client = CosS3Client(config)

    return client


# 上传PNG图片到腾讯云
def save_tx(file_data,
            file_name='1398635107044175872.png',
            bucket_name='fashion-person-1254150807'):
    # 创建腾讯云客户端
    client = config_tx()

    # 发送请求上传数据
    response = client.put_object(
        Bucket=bucket_name,
        Body=file_data,  # 二进制数据
        Key=file_name
    )


# 从腾讯云取出图片URL
def generate_tx_presigned_url(img_name='1398635107044175872.png',
                              bucket_name='fashion-person-1254150807'):
    # 创建腾讯云客户端
    client = config_tx()

    # 生成预签名URL
    try:
        img_url = client.get_presigned_url(
            Method='GET',
            Bucket=bucket_name,
            Key=img_name,
            Expired=315360000
        )
        return img_url
    except Exception as e:
        print("生成预签名URL失败：", e)
        return None


# 从腾讯云下载图片
def download_tx(file_name='1398635107044175872.png',
                local_file_path='../data/temp/temp.png',
                bucket_name='fashion-person-1254150807'):
    # 创建腾讯云客户端
    client = config_tx()

    # 从指定桶下载指定图片
    try:
        response = client.get_object(
            Bucket=bucket_name,
            Key=file_name
        )

        with open(local_file_path, 'wb') as file:
            file.write(response['Body'].get_raw_stream().read())

        print(f"成功下载文件到 {local_file_path}")
    except Exception as e:
        print(f"下载图片失败: {e}")


# 检查腾讯云存储桶中是否存在指定名称的图片（忽略文件扩展名）
def check_image_name_in_tx(image_name,
                           bucket_name='fashion-guangzhou-dataset'):
    # 创建腾讯云客户端
    client = config_tx()

    # 获取存储桶中的所有对象
    response = client.list_objects(Bucket=bucket_name)

    # 检查是否有与给定图片名称相同的结果
    for content in response.get('Contents', []):
        if content['Key'] == image_name:
            return True

    return False


if __name__ == '__main__':
    # save_tx(png_to_binary())
    # print(generate_tx_presigned_url())
    # download_tx()
    print(datetime.now())
    print(datetime.now())