第 6 讲：Excel 自动化实战 - 批量处理与自动邮件

大家好，我是正在实战各种 AI 项目的程序员晚枫。

😫 场景导入：批量处理的噩梦

场景1：月底汇总各部门数据

1
2
3
4
5
6
7
8
9
10
老板：把各部门的月度数据汇总一下

实际情况：
- 收到30个部门的Excel文件
- 格式不完全统一
- 需要逐个打开、复制、粘贴
- 还要核对数据是否完整
- 一不小心就出错

花费时间：3-4小时

场景2：每周发周报

1
2
3
4
5
6
7
8
每周五下午：
- 打开销售系统导出数据
- 复制到Excel
- 做图表
- 写邮件
- 发送给10+领导

花费时间：1小时

如果自动化呢？

1
2
3
4
5
- 批量合并Excel：3秒
- 自动生成周报：10秒
- 自动发送邮件：自动

花费时间：几乎为0

📊 一、批量处理 Excel 文件

场景1：合并多个Excel文件

Python实现：

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
# 安装：pip install python-office pandas openpyxl

import office
import pandas as pd
import os

# 方法1：使用python-office一键合并
office.excel.merge2excel(
    dir_path='./各部门数据/',  # 存放Excel的文件夹
    output_file='汇总表.xlsx'  # 输出文件名
)
print('合并完成！')

# 方法2：使用pandas自定义合并
def merge_excel_files(folder_path, output_file):
    """合并文件夹中所有Excel文件"""
    all_data = []
    
    # 遍历文件夹
    for filename in os.listdir(folder_path):
        if filename.endswith('.xlsx') or filename.endswith('.xls'):
            file_path = os.path.join(folder_path, filename)
            
            # 读取数据
            df = pd.read_excel(file_path)
            
            # 添加来源列
            df['来源文件'] = filename
            
            all_data.append(df)
            print(f'已读取：{filename}')
    
    # 合并所有数据
    merged_df = pd.concat(all_data, ignore_index=True)
    
    # 保存
    merged_df.to_excel(output_file, index=False)
    print(f'合并完成！共{len(merged_df)}行数据')

# 使用
merge_excel_files('./各部门数据/', '汇总表.xlsx')

ChatExcel操作：

1
2
3
4
5
1. 打开chatexcel.com
2. 输入指令："合并多个Excel文件"
3. 上传所有文件
4. 等待处理完成
5. 下载合并后的文件

场景2：批量拆分Excel

按条件拆分成多个文件：

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
import pandas as pd

def split_excel_by_column(input_file, split_column, output_folder):
    """
    按指定列拆分Excel
    
    参数：
    - input_file: 输入文件
    - split_column: 按哪一列拆分
    - output_folder: 输出文件夹
    """
    import os
    os.makedirs(output_folder, exist_ok=True)
    
    # 读取数据
    df = pd.read_excel(input_file)
    
    # 按列分组
    grouped = df.groupby(split_column)
    
    # 拆分保存
    for name, group in grouped:
        output_file = os.path.join(output_folder, f'{name}.xlsx')
        group.to_excel(output_file, index=False)
        print(f'已生成：{output_file}')

# 示例：按部门拆分
split_excel_by_column(
    '总数据.xlsx',
    '部门',
    './拆分后文件/'
)

场景3：批量格式化

统一所有Excel的格式：

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
import pandas as pd
import openpyxl
from openpyxl.styles import Font, Alignment, PatternFill
import os

