编程进阶网
杨充
2024-12-19
目录

办公自动化实战

# 第 7 章 办公自动化实战

# 目录介绍

# 7.1 场景引入

💬 场景：你是运营部门的数据支撑。每周五下午你都重复同一件事——打开 12 个 Excel 文件，复制粘贴到汇总表，画两张图，写 Word 周报，然后邮件发给老板和全部门。

每个周五花 3 小时。一年就是 150 小时——将近 4 个工作周浪费在机械操作上。

你需要一个 Python 自动化系统：每周五自动拉数据 → 生成 Excel 汇总 → 生成 Word 报告 → 邮件发出——你连电脑都不用开。

本章按"组件 → 组装 → 定时"三步走：

阶段 内容 核心库
① Excel 自动化 读写、样式、公式、批量 openpyxl + pandas
② 办公脚本 文件操作、PDF、Word、邮件 pypdf / python-docx / smtplib
③ 定时调度 schedule / APScheduler / cron schedule / APScheduler
④ 端到端实战 周报自动化系统 全部串联

📦 安装准备：pip install openpyxl pandas pypdf python-docx schedule APScheduler

# 7.2 Excel 处理

# 7.2.1 读写 Excel

openpyxl 是纯 Python 的 Excel 读写库——不需要安装 Office：

from openpyxl import Workbook, load_workbook

# ===== 写 Excel =====
wb = Workbook()                        # 创建新工作簿
ws = wb.active                         # 获取默认工作表
ws.title = "销售数据"                  # 重命名 Sheet

# 写入数据——按单元格
ws["A1"] = "月份"
ws["B1"] = "销售额"
ws["C1"] = "利润"

# 写入数据——按行列（更高效）
data = [
    ["1月", 12000, 3600],
    ["2月", 15000, 4500],
    ["3月", 13500, 4050],
    ["4月", 18000, 5400],
    ["5月", 16000, 4800],
    ["6月", 21000, 6300],
]
for row in data:
    ws.append(row)                     # 追加一行

# 插入公式
ws["D1"] = "利润率"
for r in range(2, len(data) + 2):
    ws[f"D{r}"] = f"=C{r}/B{r}"        # 写入 Excel 公式

wb.save("sales_report.xlsx")
print("✅ 已保存：sales_report.xlsx")

# ===== 读 Excel =====
wb = load_workbook("sales_report.xlsx")
ws = wb["销售数据"]

print(f"工作表名：{ws.title}")
print(f"数据范围：{ws.dimensions}")     # A1:D7

# 遍历所有行
for row in ws.iter_rows(min_row=1, max_row=ws.max_row, values_only=True):
    print(row)
# ('月份', '销售额', '利润', '利润率')
# ('1月', 12000, 3600, '=C2/B2')
# ...

# 取特定单元格
print(f"A1 = {ws['A1'].value}")        # 月份
print(f"B2 = {ws.cell(2, 2).value}")   # 12000
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

🔑 iter_rows 参数速查：

参数 作用
min_row / max_row 行范围
min_col / max_col 列范围
values_only=True 只返回单元格值（不给 Cell 对象）

# 7.2.2 样式公式

from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side, numbers
from openpyxl.utils import get_column_letter

wb = Workbook()
ws = wb.active
ws.title = "样式示例"

# ===== 表头样式 =====
headers = ["姓名", "语文", "数学", "英语", "总分", "平均分", "等级"]
for col, h in enumerate(headers, 1):
    cell = ws.cell(1, col, h)
    cell.font = Font(name="微软雅黑", size=12, bold=True, color="FFFFFF")       # 白色粗体
    cell.fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")  # 蓝色背景
    cell.alignment = Alignment(horizontal="center", vertical="center")           # 居中
    cell.border = Border(
        bottom=Side(style="thin", color="FFFFFF")
    )

# ===== 数据行 =====
data = [
    ("张三", 85, 92, 78, None, None, None),
    ("李四", 92, 88, 95, None, None, None),
    ("王五", 78, 85, 82, None, None, None),
    ("赵六", 65, 70, 68, None, None, None),
]

for r, (name, ch, ma, en, *_) in enumerate(data, 2):
    ws.cell(r, 1, name)
    ws.cell(r, 2, ch)
    ws.cell(r, 3, ma)
    ws.cell(r, 4, en)
    # 公式——总分
    ws.cell(r, 5).value = f"=SUM(B{r}:D{r})"
    # 公式——平均分（保留一位小数）
    ws.cell(r, 6).value = f"=ROUND(E{r}/3, 1)"
    # 公式——等级
    ws.cell(r, 7).value = (
        f'=IF(F{r}>=90,"A",IF(F{r}>=80,"B",IF(F{r}>=70,"C",IF(F{r}>=60,"D","F"))))'
    )

