IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

实用脚本如何实现自动发送邮件？

wen 实用脚本 2

本文目录导读：

实用脚本如何实现自动发送邮件？

  1. Python + SMTP 方案（最常用、灵活）
  2. 命令行脚本方案
  3. Node.js 方案
  4. 配置文件管理
  5. 最佳实践建议
  6. 实际使用场景示例

Python + SMTP 方案（最常用、灵活）

基础发送脚本

import smtplib
import schedule
import time
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import json
import csv
class EmailSender:
    def __init__(self, config_file='email_config.json'):
        self.config = self.load_config(config_file)
        self.setup_server()
    def load_config(self, config_file):
        with open(config_file, 'r') as f:
            return json.load(f)
    def setup_server(self):
        self.smtp_server = self.config['smtp_server']
        self.smtp_port = self.config['smtp_port']
        self.username = self.config['username']
        self.password = self.config['password']
    def send_email(self, to_emails, subject, body, 
                   attachments=None, html_content=False):
        """发送邮件"""
        try:
            # 创建邮件对象
            msg = MIMEMultipart('alternative')
            msg['From'] = self.username
            msg['To'] = ', '.join(to_emails) if isinstance(to_emails, list) else to_emails
            msg['Subject'] = subject
            # 添加正文
            if html_content:
                msg.attach(MIMEText(body, 'html', 'utf-8'))
            else:
                msg.attach(MIMEText(body, 'plain', 'utf-8'))
            # 添加附件
            if attachments:
                for file_path in attachments:
                    self.attach_file(msg, file_path)
            # 发送邮件
            with smtplib.SMTP_SSL(self.smtp_server, self.smtp_port) as server:
                server.login(self.username, self.password)
                server.send_message(msg)
            print(f"✔ 邮件发送成功: {subject}")
            return True
        except Exception as e:
            print(f"✘ 发送失败: {str(e)}")
            return False
    def attach_file(self, msg, file_path):
        """添加附件"""
        with open(file_path, 'rb') as attachment:
            part = MIMEBase('application', 'octet-stream')
            part.set_payload(attachment.read())
            encoders.encode_base64(part)
            part.add_header(
                'Content-Disposition',
                f'attachment; filename= {file_path.split("/")[-1]}'
            )
            msg.attach(part)
    def send_html_email(self, to_emails, subject, html_content):
        """发送HTML格式邮件"""
        return self.send_email(to_emails, subject, html_content, html_content=True)
    def send_bulk(self, recipients, subject_template, body_template):
        """批量发送邮件"""
        successful = 0
        failed = 0
        for recipient in recipients:
            subject = subject_template.format(**recipient)
            body = body_template.format(**recipient)
            if self.send_email(recipient['email'], subject, body):
                successful += 1
            else:
                failed += 1
        return successful, failed
# 配置示例 (email_config.json)
config_template = {
    "smtp_server": "smtp.gmail.com",  # 或其他SMTP服务器
    "smtp_port": 465,  # SSL端口
    "username": "your_email@gmail.com",
    "password": "your_app_password"  # 使用应用专用密码
}
# 使用示例
if __name__ == "__main__":
    # 初始化发送器
    sender = EmailSender('email_config.json')
    # 1. 发送简单文本邮件
    sender.send_email(
        ['user1@example.com', 'user2@example.com'],
        '测试邮件',
        '这是一封自动发送的测试邮件'
    )
    # 2. 发送带附件的邮件
    sender.send_email(
        'recipient@example.com',
        '月度报告',
        '请查看附件中的月度报告',
        attachments=['report.pdf', 'data.xlsx']
    )
    # 3. 发送HTML格式邮件
    html_content = """
    <html>
        <body>
            <h1 style="color: #4CAF50;">系统通知</h1>
            <p>您的任务已完成</p>
            <table border="1">
                <tr><th>项目</th><th>状态</th></tr>
                <tr><td>备份</td><td>完成</td></tr>
                <tr><td>更新</td><td>完成</td></tr>
            </table>
        </body>
    </html>
    """
    sender.send_html_email('user@example.com', '系统通知', html_content)

定时发送功能