def batch_format_excel(folder_path):
    """批量格式化Excel文件"""
    
    for filename in os.listdir(folder_path):
        if not (filename.endswith('.xlsx') or filename.endswith('.xls')):
            continue
            
        file_path = os.path.join(folder_path, filename)
        
        # 打开工作簿
        wb = openpyxl.load_workbook(file_path)
        
        for sheet_name in wb.sheetnames:
            sheet = wb[sheet_name]
            
            # 设置表头样式
            for cell in sheet[1]:
                cell.font = Font(bold=True, name='微软雅黑', size=11)
                cell.fill = PatternFill(start_color='4472C4', 
                                       end_color='4472C4', 
                                       fill_type='solid')
                cell.font = Font(bold=True, color='FFFFFF', 
                               name='微软雅黑', size=11)
                cell.alignment = Alignment(horizontal='center', 
                                          vertical='center')
            
            # 设置数据区域样式
            for row in sheet.iter_rows(min_row=2):
                for cell in row:
                    cell.font = Font(name='微软雅黑', size=11)
                    cell.alignment = Alignment(horizontal='center', 
                                              vertical='center')
            
            # 自动调整列宽
            for column in sheet.columns:
                max_length = 0
                column_letter = column[0].column_letter
                
                for cell in column:
                    try:
                        if len(str(cell.value)) > max_length:
                            max_length = len(str(cell.value))
                    except:
                        pass
                
                adjusted_width = min(max_length + 2, 50)
                sheet.column_dimensions[column_letter].width = adjusted_width
        
        # 保存
        wb.save(file_path)
        print(f'格式化完成：{filename}')

# 使用
batch_format_excel('./待格式化文件/')

场景4：批量转换格式

Excel批量转PDF、CSV等：

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
import office
import os

# 批量Excel转PDF
def batch_excel_to_pdf(folder_path, output_folder):
    """批量转换Excel为PDF"""
    os.makedirs(output_folder, exist_ok=True)
    
    for filename in os.listdir(folder_path):
        if filename.endswith('.xlsx') or filename.endswith('.xls'):
            input_path = os.path.join(folder_path, filename)
            output_path = os.path.join(
                output_folder, 
                filename.replace('.xlsx', '.pdf').replace('.xls', '.pdf')
            )
            
            office.excel.excel2pdf(input_path, output_path)
            print(f'已转换：{filename}')

# 批量Excel转CSV
def batch_excel_to_csv(folder_path, output_folder):
    """批量转换Excel为CSV"""
    import pandas as pd
    os.makedirs(output_folder, exist_ok=True)
    
    for filename in os.listdir(folder_path):
        if filename.endswith('.xlsx') or filename.endswith('.xls'):
            input_path = os.path.join(folder_path, filename)
            output_path = os.path.join(
                output_folder, 
                filename.replace('.xlsx', '.csv').replace('.xls', '.csv')
            )
            
            df = pd.read_excel(input_path)
            df.to_csv(output_path, index=False, encoding='utf-8-sig')
            print(f'已转换：{filename}')

# 使用
batch_excel_to_pdf('./Excel文件/', './PDF文件/')
batch_excel_to_csv('./Excel文件/', './CSV文件/')

🌐 二、自动抓取网络数据

场景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
import pandas as pd
import requests
from bs4 import BeautifulSoup

def scrape_table_from_url(url, output_file):
    """从网页抓取表格数据"""
    
    # 获取网页内容
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    response = requests.get(url, headers=headers)
    response.encoding = 'utf-8'
    
    # 解析HTML
    soup = BeautifulSoup(response.text, 'html.parser')
    
    # 找到所有表格
    tables = soup.find_all('table')
    
    # 提取第一个表格
    if tables:
        df = pd.read_html(str(tables[0]))[0]
        df.to_excel(output_file, index=False)
        print(f'数据已保存：{output_file}')
        return df
    else:
        print('未找到表格')

# 使用示例（抓取某个数据网站）
# scrape_table_from_url('https://example.com/data', '抓取数据.xlsx')

场景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
import schedule
import time
import pandas as pd
from datetime import datetime

def fetch_and_save_data():
    """抓取并保存数据"""
    # 模拟抓取数据
    data = {
        '时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        '汇率': 7.23,  # 实际应从API获取
        '股价': 150.5
    }
    
    # 读取历史数据
    try:
        df = pd.read_excel('历史数据.xlsx')
    except:
        df = pd.DataFrame()
    
    # 添加新数据
    new_row = pd.DataFrame([data])
    df = pd.concat([df, new_row], ignore_index=True)
    
    # 保存
    df.to_excel('历史数据.xlsx', index=False)
    print(f'数据已更新：{data}')

# 定时任务：每小时执行一次
schedule.every().hour.do(fetch_and_save_data)

# 运行
while True:
    schedule.run_pending()
    time.sleep(60)

📧 三、自动发送邮件报表

场景1：发送带附件的邮件

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

