【解决方案实施中心】邮件拓展功能

背景

在做**实施项目过程中, 客户的需求中涉及获取邮件的抄送人;并且要求获取邮件的顺序从旧到新. 这两点官方指令咱不支持.
所以尝试通过自定义代码实现...

查找邮件代码

此代码主要通过imapclient库进行查询, 官方文档的search方法其实是支持筛选邮件主题、发件人等操作，但是实践后发现由于编码问题，均未成功。这里改用通过获取后的邮件信息，再进行筛选。

-- 此代码测试过QQ邮箱和163邮箱

# fetch_email.py

import os
import pytz
import chardet
import email.utils
from datetime import timezone
from datetime import timedelta
from imapclient import IMAPClient
from dateutil.parser import parse
from email import message_from_bytes
from email.header import decode_header
from email.utils import parsedate_to_datetime


def decode_mail_header(header):
    """获取邮件内容的原始编码
    """
    headers = decode_header(header)
    header_parts = []
    for text, encoding in headers:
        if isinstance(text, bytes):
            text = decode_unknown_encoding_string(text, encoding)
        header_parts.append(text)
    return ''.join(header_parts)

def decode_unknown_encoding_string(s, encoding=None):
    """解码未知编码
    """
    if encoding:
        # 首先使用传入的编码进行解码
        try:
            return s.decode(encoding)
        except:
            pass

    # 使用chardet检测编码
    detected_encoding = chardet.detect(s)['encoding']
    
    # 尝试用检测到的编码对字符串进行解码
    try:
        return s.decode(detected_encoding)
    except:
        pass

    # 检测失败，依次尝试其他编码方式
    encodings = ["utf8", "gbk", "gb2312", "latin1", "iso2022_jp", "cp1252", "cp850","cp437", "hz","big5","euc_jp","euc_jis_2004","shift_jis","shift_jis_2004","iso2022_jp"]
    for i in encodings:
        try:
            return s.decode(i)
        except:
            continue
    return None
    
def parse_mail_address(mail_str):
    """解析邮件地址
    """
    mail_ls = mail_str.split(",")
    mail_add_ls = []
    for i in mail_ls:
        if i:
            mail_add_ls.append(email.utils.parseaddr(i)[1])
        else:
            mail_add_ls.append(i)
    return ";".join(mail_add_ls)
    
