腾讯云COS上传

2024-07-07 | 阅读：次

SDK使用

 pip install -U cos-python-sdk-v5
 #带 -U 或 --upgrade：当使用这个选项时，pip 会**尝试升级指定的包到最新版本**。如果你已经安装了这个包，pip 将检查是否有更高版本的包可用，并将其升级到最新版本。同时，它也会升级所有依赖于该包的其他包，以保证依赖关系的兼容性。

例如，pip install -U cos-python-sdk-v5 将安装或更新到 cos-python-sdk-v5 包的最新版本。

通过 COS 默认域名访问时，SDK 会以 {bucket-appid}.cos.{region}.myqcloud.com 的域名形式访问 COS。

# -*- coding=utf-8
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
import sys
import os
import logging

# 正常情况日志级别使用 INFO，需要定位时可以修改为 DEBUG，此时 SDK 会打印和服务端的通信信息
logging.basicConfig(level=logging.INFO, stream=sys.stdout)

# 1. 设置用户属性, 包括 secret_id, secret_key, region等。Appid 已在 CosConfig 中移除，请在参数 Bucket 中带上 Appid。Bucket 由 BucketName-Appid 组成
secret_id = os.environ['COS_SECRET_ID']     
secret_key = os.environ['COS_SECRET_KEY']   
region = 'ap-beijing'      # 替换为用户的 region，已创建桶归属的 region 可以在控制台查看，https://console.cloud.tencent.com/cos5/bucket
                           # COS 支持的所有 region 列表参见 https://cloud.tencent.com/document/product/436/6224
token = None               # 如果使用永久密钥不需要填入 token，如果使用临时密钥需要填入，临时密钥生成和使用指引参见 https://cloud.tencent.com/document/product/436/14048
scheme = 'https'           # 指定使用 http/https 协议来访问 COS，默认为 https，可不填

config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token, Scheme=scheme)
client = CosS3Client(config)

创建桶：

response = client.create_bucket(
    Bucket='examplebucket-1250000000'
)

上传文件对象： 文件流上传方法

with open('picture.jpg', 'rb') as fp:
#'rb' 表示以二进制形式读取（这对于非文本文件如图片、视频等是必须的）。
#使用 with 语句的好处是文件会在代码块执行完毕后自动关闭，即使在读取文件时发生异常也是如此。

    response = client.put_object(
        Bucket='examplebucket-1250000000',
        Body=fp,
        Key='picture.jpg',
        StorageClass='STANDARD', #StorageClass 指定了对象的存储级别，这里使用 STANDARD，即标准存储
        EnableMD5=False #不启用MD5
    )
print(response['ETag'])

字节流

response = client.put_object(
    Bucket='examplebucket-1250000000',
    Body=b'bytes',
    Key='picture.jpg',
    EnableMD5=False
)
print(response['ETag'])

在使用云对象存储（COS）服务上传数据时，选择使用字节流（如 b’bytes’）还是文件流（从文件对象读取），主要取决于具体的使用场景和需求。上传字节流当你直接提供一个字节字符串（如 b’bytes’）给上传函数时，这通常适用于以下场景： 数据量较小：如果你需要上传的数据量不大，如一些配置数据、小型图像或短文本，直接使用字节流是简单且有效的。 数据已在内存中：如果数据已经以字节形式在内存中，直接上传这些数据可以避免写入和再读取文件的额外开销。 生成数据：对于动态生成的数据，如临时生成的图像或文件，可能没有必要首先写入磁盘，直接从内存上传更为高效。上传文件流从文件对象中读取数据上传，适用于以下情况：大文件处理：对于大文件，使用文件流可以有效地分批读取和上传数据，而不必将整个文件加载到内存中，这可以大幅节省内存资源。磁盘到云的直接传输：如果文件已经存在于磁盘上，使用文件流可以直接读取并上传，无需额外处理。 数据持久化：如果数据需要被保存或者已经以文件形式存在，使用文件流进行上传可以确保数据的完整性和持久化存储。 哪个方法更好？性能与资源考量：对于大文件或需要频繁读写的情况，使用文件流可以更有效地管理内存和带宽，因为它允许逐块处理文件。对于小数据片段或需要快速处理的数据，直接使用字节流可以减少文件I/O操作，提高效率。具体应用需求：考虑应用的具体需求，比如是否需要处理文件元数据，文件是否需要存储在本地等因素。

