office_toolkit/
├── config.yaml          # 配置文件
├── main.py              # 主程序入口
├── modules/
│   ├── __init__.py
│   ├── collector.py     # 数据采集
│   ├── processor.py     # 数据处理
│   ├── reporter.py      # 报表生成
│   ├── mailer.py        # 邮件发送
│   └── logger.py        # 日志配置
├── templates/
│   └── report_template.html  # 邮件 HTML 模板
├── output/              # 输出目录
│   ├── reports/         # Excel 报表
│   └── charts/          # 图表图片
├── logs/                # 日志目录
└── requirements.txt

# requirements.txt
pandas>=2.0
openpyxl>=3.1
matplotlib>=3.7
seaborn>=0.12
pyyaml>=6.0
apscheduler>=3.10

# config.yaml
app:
  name: "自动化办公工具箱"
  version: "1.0"
  timezone: "Asia/Shanghai"

data:
  source_file: "data/sales_data.csv"
  encoding: "utf-8-sig"

report:
  output_dir: "output/reports"
  chart_dir: "output/charts"
  filename_pattern: "销售报表_{date}.xlsx"

mail:
  smtp_server: "smtp.qq.com"
  smtp_port: 465
  use_ssl: true
  sender: "your@qq.com"
  password: "your_auth_code"      # 授权码
  recipients:
    - "boss@company.com"
    - "team@company.com"
  subject_pattern: "📊 销售日报 {date}"

schedule:
  enabled: true
  time: "18:00"                   # 每天 18:00 执行
  timezone: "Asia/Shanghai"

logging:
  level: "INFO"
  file: "logs/app.log"
  max_bytes: 10485760             # 10MB
  backup_count: 5

# config_loader.py
import yaml
import os

def load_config(config_path='config.yaml'):
    """加载配置文件"""
    if not os.path.exists(config_path):
        raise FileNotFoundError(f'配置文件不存在: {config_path}')

    with open(config_path, 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)

    # 确保输出目录存在
    os.makedirs(config['report']['output_dir'], exist_ok=True)
    os.makedirs(config['report']['chart_dir'], exist_ok=True)
    os.makedirs(os.path.dirname(config['logging']['file']), exist_ok=True)

    return config

# modules/collector.py
import pandas as pd
import logging

logger = logging.getLogger(__name__)

class DataCollector:
    """数据采集器"""

    def __init__(self, config):
        self.config = config

    def collect(self):
        """采集数据"""
        source = self.config['data']['source_file']
        encoding = self.config['data'].get('encoding', 'utf-8')

        logger.info(f'开始采集数据: {source}')

        # 根据文件类型选择读取方式
        if source.endswith('.csv'):
            df = pd.read_csv(source, encoding=encoding)
        elif source.endswith('.xlsx'):
            df = pd.read_excel(source)
        elif source.endswith('.json'):
            df = pd.read_json(source)
        else:
            raise ValueError(f'不支持的文件格式: {source}')

        logger.info(f'采集完成: {len(df)} 行, {len(df.columns)} 列')

        # 基础验证
        if df.empty:
            raise ValueError('数据为空')

        return df

# modules/processor.py
import pandas as pd
import numpy as np
import logging

logger = logging.getLogger(__name__)

class DataProcessor:
    """数据处理器"""

    def process(self, df):
        """处理数据：清洗 + 统计"""
        logger.info('开始处理数据...')

        # 1. 清洗
        df = self._clean(df)

        # 2. 统计
        summary = self._summarize(df)

        logger.info('数据处理完成')
        return df, summary

    def _clean(self, df):
        """数据清洗"""
        original_len = len(df)

        # 删除重复行
        df = df.drop_duplicates()

        # 删除全空行
        df = df.dropna(how='all')

        # 填充缺失值
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        df[numeric_cols] = df[numeric_cols].fillna(0)

        logger.info(f'清洗: {original_len} → {len(df)} 行 (删除 {original_len - len(df)} 行)')
        return df

    def _summarize(self, df):
        """统计汇总"""
        summary = {}

        # 总体统计
        if '金额' in df.columns:
            summary['总销售额'] = df['金额'].sum()
            summary['订单数'] = len(df)
            summary['客单价'] = df['金额'].mean()

        # 按品类统计
        if '品类' in df.columns and '金额' in df.columns:
            cat_stats = df.groupby('品类')['金额'].agg(['sum', 'count', 'mean'])
            cat_stats.columns = ['销售额', '订单数', '客单价']
            cat_stats = cat_stats.sort_values('销售额', ascending=False)
            summary['品类统计'] = cat_stats

        # 日期统计
        if '日期' in df.columns and '金额' in df.columns:
            df['日期'] = pd.to_datetime(df['日期'])
            daily = df.groupby(df['日期'].dt.date)['金额'].sum()
            summary['日销售趋势'] = daily

        logger.info(f'统计: 总销售额={summary.get("总销售额", 0):,.2f}')
        return summary