# 使用python-office发送邮件
office.email.send_email(
    to_email='boss@company.com',  # 收件人
    subject='销售日报-2026-04-08',  # 主题
    content='''领导好：

附件是今日销售日报，请查收。

关键数据：
- 今日销售额：50万元
- 环比增长：+5%

详情见附件。

此致
敬礼！''',
    attachment='销售日报.xlsx'  # 附件
)

print('邮件发送成功！')

场景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
import pandas as pd
import office

def send_personalized_emails(data_file):
    """批量发送个性化邮件"""
    
    # 读取收件人数据
    df = pd.read_excel(data_file)
    
    for index, row in df.iterrows():
        # 生成个性化内容
        content = f"""
{row['姓名']}，你好！

根据你的业绩数据，本月情况如下：
- 销售额：{row['销售额']}万元
- 目标完成率：{row['完成率']}%
- 排名：第{row['排名']}名

{'恭喜你超额完成目标！' if row['完成率'] >= 100 else '下月请继续努力！'}

此致
敬礼！
"""
        
        # 发送邮件
        office.email.send_email(
            to_email=row['邮箱'],
            subject=f'业绩通知-{row["姓名"]}',
            content=content
        )
        
        print(f'已发送给：{row["姓名"]}')

# 使用
send_personalized_emails('员工数据.xlsx')

场景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
import schedule
import time
import pandas as pd
import office
from datetime import datetime

def send_weekly_report():
    """发送周报"""
    
    # 生成报告
    df = pd.read_excel('销售数据.xlsx')
    
    # 计算关键指标
    total_sales = df['销售额'].sum()
    order_count = len(df)
    avg_order = total_sales / order_count
    
    # 邮件内容
    content = f"""
领导好：

本周销售数据汇报如下：

📊 核心指标
- 总销售额：{total_sales/10000:.1f}万元
- 订单数：{order_count}单
- 平均客单价：{avg_order:.0f}元

📅 时间：{datetime.now().strftime('%Y-%m-%d')}

详情见附件。

此致
敬礼！
"""
    
    # 发送邮件
    office.email.send_email(
        to_email='boss@company.com',
        subject=f'销售周报-{datetime.now().strftime("%Y年第%W周")}',
        content=content,
        attachment='销售数据.xlsx'
    )
    
    print(f'周报已发送：{datetime.now()}')

# 定时：每周一早上9点发送
schedule.every().monday.at("09:00").do(send_weekly_report)

# 运行
while True:
    schedule.run_pending()
    time.sleep(60)

🔄 四、完整自动化流程

案例：自动化日报系统

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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
import pandas as pd
import office
import schedule
import time
from datetime import datetime
import os