# ===== 条件格式——不及格红色 =====
from openpyxl.formatting.rule import CellIsRule
red_fill = PatternFill(start_color="FFCDD2", end_color="FFCDD2", fill_type="solid")
red_font = Font(color="B71C1C", bold=True)
ws.conditional_formatting.add(
    f"B2:D{1+len(data)}",
    CellIsRule(operator="lessThan", formula=["60"], fill=red_fill, font=red_font)
)

# ===== 自适应列宽 =====
for col in range(1, len(headers) + 1):
    ws.column_dimensions[get_column_letter(col)].width = 14

# ===== 冻结首行 =====
ws.freeze_panes = "A2"

wb.save("styled_report.xlsx")
print("✅ 已保存：styled_report.xlsx")
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

常用样式速查：

类别 类 示例
字体 Font Font(name="微软雅黑", size=12, bold=True, color="FF0000")
填充 PatternFill PatternFill(start_color="FFFF00", fill_type="solid")
对齐 Alignment Alignment(horizontal="center", vertical="center", wrap_text=True)
边框 Border + Side Border(bottom=Side(style="thin"))
数字格式 ws.cell(..).number_format "#,##0.00", "0.00%", "yyyy-mm-dd"

# 7.2.3 批量报表

真实场景——目录下有 12 个月的 Excel，需要合并统计：

"""
批量合并月度报表——真实场景最常用的脚本模式
"""
import os
from openpyxl import load_workbook, Workbook
from glob import glob

def merge_monthly_reports(input_dir: str, output_path: str):
    """合并目录下所有 .xlsx 文件到一张汇总表"""
    wb_out = Workbook()
    # ① 汇总 Sheet
    ws_summary = wb_out.active
    ws_summary.title = "汇总"
    ws_summary.append(["月份", "文件", "总销售额", "总利润", "行数"])

    all_data = []   # 收集所有数据——用于 raw data Sheet

    for filepath in sorted(glob(os.path.join(input_dir, "*.xlsx"))):
        wb = load_workbook(filepath, data_only=True)  # data_only=True 读公式结果
        ws = wb.active

        # 提取数据（假设每行是 [月份, 销售额, 利润]）
        month_name = os.path.splitext(os.path.basename(filepath))[0]
        total_sales, total_profit, row_count = 0, 0, 0

        for row in ws.iter_rows(min_row=2, values_only=True):
            if row[0] is None:
                continue
            sales = float(row[1]) if row[1] else 0
            profit = float(row[2]) if len(row) > 2 and row[2] else 0
            total_sales += sales
            total_profit += profit
            row_count += 1
            all_data.append([month_name] + list(row[:3]))

        ws_summary.append([month_name, filepath, total_sales, total_profit, row_count])
        print(f"  ✅ {month_name}：{row_count} 行，销售额 {total_sales:.0f}，利润 {total_profit:.0f}")
        wb.close()

    # ② 原始数据 Sheet——把所有月份的数据合并在一起
    ws_raw = wb_out.create_sheet("原始数据")
    ws_raw.append(["月份", "日期", "销售额", "利润"])
    for d in all_data:
        ws_raw.append(d)

    # ③ 公式——汇总行
    last_row = ws_summary.max_row + 1
    ws_summary.cell(last_row, 1, "合计")
    ws_summary.cell(last_row, 3).value = f"=SUM(C2:C{last_row-1})"
    ws_summary.cell(last_row, 4).value = f"=SUM(D2:D{last_row-1})"
    ws_summary.cell(last_row, 5).value = f"=SUM(E2:E{last_row-1})"

    # 样式：合计行加粗
    from openpyxl.styles import Font
    for col in range(1, 6):
        ws_summary.cell(last_row, col).font = Font(bold=True)

    wb_out.save(output_path)
    print(f"\n✅ 汇总完成：{output_path}（{len(all_data)} 条原始数据）")
    wb_out.close()


# 使用示例
# merge_monthly_reports("./monthly_data", "yearly_summary.xlsx")
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

# 7.2.4 pandas 读写

pandas 一行代码读出整个 Excel 到 DataFrame——比 openpyxl 快 10 倍（见 §2.3）：

import pandas as pd

# 读
df = pd.read_excel("sales_report.xlsx", sheet_name="销售数据")

# 写——多个 Sheet 一次搞定
with pd.ExcelWriter("batch_output.xlsx", engine="openpyxl") as writer:
    df.to_excel(writer, sheet_name="原始数据", index=False)
    df.groupby("月份")[["销售额", "利润"]].sum().to_excel(writer, sheet_name="月份汇总")