# modules/reporter.py
import os
from datetime import datetime
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.chart import BarChart, LineChart, Reference
from openpyxl.utils import get_column_letter
import matplotlib.pyplot as plt
import seaborn as sns
import logging

logger = logging.getLogger(__name__)

class ReportGenerator:
    """报表生成器"""

    def __init__(self, config):
        self.config = config
        self.output_dir = config['report']['output_dir']
        self.chart_dir = config['report']['chart_dir']

        plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
        plt.rcParams['axes.unicode_minus'] = False

    def generate(self, df, summary):
        """生成完整报表"""
        date_str = datetime.now().strftime('%Y%m%d')

        # 1. 生成 Excel 报表
        excel_file = self._generate_excel(df, summary, date_str)

        # 2. 生成图表
        chart_file = self._generate_chart(df, summary, date_str)

        logger.info(f'报表生成完成: {excel_file}')
        return excel_file, chart_file

    def _generate_excel(self, df, summary, date_str):
        """生成 Excel 报表"""
        filename = f'销售报表_{date_str}.xlsx'
        filepath = os.path.join(self.output_dir, filename)

        wb = Workbook()

        # --- Sheet1: 概览 ---
        ws_overview = wb.active
        ws_overview.title = '概览'

        # 样式
        header_font = Font(name='微软雅黑', size=12, bold=True, color='FFFFFF')
        header_fill = PatternFill(start_color='07C160', fill_type='solid')
        center = Alignment(horizontal='center', vertical='center')
        border = Border(
            left=Side(style='thin'), right=Side(style='thin'),
            top=Side(style='thin'), bottom=Side(style='thin')
        )

        # 写入 KPI
        kpis = [
            ('指标', '数值'),
            ('总销售额', f'¥{summary.get("总销售额", 0):,.2f}'),
            ('总订单数', f'{summary.get("订单数", 0):,}'),
            ('客单价', f'¥{summary.get("客单价", 0):,.2f}'),
            ('报表日期', datetime.now().strftime('%Y-%m-%d')),
        ]

        for row_idx, (label, value) in enumerate(kpis, 1):
            for col_idx, text in enumerate([label, value], 1):
                cell = ws_overview.cell(row=row_idx, column=col_idx, value=text)
                if row_idx == 1:
                    cell.font = header_font
                    cell.fill = header_fill
                cell.alignment = center
                cell.border = border

        ws_overview.column_dimensions['A'].width = 20
        ws_overview.column_dimensions['B'].width = 25

        # --- Sheet2: 品类统计 ---
        if '品类统计' in summary:
            ws_cat = wb.create_sheet('品类统计')
            cat_stats = summary['品类统计']

            # 表头
            headers = ['品类'] + list(cat_stats.columns)
            for col_idx, header in enumerate(headers, 1):
                cell = ws_cat.cell(row=1, column=col_idx, value=header)
                cell.font = header_font
                cell.fill = header_fill
                cell.alignment = center
                cell.border = border

            # 数据
            for row_idx, (cat, row) in enumerate(cat_stats.iterrows(), 2):
                ws_cat.cell(row=row_idx, column=1, value=cat).border = border
                for col_idx, val in enumerate(row, 2):
                    cell = ws_cat.cell(row=row_idx, column=col_idx, value=round(val, 2))
                    cell.border = border
                    cell.alignment = center

            # 列宽
            for col in range(1, len(headers) + 1):
                ws_cat.column_dimensions[get_column_letter(col)].width = 15

            # 添加柱状图
            chart = BarChart()
            chart.title = '品类销售额'
            chart.y_axis.title = '销售额'
            chart.x_axis.title = '品类'
            data_ref = Reference(ws_cat, min_col=2, min_row=1, max_row=len(cat_stats) + 1)
            cats_ref = Reference(ws_cat, min_col=1, min_row=2, max_row=len(cat_stats) + 1)
            chart.add_data(data_ref, titles_from_data=True)
            chart.set_categories(cats_ref)
            chart.width = 18
            chart.height = 10
            ws_cat.add_chart(chart, 'F2')

        # --- Sheet3: 原始数据 ---
        ws_data = wb.create_sheet('原始数据')
        # 表头
        for col_idx, col_name in enumerate(df.columns, 1):
            cell = ws_data.cell(row=1, column=col_idx, value=col_name)
            cell.font = header_font
            cell.fill = header_fill
            cell.alignment = center
            cell.border = border
        # 数据
        for row_idx, row in enumerate(df.values, 2):
            for col_idx, val in enumerate(row, 1):
                cell = ws_data.cell(row=row_idx, column=col_idx, value=val)
                cell.border = border

        wb.save(filepath)
        wb.close()

        return filepath

    def _generate_chart(self, df, summary, date_str):
        """生成分析图表"""
        chart_path = os.path.join(self.chart_dir, f'chart_{date_str}.png')

        fig, axes = plt.subplots(1, 2, figsize=(16, 6))

        # 图1：品类柱状图
        if '品类统计' in summary:
            cat_stats = summary['品类统计']
            axes[0].bar(cat_stats.index, cat_stats['销售额'], color='#07c160', alpha=0.8)
            axes[0].set_title('品类销售额排名', fontsize=14)
            axes[0].set_ylabel('销售额')
            axes[0].tick_params(axis='x', rotation=45)

        # 图2：日销售趋势
        if '日销售趋势' in summary:
            daily = summary['日销售趋势']
            axes[1].plot(daily.index, daily.values, color='#4d96ff', linewidth=2, marker='o', markersize=4)
            axes[1].set_title('日销售趋势', fontsize=14)
            axes[1].set_ylabel('销售额')
            axes[1].tick_params(axis='x', rotation=45)

        plt.tight_layout()
        plt.savefig(chart_path, dpi=150, bbox_inches='tight')
        plt.close()

        return chart_path

