API开放平台场景示例:列出邮件和附件并下载附件

调用“批量获取指定文件夹下的邮件列表”接口,获取邮件基本信息,选取邮件ID,传入“列出邮件的全部附件”接口,获取附件ID和名称,传入“创建下载会话(附件)并下载附件”接口并调用,完成下载。

相关接口

批量获取指定文件夹下的邮件列表

列出邮件的全部附件

创建下载会话(附件)并下载附件

基本流程

image

Python示例代码

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

重要

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

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


def list_mails(email_account, cursor, folder_id):
    """
        批量获取指定文件夹下的邮件列表,每次最多获取100封邮件
        文档:https://mailhelp.aliyun.com/openapi/index.html#/operations/alimailpb_alimail_mailagent_open_MailService_ListMessages
        参数:
        email_account -- 邮箱账号
        cursor -- 游标,用于分页获取数据
        folder_id -- 邮件文件夹ID

        返回:
        邮件列表的JSON响应
        """
    print('接口名称:,批量获取指定文件夹下的邮件列表,每次最多获取100封邮件')
    url = f"https://alimail-cn.aliyuncs.com/v2/users/{email_account}/mailFolders/{folder_id}/messages"

    querystring = {"cursor": cursor, "size": "100", "orderby": "DES"}

    headers = {'Content-Type': 'application/json', 'Authorization': 'bearer ' + access_token}

    response = requests.request("GET", url, headers=headers, params=querystring)

    print(response.text.encode('GBK', 'ignore'))
    return response.json()


def list_mails_attachments(v_email, v_id):
    """
        列出邮件的全部附件
        接口名称:,列出邮件的全部附件
        文档:https://mailhelp.aliyun.com/openapi/index.html#/operations/alimailpb_alimail_mailagent_open_MailService_ListAttachments
        参数:
        v_email -- 邮箱账号
        v_id -- 邮件ID

        返回:
        邮件附件列表的JSON响应
        """

    url = f"https://alimail-cn.aliyuncs.com/v2/users/{v_email}/messages/{v_id}/attachments"

    querystring = {}

    headers = {'Content-Type': 'application/json', 'Authorization': 'bearer ' + access_token}

    response = requests.request("GET", url, headers=headers, params=querystring)

    print('##########################################################################')
    print('请求参数:', querystring)
    print('返回参数:', response.status_code, response.text)
    print('##########################################################################')
    return response.json()['attachments']


def download_mails_attachments(v_email, v_id, attachment_id, attachment_name):
    """
        创建下载会话(附件)并下载附件
        接口名称:创建下载会话(附件)
        文档:https://mailhelp.aliyun.com/openapi/index.html#/operations/alimailpb_alimail_mailagent_open_MailService_CreateAttachmentDownloadSession
        参数:
        v_email -- 邮箱账号
        v_id -- 邮件ID
        attachment_id -- 附件ID
        attachment_name -- 附件名称,用于保存附件
        """

    url = f"https://alimail-cn.aliyuncs.com/v2/users/{v_email}/messages/{v_id}/attachments/{attachment_id}/$value"

    headers = {
        "Content-Type": "application/json",
        "Authorization": 'bearer ' + access_token
    }

    response = requests.request("GET", url, headers=headers)
    with open(attachment_name, 'wb') as f:
        f.write(response.content)


def get_all_mails(email_account, v_folder_id):
    """
        获取邮件组的基本信息

        参数:
        email_account -- 邮箱账号
        v_folder_id -- 邮件文件夹ID

        返回:
        邮件组基本信息列表
    """
    records = []
    cursor = ""
    while True:
        # 获取数据
        parsed_data = list_mails(email_account, cursor, v_folder_id)

        # 提取所需字段
        for v_mail in parsed_data['messages']:
            record = {
                'id': v_mail['id'],
                'mailId': v_mail['mailId'],
                'sentDateTime': v_mail['sentDateTime'],  # 和北京时间时差8小时
                'hasAttachments': v_mail['hasAttachments']
            }
            records.append(record)
        # 更新游标
        cursor = parsed_data["nextCursor"]
        print(f'hasMore={parsed_data["hasMore"]},nextCursor={cursor}')
        if not parsed_data["hasMore"]:
            print(f'没有更多数据')
            break
    return records


print('======================================')
folder_id = {
    "发件箱": "1",
    "收件箱": "2",
    "垃圾箱": "3",
    "草稿箱": "5",
    "已删除": "6"
}

email_account = 'test@example.com'
all_data = get_all_mails(email_account, folder_id.get('收件箱'))
print(f'邮件基本信息:共{len(all_data)}封,{all_data}')
print('======================================')
# 根据邮件时间等信息选择需要下载的附件的邮件id,如'DzzzzzzNpQ9'
email_id = 'DzzzzzzNpQ9'
list_attachments = list_mails_attachments(email_account, email_id)  # 列出邮件的全部附件,成功
print('======================================')
for v_file in list_attachments:
    print(f'附件名称:{v_file["name"]}')
    download_mails_attachments(email_account, email_id, v_file["id"], v_file["name"])
    print('下载完成')
print('======================================')

运行结果

image