class DailyReportSystem:
    """自动化日报系统"""
    
    def __init__(self, data_source, recipients):
        self.data_source = data_source
        self.recipients = recipients
    
    def collect_data(self):
        """收集数据"""
        # 合并各来源数据
        office.excel.merge2excel(
            dir_path=self.data_source,
            output_file='日报_原始数据.xlsx'
        )
        
        # 读取并处理
        df = pd.read_excel('日报_原始数据.xlsx')
        return df
    
    def analyze_data(self, df):
        """分析数据"""
        analysis = {
            '日期': datetime.now().strftime('%Y-%m-%d'),
            '总销售额': df['销售额'].sum(),
            '订单数': len(df),
            '客单价': df['销售额'].mean(),
            'top_products': df.groupby('产品')['销售额'].sum().nlargest(5)
        }
        return analysis
    
    def generate_report(self, analysis):
        """生成报告"""
        report = f"""
# 销售日报

日期：{analysis['日期']}

## 核心指标

| 指标 | 数值 |
|------|------|
| 总销售额 | {analysis['总销售额']/10000:.1f}万元 |
| 订单数 | {analysis['订单数']}单 |
| 客单价 | {analysis['客单价']:.0f}元 |

## Top5产品

{analysis['top_products'].to_markdown()}

---
生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
        
        # 保存报告
        with open('日报.md', 'w', encoding='utf-8') as f:
            f.write(report)
        
        return report
    
    def send_email(self, report):
        """发送邮件"""
        for recipient in self.recipients:
            office.email.send_email(
                to_email=recipient,
                subject=f'销售日报-{datetime.now().strftime("%Y-%m-%d")}',
                content=report,
                attachment='日报_原始数据.xlsx'
            )
            print(f'已发送给：{recipient}')
    
    def run(self):
        """运行完整流程"""
        print('开始生成日报...')
        
        # 1. 收集数据
        df = self.collect_data()
        print('数据收集完成')
        
        # 2. 分析数据
        analysis = self.analyze_data(df)
        print('数据分析完成')
        
        # 3. 生成报告
        report = self.generate_report(analysis)
        print('报告生成完成')
        
        # 4. 发送邮件
        self.send_email(report)
        print('日报发送完成！')

# 使用
system = DailyReportSystem(
    data_source='./各区域数据/',
    recipients=['boss@company.com', 'manager@company.com']
)

# 手动运行
# system.run()

# 定时运行：每天下午5点
schedule.every().day.at("17:00").do(system.run)

while True:
    schedule.run_pending()
    time.sleep(60)

⚠️ 五、避坑指南

坑1：没有错误处理

错误示例：

1
2
3
4
# 没有错误处理，一旦出错整个流程中断
for file in files:
    df = pd.read_excel(file)
    process(df)

正确做法：

1
2
3
4
5
6
7
8
9
10
11
import logging

logging.basicConfig(filename='error.log', level=logging.ERROR)

for file in files:
    try:
        df = pd.read_excel(file)
        process(df)
    except Exception as e:
        logging.error(f'处理{file}时出错：{str(e)}')
        continue  # 跳过错误文件，继续处理

坑2：没有日志记录

问题： 不知道处理了多少、失败了几次

正确做法：

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

logging.basicConfig(
    filename='process.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

total = 0
success = 0
failed = 0

for file in files:
    total += 1
    try:
        process(file)
        success += 1
        logging.info(f'成功处理：{file}')
    except:
        failed += 1
        logging.error(f'处理失败：{file}')

logging.info(f'总计：{total}，成功：{success}，失败：{failed}')

坑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
import schedule
import time
import requests

def job():
    try:
        # 执行任务
        process()
        
        # 发送成功通知
        requests.post(
            'https://api.example.com/notify',
            json={'status': 'success', 'message': '任务完成'}
        )
    except Exception as e:
        # 发送失败通知
        requests.post(
            'https://api.example.com/notify',
            json={'status': 'failed', 'message': str(e)}
        )

schedule.every().day.at("09:00").do(job)

while True:
    schedule.run_pending()
    time.sleep(60)

💡 六、本讲作业

作业1：批量处理实践

准备10个Excel文件，完成以下任务：

  1. 批量合并成一个文件
  2. 按某列拆分成多个文件
  3. 统一格式化

作业2：定时任务

编写定时任务：

  1. 每天9点自动汇总前一天数据
  2. 生成报表
  3. 发送邮件

作业3：完整自动化

设计并实现一个完整的自动化流程：

  1. 数据采集
  2. 数据处理
  3. 报告生成
  4. 自动发送

🎯 本讲总结

核心要点

  1. 批量处理：合并、拆分、格式化、转换
  2. 数据抓取：网页表格、定时采集
  3. 自动邮件：单发、群发、定时发送
  4. 完整流程：采集→处理→报告→发送

工具清单

1
2
3
4
5
6
7
8
9
✅ Python库：
- python-office：一键办公自动化
- pandas：数据处理
- openpyxl：Excel操作
- schedule：定时任务

✅ 在线工具：
- 集简云：工作流自动化
- ChatExcel：在线Excel处理

🔗 课程导航

← 上一讲：AI 生成数据报表 | 下一讲：AI 写 Word 文档 →

💬 加入学习交流群

👉 点击加入交流群

💬 联系我

平台账号/链接
微信扫码加好友
微博@程序员晚枫
知乎@程序员晚枫
抖音@程序员晚枫
小红书@程序员晚枫
B 站Python 自动化办公社区

主营业务：AI 编程培训、企业内训、技术咨询

科技不高冷，AI很好用。

下一讲，我们学习如何用AI写Word文档！

🎓 AI 编程实战课程

想系统学习 AI 编程？程序员晚枫的 AI 编程实战课 帮你从零上手！