7. 邮件发送模块

# modules/mailer.py
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from email import encoders
from email.header import Header
from datetime import datetime
import os
import logging

logger = logging.getLogger(__name__)

class Mailer:
    """邮件发送器"""

    def __init__(self, config):
        self.smtp_server = config['mail']['smtp_server']
        self.smtp_port = config['mail']['smtp_port']
        self.use_ssl = config['mail'].get('use_ssl', True)
        self.sender = config['mail']['sender']
        self.password = config['mail']['password']
        self.recipients = config['mail']['recipients']

    def send_report(self, summary, excel_path, chart_path):
        """发送报表邮件"""
        date_str = datetime.now().strftime('%Y-%m-%d')

        # 构造 HTML 正文
        html = self._build_html(summary, date_str)

        # 构造邮件
        msg = MIMEMultipart('mixed')
        msg['From'] = self.sender
        msg['To'] = ', '.join(self.recipients)
        msg['Subject'] = Header(f'📊 销售日报 {date_str}', 'utf-8')

        # HTML 正文
        html_part = MIMEText(html, 'html', 'utf-8')
        msg.attach(html_part)

        # 内嵌图表
        if chart_path and os.path.exists(chart_path):
            with open(chart_path, 'rb') as f:
                img = MIMEImage(f.read())
                img.add_header('Content-ID', '')
                msg.attach(img)

        # Excel 附件
        if excel_path and os.path.exists(excel_path):
            with open(excel_path, 'rb') as f:
                attachment = MIMEBase('application', 'octet-stream')
                attachment.set_payload(f.read())
                encoders.encode_base64(attachment)
                filename = os.path.basename(excel_path)
                attachment.add_header('Content-Disposition', 'attachment',
                                     filename=('utf-8', '', filename))
                msg.attach(attachment)

        # 发送
        server = None
        try:
            if self.use_ssl:
                server = smtplib.SMTP_SSL(self.smtp_server, self.smtp_port)
            else:
                server = smtplib.SMTP(self.smtp_server, self.smtp_port)
                server.starttls()

            server.login(self.sender, self.password)
            server.sendmail(self.sender, self.recipients, msg.as_string())
            logger.info(f'邮件发送成功: {self.recipients}')
            return True
        except Exception as e:
            logger.error(f'邮件发送失败: {e}')
            return False
        finally:
            if server:
                try:
                    server.quit()
                except Exception:
                    pass

    def _build_html(self, summary, date_str):
        """构造 HTML 邮件正文"""
        total_sales = summary.get('总销售额', 0)
        order_count = summary.get('订单数', 0)
        avg_order = summary.get('客单价', 0)

        # 品类排名表
        cat_rows = ''
        if '品类统计' in summary:
            for i, (cat, row) in enumerate(summary['品类统计'].iterrows(), 1):
                cat_rows += f'''

                    {i}
                    {cat}
                    ¥{row["销售额"]:,.2f}
                    {int(row["订单数"])}
                    ¥{row["客单价"]:,.2f}
                '''

        html = f'''


📊 销售日报
{date_str}


📈 今日概览

¥{total_sales:,.0f}

总销售额

{order_count:,}

订单数

¥{avg_order:,.0f}

客单价
🏆 品类排名


                    {cat_rows}
排名
品类
销售额
订单数
客单价
📊 趋势图



                本报告由 Python 自动生成 | 完整数据见附件 Excel



        '''
        return html

