第17讲：邮件自动化 Skill 开发

第17讲：邮件自动化 Skill 开发

掌握邮件自动化技能，实现邮件自动发送、模板管理、批量投递等功能，让邮件处理效率大幅提升。

一、场景分析

1.1 用户痛点

邮件是商务沟通的主要方式，但重复性工作很多：

  • 批量发送繁琐：需要给多个客户发送相似内容的邮件，逐一发送效率低
  • 模板管理混乱：常用邮件模板散落在各处，查找使用不便
  • 附件处理耗时：每次发送都要手动添加附件，容易遗漏
  • 跟进提醒困难：需要定时跟进客户，但容易忘记
  • 发送记录难查：历史邮件分散，难以统一管理和统计

1.2 典型应用场景

场景需求描述Skill 价值
营销邮件向客户群发产品推广邮件批量个性化发送
会议通知自动发送会议邀请和提醒定时自动发送
报表投递定期发送数据报表给相关人员自动附件处理
客户跟进按设定时间自动跟进客户智能提醒机制
自动回复根据邮件内容自动回复智能应答处理

二、核心功能设计

2.1 Skill 功能架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
📧 邮件自动化助手
├── 邮件发送
│   ├── 单封发送
│   ├── 批量发送
│   ├── 定时发送
│   └── 抄送/密送
├── 模板管理
│   ├── 模板创建
│   ├── 变量替换
│   ├── 模板分类
│   └── 模板预览
├── 附件处理
│   ├── 自动附加
│   ├── 批量附加
│   ├── 压缩打包
│   └── 云存储链接
├── 收件管理
│   ├── 邮件读取
│   ├── 自动分类
│   ├── 智能回复
│   └── 跟进提醒
└── 统计分析
    ├── 发送记录
    ├── 送达统计
    ├── 打开追踪
    └── 效果分析

2.2 技术选型

邮件处理的核心技术栈：

功能技术方案说明
邮件发送smtplib / yagmailPython 标准库和第三方库
邮件接收imaplib / poplib读取收件箱邮件
邮件解析email / mail-parser解析邮件内容
定时任务schedule / APScheduler定时发送邮件
邮件服务SendGrid / Mailgun专业邮件发送服务

三、技术实现

3.1 Coze 平台实现

3.1.1 基础发送代码

使用 smtplib：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import os

def send_email(smtp_config, to_addrs, subject, body, 
               attachments=None, cc_addrs=None):
    """
    发送邮件
    
    Args:
        smtp_config: SMTP 配置
            {
                'host': 'smtp.example.com',
                'port': 587,
                'user': 'user@example.com',
                'password': 'password',
                'use_tls': True
            }
        to_addrs: 收件人地址列表
        subject: 邮件主题
        body: 邮件正文（支持 HTML）
        attachments: 附件路径列表
        cc_addrs: 抄送地址列表
    """
    # 创建邮件对象
    msg = MIMEMultipart()
    msg['From'] = smtp_config['user']
    msg['To'] = ', '.join(to_addrs)
    msg['Subject'] = subject
    
    if cc_addrs:
        msg['Cc'] = ', '.join(cc_addrs)
    
    # 添加正文
    msg.attach(MIMEText(body, 'html', 'utf-8'))
    
    # 添加附件
    if attachments:
        for file_path in attachments:
            if os.path.exists(file_path):
                with open(file_path, 'rb') as f:
                    attachment = MIMEBase('application', 'octet-stream')
                    attachment.set_payload(f.read())
                
                encoders.encode_base64(attachment)
                filename = os.path.basename(file_path)
                attachment.add_header(
                    'Content-Disposition',
                    f'attachment; filename="{filename}"'
                )
                msg.attach(attachment)
    
    # 发送邮件
    with smtplib.SMTP(smtp_config['host'], smtp_config['port']) as server:
        if smtp_config.get('use_tls', True):
            server.starttls()
        
        server.login(smtp_config['user'], smtp_config['password'])
        
        all_recipients = to_addrs + (cc_addrs or [])
        server.sendmail(smtp_config['user'], all_recipients, msg.as_string())
    
    return True

使用 yagmail（更简单）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import yagmail

def send_email_simple(user, password, to, subject, body, attachments=None):
    """
    使用 yagmail 简化发送
    
    Args:
        user: 发件人邮箱
        password: 邮箱密码或授权码
        to: 收件人地址
        subject: 邮件主题
        body: 邮件正文
        attachments: 附件路径列表
    """
    yag = yagmail.SMTP(user, password)
    
    yag.send(
        to=to,
        subject=subject,
        contents=body,
        attachments=attachments
    )
    
    return True

3.1.2 模板邮件代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
from jinja2 import Template