# 读的时候只取需要的列
df = pd.read_excel("huge.xlsx", usecols=["月份", "销售额"], nrows=1000)
1
2
3
4
5
6
7
8
9
10
11
12

🔑 openpyxl vs pandas 的选择：

场景 用 原因
数据量大、需要分析 pandas.read_excel() 快 + 直接进 DataFrame
需要精确控制样式/公式 openpyxl pandas 的样式能力有限
需要改已有 Excel 的样式 openpyxl.load_workbook() pandas 会覆盖样式
批量写入大量 DataFrame pd.ExcelWriter 性能最优

# 7.3 常用脚本集

# 7.3.1 批量重命名

import os
from pathlib import Path

def batch_rename(directory: str, prefix: str = "", suffix: str = "",
                 replace_old: str = "", replace_new: str = "",
                 dry_run: bool = True):
    """批量重命名文件。

    Args:
        directory: 目标目录
        prefix: 加前缀（如 "2025Q1_"）
        suffix: 加后缀
        replace_old / replace_new: 替换文件名中的文本
        dry_run: True=模拟（只打印不改），False=执行
    """
    dir_path = Path(directory)
    if not dir_path.exists():
        print(f"❌ 目录不存在：{directory}")
        return

    files = sorted(f for f in dir_path.iterdir() if f.is_file())
    renamed = 0

    for f in files:
        name, ext = f.stem, f.suffix
        new_name = name

        if replace_old:
            new_name = new_name.replace(replace_old, replace_new)
        new_name = f"{prefix}{new_name}{suffix}{ext}"

        new_path = f.parent / new_name

        if new_path == f:
            continue   # 没变化——跳过

        if dry_run:
            print(f"  [模拟] {f.name} → {new_name}")
        else:
            f.rename(new_path)
            print(f"  ✅ {f.name} → {new_name}")
        renamed += 1

    action = "模拟" if dry_run else "完成"
    print(f"\n{action}：{renamed} 个文件{'（没有实际修改）' if dry_run else ''}")


# 示例用法
# 1. 加前缀（先模拟）
# batch_rename("./invoices", prefix="2025Q2_", dry_run=True)
# 2. 替换文本——把所有"草稿"改成"终版"
# batch_rename("./reports", replace_old="草稿", replace_new="终版", dry_run=True)
# 3. 确认无误后执行
# batch_rename("./invoices", prefix="2025Q2_", dry_run=False)
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

# 7.3.2 PDF 合并拆分

from pypdf import PdfReader, PdfWriter

def merge_pdfs(input_paths: list[str], output_path: str):
    """合并多个 PDF 文件"""
    writer = PdfWriter()
    for path in input_paths:
        reader = PdfReader(path)
        for page in reader.pages:
            writer.add_page(page)
        print(f"  ✅ 已添加：{path}（{len(reader.pages)} 页）")

    with open(output_path, "wb") as f:
        writer.write(f)
    print(f"\n✅ 合并完成：{output_path}（共 {len(writer.pages)} 页）")


def split_pdf(input_path: str, output_dir: str, pages_per_file: int = 1):
    """拆分 PDF——每 N 页输出一个文件"""
    reader = PdfReader(input_path)
    total = len(reader.pages)
    os.makedirs(output_dir, exist_ok=True)

    for i in range(0, total, pages_per_file):
        writer = PdfWriter()
        end = min(i + pages_per_file, total)
        for j in range(i, end):
            writer.add_page(reader.pages[j])

        output_name = f"split_{i+1:04d}-{end:04d}.pdf"
        output_path = os.path.join(output_dir, output_name)
        with open(output_path, "wb") as f:
            writer.write(f)
        print(f"  ✅ {output_name}（页 {i+1}~{end}）")

    print(f"\n✅ 拆分完成：{total} 页 → {((total + pages_per_file - 1) // pages_per_file)} 个文件")


def rotate_pdf(input_path: str, output_path: str, rotation: int = 90):
    """旋转 PDF 所有页面"""
    reader = PdfReader(input_path)
    writer = PdfWriter()
    for page in reader.pages:
        page.rotate(rotation)
        writer.add_page(page)
    with open(output_path, "wb") as f:
        writer.write(f)
    print(f"✅ 旋转完成：{output_path}（旋转 {rotation}°）")
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

# 7.3.3 Word 文档生成

from docx import Document
from docx.shared import Inches, Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from datetime import datetime

