API开放平台场景示例：创建草稿并上传附件

调用“创建草稿”接口，从返回值中获取id，也就是草稿id，传入“创建上传会话（附件）”接口，获取uploadUrl中的session值，传入“上传文件流”接口，上传文件流，前往网页版邮箱的草稿箱，检查附件是否已上传完成。

相关接口

创建草稿

创建上传会话（附件）

上传文件流

基本流程

Python示例代码

API开放平台代码示例：获取访问凭证

# -*- coding: utf-8 -*-
import email
import os

import requests
# 导入获取访问令牌的函数，路径根据实际情况进行修改，或直接给access_token赋值用于测试
from api_demo.get_access_token import access_token


def create_draft(email_account, payload):
    """
    接口名称：创建草稿
    文档：https://mailhelp.aliyun.com/openapi/index.html#/operations/alimailpb_alimail_mailagent_open_MailService_CreateMessage

    参数：
    email (str)： 用户邮箱地址
    payload (dict)： 邮件内容的JSON格式数据

    返回：
    str：: 创建的邮件草稿的ID
    """
    # 构造请求URL
    url = f"https://alimail-cn.aliyuncs.com/v2/users/{email_account}/messages"

    # 构造请求头，包括内容类型和授权令牌
    headers = {'Content-Type': 'application/json', 'Authorization': 'bearer ' + access_token}

    # 发送POST请求，创建邮件草稿
    response = requests.request("POST", url, json=payload, headers=headers)

    # 打印请求和响应的详细信息
    print('##########################################################################')
    print('请求参数：', payload)
    print('返回参数：', response.status_code, response.text)
    print('##########################################################################')

    # 返回邮件草稿的ID
    return response.json()["message"]["id"]

def create_upload_session(email_account, v_id, file_name):
    """
    创建上传会话（附件），为草稿邮件添加一个附件
    文档：https://mailhelp.aliyun.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('请求参数：', payload)
    print('返回参数：', response.status_code, response.text)
    print('##########################################################################')
    return json_payload['uploadUrl'].split('stream/')[1]


def upload_file(v_file, v_id):
    """
    上传文件流
    https://mailhelp.aliyun.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('请求参数：', payload)
    print('返回参数：', response.status_code, response.text)
    print('##########################################################################')

# 定义邮箱地址和邮件内容的JSON格式数据
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"]
    }
}

# 调用函数创建邮件草稿并获取草稿ID
email_id = create_draft(email_account, v_payload)
print(f'email_id: {email_id}')
print('草稿创建完成')

# 本地附件
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}')

    # 创建上传会话(附件)
    session_id = create_upload_session(email_account, email_id, file_name)
    # print(f'session_id={session_id}')
    print('创建上传会话（附件）完成')

    # 文件流上传
    upload_file(v_file, session_id)
    print('上传文件流完成')

运行结果