def render_email_template(template_str, variables):
    """
    渲染邮件模板
    
    Args:
        template_str: 模板字符串
        variables: 变量字典
    
    Returns:
        渲染后的邮件内容
    """
    template = Template(template_str)
    return template.render(**variables)

# 示例模板
EMAIL_TEMPLATES = {
    'welcome': """
    <html>
    <body>
        <h2>欢迎 {{ name }}！</h2>
        <p>您好，感谢您注册我们的服务。</p>
        <p>您的账户信息：</p>
        <ul>
            <li>用户名：{{ username }}</li>
            <li>注册时间：{{ register_time }}</li>
        </ul>
        <p>如有任何问题，请随时联系我们。</p>
        <p>祝好！</p>
    </body>
    </html>
    """,
    
    'meeting_invite': """
    <html>
    <body>
        <h2>会议邀请</h2>
        <p>您好 {{ name }}，</p>
        <p>诚邀您参加以下会议：</p>
        <table border="1" cellpadding="10">
            <tr><td>会议主题</td><td>{{ meeting_title }}</td></tr>
            <tr><td>时间</td><td>{{ meeting_time }}</td></tr>
            <tr><td>地点</td><td>{{ meeting_location }}</td></tr>
            <tr><td>主持人</td><td>{{ host }}</td></tr>
        </table>
        <p>请确认是否能参加。</p>
    </body>
    </html>
    """,
    
    'report_delivery': """
    <html>
    <body>
        <h2>{{ report_name }}</h2>
        <p>您好 {{ name }}，</p>
        <p>{{ report_period }} 的数据报表已生成，请查收附件。</p>
        <p>报表概要：</p>
        <ul>
            {% for item in summary %}
            <li>{{ item }}</li>
            {% endfor %}
        </ul>
        <p>如有疑问，请联系数据分析团队。</p>
    </body>
    </html>
    """
}

def send_template_email(smtp_config, template_name, to, variables, 
                        attachments=None):
    """
    发送模板邮件
    
    Args:
        smtp_config: SMTP 配置
        template_name: 模板名称
        to: 收件人地址
        variables: 模板变量
        attachments: 附件列表
    """
    template = EMAIL_TEMPLATES.get(template_name)
    if not template:
        raise ValueError(f"模板 {template_name} 不存在")
    
    # 渲染模板
    body = render_email_template(template, variables)
    
    # 生成主题
    subject_map = {
        'welcome': '欢迎加入！',
        'meeting_invite': f"会议邀请：{variables.get('meeting_title', '')}",
        'report_delivery': f"报表：{variables.get('report_name', '')}"
    }
    subject = subject_map.get(template_name, '邮件通知')
    
    # 发送邮件
    return send_email(smtp_config, [to], subject, body, attachments)

3.1.3 批量发送代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def batch_send_emails(smtp_config, recipient_list, template_name, 
                      delay=1, progress_callback=None):
    """
    批量发送邮件
    
    Args:
        smtp_config: SMTP 配置
        recipient_list: 收件人列表
            [
                {
                    'email': 'user@example.com',
                    'name': '用户名',
                    'variables': {...}
                }
            ]
        template_name: 模板名称
        delay: 每封邮件间隔（秒）
        progress_callback: 进度回调函数
    """
    import time
    
    results = []
    total = len(recipient_list)
    
    for i, recipient in enumerate(recipient_list):
        try:
            variables = recipient.get('variables', {})
            variables['name'] = recipient.get('name', '')
            
            send_template_email(
                smtp_config,
                template_name,
                recipient['email'],
                variables
            )
            
            results.append({
                'email': recipient['email'],
                'status': 'success',
                'message': '发送成功'
            })
            
        except Exception as e:
            results.append({
                'email': recipient['email'],
                'status': 'failed',
                'message': str(e)
            })
        
        # 进度回调
        if progress_callback:
            progress_callback(i + 1, total)
        
        # 延迟，避免发送过快
        if delay > 0 and i < total - 1:
            time.sleep(delay)
    
    return results

3.1.4 定时发送代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import schedule
import time
from datetime import datetime

def schedule_email(smtp_config, send_time, to, subject, body, attachments=None):
    """
    定时发送邮件
    
    Args:
        smtp_config: SMTP 配置
        send_time: 发送时间（datetime 对象）
        to: 收件人地址
        subject: 邮件主题
        body: 邮件正文
        attachments: 附件列表
    """
    def job():
        send_email(smtp_config, [to], subject, body, attachments)
        print(f"邮件已发送给 {to} 于 {datetime.now()}")
    
    # 设置定时任务
    schedule_time = send_time.strftime("%H:%M")
    schedule.every().day.at(schedule_time).do(job)
    
    print(f"已设置定时任务，将在 {send_time} 发送邮件")
    
    # 运行调度器（需要在单独线程或进程中运行）
    # while True:
    #     schedule.run_pending()
    #     time.sleep(60)