def create_monthly_report():
    """生成一份格式规范的月度报告"""
    doc = Document()

    # ① 标题
    title = doc.add_heading("月度销售分析报告", level=0)
    title.alignment = WD_ALIGN_PARAGRAPH.CENTER

    # ② 元信息
    meta = doc.add_paragraph()
    meta.alignment = WD_ALIGN_PARAGRAPH.RIGHT
    meta.add_run(f"报告日期：{datetime.now().strftime('%Y-%m-%d')}").italic = True
    meta.add_run("\n").bold = False
    meta.add_run("编制：数据分析部").font.size = Pt(9)

    # ③ 一级标题 + 正文
    doc.add_heading("一、销售概况", level=1)
    p = doc.add_paragraph()
    p.add_run("本月总销售额为 ").font.size = Pt(12)
    run_highlight = p.add_run("¥215,000")
    run_highlight.font.size = Pt(14)
    run_highlight.bold = True
    run_highlight.font.color.rgb = RGBColor(0xE9, 0x1E, 0x63)
    p.add_run("，较上月增长 ").font.size = Pt(12)
    run_up = p.add_run("8.5%")
    run_up.font.size = Pt(14)
    run_up.bold = True
    run_up.font.color.rgb = RGBColor(0x4C, 0xAF, 0x50)

    # ④ 表格
    doc.add_heading("二、各品类数据", level=1)
    table = doc.add_table(rows=1, cols=4, style="Light Grid Accent 1")
    headers = ["品类", "销售额", "环比增长", "占比"]
    for i, h in enumerate(headers):
        table.rows[0].cells[i].text = h
        for p in table.rows[0].cells[i].paragraphs:
            p.runs[0].bold = True

    data = [
        ("电子", "¥82,000", "+12.3%", "38.1%"),
        ("服装", "¥45,000", "+5.8%", "20.9%"),
        ("食品", "¥38,000", "+3.2%", "17.7%"),
        ("家居", "¥29,000", "+8.1%", "13.5%"),
        ("运动", "¥21,000", "+15.6%", "9.8%"),
    ]
    for row_data in data:
        row = table.add_row()
        for i, val in enumerate(row_data):
            row.cells[i].text = val

    # ⑤ 结论
    doc.add_heading("三、结论与建议", level=1)
    conclusions = [
        "电子品类持续领跑，建议下半年加大库存备货。",
        "运动品类增速最快（+15.6%），可考虑增加营销投入。",
        "食品品类增速放缓，需关注供应链效率。",
    ]
    for c in conclusions:
        doc.add_paragraph(c, style="List Bullet")

    # ⑥ 保存
    doc.save("monthly_report.docx")
    print("✅ 已保存：monthly_report.docx")
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

# 7.3.4 邮件发送

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: dict,
               to_addrs: list[str],
               subject: str,
               body: str,
               attachments: list[str] = None,
               is_html: bool = False):
    """发送邮件——支持附件和 HTML。

    Args:
        smtp_config: {"host", "port", "user", "password"}
        to_addrs: 收件人列表
        subject: 邮件主题
        body: 邮件正文
        attachments: 附件路径列表（可选）
        is_html: 正文是否为 HTML
    """
    msg = MIMEMultipart()
    msg["From"] = smtp_config["user"]
    msg["To"] = ", ".join(to_addrs)
    msg["Subject"] = subject

    # 正文
    subtype = "html" if is_html else "plain"
    msg.attach(MIMEText(body, subtype, "utf-8"))

    # 附件
    if attachments:
        for filepath in attachments:
            if not os.path.exists(filepath):
                print(f"  ⚠️ 附件不存在，跳过：{filepath}")
                continue
            with open(filepath, "rb") as f:
                part = MIMEBase("application", "octet-stream")
                part.set_payload(f.read())
            encoders.encode_base64(part)
            filename = os.path.basename(filepath)
            part.add_header(
                "Content-Disposition",
                f'attachment; filename="{filename}"'
            )
            msg.attach(part)
            print(f"  📎 已附加：{filename}")

    # 发送
    try:
        with smtplib.SMTP_SSL(smtp_config["host"], smtp_config["port"]) as server:
            server.login(smtp_config["user"], smtp_config["password"])
            server.send_message(msg)
        print(f"✅ 邮件已发送至：{', '.join(to_addrs)}")
    except smtplib.SMTPException as e:
        print(f"❌ 邮件发送失败：{e}")


# 配置（实际使用前请替换为真实信息）
# ⚠️ Gmail/QQ邮箱需要开启 SMTP 服务 + 使用授权码（不是登录密码！）
SMTP_CONFIG = {
    "host": "smtp.qq.com",              # QQ邮箱 SMTP
    "port": 465,                         # SSL 端口
    "user": "your_email@qq.com",        # 你的邮箱
    "password": "your_auth_code",        # 邮箱授权码（不是密码！）
}

