Files
life-echo/docs/tencent-cos-python-sdk-guide.md
2026-03-11 11:27:12 +08:00

1118 lines
26 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 腾讯云 COS Python SDK 开发指南
> 基于腾讯云对象存储Cloud Object Storage, COSPython SDK 官方文档整理。
---
## 官方文档链接
| 章节 | 官方文档 |
| --- | --- |
| 快速入门 | https://cloud.tencent.com/document/product/436/12269 |
| Python SDK 常见问题 | https://cloud.tencent.com/document/product/436/56531 |
| 存储桶操作 | https://cloud.tencent.com/document/product/436/35150 |
| 上传对象 | https://cloud.tencent.com/document/product/436/65820 |
| 复制和移动对象 | https://cloud.tencent.com/document/product/436/65826 |
| 列出对象 | https://cloud.tencent.com/document/product/436/65823 |
| 判断对象是否存在 | https://cloud.tencent.com/document/product/436/65830 |
| 查询对象元数据 | https://cloud.tencent.com/document/product/436/65828 |
| 修改对象元数据 | https://cloud.tencent.com/document/product/436/65822 |
| 对象访问 URL | https://cloud.tencent.com/document/product/436/56882 |
| 生成预签名 URL | https://cloud.tencent.com/document/product/436/35153 |
| 恢复归档对象 | https://cloud.tencent.com/document/product/436/65827 |
| 检索对象内容 | https://cloud.tencent.com/document/product/436/65829 |
| 对象加密 | https://cloud.tencent.com/document/product/436/63744 |
| 存储桶加密 | https://cloud.tencent.com/document/product/436/63745 |
| 客户端加密 | https://cloud.tencent.com/document/product/436/65834 |
| 单链接限速 | https://cloud.tencent.com/document/product/436/65825 |
| 图片持久化处理 | https://cloud.tencent.com/document/product/436/55344 |
| 图片高级压缩 | https://cloud.tencent.com/document/product/436/120479 |
| 盲水印 | https://cloud.tencent.com/document/product/436/55346 |
---
## 目录
1. [环境配置与安装](#1-环境配置与安装)
2. [初始化 COS 客户端](#2-初始化-cos-客户端)
3. [存储桶操作](#3-存储桶操作)
4. [上传对象](#4-上传对象)
5. [下载对象](#5-下载对象)
6. [复制与移动对象](#6-复制与移动对象)
7. [列出对象](#7-列出对象)
8. [判断对象是否存在](#8-判断对象是否存在)
9. [查询对象元数据](#9-查询对象元数据)
10. [修改对象元数据](#10-修改对象元数据)
11. [对象访问 URL](#11-对象访问-url)
12. [生成预签名 URL](#12-生成预签名-url)
13. [恢复归档对象](#13-恢复归档对象)
14. [检索对象内容](#14-检索对象内容)
15. [服务端加密](#15-服务端加密)
16. [客户端加密](#16-客户端加密)
17. [单链接限速](#17-单链接限速)
18. [图片持久化处理](#18-图片持久化处理)
19. [图片高级压缩](#19-图片高级压缩)
20. [盲水印](#20-盲水印)
21. [常见问题](#21-常见问题)
---
## 1. 环境配置与安装
> 官方文档https://cloud.tencent.com/document/product/436/12269
### 环境要求
- Python 2.7 及 Python 3.4 及以上
### 安装 SDK
```bash
# 推荐: pip 安装
pip install -U cos-python-sdk-v5
# 手动安装 (从源码)
python setup.py install
# 离线安装
mkdir cos-python-sdk-packages
pip download cos-python-sdk-v5 -d cos-python-sdk-packages
tar -czvf cos-python-sdk-packages.tar.gz cos-python-sdk-packages
# 拷贝到目标机器后:
tar -xzvf cos-python-sdk-packages.tar.gz
pip install cos-python-sdk-v5 --no-index -f cos-python-sdk-packages
```
---
## 2. 初始化 COS 客户端
### 基本初始化(默认域名)
```python
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
import os
secret_id = os.environ['COS_SECRET_ID']
secret_key = os.environ['COS_SECRET_KEY']
region = 'ap-beijing' # 替换为实际 region
token = None # 永久密钥不填, 临时密钥需填入
scheme = 'https'
config = CosConfig(
Region=region,
SecretId=secret_id,
SecretKey=secret_key,
Token=token,
Scheme=scheme
)
client = CosS3Client(config)
```
> **注意:** 一个 region 只需生成一个 `CosS3Client` 实例并复用,不要每次请求都创建新实例。
### 临时密钥初始化
```python
config = CosConfig(
Region=region,
SecretId=tmp_secret_id,
SecretKey=tmp_secret_key,
Token=token,
Scheme=scheme
)
client = CosS3Client(config)
```
### 全球加速域名初始化
```python
config = CosConfig(
Region=None,
SecretId=secret_id,
SecretKey=secret_key,
Token=token,
Endpoint='cos.accelerate.myqcloud.com',
Scheme=scheme
)
client = CosS3Client(config)
```
### 自定义域名初始化
```python
config = CosConfig(
Region=None,
SecretId=secret_id,
SecretKey=secret_key,
Token=token,
Domain='user-define.example.com',
Scheme=scheme
)
client = CosS3Client(config)
```
### 代理初始化
```python
proxies = {
'http': '127.0.0.1:80',
'https': '127.0.0.1:443'
}
config = CosConfig(
Region=region,
SecretId=secret_id,
SecretKey=secret_key,
Token=token,
Proxies=proxies
)
client = CosS3Client(config)
```
---
## 3. 存储桶操作
> 官方文档https://cloud.tencent.com/document/product/436/35150
### 查询存储桶列表
```python
response = client.list_buckets()
for bucket in response['Buckets']['Bucket']:
print(bucket['Name'], bucket['Location'], bucket['CreationDate'])
```
### 创建存储桶
```python
response = client.create_bucket(
Bucket='examplebucket-1250000000',
ACL='private' # 可选: 'private', 'public-read', 'public-read-write'
)
```
### 检索存储桶
```python
response = client.head_bucket(Bucket='examplebucket-1250000000')
print(response)
```
### 删除存储桶
```python
client.delete_bucket(Bucket='examplebucket-1250000000')
```
### 判断存储桶是否存在
```python
exists = client.bucket_exists(Bucket='examplebucket-1250000000')
print(exists) # True / False
```
---
## 4. 上传对象
> 官方文档https://cloud.tencent.com/document/product/436/65820
### 高级接口(推荐,支持断点续传)
根据文件大小自动选择简单上传或分块上传,分块上传支持断点续传。
```python
response = client.upload_file(
Bucket='examplebucket-1250000000',
Key='exampleobject',
LocalFilePath='local.txt',
PartSize=1, # 分块大小,单位 MB默认 1
MAXThread=5, # 并发线程数,默认 5
EnableMD5=False
)
print(response['ETag'])
```
#### 带进度回调
```python
import sys
def upload_percentage(consumed_bytes, total_bytes):
if total_bytes:
rate = int(100 * (float(consumed_bytes) / float(total_bytes)))
print('\r{0}% '.format(rate))
sys.stdout.flush()
response = client.upload_file(
Bucket='examplebucket-1250000000',
Key='exampleobject',
LocalFilePath='local.txt',
PartSize=1,
MAXThread=5,
progress_callback=upload_percentage,
StorageClass='STANDARD',
Metadata={'x-cos-meta-key1': 'value1'}
)
```
#### 断点续传(失败重试)
```python
from qcloud_cos.cos_exception import CosClientError, CosServiceError
for i in range(10):
try:
response = client.upload_file(
Bucket='examplebucket-1250000000',
Key='exampleobject',
LocalFilePath='local.txt'
)
break
except (CosClientError, CosServiceError) as e:
print(e)
```
### 简单上传(不超过 5GB
```python
# 文件流上传
with open('test.txt', 'rb') as fp:
response = client.put_object(
Bucket='examplebucket-1250000000',
Body=fp,
Key='test.txt',
StorageClass='STANDARD',
ContentType='text/html; charset=utf-8'
)
print(response['ETag'])
# 字节流上传
response = client.put_object(
Bucket='examplebucket-1250000000',
Body=b'abcdefg',
Key='test.txt'
)
# 本地路径上传
response = client.put_object_from_local_file(
Bucket='examplebucket-1250000000',
LocalFilePath='local.txt',
Key='test.txt'
)
```
### 创建目录
```python
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='path/to/create/dir/',
Body=b''
)
```
### 批量上传(文件夹上传)
```python
from qcloud_cos.cos_threadpool import SimpleThreadPool
uploadDir = '/root/logs'
bucket = 'examplebucket-1250000000'
pool = SimpleThreadPool()
for path, dir_list, file_list in os.walk(uploadDir):
for file_name in file_list:
localFilePath = os.path.join(path, file_name)
cosObjectKey = localFilePath.strip('/')
pool.add_task(client.upload_file, bucket, cosObjectKey, localFilePath)
pool.wait_completion()
result = pool.get_result()
if not result['success_all']:
print("Not all files upload successed. you should retry")
```
### 分块上传(手动控制)
```python
# 1. 初始化分块上传
response = client.create_multipart_upload(
Bucket='examplebucket-1250000000',
Key='exampleobject',
StorageClass='STANDARD'
)
uploadId = response['UploadId']
# 2. 上传分块(最多 10000 块)
response = client.upload_part(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Body=b'b' * 1024 * 1024,
PartNumber=1,
UploadId=uploadId
)
etag = response['ETag']
# 3. 查询已上传分块
response = client.list_parts(
Bucket='examplebucket-1250000000',
Key='exampleobject',
UploadId=uploadId
)
# 4. 完成分块上传
response = client.complete_multipart_upload(
Bucket='examplebucket-1250000000',
Key='exampleobject',
UploadId=uploadId,
MultipartUpload={
'Part': [
{'ETag': 'string', 'PartNumber': 1},
{'ETag': 'string', 'PartNumber': 2},
]
}
)
# 终止分块上传
client.abort_multipart_upload(
Bucket='examplebucket-1250000000',
Key='exampleobject',
UploadId=uploadId
)
```
---
## 5. 下载对象
> 官方文档https://cloud.tencent.com/document/product/436/65820
```python
# 下载到本地文件
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
response['Body'].get_stream_to_file('output.txt')
# 获取文件流
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
fp = response['Body'].get_raw_stream()
print(fp.read(2))
# 指定下载范围
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Range='bytes=0-10'
)
# 设置响应头部
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
ResponseContentType='text/html; charset=utf-8'
)
```
---
## 6. 复制与移动对象
> 官方文档https://cloud.tencent.com/document/product/436/65826
### 高级复制接口(推荐)
小于 5GB 调用 `copy_object`,大于等于 5GB 调用分块复制。
```python
response = client.copy(
Bucket='examplebucket-1250000000',
Key='exampleobject',
CopySource={
'Bucket': 'sourcebucket-1250000000',
'Key': 'sourceobject',
'Region': 'ap-guangzhou'
}
)
```
### 移动对象(复制 + 删除)
```python
bucket = 'examplebucket-1250000000'
srcKey = 'src_object_key'
destKey = 'dest_object_key'
response = client.copy(
Bucket=bucket,
Key=destKey,
CopySource={
'Bucket': bucket,
'Key': srcKey,
'Region': 'ap-guangzhou'
}
)
client.delete_object(Bucket=bucket, Key=srcKey)
```
### 简单复制(不超过 5GB
```python
response = client.copy_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
CopySource={
'Bucket': 'sourcebucket-1250000000',
'Key': 'sourceobject',
'Region': 'ap-guangzhou'
}
)
```
---
## 7. 列出对象
> 官方文档https://cloud.tencent.com/document/product/436/65823
### 列出第一页对象
```python
response = client.list_objects(Bucket='examplebucket-1250000000')
if 'Contents' in response:
for content in response['Contents']:
print(content['Key'])
```
### 按前缀列出
```python
response = client.list_objects(
Bucket='examplebucket-1250000000',
Prefix='folder1/'
)
```
### 分页列出全部对象
```python
marker = ""
while True:
response = client.list_objects(
Bucket='examplebucket-1250000000',
Prefix='folder1/',
Marker=marker,
MaxKeys=10
)
if 'Contents' in response:
for content in response['Contents']:
print(content['Key'])
if response['IsTruncated'] == 'false':
break
marker = response["NextMarker"]
```
### 列出目录下的对象和子目录
```python
response = client.list_objects(
Bucket='examplebucket-1250000000',
Prefix='folder1/',
Delimiter='/'
)
# 文件列表
if 'Contents' in response:
for content in response['Contents']:
print(content['Key'])
# 子目录
if 'CommonPrefixes' in response:
for folder in response['CommonPrefixes']:
print(folder['Prefix'])
```
### 列出对象历史版本
```python
response = client.list_objects_versions(
Bucket='examplebucket-1250000000',
Prefix='string'
)
```
---
## 8. 判断对象是否存在
> 官方文档https://cloud.tencent.com/document/product/436/65830
```python
exists = client.object_exists(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
print(exists) # True / False
```
---
## 9. 查询对象元数据
> 官方文档https://cloud.tencent.com/document/product/436/65828
```python
response = client.head_object(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
print(response['Content-Length'])
print(response['ETag'])
print(response['Last-Modified'])
print(response.get('x-cos-meta-test'))
```
---
## 10. 修改对象元数据
> 官方文档https://cloud.tencent.com/document/product/436/65822
通过 `copy_object` 并设置 `CopyStatus='Replaced'` 实现,源和目标是同一个对象:
```python
response = client.copy_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
CopySource={
'Bucket': 'examplebucket-1250000000',
'Key': 'exampleobject',
'Region': 'ap-guangzhou'
},
CopyStatus='Replaced',
ContentType='text/plain',
Metadata={'x-cos-meta-key1': 'value1', 'x-cos-meta-key2': 'value2'}
)
```
### 修改存储类型
```python
response = client.copy_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
CopySource={
'Bucket': 'examplebucket-1250000000',
'Key': 'exampleobject',
'Region': 'ap-guangzhou'
},
CopyStatus='Replaced',
StorageClass='STANDARD_IA'
)
```
---
## 11. 对象访问 URL
> 官方文档https://cloud.tencent.com/document/product/436/56882
生成匿名下载或分发的对象访问 URL**公有读**对象可直接访问,**私有读**需要预签名 URL
```python
url = client.get_object_url(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
print(url)
```
---
## 12. 生成预签名 URL
> 官方文档https://cloud.tencent.com/document/product/436/35153
### 上传预签名 URL
```python
url = client.get_presigned_url(
Method='PUT',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Expired=120 # 秒
)
print(url)
# 使用 URL 上传
import requests
response = requests.put(url=url, data=b'123')
```
#### 限制存储类型和上传速度
```python
url = client.get_presigned_url(
Method='PUT',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Headers={
'x-cos-storage-class': 'STANDARD_IA',
'x-cos-traffic-limit': '819200'
},
Expired=300
)
```
### 下载预签名 URL
```python
url = client.get_presigned_url(
Method='GET',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Expired=120
)
# 指定下载文件名
url = client.get_presigned_url(
Method='GET',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Params={
'response-content-disposition': 'attachment; filename=example.xlsx'
},
Expired=120
)
```
### 便捷下载预签名接口
```python
url = client.get_presigned_download_url(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Expired=120
)
```
### 临时密钥预签名
```python
url = client.get_presigned_url(
Method='GET',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Params={'x-cos-security-token': 'string'},
Expired=120,
SignHost=False
)
```
### 域名不签入签名
当分发后使用者需修改请求域名时:
```python
url = client.get_presigned_url(
Method='GET',
Bucket='examplebucket-1250000000',
Key='exampleobject',
SignHost=False,
Expired=120
)
```
---
## 13. 恢复归档对象
> 官方文档https://cloud.tencent.com/document/product/436/65827
```python
response = client.restore_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
RestoreRequest={
'Days': 100,
'CASJobParameters': {
'Tier': 'Standard' # 归档: Expedited/Standard/Bulk; 深度归档: Standard/Bulk
}
}
)
```
---
## 14. 检索对象内容
> 官方文档https://cloud.tencent.com/document/product/436/65829
使用 SQL 从 CSV/JSON 对象中检索内容(对象需 UTF-8 编码)。
```python
response = client.select_object_content(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Expression='Select * from COSObject',
ExpressionType='SQL',
InputSerialization={
'CompressionType': 'NONE',
'JSON': {'Type': 'LINES'}
},
OutputSerialization={
'CSV': {'RecordDelimiter': '\n'}
}
)
event_stream = response['Payload']
result = event_stream.get_select_result()
print(result)
```
---
## 15. 服务端加密
> 官方文档对象加密https://cloud.tencent.com/document/product/436/63744
> 官方文档存储桶加密https://cloud.tencent.com/document/product/436/63745
### SSE-COS 加密COS 托管密钥)
```python
# 上传
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-cos',
Body='123',
ServerSideEncryption='AES256'
)
# 下载(自动解密,无需额外参数)
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-cos'
)
```
### SSE-KMS 加密KMS 托管密钥,必须 HTTPS
```python
import base64
from qcloud_cos.cos_comm import to_bytes
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-kms',
Body='123',
ServerSideEncryption='cos/kms',
SSEKMSKeyId='kms-key-id', # 可选,不填则用 COS 默认 CMK
SSEKMSContext=base64.standard_b64encode(to_bytes('{"test":"test"}'))
)
```
### SSE-C 加密(用户提供密钥,必须 HTTPS
```python
import base64
from qcloud_cos.cos_comm import get_md5, to_bytes
ssec_secret = '00000000000000000000000000000001' # 32 字节
ssec_key = base64.standard_b64encode(to_bytes(ssec_secret))
ssec_key_md5 = get_md5(ssec_secret)
# 上传
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-c',
Body='00000',
SSECustomerAlgorithm='AES256',
SSECustomerKey=ssec_key,
SSECustomerKeyMD5=ssec_key_md5
)
# 下载(必须携带相同密钥)
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-c',
SSECustomerAlgorithm='AES256',
SSECustomerKey=ssec_key,
SSECustomerKeyMD5=ssec_key_md5
)
```
### 存储桶加密
```python
# 设置
client.put_bucket_encryption(
Bucket='examplebucket-1250000000',
ServerSideEncryptionConfiguration={
'Rule': [{'ApplySideEncryptionConfiguration': {'SSEAlgorithm': 'AES256'}}]
}
)
# 查询
response = client.get_bucket_encryption(Bucket='examplebucket-1250000000')
# 删除
client.delete_bucket_encryption(Bucket='examplebucket-1250000000')
```
---
## 16. 客户端加密
> 官方文档https://cloud.tencent.com/document/product/436/65834
### 对称 AES256 加密
```python
from qcloud_cos.cos_encryption_client import CosEncryptionClient
from qcloud_cos.crypto import AESProvider
# 通过密钥值初始化aes_key_value 需 base64 编码后的 32 字节密钥)
aes_provider = AESProvider(aes_key='aes_key_value')
# 或通过密钥文件路径
# aes_provider = AESProvider(aes_key_path='aes_key_path')
client_for_aes = CosEncryptionClient(config, aes_provider)
# 上传
client_for_aes.put_object(
Bucket='examplebucket-1250000000',
Body=b'bytes',
Key='exampleobject'
)
# 下载
response = client_for_aes.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
# 断点续传上传partsize 必须为 16 字节整数倍)
client_for_aes.upload_file(
Bucket='examplebucket-1250000000',
LocalFilePath='local.txt',
Key='exampleobject',
PartSize=10,
MAXThread=10
)
```
### 非对称 RSA 加密
```python
from qcloud_cos.cos_encryption_client import CosEncryptionClient
from qcloud_cos.crypto import RSAProvider
rsa_key_pair = RSAProvider.get_rsa_key_pair('public_key_value', 'private_key_value')
# 或通过路径
# rsa_key_pair = RSAProvider.get_rsa_key_pair_path('public_key_path', 'private_key_path')
rsa_provider = RSAProvider(key_pair_info=rsa_key_pair)
client_for_rsa = CosEncryptionClient(config, rsa_provider)
# 上传 / 下载 / 分块上传 接口与 AES 加密客户端一致
```
---
## 17. 单链接限速
> 官方文档https://cloud.tencent.com/document/product/436/65825
限速值范围:**819200 - 838860800**bit/s即 800Kb/s - 800Mb/s。
```python
# 上传限速
with open('test.bin', 'rb') as fp:
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Body=fp,
TrafficLimit='819200'
)
# 下载限速
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
TrafficLimit='819200'
)
response['Body'].get_stream_to_file('exampleobject')
```
---
## 18. 图片持久化处理
> 官方文档https://cloud.tencent.com/document/product/436/55344
### 上传时处理
```python
import json
response, data = client.ci_put_object(
Bucket='examplebucket-1250000000',
Body=open('local.jpg', 'rb'),
Key='local.jpg',
PicOperations=json.dumps({
"is_pic_info": 1,
"rules": [{
"fileid": "format.png",
"rule": "imageView2/format/png"
}]
})
)
```
### 云上数据处理
```python
response, data = client.ci_image_process(
Bucket='examplebucket-1250000000',
Key='local.jpg',
PicOperations=json.dumps({
"is_pic_info": 1,
"rules": [{
"fileid": "format.png",
"rule": "imageView2/format/png"
}]
})
)
```
**PicOperations rules 参数说明:**
| 参数 | 描述 | 必选 |
| ------ | ---------------------------------------------------------- | ---- |
| bucket | 目标存储桶名称,不指定则保存到当前桶 | 否 |
| fileid | 结果文件路径,以 `/` 开头存入指定文件夹,否则存入同目录 | 是 |
| rule | 处理参数,参见数据万象图片处理 API | 是 |
---
## 19. 图片高级压缩
> 官方文档https://cloud.tencent.com/document/product/436/120479
>
> 需先在控制台开通图片高级压缩服务。
```python
# TPG 压缩
client.ci_download_compress_image(
Bucket='examplebucket-1250000000',
Key='sample.png',
DestImagePath='sample.tpg',
CompressType='tpg'
)
# HEIF 压缩
client.ci_download_compress_image(
Bucket='examplebucket-1250000000',
Key='sample.png',
DestImagePath='sample.heif',
CompressType='heif'
)
# AVIF 压缩
client.ci_download_compress_image(
Bucket='examplebucket-1250000000',
Key='sample.png',
DestImagePath='sample.avif',
CompressType='avif'
)
```
支持格式:`tpg``heif``avif``svg`
---
## 20. 盲水印
> 官方文档https://cloud.tencent.com/document/product/436/55346
```python
import base64
# 添加盲水印
watermark_url = 'http://examplebucket-1250000000.cos.ap-beijing.myqcloud.com/watermark.png'
watermark_url_base64 = base64.b64encode(watermark_url.encode()).decode()
response, data = client.ci_put_object_from_local_file(
Bucket='examplebucket-1250000000',
LocalFilePath='sample.png',
Key='sample.png',
PicOperations='{"is_pic_info":1,"rules":[{"fileid":"format.png","rule":"watermark/3/type/1/image/' + watermark_url_base64 + '"}]}'
)
# 提取盲水印
sample_url = 'http://examplebucket-1250000000.cos.ap-beijing.myqcloud.com/sample.png'
sample_url_base64 = base64.b64encode(sample_url.encode()).decode()
response, data = client.ci_put_object_from_local_file(
Bucket='examplebucket-1250000000',
LocalFilePath='format.png',
Key='format.png',
PicOperations='{"is_pic_info":1,"rules":[{"fileid":"watermark.png","rule":"watermark/4/type/1/image/' + sample_url_base64 + '"}]}'
)
```
---
## 21. 常见问题
> 官方文档https://cloud.tencent.com/document/product/436/56531
### Q: 升级后无法执行"移动文件"操作?
COS V5 不支持移动文件接口。使用 `copy_object` + `delete_object` 实现。建议删除前校验数据 MD5 一致性。
### Q: 如何获取下载文件的临时链接?
使用 `get_presigned_url``get_presigned_download_url`。使用临时密钥时需在 header 或 query_string 中加上 `x-cos-security-token`
### Q: 出现异常如何处理?
操作成功返回 `dict``None`。失败抛出:
- `CosClientError`:客户端异常(网络、文件 IO、参数校验等
- `CosServiceError`:服务端异常(文件不存在、无权限等)
### Q: 上传时报 `Request has expired`
签名过期,重新生成签名。若仍报错,检查机器本地时间是否为标准北京时间。
---
## 附录:存储类型
| 值 | 说明 |
| ------------- | ---------- |
| STANDARD | 标准存储 |
| STANDARD_IA | 低频存储 |
| ARCHIVE | 归档存储 |
## 附录Region 列表
完整列表参见https://cloud.tencent.com/document/product/436/6224
常用 Region
| Region | 地域 |
| -------------- | ---------- |
| ap-beijing | 北京 |
| ap-shanghai | 上海 |
| ap-guangzhou | 广州 |
| ap-chengdu | 成都 |
| ap-hongkong | 中国香港 |
| ap-singapore | 新加坡 |
| na-siliconvalley | 硅谷 |
| eu-frankfurt | 法兰克福 |