import schedule
import time
from datetime import datetime, timedelta
class ScheduledEmailSender:
    def __init__(self, email_sender):
        self.sender = email_sender
        self.tasks = []
    def daily_report(self):
        """每日报告"""
        today = datetime.now().strftime('%Y-%m-%d')
        subject = f"日报告 - {today}"
        # 从数据库或文件生成报告
        report_data = self.generate_report()
        body = f"""
        日报告 {today}
        ============
        新增用户: {report_data['new_users']}
        活跃用户: {report_data['active_users']}
        订单数量: {report_data['orders']}
        总收入: ¥{report_data['revenue']}
        """
        self.sender.send_email(['manager@company.com'], subject, body)
    def weekly_summary(self):
        """每周摘要"""
        week_start = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
        week_end = datetime.now().strftime('%Y-%m-%d')
        email_content = self.prepare_weekly_html(week_start, week_end)
        self.sender.send_html_email(
            ['team@company.com'],
            f"周报 - {week_start} 至 {week_end}",
            email_content
        )
    def setup_schedule(self):
        """设置定时任务"""
        # 每天早上9点发送日报
        schedule.every().day.at("09:00").do(self.daily_report)
        # 每周一发送周报
        schedule.every().monday.at("10:00").do(self.weekly_summary)
        # 每半小时检查一次
        schedule.every(30).minutes.do(self.check_system_status)
        # 自定义定时
        self.schedule_custom_tasks()
    def schedule_custom_tasks(self):
        """自定义定时任务"""
        # 每5分钟发送一次
        schedule.every(5).minutes.do(
            self.send_notification, 
            subject="系统状态检查",
            body="系统运行正常"
        )
        # 特定时间
        schedule.every().day.at("18:00").do(
            self.send_eod_report
        )
    def run(self):
        """运行调度器"""
        self.setup_schedule()
        while True:
            schedule.run_pending()
            time.sleep(1)
    def generate_report(self):
        """生成报告数据（示例）"""
        return {
            'new_users': 150,
            'active_users': 2340,
            'orders': 567,
            'revenue': 45678.50
        }
    def prepare_weekly_html(self, start, end):
        """准备HTML格式周报"""
        return f"""
        <html>
        <body>
            <h2>周报告: {start} - {end}</h2>
            <!-- 添加更多HTML内容 -->
        </body>
        </html>
        """
    def send_notification(self, subject, body):
        """发送通知"""
        self.sender.send_email(['admin@company.com'], subject, body)
    def send_eod_report(self):
        """发送收盘报告"""
        # 具体的收盘报告逻辑
        pass
# 使用定时发送
if __name__ == "__main__":
    sender = EmailSender('config.json')
    scheduler = ScheduledEmailSender(sender)
    scheduler.run()  # 启动定时任务

命令行脚本方案

使用 sendmail (Linux/Unix)

#!/bin/bash
# send_email.sh
# 配置
SMTP_SERVER="smtp.gmail.com:587"
USERNAME="your_email@gmail.com"
PASSWORD="your_app_password"
# 发送简单邮件
send_simple_email() {
    local to="$1"
    local subject="$2"
    local body="$3"
    echo -e "Subject: $subject\n\n$body" | \
    mail -s "$subject" "$to" <<< "$body"
}
# 发送带附件的邮件
send_email_with_attachment() {
    local to="$1"
    local subject="$2"
    local body="$3"
    local attachment="$4"
    echo "$body" | mail -s "$subject" \
        -a "$attachment" \
        "$to"
}
# 批量发送
send_bulk_emails() {
    local csv_file="$1"
    while IFS=',' read -r email name
    do
        subject="Hello $name"
        body="Dear $name,\n\nThis is a test email."
        send_simple_email "$email" "$subject" "$body"
        echo "Sent to $email"
    done < "$csv_file"
}
# 使用示例
if [ "$1" = "send" ]; then
    send_simple_email "$2" "$3" "$4"
elif [ "$1" = "bulk" ]; then
    send_bulk_emails "$2"
fi

Node.js 方案