# 发送示例
# send_email(
#     SMTP_CONFIG,
#     to_addrs=["boss@company.com", "team@company.com"],
#     subject="2025年第23周 销售周报",
#     body="<h2>本周销售摘要</h2><p>总销售额 ¥215,000，环比增长 8.5%。</p>",
#     attachments=["weekly_report.xlsx", "weekly_chart.png"],
#     is_html=True,
# )
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

# 7.4 定时调度

# 7.4.1 schedule 定时

import schedule
import time
from datetime import datetime


def job_collect_data():
    """模拟数据采集任务"""
    print(f"[{datetime.now():%H:%M:%S}] 📥 正在采集数据...")


def job_generate_report():
    """模拟报告生成任务"""
    print(f"[{datetime.now():%H:%M:%S}] 📊 正在生成报告...")


def job_send_email():
    """模拟邮件发送任务"""
    print(f"[{datetime.now():%H:%M:%S}] ✉️ 正在发送邮件...")


# ===== 注册定时任务 =====
# 每天 9:00 采集数据
schedule.every().day.at("09:00").do(job_collect_data)
# 每周五 18:00 生成报告
schedule.every().friday.at("18:00").do(job_generate_report)
# 每 10 分钟检查一次健康状态
schedule.every(10).minutes.do(lambda: print(f"[{datetime.now():%H:%M:%S}] ❤️ 心跳"))
# 每小时重试失败任务
schedule.every().hour.do(lambda: print(f"[{datetime.now():%H:%M:%S}] 🔄 重试中..."))

# 启动调度循环
print("⏳ 调度器已启动，按 Ctrl+C 停止...")
try:
    while True:
        schedule.run_pending()
        time.sleep(1)
except KeyboardInterrupt:
    print("\n👋 调度器已停止")
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

# 7.4.2 APScheduler

APScheduler 比 schedule 强在持久化、进程内/外、日期调度：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime


scheduler = BlockingScheduler()


@scheduler.scheduled_job(CronTrigger(day_of_week="fri", hour=18, minute=0))
def weekly_job():
    """每周五 18:00 执行——APScheduler 版"""
    print(f"[{datetime.now()}] 📊 周末报告已生成")


@scheduler.scheduled_job(CronTrigger(day="last", hour=23, minute=0))
def monthly_job():
    """每月最后一天 23:00 执行"""
    print(f"[{datetime.now()}] 📈 月度汇总已生成")


@scheduler.scheduled_job("interval", minutes=30)
def health_check():
    """每 30 分钟执行一次"""
    print(f"[{datetime.now():%H:%M}] ❤️ 健康检查通过")


# scheduler.start()  # 阻塞式启动
print("APScheduler 配置完成（示例未启动——调用 scheduler.start() 启动）")
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

🔑 schedule vs APScheduler vs cron：

维度 schedule APScheduler cron
安装 pip install schedule pip install APScheduler 系统自带（Linux/macOS）
持久化 ❌ 进程重启丢失 ✅ 支持 SQLite/Redis ✅ 系统重启保留
复杂调度 基本 ✅ 秒级/日期/一次 ✅ 全面
平台 跨平台 跨平台 仅 Linux/macOS
适用 原型/轻量脚本 生产级调度 系统运维

# 7.4.3 系统级定时

# ===== Linux/macOS cron =====
# 编辑 crontab
crontab -e

# 每周五 18:00 执行 Python 脚本
0 18 * * 5 /usr/bin/python3 /home/user/weekly_report.py >> /var/log/report.log 2>&1

# 每天 9:00 执行
0 9 * * * /usr/bin/python3 /home/user/daily_sync.py

# ===== cron 语法速查 =====
# 分  时  日  月  周   命令
# *   *   *   *   *    /path/to/script
# 0   9   *   *   1-5  工作日 9:00
# */15 * * * *          每 15 分钟
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# ===== Windows 任务计划 =====
# 命令行创建（管理员权限）
schtasks /create /tn "WeeklyReport" /tr "python C:\scripts\weekly_report.py" /sc weekly /d FRI /st 18:00
1
2
3

# 7.5 综合实战

现在把全部组件串联——每周五 18:00 自动：采集数据 → 生成 Excel → 生成 Word → 发送邮件。

# 7.5.1 数据采集

"""
module_data.py —— 数据采集模块
模拟从数据库/API/文件拉取销售数据
"""

import random
from datetime import datetime, timedelta


