如何写一个脚本自动转发邮件

wen 实用脚本 2026-07-28 227

本文目录导读：

如何写一个脚本自动转发邮件

方案一：使用Python（最灵活通用）
方案二：使用Shell脚本（简单快速）
方案三：使用Node.js
部署建议
安全注意事项
常见问题解决

我来为您介绍几种自动转发邮件的脚本方案,从简单到复杂：

使用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小时