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

本文介绍如何创建草稿并上传附件。

调用“创建草稿”接口，从返回值中获取id，也就是草稿ID；

草稿ID传入“创建上传会话（附件）”接口，截取出uploadUrl中的session值；

session值传入“上传文件流”接口，上传文件流，前往网页版邮箱的草稿箱，检查附件是否已上传完成。

相关接口

创建草稿

创建上传会话（附件）

上传文件流

基本流程

Python示例代码

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

重要

风险提示：下述代码在Python 3.11.9进行的测试，用于生产环境之前请务必先做好测试。

# -*- coding: utf-8 -*-import emailimport osimport requests# 导入获取访问令牌的函数，路径根据实际情况进行修改，或直接给access_token赋值用于测试from api_demo.get_access_token import access_tokendef 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"]
    }
}# 调用函数创建邮件草稿并获取草稿IDemail_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('上传文件流完成')

运行结果

常见问题

上传的附件是什么数据类型，如何在网页上测试？

在代码中实现时，上传的附件是字节流bytes类型。

try:    with open(v_file, 'rb') as file:
        payload = file.read()except FileNotFoundError as e:    print(f"文件未找到： {e}")

网页中测试txt附件时，直接输入文件内容的字符串即可。