API開放平臺場景示例:創(chuàng)建草稿并上傳附件
更新時間:
本文介紹如何創(chuàng)建草稿并上傳附件。
調(diào)用“創(chuàng)建草稿”接口,從返回值中獲取id,也就是草稿ID;
草稿ID傳入“創(chuàng)建上傳會話(附件)”接口,截取出uploadUrl中的session值;
session值傳入“上傳文件流”接口,上傳文件流,前往網(wǎng)頁版郵箱的草稿箱,檢查附件是否已上傳完成。
相關(guān)接口
基本流程
Python示例代碼
重要
風(fēng)險提示:下述代碼在Python 3.11.9進行的測試,用于生產(chǎn)環(huán)境之前請務(wù)必先做好測試。
# -*- coding: utf-8 -*-
import email
import os
import requests
# 導(dǎo)入獲取訪問令牌的函數(shù),路徑根據(jù)實際情況進行修改,或直接給access_token賦值用于測試
from api_demo.get_access_token import access_token
def create_draft(email_account, payload):
"""
接口名稱:創(chuàng)建草稿
文檔:https://mailbestwisewords.com/openapi/index.html#/operations/alimailpb_alimail_mailagent_open_MailService_CreateMessage
參數(shù):
email (str): 用戶郵箱地址
payload (dict): 郵件內(nèi)容的JSON格式數(shù)據(jù)
返回:
str:: 創(chuàng)建的郵件草稿的ID
"""
# 構(gòu)造請求URL
url = f"https://alimail-cn.aliyuncs.com/v2/users/{email_account}/messages"
# 構(gòu)造請求頭,包括內(nèi)容類型和授權(quán)令牌
headers = {'Content-Type': 'application/json', 'Authorization': 'bearer ' + access_token}
# 發(fā)送POST請求,創(chuàng)建郵件草稿
response = requests.request("POST", url, json=payload, headers=headers)
# 打印請求和響應(yīng)的詳細(xì)信息
print('##########################################################################')
print('請求參數(shù):', payload)
print('返回參數(shù):', response.status_code, response.text)
print('##########################################################################')
# 返回郵件草稿的ID
return response.json()["message"]["id"]
def create_upload_session(email_account, v_id, file_name):
"""
創(chuàng)建上傳會話(附件),為草稿郵件添加一個附件
文檔:https://mailbestwisewords.com/openapi/index.html#/operations/alimailpb_alimail_mailagent_open_MailService_CreateAttachmentUploadSession
"""
url = f"https://alimail-cn.aliyuncs.com/v2/users/{email_account}/messages/{v_id}/attachments/createUploadSession"
payload = {"attachment": {
"name": file_name,
"contentId": "string",
"isInline": True,
"extHeaders": {
"property1": "string",
"property2": "string"
}
}}
headers = {'Content-Type': 'application/json', 'Authorization': 'bearer ' + access_token}
response = requests.request("POST", url, json=payload, headers=headers)
print(response.text)
json_payload = response.json()
print('##########################################################################')
print('請求參數(shù):', payload)
print('返回參數(shù):', response.status_code, response.text)
print('##########################################################################')
return json_payload['uploadUrl'].split('stream/')[1]
def upload_file(v_file, v_id):
"""
上傳文件流
https://mailbestwisewords.com/openapi/index.html#/operations/alimailpb_stream_StreamService_DoUpload
:param v_file:
:param v_id:
:return:
"""
print('接口名稱:', '上傳文件流,upload_file')
url = f"https://alimail-cn.aliyuncs.com/v2/stream/{v_id}"
try:
with open(v_file, 'rb') as file:
payload = file.read()
except FileNotFoundError as e:
print(f"文件未找到: {e}")
return None
print(f'payload={type(payload)},{payload}') # <class 'bytes'>
headers = {
"Content-Type": "application/octet-stream",
"Authorization": 'Bearer ' + access_token
}
response = requests.request("POST", url, data=payload, headers=headers)
print('##########################################################################')
print('請求參數(shù):', payload)
print('返回參數(shù):', response.status_code, response.text)
print('##########################################################################')
# 定義郵箱地址和郵件內(nèi)容的JSON格式數(shù)據(jù)
email_account = "test@example.com"
v_payload = {
"message": {
"internetMessageId": email.utils.make_msgid(),
"subject": "主題",
"summary": "摘要",
"priority": "PRY_NORMAL",
"isReadReceiptRequested": True,
"from": {
"email": "test@example.com",
"name": "test"
},
"toRecipients": [
{
"email": "test@example.com",
"name": "to"
}
],
"ccRecipients": [
{
"email": "test@example.com",
"name": "cc"
}
],
"bccRecipients": [
{
"email": "test@example.com",
"name": "bcc"
}
],
"replyTo": [
{
"email": "test@example.com",
"name": "replyTo"
}
],
"body": {
"bodyText": "bodyText",
"bodyHtml": "<h1>bodyHtml</h1>"
},
"internetMessageHeaders": {
"property1": "string",
"property2": "string"
},
"tags": ["string"]
}
}
# 調(diào)用函數(shù)創(chuàng)建郵件草稿并獲取草稿ID
email_id = create_draft(email_account, v_payload)
print(f'email_id: {email_id}')
print('草稿創(chuàng)建完成')
# 本地附件
files = [r'C:\Users\PycharmProjects\aliyun\api_demo\123.txt',
r'C:\Users\PycharmProjects\aliyun\api_demo\234.txt']
for v_file in files:
file_name = os.path.basename(v_file)
print(f'附件名稱:{file_name}')
# 創(chuàng)建上傳會話(附件)
session_id = create_upload_session(email_account, email_id, file_name)
# print(f'session_id={session_id}')
print('創(chuàng)建上傳會話(附件)完成')
# 文件流上傳
upload_file(v_file, session_id)
print('上傳文件流完成')
運行結(jié)果
常見問題
上傳的附件是什么數(shù)據(jù)類型,如何在網(wǎng)頁上測試?
在代碼中實現(xiàn)時,上傳的附件是字節(jié)流bytes類型。
try:
with open(v_file, 'rb') as file:
payload = file.read()
except FileNotFoundError as e:
print(f"文件未找到: {e}")
網(wǎng)頁中測試txt附件時,直接輸入文件內(nèi)容的字符串即可。
文檔內(nèi)容是否對您有幫助?