图片这类数据，推荐使用文件流来进行上传，尤其是当处理的图片文件较大或者数量较多时。 以下是使用文件流上传图片的几个主要优点：内存效率：使用文件流上传图片可以逐块读取数据，不需要一次性将整个图片加载到内存中。这对于大型图片或高分辨率图片尤其重要，因为这些图片完全加载到内存中可能消耗大量资源。数据完整性：通过文件流，可以确保从源文件直接读取数据，减少了数据在处理过程中可能出现的损坏或变更的风险。处理大量文件：如果你需要上传大量图片，使用文件流可以帮助你有效管理系统资源，防止因大量数据同时处理而导致的内存溢出或性能问题。简化工作流：如果图片已经以文件形式存储在磁盘上，直接使用文件流进行上传可以简化处理流程，无需额外的步骤将图片转换成字节流。

其他传输：（块传输与高级分块传输）

#### chunk 简单上传
import requests
stream = requests.get('https://cloud.tencent.com/document/product/436/7778')


# 网络流将以 Transfer-Encoding:chunked 的方式传输到 COS
response = client.put_object(
    Bucket='examplebucket-1250000000',
    Body=stream,
    Key='picture.jpg'
)
print(response['ETag'])


#### 高级上传接口（推荐）
# 根据文件大小自动选择简单上传或分块上传，分块上传具备断点续传功能。
response = client.upload_file(
    Bucket='examplebucket-1250000000',
    LocalFilePath='local.txt',
    Key='picture.jpg',
    PartSize=1,
    MAXThread=10,
    EnableMD5=False
)
print(response['ETag'])

异步上传文件

import os
import uuid

from aiobotocore.session import AioSession
from qcloud_cos import CosConfig, CosS3Client

from config_txt import (
    TENCENT_COS_BUCKET,
    TENCENT_COS_BUCKET_NAME,
    TENCENT_COS_REGION,
    TENCENT_COS_SECRET_ID,
    TENCENT_COS_SECRET_KEY,
)

config = CosConfig(Region=TENCENT_COS_REGION, SecretId=TENCENT_COS_SECRET_ID, SecretKey=TENCENT_COS_SECRET_KEY)
client = CosS3Client(config)


def generate_unique_filename(original_path):
    """生成包含UUID的唯一文件名，保留原始扩展名"""
    basename = os.path.basename(original_path)
    extension = basename.split('.')[-1]
    uuid_name = f"{uuid.uuid4()}.{extension}"
    return uuid_name


async def upload_image(local_path):
    """异步上传图片到腾讯云，并返回图片URL"""
    session = AioSession()
    endpoint_url = f'https://{TENCENT_COS_BUCKET}.cos.{TENCENT_COS_REGION}.myqcloud.com'
    async with session.create_client('s3', region_name=TENCENT_COS_REGION, aws_secret_access_key=TENCENT_COS_SECRET_KEY,
                                     aws_access_key_id=TENCENT_COS_SECRET_ID, endpoint_url=endpoint_url) as client:
        key = generate_unique_filename(local_path)
        with open(local_path, 'rb') as file_data:
            response = await client.put_object(Bucket=TENCENT_COS_BUCKET_NAME, Key=key, Body=file_data)
        if 'ETag' in response:
            url = f"{endpoint_url}/{TENCENT_COS_BUCKET_NAME}/{key}"
            return url
        else:
            return "Upload failed."
            
async def upload_image_byte(data_stream, filename):
    """异步上传图片到腾讯云，并返回图片URL，使用数据流"""
    session = AioSession()
    endpoint_url = f'https://{TENCENT_COS_BUCKET}.cos.{TENCENT_COS_REGION}.myqcloud.com'
    key = generate_unique_filename(filename)
    async with session.create_client('s3', region_name=TENCENT_COS_REGION, aws_secret_access_key=TENCENT_COS_SECRET_KEY,
                                     aws_access_key_id=TENCENT_COS_SECRET_ID, endpoint_url=endpoint_url) as client:
        response = await client.put_object(Bucket=TENCENT_COS_BUCKET_NAME, Key=key, Body=data_stream)
        if 'ETag' in response:
            url = f"{endpoint_url}/{TENCENT_COS_BUCKET_NAME}/{key}"
            return url
        else:
            return None