def schedule_recurring_email(smtp_config, schedule_config, to, subject, body):
    """
    设置周期性邮件
    
    Args:
        smtp_config: SMTP 配置
        schedule_config: 调度配置
            {
                'frequency': 'daily' / 'weekly' / 'monthly',
                'day': 'monday' / 'tuesday' / ... (weekly 时),
                'date': 1-31 (monthly 时),
                'time': '09:00'
            }
    """
    def job():
        send_email(smtp_config, [to], subject, body)
    
    freq = schedule_config['frequency']
    send_time = schedule_config['time']
    
    if freq == 'daily':
        schedule.every().day.at(send_time).do(job)
    
    elif freq == 'weekly':
        day = schedule_config.get('day', 'monday')
        getattr(schedule.every(), day).at(send_time).do(job)
    
    elif freq == 'monthly':
        # schedule 库不直接支持每月，需要自定义
        def monthly_job():
            today = datetime.now()
            if today.day == schedule_config.get('date', 1):
                job()
        
        schedule.every().day.at(send_time).do(monthly_job)

3.2 OpenClaw 平台实现

OpenClaw 的邮件 Skill 示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from openclaw import Skill, Tool
import yagmail

class EmailSkill(Skill):
    name = "邮件自动化助手"
    description = "自动化发送和管理邮件"
    
    def __init__(self, smtp_config):
        self.smtp_config = smtp_config
        self.yag = yagmail.SMTP(smtp_config['user'], smtp_config['password'])
    
    @Tool
    def send(self, to: str, subject: str, body: str, attachments: list = None) -> str:
        """发送单封邮件"""
        self.yag.send(to=to, subject=subject, contents=body, attachments=attachments)
        return f"邮件已发送至 {to}"
    
    @Tool
    def send_template(self, to: str, template: str, variables: dict, 
                      attachments: list = None) -> str:
        """发送模板邮件"""
        from jinja2 import Template
        
        t = Template(template)
        body = t.render(**variables)
        
        return self.send(to, variables.get('subject', '邮件通知'), body, attachments)
    
    @Tool
    def batch_send(self, recipients: list, template: str, 
                   variables_list: list, delay: int = 1) -> list:
        """批量发送邮件"""
        results = []
        for recipient, vars in zip(recipients, variables_list):
            try:
                result = self.send_template(recipient, template, vars)
                results.append({'email': recipient, 'status': 'success'})
            except Exception as e:
                results.append({'email': recipient, 'status': 'failed', 'error': str(e)})
            time.sleep(delay)
        return results

四、Prompt 设计

4.1 系统 Prompt

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
你是邮件自动化助手，专门帮助用户高效处理邮件发送任务。

你可以执行以下操作：
1. 单封发送：发送单封邮件，支持 HTML 格式和附件
2. 批量发送：批量发送个性化邮件
3. 模板邮件：使用预设模板发送邮件
4. 定时发送：设置邮件定时发送
5. 邮件管理：查看发送记录和统计

工作流程：
1. 理解用户的邮件发送需求
2. 收集必要信息（收件人、主题、内容、附件等）
3. 确认邮件内容
4. 执行发送
5. 返回发送结果

注意事项：
- 批量发送时注意发送频率，避免被判定为垃圾邮件
- 重要邮件建议添加已读回执
- 敏感信息注意加密保护
- 遵守邮件发送规范和法律法规

4.2 意图识别示例

用户输入识别意图提取参数
“给张三发封邮件”单封发送收件人、主题、内容
“批量发送会议邀请”批量发送收件人列表、模板
“明天上午9点发提醒邮件”定时发送时间、收件人、内容
“用欢迎模板给新用户发邮件”模板邮件模板名称、收件人、变量
“查看邮件发送记录”记录查询查询条件

五、实战案例

5.1 案例一：营销邮件群发

场景：市场部门需要向客户群发新品推广邮件。

