本文目录导读：
我来为您介绍几种自动转发邮件的脚本方案,从简单到复杂：
使用Python（最灵活通用）
基础转发脚本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import imaplib
import smtplib
import email
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import time
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class EmailForwarder:
def __init__(self, source_config, target_config):
"""
初始化邮件转发器
source_config: 源邮箱配置
target_config: 目标邮箱配置
"""
self.source_config = source_config
self.target_config = target_config
def connect_source(self):
"""连接源邮箱"""
try:
mail = imaplib.IMAP4_SSL(self.source_config['imap_server'])
mail.login(self.source_config['username'], self.source_config['password'])
mail.select('INBOX')
return mail
except Exception as e:
logging.error(f"连接源邮箱失败: {e}")
return None
def fetch_emails(self, mail, criteria='UNSEEN'):
"""获取未读邮件"""
try:
status, messages = mail.search(None, criteria)
if status != 'OK':
return []
email_ids = messages[0].split()
emails = []
for e_id in email_ids[-10:]: # 限制每次处理10封
status, msg_data = mail.fetch(e_id, '(RFC822)')
if status == 'OK':
email_body = msg_data[0][1]
msg = email.message_from_bytes(email_body)
emails.append((e_id, msg))
return emails
except Exception as e:
logging.error(f"获取邮件失败: {e}")
return []
def forward_email(self, original_msg):
"""转发邮件"""
try:
# 创建新邮件
forward_msg = MIMEMultipart()
forward_msg['From'] = self.source_config['username']
forward_msg['To'] = self.target_config['username']
forward_msg['Subject'] = f"Fwd: {original_msg['Subject']}"
# 添加原始邮件的头部信息
body = f"""
原始发件人: {original_msg['From']}
原始主题: {original_msg['Subject']}
原始日期: {original_msg['Date']}
---
"""
# 获取邮件内容
if original_msg.is_multipart():
for part in original_msg.walk():
if part.get_content_type() == "text/plain":
body += part.get_payload(decode=True).decode()
break
else:
body += original_msg.get_payload(decode=True).decode()
forward_msg.attach(MIMEText(body, 'plain', 'utf-8'))
# 发送邮件
smtp = smtplib.SMTP_SSL(self.target_config['smtp_server'])
smtp.login(self.target_config['username'], self.target_config['password'])
smtp.send_message(forward_msg)
smtp.quit()
logging.info(f"成功转发邮件: {original_msg['Subject']}")
return True
except Exception as e:
logging.error(f"转发邮件失败: {e}")
return False
def process_emails(self):
"""处理邮件转发"""
mail = self.connect_source()
if not mail:
return
try:
emails = self.fetch_emails(mail)
logging.info(f"发现 {len(emails)} 封新邮件")
for e_id, msg in emails:
if self.forward_email(msg):
# 可选：标记为已读
mail.store(e_id, '+FLAGS', '\\Seen')
time.sleep(1) # 避免过快操作
finally:
mail.logout()
def run_once(self):
"""单次运行"""
logging.info("开始检查邮件...")
self.process_emails()
logging.info("检查完成")
def run_loop(self, interval=60):
"""循环运行模式"""
logging.info(f"启动监控模式，每隔 {interval} 秒检查一次")
while True:
try:
self.process_emails()
time.sleep(interval)
except KeyboardInterrupt:
logging.info("收到停止信号，退出程序")
break
except Exception as e:
logging.error(f"运行错误: {e}")
time.sleep(interval)
# 配置示例
if __name__ == "__main__":
# 源邮箱配置
source_config = {
'imap_server': 'imap.gmail.com',
'username': 'your_email@gmail.com',
'password': 'your_app_password', # Gmail需要应用专用密码
'smtp_server': 'smtp.gmail.com'
}
# 目标邮箱配置
target_config = {
'smtp_server': 'smtp.gmail.com',
'username': 'target_email@gmail.com',
'password': 'target_app_password'
}
# 创建转发器
forwarder = EmailForwarder(source_config, target_config)
# 单次运行
forwarder.run_once()
# 或循环运行（每5分钟检查一次）
# forwarder.run_loop(interval=300)
带过滤器的增强版
import re
from typing import List, Dict
class SmartEmailForwarder(EmailForwarder):
def __init__(self, source_config, target_config, filters=None):
super().__init__(source_config, target_config)
self.filters = filters or {}
def fetch_emails_with_filters(self, mail):
"""带过滤条件的邮件获取"""
emails = self.fetch_emails(mail, 'ALL')
filtered_emails = []
for e_id, msg in emails:
if self.passes_filters(msg):
filtered_emails.append((e_id, msg))
return filtered_emails
def passes_filters(self, msg):
"""检查邮件是否通过过滤器"""
# 关键词过滤
if 'keywords' in self.filters:
subject = msg['Subject'] or ''
body = self.get_email_body(msg)
has_keyword = any(
keyword.lower() in subject.lower() or
keyword.lower() in body.lower()
for keyword in self.filters['keywords']
)
if self.filters.get('keyword_mode') == 'require' and not has_keyword:
return False
elif self.filters.get('keyword_mode') == 'exclude' and has_keyword:
return False
# 发件人过滤
if 'senders' in self.filters:
sender = msg['From'] or ''
if 'whitelist' in self.filters['senders']:
if not any(s in sender for s in self.filters['senders']['whitelist']):
return False
if 'blacklist' in self.filters['senders']:
if any(s in sender for s in self.filters['senders']['blacklist']):
return False
return True
@staticmethod
def get_email_body(msg):
"""提取邮件正文"""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain":
return part.get_payload(decode=True).decode()
else:
return msg.get_payload(decode=True).decode()
return ""
# 使用示例
filters = {
'keywords': ['urgent', 'important', '发票', '合同'],
'keyword_mode': 'require', # 'require' 或 'exclude'
'senders': {
'whitelist': ['@company.com', '@partner.com'],
# 'blacklist': ['spam@example.com']
}
}
使用Shell脚本（简单快速）
#!/bin/bash
# 邮件转发脚本
# 使用前提：需要安装 mailx 或 mutt
# 配置
SOURCE_EMAIL="source@gmail.com"
SOURCE_PASSWORD="password"
TARGET_EMAIL="target@gmail.com"
IMAP_SERVER="imap.gmail.com"
SMTP_SERVER="smtp.gmail.com"
# 获取新邮件
fetch_and_forward() {
echo "检查新邮件..."
# 使用 fetchmail 获取邮件
fetchmail -c
# 如果有新邮件，转发
if [ $? -eq 0 ]; then
# 获取最新邮件并转发
echo "发现新邮件，开始转发..."
# 使用 procmail 或 mail 命令转发
mail -s "Forwarded from $SOURCE_EMAIL" "$TARGET_EMAIL" < /tmp/latest_email.txt
echo "邮件已转发"
fi
}
# 主循环
while true; do
fetch_and_forward
sleep 300 # 每5分钟检查一次
done
使用Node.js
// forward-email.js
const Imap = require('imap');
const { simpleParser } = require('mailparser');
const nodemailer = require('nodemailer');
const util = require('util');
class EmailForwarder {
constructor(config) {
this.config = config;
this.imap = new Imap({
user: config.source.user,
password: config.source.password,
host: config.source.imapHost,
port: 993,
tls: true,
tlsOptions: { rejectUnauthorized: false }
});
this.transporter = nodemailer.createTransport({
host: config.target.smtpHost,
port: 465,
secure: true,
auth: {
user: config.target.user,
pass: config.target.password
}
});
}
async connect() {
return new Promise((resolve, reject) => {
this.imap.once('ready', resolve);
this.imap.once('error', reject);
this.imap.connect();
});
}
async getUnseenEmails() {
return new Promise((resolve, reject) => {
this.imap.openBox('INBOX', false, (err, box) => {
if (err) return reject(err);
const searchCriteria = ['UNSEEN'];
this.imap.search(searchCriteria, (err, results) => {
if (err) return reject(err);
if (results.length === 0) {
return resolve([]);
}
const fetch = this.imap.fetch(results, { bodies: '' });
const emails = [];
fetch.on('message', (msg) => {
msg.on('body', (stream) => {
simpleParser(stream, (err, parsed) => {
if (!err) {
emails.push(parsed);
}
});
});
});
fetch.once('end', () => {
resolve(emails);
});
});
});
});
}
async forwardEmail(email) {
const mailOptions = {
from: this.config.source.user,
to: this.config.target.user,
subject: `Fwd: ${email.subject}`,
text: `
Original From: ${email.from.text}
Original Subject: ${email.subject}
Original Date: ${email.date}
---
Original Message:
${email.text}
`
};
return this.transporter.sendMail(mailOptions);
}
async start(interval = 60000) {
console.log('Email forwarder started');
setInterval(async () => {
try {
await this.connect();
const emails = await this.getUnseenEmails();
for (const email of emails) {
await this.forwardEmail(email);
console.log(`Forwarded: ${email.subject}`);
}
this.imap.end();
} catch (error) {
console.error('Error:', error);
}
}, interval);
}
}
// 使用示例
const forwarder = new EmailForwarder({
source: {
user: 'your_email@gmail.com',
password: 'your_app_password',
imapHost: 'imap.gmail.com'
},
target: {
user: 'target@example.com',
password: 'target_password',
smtpHost: 'smtp.gmail.com'
}
});
forwarder.start(60000); // 每分钟检查一次
部署建议
作为服务运行（Linux）
创建 systemd 服务文件 /etc/systemd/system/email-forwarder.service：
[Unit] Description=Email Forwarder Service After=network.target [Service] Type=simple User=your_user WorkingDirectory=/path/to/script ExecStart=/usr/bin/python3 /path/to/script/forward_email.py Restart=always RestartSec=10 [Install] WantedBy=multi-user.target
使用Docker部署
FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY forward_email.py . CMD ["python", "forward_email.py"]
使用Crontab定时运行
# 每5分钟运行一次 */5 * * * * /usr/bin/python3 /path/to/forward_email.py --run-once # 或更复杂的调度 0 */2 * * * /usr/bin/python3 /path/to/forward_email.py --run-once # 每2小时
安全注意事项
- 密码安全：使用环境变量或加密配置
- 使用应用专用密码：特别是Gmail等
- 启用SSL/TSL：确保通信加密
- 限制处理数量：避免邮箱被封
- 添加日志监控：记录所有操作
常见问题解决
- Gmail特殊处理：需要开启IMAP并生成应用专用密码
- QQ/163邮箱：需要开启POP3/IMAP服务
- 企业邮箱：可能需要特殊配置
- 附件处理：如需转发附件，代码需要相应修改
- 并发控制：避免同时处理过多邮件
选择哪种方案取决于您的具体需求和技术环境,Python方案最为灵活，Shell最简单，Node.js适合前后端统一的技术栈。