// sendEmail.js
const nodemailer = require('nodemailer');
const fs = require('fs');
const path = require('path');
class EmailService {
    constructor(config) {
        this.transporter = nodemailer.createTransport({
            host: config.host || 'smtp.gmail.com',
            port: config.port || 465,
            secure: true,
            auth: {
                user: config.user,
                pass: config.pass
            }
        });
    }
    async send(options) {
        try {
            const info = await this.transporter.sendMail({
                from: options.from,
                to: options.to,
                subject: options.subject,
                text: options.text,
                html: options.html,
                attachments: options.attachments
            });
            console.log('邮件发送成功:', info.messageId);
            return info;
        } catch (error) {
            console.error('发送失败:', error);
            throw error;
        }
    }
    async sendBatch(recipients, options) {
        const results = [];
        for (const recipient of recipients) {
            try {
                const result = await this.send({
                    ...options,
                    to: recipient.email,
                    subject: options.subject.replace('{name}', recipient.name)
                });
                results.push({ recipient, success: true, result });
            } catch (error) {
                results.push({ recipient, success: false, error });
            }
        }
        return results;
    }
}
// 使用示例
const emailService = new EmailService({
    user: 'your_email@gmail.com',
    pass: 'your_app_password'
});
// 发送邮件
emailService.send({
    to: 'recipient@example.com',
    subject: '自动发送测试',
    html: '<h1>Hello!</h1><p>这是一封自动发送的邮件</p>',
    attachments: [
        { filename: 'report.pdf', path: './reports/monthly.pdf' }
    ]
});

配置文件管理

JSON 配置模板

{
  "email": {
    "smtp": {
      "server": "smtp.gmail.com",
      "port": 465,
      "use_ssl": true
    },
    "credentials": {
      "username": "your_email@gmail.com",
      "password": "your_app_password"
    },
    "settings": {
      "from_name": "自动通知系统",
      "reply_to": "noreply@yourdomain.com",
      "max_attachments": 5,
      "max_attachment_size": 10485760
    }
  },
  "templates": {
    "notification": {
      "subject": "系统通知 - {timestamp}",
      "template_file": "templates/notification.html"
    },
    "report": {
      "subject": "报告 - {date}",
      "template_file": "templates/report.html"
    }
  },
  "scheduling": {
    "daily_report": {
      "time": "09:00",
      "recipients": ["manager@company.com", "team@company.com"]
    },
    "weekly_summary": {
      "day": "monday",
      "time": "10:00",
      "recipients": ["all@company.com"]
    }
  }
}

最佳实践建议

安全性

# 使用环境变量存储敏感信息
import os
from dotenv import load_dotenv
load_dotenv()
EMAIL_USER = os.getenv('EMAIL_USER')
EMAIL_PASS = os.getenv('EMAIL_PASS')
EMAIL_SERVER = os.getenv('EMAIL_SERVER')

错误处理和日志

import logging
from datetime import datetime
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'email_log_{datetime.now().strftime("%Y%m")}.log'),
        logging.StreamHandler()
    ]
)
def send_with_retry(email_sender, max_retries=3):
    """带重试机制的发送函数"""
    for attempt in range(max_retries):
        try:
            result = email_sender.send_email()
            logging.info(f"发送成功 (尝试 {attempt + 1}/{max_retries})")
            return result
        except Exception as e:
            logging.error(f"第 {attempt + 1} 次尝试失败: {str(e)}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # 指数退避
    logging.critical("发送失败，已达到最大重试次数")
    return False

Docker 部署

FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# 使用 supervisor 管理进程
RUN apt-get update && apt-get install -y supervisor
COPY supervisord.conf /etc/supervisor/conf.d/supervisord.conf
CMD ["supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"]

实际使用场景示例

服务器监控通知

def monitor_and_notify():
    """监控系统状态并发送通知"""
    import psutil
    cpu_usage = psutil.cpu_percent(interval=1)
    memory = psutil.virtual_memory()
    disk = psutil.disk_usage('/')
    if cpu_usage > 80 or memory.percent > 80 or disk.percent > 80:
        alert_msg = f"""
        系统警告!
        CPU: {cpu_usage}%
        内存: {memory.percent}%
        磁盘: {disk.percent}%
        """
        email_sender.send_email(
            ['admin@company.com'],
            '系统资源警告',
            alert_msg
        )

这个脚本可以根据你的具体需求进行调整,关键是要：

  1. 使用应用专用密码而不是真实密码
  2. 实现错误处理和重试机制
  3. 添加日志记录
  4. 考虑使用模板系统管理邮件内容
  5. 实现配置管理

选择哪种方案取决于你的具体使用场景和技术栈。

上一篇如何用实用脚本监控服务器状态？

下一篇哪些实用脚本适合Linux系统管理员？

相关文章

抱歉，评论功能暂时关闭!