def collect_weekly_data() -> list[dict]:
    """采集本周销售数据——返回列表 of dict。
    真实场景：从数据库/SQL/API 拉取，这里用模拟数据。
    """
    categories = ["服装", "电子", "食品", "家居", "运动", "美妆"]
    cities = ["北京", "上海", "深圳", "杭州", "广州"]
    data = []

    for day_offset in range(7):   # 最近 7 天
        date = datetime.now() - timedelta(days=day_offset)
        for cat in categories:
            for _ in range(random.randint(10, 30)):  # 每天每品类 10~30 条
                data.append({
                    "date": date.strftime("%Y-%m-%d"),
                    "weekday": date.strftime("%A"),
                    "category": cat,
                    "city": random.choice(cities),
                    "amount": random.randint(500, 5000),
                    "quantity": random.randint(1, 20),
                })
    print(f"✅ 采集完成：{len(data)} 条交易记录")
    return data
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

# 7.5.2 Excel 生成

"""
module_excel.py —— Excel 报告生成
"""

import pandas as pd
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.chart import BarChart, Reference, PieChart
from openpyxl.utils import get_column_letter


def generate_excel_report(data: list[dict], output_path: str):
    """生成带图表和样式的 Excel 周报。"""
    df = pd.DataFrame(data)

    wb = Workbook()

    # ===== Sheet 1：原始数据 =====
    ws1 = wb.active
    ws1.title = "原始数据"
    df.to_excel  # 实际代码：写入 DataFrame

    # ===== Sheet 2：分析汇总 =====
    ws2 = wb.create_sheet("分析汇总")

    # 按品类汇总
    cat_summary = df.groupby("category").agg(
        销售额=("amount", "sum"), 订单数=("quantity", "sum")
    ).round(0)
    cat_summary["客单价"] = (cat_summary["销售额"] / cat_summary["订单数"]).round(0)
    cat_summary.sort_values("销售额", ascending=False, inplace=True)

    # 写入表头
    headers = ["品类", "销售额", "订单数", "客单价"]
    header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
    header_font = Font(name="微软雅黑", size=11, bold=True, color="FFFFFF")

    for col, h in enumerate(headers, 1):
        cell = ws2.cell(1, col, h)
        cell.fill = header_fill
        cell.font = header_font
        cell.alignment = Alignment(horizontal="center")

    # 写入数据
    for r, (idx, row) in enumerate(cat_summary.iterrows(), 2):
        ws2.cell(r, 1, idx)
        ws2.cell(r, 2, row["销售额"]).number_format = "#,##0"
        ws2.cell(r, 3, row["订单数"]).number_format = "#,##0"
        ws2.cell(r, 4, row["客单价"]).number_format = "#,##0"

    # 添加柱状图
    chart = BarChart()
    chart.title = "各品类销售额"
    chart.y_axis.title = "销售额（元）"
    chart.x_axis.title = "品类"
    chart.style = 10
    data_ref = Reference(ws2, min_col=2, min_row=1, max_row=len(cat_summary)+1)
    cats_ref = Reference(ws2, min_col=1, min_row=2, max_row=len(cat_summary)+1)
    chart.add_data(data_ref, titles_from_data=True)
    chart.set_categories(cats_ref)
    ws2.add_chart(chart, "F2")

    # 饼图——按城市
    ws3 = wb.create_sheet("城市分析")
    city_summary = df.groupby("city", observed=False)["amount"].sum().sort_values(ascending=False)
    ws3.append(["城市", "销售额"])
    for city, val in city_summary.items():
        ws3.append([city, val])

    pie = PieChart()
    pie.title = "各城市销售占比"
    data_ref = Reference(ws3, min_col=2, min_row=1, max_row=len(city_summary)+1)
    cats_ref = Reference(ws3, min_col=1, min_row=2, max_row=len(city_summary)+1)
    pie.add_data(data_ref, titles_from_data=True)
    pie.set_categories(cats_ref)
    ws3.add_chart(pie, "D1")

    # 自适应列宽
    for ws in [ws2, ws3]:
        for col in range(1, ws.max_column + 1):
            ws.column_dimensions[get_column_letter(col)].width = 16

    wb.save(output_path)
    print(f"✅ Excel 已保存：{output_path}")
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

# 7.5.3 Word 报告

"""
module_word.py —— Word 周报生成
"""

from docx import Document
from docx.shared import Inches, Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from datetime import datetime


