Python基于smtplib和email构建可靠的邮件自动化系统

# 基础依赖安装
pip install schedule python-keyring

# 配置字典示例
SMTP_CONFIG = {
    '163': {
        'server': 'smtp.163.com',
        'port': 465,
        'ssl': True
    },
    'qq': {
        'server': 'smtp.qq.com',
        'port': 587,
        'ssl': False,
        'tls': True
    },
    'gmail': {
        'server': 'smtp.gmail.com',
        'port': 587,
        'tls': True
    }
}

def create_smtp_conn(email_type, username, password):
    config = SMTP_CONFIG[email_type]
    if config['ssl']:
        server = smtplib.SMTP_SSL(config['server'], config['port'])
    else:
        server = smtplib.SMTP(config['server'], config['port'])
        if config.get('tls'):
            server.starttls()
    
    server.login(username, password)
    return server

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

def create_html_email(sender, receiver, subject):
    msg = MIMEMultipart('alternative')
    msg['From'] = sender
    msg['To'] = receiver
    msg['Subject'] = subject
    
    # 纯文本备用内容
    text = "您的邮箱不支持HTML显示"
    
    # HTML内容
    html = """
    <html>
      <body style="font-family: Arial; color: #333;">
        <h1 style="color: #0066cc;">每日报告</h1>
        <p>点击查看<a href="https://example.com" rel="external nofollow" >详细数据</a></p>
        <table border="1">
          <tr><th>项目</th><th>数值</th></tr>
          <tr><td>访问量</td><td style="text-align: right;">1,024</td></tr>
        </table>
      </body>
    </html>
    """
    
    msg.attach(MIMEText(text, 'plain'))
    msg.attach(MIMEText(html, 'html'))
    return msg

from email.mime.application import MIMEApplication

def add_attachment(msg, filepath):
    with open(filepath, 'rb') as f:
        part = MIMEApplication(
            f.read(),
            Name=os.path.basename(filepath)
        )
    part['Content-Disposition'] = f'attachment; filename="{os.path.basename(filepath)}"'
    msg.attach(part)

import time
from concurrent.futures import ThreadPoolExecutor

def batch_send(server, sender, receivers, msg_template, max_workers=5):
    def send_one(receiver):
        try:
            msg = msg_template.copy()
            msg.replace_header('To', receiver)
            server.sendmail(sender, receiver, msg.as_string())
            return True
        except Exception as e:
            print(f"发送失败 {receiver}: {str(e)}")
            return False
    
    with ThreadPoolExecutor(max_workers) as executor:
        results = list(executor.map(send_one, receivers))
    
    success_rate = sum(results)/len(results)
    print(f"发送完成 成功率: {success_rate:.1%}")
    return success_rate

# 使用示例
receivers = ['user1@example.com', 'user2@example.com']  # 实际应从文件或数据库读取
msg = create_html_email(...)
with create_smtp_conn('163', 'user@163.com', 'password') as server:
    batch_send(server, 'user@163.com', receivers, msg)

import imaplib
import email
from email.header import decode_header

def fetch_unseen_emails(username, password, mailbox='INBOX'):
    # 连接IMAP服务器
    imap = imaplib.IMAP4_SSL("imap.163.com")
    imap.login(username, password)
    imap.select(mailbox)
    
    # 搜索未读邮件
    status, messages = imap.search(None, 'UNSEEN')
    if status != 'OK':
        return []
    
    emails = []
    for num in messages[0].split():
        # 获取邮件内容
        status, data = imap.fetch(num, '(RFC822)')
        if status != 'OK':
            continue
            
        raw_email = data[0][1]
        msg = email.message_from_bytes(raw_email)
        
        # 解析邮件头
        subject = decode_header(msg['Subject'])[0][0]
        if isinstance(subject, bytes):
            subject = subject.decode()
        
        from_ = msg.get('From')
        date = msg.get('Date')
        
        # 处理邮件正文
        body = ""
        if msg.is_multipart():
            for part in msg.walk():
                content_type = part.get_content_type()
                if content_type == 'text/plain':
                    body = part.get_payload(decode=True).decode()
                    break
        else:
            body = msg.get_payload(decode=True).decode()
        
        emails.append({
            'num': num,
            'subject': subject,
            'from': from_,
            'date': date,
            'body': body[:200]  # 截取前200字符
        })
    
    imap.close()
    imap.logout()
    return emails

RULES = [
    {
        'condition': lambda mail: '报价单' in mail['subject'],
        'action': lambda mail: send_quote_reply(mail['from'])
    },
    {
        'condition': lambda mail: any(word in mail['body'].lower() 
                                    for word in ['合作', '洽谈']),
        'action': lambda mail: send_business_contact(mail['from'])
    }
]

def process_incoming_emails(username, password):
    emails = fetch_unseen_emails(username, password)
    for mail in emails:
        for rule in RULES:
            if rule['condition'](mail):
                rule['action'](mail)
                mark_as_read(mail['num'])  # 需要实现标记已读
                break

import keyring

# 存储密码
keyring.set_password('email_system', 'user@163.com', 'your_password')

# 获取密码
password = keyring.get_password('email_system', 'user@163.com')

import ssl

context = ssl.create_default_context()
server = smtplib.SMTP_SSL('smtp.163.com', 465, context=context)

import logging
from logging.handlers import RotatingFileHandler

# 配置日志
logger = logging.getLogger('email_system')
logger.setLevel(logging.INFO)
handler = RotatingFileHandler('email.log', maxBytes=1e6, backupCount=3)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# 带错误处理的发送函数
def safe_send(server, msg, receiver):
    try:
        server.sendmail(msg['From'], receiver, msg.as_string())
        logger.info(f"发送成功至 {receiver}")
        return True
    except smtplib.SMTPException as e:
        logger.error(f"发送失败至 {receiver}: {str(e)}")
        return False

from queue import Queue

class SMTPConnectionPool:
    def __init__(self, max_connections=5):
        self._pool = Queue(max_connections)
        for _ in range(max_connections):
            conn = create_smtp_conn(...)
            self._pool.put(conn)
    
    def get_connection(self):
        return self._pool.get()
    
    def release_connection(self, conn):
        self._pool.put(conn)

import asyncio
import aiosmtplib

async def async_send(sender, receiver, msg):
    try:
        await aiosmtplib.send(
            msg,
            sender=sender,
            recipients=[receiver],
            hostname='smtp.163.com',
            port=465,
            use_tls=True
        )
        return True
    except Exception as e:
        print(f"异步发送失败: {e}")
        return False

server.set_debuglevel(1)

from email.parser import Parser
msg = Parser().parsestr(msg.as_string())
print(msg.keys())  # 检查必要头信息

python -m smtpd -n -c DebuggingServer localhost:1025

from flask import Flask, request
import threading

app = Flask(__name__)
mail_queue = Queue()

@app.route('/submit', methods=['POST'])
def handle_form():
    data = request.form
    mail_queue.put(data)
    return "提交成功"

def background_mailer():
    while True:
        data = mail_queue.get()
        send_custom_email(data)
        mail_queue.task_done()

threading.Thread(target=background_mailer, daemon=True).start()

import pandas as pd

def analyze_sent_emails(log_file):
    logs = []
    with open(log_file) as f:
        for line in f:
            if '发送成功' in line:
                logs.append({'status': 'success', **parse_log(line)})
            else:
                logs.append({'status': 'fail', **parse_log(line)})
    
    df = pd.DataFrame(logs)
    print(f"成功率: {df[df.status=='success'].shape[0]/len(df):.1%}")
    print("失败原因分布:")
    print(df[df.status=='fail']['error'].value_counts())