8. 主程序入口

# main.py
import sys
import os
from datetime import datetime
from config_loader import load_config
from modules.collector import DataCollector
from modules.processor import DataProcessor
from modules.reporter import ReportGenerator
from modules.mailer import Mailer
import logging

def setup_logging(config):
    """配置日志"""
    log_config = config['logging']
    log_file = log_config['file']
    os.makedirs(os.path.dirname(log_file), exist_ok=True)

    logging.basicConfig(
        level=getattr(logging, log_config['level']),
        format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S',
        handlers=[
            logging.FileHandler(log_file, encoding='utf-8'),
            logging.StreamHandler(sys.stdout),
        ]
    )

def run_pipeline(config):
    """执行完整流水线"""
    logger = logging.getLogger('pipeline')
    start_time = datetime.now()

    logger.info('=' * 50)
    logger.info('自动化办公流水线开始执行')
    logger.info('=' * 50)

    try:
        # 1. 数据采集
        logger.info('步骤 1/4: 数据采集...')
        collector = DataCollector(config)
        df = collector.collect()

        # 2. 数据处理
        logger.info('步骤 2/4: 数据处理...')
        processor = DataProcessor()
        df, summary = processor.process(df)

        # 3. 生成报表
        logger.info('步骤 3/4: 生成报表...')
        reporter = ReportGenerator(config)
        excel_path, chart_path = reporter.generate(df, summary)

        # 4. 发送邮件
        if config['mail'].get('enabled', True):
            logger.info('步骤 4/4: 发送邮件...')
            mailer = Mailer(config)
            mailer.send_report(summary, excel_path, chart_path)
        else:
            logger.info('步骤 4/4: 邮件发送已禁用，跳过')

        # 完成
        elapsed = (datetime.now() - start_time).total_seconds()
        logger.info(f'流水线执行完成 ✅ (耗时 {elapsed:.1f} 秒)')

        return True

    except Exception as e:
        elapsed = (datetime.now() - start_time).total_seconds()
        logger.error(f'流水线执行失败 ❌ (耗时 {elapsed:.1f} 秒)')
        logger.error(f'错误: {e}')
        import traceback
        logger.error(traceback.format_exc())
        return False

def main():
    """主函数"""
    config = load_config()
    setup_logging(config)

    # 检查命令行参数
    if len(sys.argv) > 1 and sys.argv[1] == '--schedule':
        # 定时模式
        from apscheduler.schedulers.blocking import BlockingScheduler
        from apscheduler.triggers.cron import CronTrigger

        scheduler = BlockingScheduler()
        schedule_config = config['schedule']

        time_parts = schedule_config['time'].split(':')
        hour = int(time_parts[0])
        minute = int(time_parts[1]) if len(time_parts) > 1 else 0

        scheduler.add_job(
            run_pipeline,
            CronTrigger(hour=hour, minute=minute),
            args=[config],
            id='daily_report',
        )

        logging.getLogger('main').info(f'定时模式: 每天 {schedule_config["time"]} 执行')
        scheduler.start()
    else:
        # 立即执行一次
        run_pipeline(config)

if __name__ == '__main__':
    main()

9. 完整使用方法

# 1. 安装依赖
pip install pandas openpyxl matplotlib seaborn pyyaml apscheduler

# 2. 准备配置文件 config.yaml（修改邮箱、路径等）

# 3. 准备数据文件 data/sales_data.csv

# 4. 立即执行一次
python main.py

# 5. 定时模式（每天 18:00 自动执行）
python main.py --schedule

# 6. 后台运行（Linux）
nohup python main.py --schedule > /dev/null 2>&1 &

# 7. 查看日志
tail -f logs/app.log

10. 今日小结

🎉 自动化办公阶段完结！

Day 81-84 你掌握了完整的自动化办公技能栈：
✅ Excel 自动化（openpyxl）· 邮件自动化（smtplib）
✅ 定时任务（APScheduler / crontab）· 综合实战项目

接下来进入机器学习入门（Day 85-88）—— Scikit-learn、模型训练、预测。Python 在 AI 领域的核心能力即将解锁！