def generate_word_report(stats: dict, output_path: str):
    """根据统计字典生成 Word 周报。

    stats = {
        "total_amount": 综合销售总金额,
        "top_category": (品类名, 销售额),
        "top_city": (城市名, 销售额),
        "category_data": [(品类, 销售额), ...],
        "highlights": ["亮点1", "亮点2", ...],
    }
    """
    doc = Document()

    # 标题
    title = doc.add_heading(f"销售周报（{datetime.now().strftime('%Y年第%W周')}）", level=0)
    title.alignment = WD_ALIGN_PARAGRAPH.CENTER

    # 本周概要
    doc.add_heading("一、本周概览", level=1)
    p = doc.add_paragraph()
    p.add_run(f"本周总销售额：").font.size = Pt(12)
    run_total = p.add_run(f"¥{stats['total_amount']:,.0f}")
    run_total.font.size = Pt(14)
    run_total.bold = True
    run_total.font.color.rgb = RGBColor(0xE9, 0x1E, 0x63)

    p2 = doc.add_paragraph()
    p2.add_run(f"最佳品类：{stats['top_category'][0]}（¥{stats['top_category'][1]:,}）").font.size = Pt(11)
    p2.add_run(f"\n最佳城市：{stats['top_city'][0]}（¥{stats['top_city'][1]:,}）").font.size = Pt(11)

    # 品类数据表
    doc.add_heading("二、各品类数据", level=1)
    table = doc.add_table(rows=1, cols=3, style="Light Grid Accent 1")
    for i, h in enumerate(["品类", "销售额", "占比"]):
        table.rows[0].cells[i].text = h
    for cat_name, cat_amount in stats["category_data"]:
        row = table.add_row()
        row.cells[0].text = cat_name
        row.cells[1].text = f"¥{cat_amount:,}"
        pct = cat_amount / stats["total_amount"] * 100 if stats["total_amount"] else 0
        row.cells[2].text = f"{pct:.1f}%"

    # 本周亮点
    doc.add_heading("三、本周亮点", level=1)
    for h in stats.get("highlights", []):
        doc.add_paragraph(h, style="List Bullet")

    # 下周重点
    doc.add_heading("四、下周重点", level=1)
    doc.add_paragraph("继续跟踪电子品类库存，备战促销季。", style="List Bullet")
    doc.add_paragraph("跟进运动品类增长趋势，评估扩品可行性。", style="List Bullet")

    doc.save(output_path)
    print(f"✅ Word 已保存：{output_path}")
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

# 7.5.4 邮件与调度

"""
module_main.py —— 主入口：一键串联 + 定时调度
"""

import os
import schedule
import time
from datetime import datetime, timedelta

# 导入各模块
from module_data import collect_weekly_data
from module_excel import generate_excel_report
from module_word import generate_word_report
# smtp 封装（§3.3.4 的 send_email 函数）
from module_email import send_email, SMTP_CONFIG


def run_weekly_report():
    """一键运行：采集 → Excel → Word → 邮件"""
    print("\n" + "=" * 55)
    print(f"  周报自动化系统启动 - {datetime.now():%Y-%m-%d %H:%M}")
    print("=" * 55)

    # 1. 采集
    print("\n[1/4] 数据采集...")
    data = collect_weekly_data()
    if not data:
        print("❌ 未采集到数据，终止")
        return

    # 2. Excel
    print("\n[2/4] 生成 Excel 报告...")
    excel_path = f"weekly_report_{datetime.now():%Y%m%d}.xlsx"
    generate_excel_report(data, excel_path)

    # 3. Word（从 data 提取统计信息）
    from collections import Counter
    total_amount = sum(d["amount"] for d in data)
    cat_counter = Counter()
    city_counter = Counter()
    highlights = []

    for d in data:
        cat_counter[d["category"]] += d["amount"]
        city_counter[d["city"]] += d["amount"]

    top_cat = cat_counter.most_common(1)[0]
    top_city = city_counter.most_common(1)[0]

    # 生成亮点
    if top_cat[1] > 50000:
        highlights.append(f"🔥 {top_cat[0]}品类销售突破 ¥{top_cat[1]:,}")
    if len(set(d["city"] for d in data)) >= 4:
        highlights.append("🌍 本周覆盖 4+ 城市，渠道拓展良好")

    stats = {
        "total_amount": total_amount,
        "top_category": top_cat,
        "top_city": top_city,
        "category_data": cat_counter.most_common(),
        "highlights": highlights or ["本周数据正常，各项指标平稳"],
    }

    print("\n[3/4] 生成 Word 报告...")
    word_path = f"weekly_report_{datetime.now():%Y%m%d}.docx"
    generate_word_report(stats, word_path)

    # 4. 邮件
    print("\n[4/4] 发送邮件...")
    send_email(
        SMTP_CONFIG,
        to_addrs=["boss@company.com"],
        subject=f"销售周报（{datetime.now():%Y年第%W周}）",
        body=(
            f"<h3>📊 本周销售摘要</h3>"
            f"<p>总销售额：<b>¥{total_amount:,}</b></p>"
            f"<p>最佳品类：{top_cat[0]}（¥{top_cat[1]:,}）</p>"
            f"<p>最佳城市：{top_city[0]}（¥{top_city[1]:,}）</p>"
            f"<hr><p><i>本邮件由 Python 自动化系统发送</i></p>"
        ),
        attachments=[excel_path, word_path],
        is_html=True,
    )

    print("\n" + "=" * 55)
    print(f"  ✅ 周报自动化完成 - {datetime.now():%H:%M}")
    print("=" * 55 + "\n")