解决方案：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def send_marketing_campaign(smtp_config, customer_data, product_info):
    """
    发送营销邮件活动
    
    Args:
        smtp_config: SMTP 配置
        customer_data: 客户数据列表
        product_info: 产品信息
    """
    template = """
    <html>
    <body style="font-family: Arial, sans-serif;">
        <h2 style="color: #333;">🎉 新品上市：{{ product_name }}</h2>
        <p>尊敬的 {{ name }}，</p>
        <p>我们很高兴向您推荐我们的最新产品！</p>
        <div style="border: 1px solid #ddd; padding: 15px; margin: 15px 0;">
            <h3>{{ product_name }}</h3>
            <p>{{ product_description }}</p>
            <p style="color: #e74c3c; font-size: 18px;">
                特惠价：¥{{ price }} 
                <span style="text-decoration: line-through; color: #999;">¥{{ original_price }}</span>
            </p>
        </div>
        <p>限时优惠，先到先得！</p>
        <p><a href="{{ link }}" style="background: #3498db; color: white; padding: 10px 20px; text-decoration: none; border-radius: 5px;">立即购买</a></p>
        <p>如您不感兴趣，<a href="{{ unsubscribe_link }}">点击退订</a></p>
    </body>
    </html>
    """
    
    recipient_list = []
    for customer in customer_data:
        recipient_list.append({
            'email': customer['email'],
            'name': customer['name'],
            'variables': {
                'name': customer['name'],
                'product_name': product_info['name'],
                'product_description': product_info['description'],
                'price': product_info['price'],
                'original_price': product_info['original_price'],
                'link': product_info['purchase_link'],
                'unsubscribe_link': f"https://example.com/unsubscribe?email={customer['email']}"
            }
        })
    
    # 批量发送，每5秒一封
    results = batch_send_emails(
        smtp_config,
        recipient_list,
        'marketing',
        delay=5
    )
    
    # 统计结果
    success_count = sum(1 for r in results if r['status'] == 'success')
    print(f"发送完成：成功 {success_count}/{len(results)}")
    
    return results

5.2 案例二：自动报表投递

场景：需要定期向管理层发送数据报表。

解决方案：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def schedule_report_delivery(smtp_config, report_config):
    """
    设置自动报表投递
    
    Args:
        smtp_config: SMTP 配置
        report_config: 报表配置
            {
                'recipients': ['manager@example.com'],
                'report_generator': generate_weekly_report,  # 报表生成函数
                'schedule': {
                    'frequency': 'weekly',
                    'day': 'monday',
                    'time': '09:00'
                }
            }
    """
    import schedule
    
    def job():
        # 生成报表
        report_path = report_config['report_generator']()
        
        # 获取报表信息
        report_name = os.path.basename(report_path)
        report_date = datetime.now().strftime("%Y年%m月%d日")
        
        # 发送邮件
        for recipient in report_config['recipients']:
            send_template_email(
                smtp_config,
                'report_delivery',
                recipient,
                {
                    'name': '领导',
                    'report_name': report_name,
                    'report_period': report_date,
                    'summary': [
                        '本周销售额同比增长 15%',
                        '新增客户 23 家',
                        '订单转化率提升 3%'
                    ]
                },
                attachments=[report_path]
            )
        
        print(f"报表已发送：{report_date}")
    
    # 设置定时任务
    schedule_config = report_config['schedule']
    day = schedule_config['day']
    time = schedule_config['time']
    
    getattr(schedule.every(), day).at(time).do(job)
    
    print(f"已设置定时报表投递：每周{day} {time}")

六、实战练习

练习 1：欢迎邮件发送

创建一个 Skill，实现以下功能：

  1. 使用欢迎邮件模板
  2. 接收新用户信息（姓名、邮箱等）
  3. 个性化填充模板变量
  4. 发送欢迎邮件

练习 2：会议邀请批量发送

创建一个 Skill，实现以下功能：

  1. 读取参会人员列表
  2. 使用会议邀请模板
  3. 批量发送个性化邀请邮件
  4. 记录发送状态

练习 3：定时提醒邮件

创建一个 Skill，实现以下功能：

  1. 设置提醒任务（时间、收件人、内容）
  2. 定时检查并发送提醒邮件
  3. 支持周期性提醒（每日/每周/每月）
  4. 管理提醒任务列表

七、常见问题

Q1：邮件被判定为垃圾邮件怎么办？

解决方案：

  • 使用专业的邮件发送服务（SendGrid、Mailgun）
  • 控制发送频率，避免短时间内大量发送
  • 添加退订链接
  • 使用 SPF、DKIM、DMARC 等邮件验证
  • 保持邮件内容质量，避免敏感词汇

Q2：附件太大无法发送？

解决方案：

  • 压缩附件
  • 使用云存储链接代替附件
  • 分割大文件
  • 使用支持大附件的邮件服务

Q3：如何确保邮件送达？

解决方案：

  • 使用可靠的 SMTP 服务
  • 实现重试机制
  • 添加送达回执
  • 监控退信和投诉率

八、下节预告

下一讲我们将进入 进阶开发技巧 章节，学习：

  • 多平台 Skill 适配
  • 数据持久化
  • 安全与权限
  • 性能优化

加入学习群

👉 加入AI编程学习交流群

点击加入

本讲是《AI Skills 从入门到实践》系列课程的第17讲。

进群

嗨，扫码 进群

微信

扫一扫，分享到微信

微信分享二维码
本站总访问量次

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置：

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

编程博主，全网粉丝30w+，开源项目：python-office，我的微信：python-office