def parse_mail_date(date_str):
    """解析邮件时间
    """
    if date_str:
        dt = parse(date_str)
        dt_utc = dt.astimezone(pytz.UTC).replace(tzinfo=None)  # 将邮件的时间转为 UTC 时间
        local_dt = dt_utc + timedelta(hours=(dt.utcoffset().total_seconds()//3600))  # 转换为本地时间
        return local_dt.strftime('%Y-%m-%d %H:%M:%S')
    else:
        return date_str

def get_mail_content(server, mail_id, is_save_attachment, download_dir, from_match, to_match, subject_match, body_match):
    """获取邮件信息
    """
    res = server.fetch(mail_id, ['BODY.PEEK[]'])
    raw_message = res[mail_id].get(b'BODY[]', None)
    # 有些邮件比较奇怪, 没有b'BODY[]'参数, 跳过...
    if raw_message is None:
        return {}
    
    # 解析邮件对象
    message = message_from_bytes(raw_message)
    
    # 初始化一个空字典来存储邮件内容
    mail_content = {}

    
    # 获取邮件的主题、发件人、收件人信息，并解码，如果字段不存在则设为一个空字符串
    for header in ['Subject', 'From', 'To', 'Cc']:
        content = message.get(header)
        parsed_content = decode_mail_header(content) if content else ""
        if header in ['From', 'To', 'Cc']:
            parsed_content = parse_mail_address(parsed_content)
        elif header == 'Date':
            parsed_content = parsed_content
        mail_content[header] = parsed_content

    # 获取时间
    date_tuple = parsedate_to_datetime(message['Date'])
    if date_tuple.tzinfo is None:
        # 设置时区信息，如果缺失，假设为UTC（你可以根据需求调整时区）
        date_tuple = date_tuple.replace(tzinfo=timezone.utc)
    # 转换为时间字符串
    standard_time_str = date_tuple.strftime('%Y-%m-%d %H:%M:%S')
    mail_content['Date'] = standard_time_str
    

    # 初始化用来存储附件路径的列表
    mail_content['Attachments'] = []
    mail_content['Body'] = ""
    mail_content['HtmlBody'] = ""

    # 获取邮件的正文内容
    if message.is_multipart():
        for part in message.walk():
            content_type = part.get_content_type()
            if "plain" in content_type:
                # 文本邮件内容
                payload = part.get_payload(decode=True)
                mail_content['Body'] = decode_unknown_encoding_string(payload) if payload else ""
            elif "html" in content_type:
                payload = part.get_payload(decode=True)
                mail_content['HtmlBody'] = decode_unknown_encoding_string(payload) if payload else ""
            elif part.get_content_disposition() == "attachment":  # 下载附件
                if is_save_attachment and subject_match in mail_content.get("Subject", "") and from_match in mail_content.get("From", "") and to_match in mail_content.get("To", "") and (body_match in mail_content.get("Body", "") or body_match in mail_content.get("HtmlBody", "")):
                    filename = decode_mail_header(part.get_filename())
                    if not os.path.isdir(download_dir):
                        os.makedirs(download_dir, exist_ok=True)
                    filepath = os.path.join(download_dir, filename)
                    with open(filepath, 'wb') as f:
                        f.write(part.get_payload(decode=True))
                    mail_content['Attachments'].append(filepath)
    else:
        payload = message.get_payload(decode=True)
        mail_content['Body'] = decode_unknown_encoding_string(payload) if payload else ""
        
    return mail_content


def main(host, port, user, password, fetch_count: int, download_dir, only_unseen, mark_read, from_match, to_match, subject_match, body_match, is_save_attachment, reverse = False, is_ssl=True, folder="INBOX"):
    """主函数

    Args:
        host (str): 邮箱服务器地址
        port (int): 邮箱服务器端口
        user (str): 邮箱用户名
        password (str): 邮箱密码
        fetch_count (int): 获取的邮件数量
        download_dir (str): 附件下载文件夹
        only_unseen (bool): 是否获取未读邮件
        mark_read (bool): 是否把邮件标注为已读
        from_match (str): 筛选发件人
        to_match (str): 筛选收件人
        subject_match (str): 筛选主题
        body_match (str): 筛选正文
        reverse (bool, optional): 是否反转邮件获取的顺序. Defaults to False.
        is_ssl (bool, optional): 是否启用ssl. Defaults to True.
        folder (str, optional): 邮箱文件夹名称. Defaults to "INBOX".
    
    Returns:
        [
            {
                "Subject": 邮件主题,
                "From": 发件人,
                "To": 收件人,
                "Cc": 抄送人,
                "Date": 日期,
                "Attachments": 附件列表,
                "Body": 正文,
                "HtmlBody": html正文,
                "EmailId": 邮件id
            }
        ]
    """
    with IMAPClient(host, port, ssl=is_ssl, timeout=60) as server:
        # 登录
        server.login(user, password)

        # 选择邮箱文件夹
        server.select_folder(folder)

        # 获取邮件ID列表
        if only_unseen:
            ids = server.search('UNSEEN')
        else:
            ids = server.search('ALL')

        # 选择获取的顺序
        if reverse:
            ids = ids[::-1]

        # 获取结果
        result_ls = []
        result_ids = []
        for id in ids:
            ret = get_mail_content(server, id, is_save_attachment, download_dir, from_match, to_match, subject_match, body_match)
            if ret == {}:
                continue
            if subject_match in ret.get("Subject", "") and from_match in ret.get("From", "") and to_match in ret.get("To", "") and (body_match in ret.get("Body", "") or body_match in ret.get("HtmlBody", "")):
                result_ids.append(id)
                ret["EmailId"] = id
                result_ls.append(ret)
                if len(result_ls) == fetch_count:
                    break
        if mark_read:
            server.add_flags(result_ids, '\\Seen')
        return result_ls


if __name__ == '__main__':
    ret = main(
        host="imap.qq.com",
        port=993,
        user="***@**.com",
        password="***",
        fetch_count=1,
        download_dir=r"C:\Users\无尘\Desktop\测试",
        only_unseen=True,
        mark_read=False,
        from_match="",
        to_match="",
        subject_match="",
        body_match="",
        is_save_attachment=True,
        reverse = True,
        is_ssl=True,
        folder="INBOX"
        )
    print(ret)

拓展功能

此代码主要是对查找邮件返回的结果做操作, 支持标记为已读、标记为未读、删除邮件。

-- excute_email的mail_id参数(列表类型)可以从查找邮件结果中提取EmailId字段拼接成列表.

# utils.py

from typing import List
from imapclient import IMAPClient


def login(host, port, user, password, folder, is_ssl=True):
    """登录, 返回"server"
    """
    server = IMAPClient(host=host, port=port, ssl=is_ssl)
    server.login(user, password)
    server.select_folder(folder)
    return server


def list_mailbox(server):
    """列出所有的邮件目录：创建一个函数用于列出所有的邮件目录
    """
    return server.list_folders()


def excute_email(server, mail_id: List[int], flag) -> None:
    """邮件操作
    """
    if flag == "seen":  # 标记为已读
        server.add_flags(mail_id, '\\Seen')
    elif flag == "unseen":  # 标记为未读
        server.remove_flags(mail_id, '\\Seen')
    elif flag == "delete":  # 删除邮件
        server.delete_messages(mail_id)  # 标记为删除状态
        server.expunge()  # 提交删除操作


# 退出：退出并关闭与服务器的连接。
def logout(server):
    """退出：退出并关闭与服务器的连接
    """
    server.logout()

结语

1. 代码有啥问题请多指教.
2. 有更好的方法相互share.