# ===== 定时调度 =====
if __name__ == "__main__":
    print("⏳ 周报自动化调度器已启动...")
    print("   每周五 18:00 自动执行")
    print("   现在手动执行一次：\n")

    # 先手动跑一次——验证流程
    run_weekly_report()

    # 注册定时任务
    schedule.every().friday.at("18:00").do(run_weekly_report)

    print("按 Ctrl+C 停止调度...\n")
    try:
        while True:
            schedule.run_pending()
            time.sleep(30)
    except KeyboardInterrupt:
        print("\n👋 调度器已停止")
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

运行输出示例：

=======================================================
  周报自动化系统启动 - 2025-06-13 18:00
=======================================================

[1/4] 数据采集...
✅ 采集完成：1,050 条交易记录

[2/4] 生成 Excel 报告...
✅ Excel 已保存：weekly_report_20250613.xlsx

[3/4] 生成 Word 报告...
✅ Word 已保存：weekly_report_20250613.docx

[4/4] 发送邮件...
  📎 已附加：weekly_report_20250613.xlsx
  📎 已附加：weekly_report_20250613.docx
✅ 邮件已发送至：boss@company.com

=======================================================
  ✅ 周报自动化完成 - 18:01
=======================================================
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# 7.6 新手陷阱

# 陷阱 说明
1 关闭 Excel 后文件才释放 load_workbook() 后忘记 wb.close() 导致文件锁——或直接用 with 语法
2 openpyxl vs pandas 样式覆盖 pandas to_excel() 会覆盖已有 Excel 的全部样式——需要用 load_workbook+ExcelWriter 组合
3 邮件附件中文名乱码 add_header 时 FileName 需 URL 编码——from email.utils import formataddr; formataddr(("", filename))
4 schedule 单线程阻塞 任务执行时间过长会延迟后续任务——耗时任务用 threading.Thread(target=job).start() 异步
5 cron 环境变量缺失 cron 执行时 PATH、PYTHONPATH 与终端不同——脚本中使用绝对路径

陷阱 4 详解：

# ❌ schedule 是单线程——如果 job 跑了 5 分钟，后续任务都会延迟
def slow_job():
    time.sleep(300)    # 模拟耗时任务
    print("慢任务完成")

# ✅ 异步化——不阻塞调度循环
import threading
def async_job():
    threading.Thread(target=slow_job).start()

schedule.every().hour.do(async_job)
1
2
3
4
5
6
7
8
9
10
11

# 7.7 综合思考题

  1. pandas vs openpyxl 的职责划分：openpyxl 能读 Excel、pandas 也能读——但 pandas 读完丢掉所有样式和图表。如果一份报表已有 12 张图和复杂样式，你要在不破坏样式的前提下更新数据——该用哪个库？怎么做？

  2. 大 PDF 的内存问题：pypdf 把整个 PDF 全部读入内存——如果合并 100 个 50MB 的 PDF（总计 5GB），你的 8GB 内存够吗？有哪些方案可以"流式处理"大 PDF？

  3. 邮件发送的安全配置：QQ 邮箱的 SMTP 密码是授权码而非登录密码——这是为什么？Google 的 Gmail SMTP 还需要开启"允许不够安全的应用"——这些安全机制在保护什么？

  4. Excel 文件格式差异：.xlsx（OOXML）是 ZIP 压缩的 XML 文件，.xls（旧格式）是二进制格式——openpyxl 只支持 .xlsx。如果公司有大量旧 .xls 文件需要处理，应该用什么库？如果有一份 .xlsm（带宏的文件），直接用 openpyxl 打开会丢失什么？

  5. 调度系统的容错：每周五的定时任务突然失败（网络断了 / 数据源挂了）——你如何确保任务不会静默失败？你需要设计哪些容错机制（重试、告警、日志、手动触发）？

#Python#实战
上次更新: 2026/06/17, 12:47:39
数据分析三件套
开发环境与规范

← 开发环境与规范→

最近更新
01
信号崩溃快速排查
06-15
02
CoreDump破案
06-15
03
perf火焰图实战
06-15
更多文章>
Theme by Vdoing | Copyright © 2019-2026 杨充 | MIT License | 桂ICP备2024034950号 | 桂公网安备45142202000030