Day 38-7：自动化分析流程 — 让程序替你工作

开篇：一个让人羡慕的场景

2024年2月，我的一位学员王明（区域运营总监）给我发来消息：

老师，我现在每周一早上9点，只需要点击一个按钮，20秒后，一份完整的上周运营周报就自动生成好了，包括50个门店的所有数据、分析图表和管理建议。我的老板都惊呆了！

他的同事们还在周末加班整理数据，而他已经在享受周末时光。

这就是自动化的威力。今天，我将教你如何把Day 38-5和38-6中的分析流程完全自动化。

自动化的三个层次

层次1：手工分析（目前大多数人）

• 每周手动打开150个Excel文件
• 手动复制粘贴数据
• 手动制作图表
• 耗时：2天

层次2：半自动化（Day 38-5和38-6的水平）

• 用Python脚本批量处理数据
• 自动清洗和分析
• 但还需要手动运行脚本
• 耗时：2小时

层次3：全自动化（本节目标）

• 一键运行完整分析流程
• 自动生成Excel报告和图表
• 自动发送邮件给相关人员
• 耗时：20秒

我们的目标：从层次1跃升到层次3。

第一步：创建主控脚本

什么是主控脚本？

主控脚本就像一个总指挥，它负责：

1. 调用数据读取模块
2. 调用数据清洗模块
3. 调用数据分析模块
4. 调用报告生成模块

项目文件结构

首先，我们要把代码组织成模块化结构：

汽车售后分析系统/
├── [main.py](http://main.py)                 # 主控脚本
├── [config.py](http://config.py)              # 配置文件
├── modules/               # 功能模块
│   ├── data_[loader.py](http://loader.py)     # 数据读取
│   ├── data_[cleaner.py](http://cleaner.py)    # 数据清洗
│   ├── [analyzer.py](http://analyzer.py)        # 数据分析
│   └── [reporter.py](http://reporter.py)        # 报告生成
├── data/                  # 数据文件夹
│   └── Q4/               # 原始数据
└── output/               # 输出文件夹
    ├── reports/          # 分析报告
    └── logs/             # 运行日志

创建配置文件 config.py

配置文件集中管理所有设置，方便修改：

# 路径配置
DATA_FOLDER = 'data/Q4'
OUTPUT_FOLDER = 'output/reports'
LOG_FOLDER = 'output/logs'

# 数据质量阈值
MAX_LABOR_FEE = 10000      # 工时费上限
MAX_PARTS_FEE = 50000      # 配件费上限
MIN_RATING = 1             # 评分下限
MAX_RATING = 5             # 评分上限

# 门店分级标准
GRADE_A_NPS = 60           # A级门店NPS标准
GRADE_A_REVENUE = 3000000  # A级门店营收标准
GRADE_B_NPS = 50
GRADE_B_REVENUE = 2500000
GRADE_C_NPS = 40
GRADE_C_REVENUE = 2000000

# 门店战区映射
STORE_REGION_MAP = {
    '北京朝阳店': '华北战区',
    '天津和平店': '华北战区',
    '上海浦东店': '华东战区',
    '杭州西湖店': '华东战区',
    # ... 更多映射
}

# 列名映射
COLUMN_MAPPING = {
    '工单编号': 'order_id',
    '日期': 'date',
    '客户姓名': 'customer_name',
    # ... 更多映射
}

为什么要用配置文件？

想象一下，如果A级门店的标准从NPS 60分调整到65分：

• 没有配置文件：需要在代码的多个地方查找和修改
• 有配置文件：只需要修改config.py中的一个数字

第二步：模块化开发

模块1：数据读取模块 data_loader.py

这个模块负责批量读取Excel文件：

import pandas as pd
import glob
import os
from config import DATA_FOLDER, COLUMN_MAPPING
import logging

def load_all_data():
    # 获取所有文件
    file_list = glob.glob(f'{DATA_FOLDER}/*.xlsx')
    [logging.info](http://logging.info)(f'找到 {len(file_list)} 个数据文件')

    df_list = []
    for file_path in file_list:
        try:
            df = [pd.read](http://pd.read)_excel(file_path)
            # 提取门店名和月份
            filename = os.path.basename(file_path)
            store_name = filename.split('_')[0]
            df['门店'] = store_name

            # 统一列名
            df = df.rename(columns=COLUMN_MAPPING)
            df_list.append(df)

        except Exception as e:
            logging.error(f'读取文件失败: {file_path}, 错误: {str(e)}')

    # 合并所有数据
    df_all = pd.concat(df_list, ignore_index=True)
    [logging.info](http://logging.info)(f'成功读取 {len(df_all)} 条记录')

    return df_all

模块2：数据清洗模块 data_cleaner.py

这个模块负责数据清洗：

import pandas as pd
from config import MAX_LABOR_FEE, MAX_PARTS_FEE
import logging

def clean_data(df):
    [logging.info](http://logging.info)('开始数据清洗...')

    # 1. 删除重复工单
    before = len(df)
    df = df.drop_duplicates(subset=['order_id'], keep='first')
    [logging.info](http://logging.info)(f'删除重复工单: {before - len(df)} 条')

    # 2. 处理缺失值
    df['customer_name'] = df['customer_name'].fillna('未知客户')
    df['labor_fee'] = df['labor_fee'].fillna(0)
    df['parts_fee'] = df['parts_fee'].fillna(0)

    # 3. 处理异常值
    df['labor_fee'] = df['labor_fee'].abs()
    df['parts_fee'] = df['parts_fee'].abs()
    df = df[(df['labor_fee'] <= MAX_LABOR_FEE) & 
            (df['parts_fee'] <= MAX_PARTS_FEE)]

    # 4. 特征工程
    df['total_amount'] = df['labor_fee'] + df['parts_fee']
    df['date_clean'] = [pd.to](http://pd.to)_datetime(df['date'], errors='coerce')

    [logging.info](http://logging.info)(f'清洗完成，剩余 {len(df)} 条记录')
    return df

模块3：分析模块 analyzer.py

这个模块负责核心分析：

import pandas as pd
import logging

def calculate_nps(df):
    # NPS计算逻辑
    total = len(df)
    if total == 0:
        return None
    promoters = len(df[df['rating'] == 5])
    detractors = len(df[df['rating'] <= 3])
    return round((promoters - detractors) / total * 100, 1)

def analyze_stores(df):
    # 门店维度分析
    [logging.info](http://logging.info)('开始门店分析...')

    store_metrics = df.groupby('门店').agg({
        'order_id': 'count',
        'total_amount': 'sum',
        'rating': 'mean'
    }).round(2)

    store_metrics['NPS'] = df.groupby('门店').apply(calculate_nps)

    return store_metrics

def analyze_regions(df):
    # 战区维度分析
    [logging.info](http://logging.info)('开始战区分析...')
    # 分析逻辑...
    pass

def analyze_service_types(df):
    # 服务类型分析
    [logging.info](http://logging.info)('开始服务类型分析...')
    # 分析逻辑...
    pass

第三步：创建主控脚本

main.py - 一键运行的核心

import logging
from datetime import datetime
from [modules.data](http://modules.data)_loader import load_all_data
from [modules.data](http://modules.data)_cleaner import clean_data
from modules.analyzer import analyze_stores, analyze_regions
from modules.reporter import generate_excel_report
from config import LOG_FOLDER

# 设置日志
log_file = f'{LOG_FOLDER}/run_{[datetime.now](http://datetime.now)().strftime("%Y%m%d_%H%M%S")}.log'
logging.basicConfig(
    level=[logging.INFO](http://logging.INFO),
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(log_file, encoding='utf-8'),
        logging.StreamHandler()
    ]
)

def main():
    [logging.info](http://logging.info)('='*60)
    [logging.info](http://logging.info)('汽车售后数据自动化分析系统 启动')
    [logging.info](http://logging.info)('='*60)

    try:
        # 第1步：读取数据
        [logging.info](http://logging.info)('\n步骤1: 读取数据...')
        df_raw = load_all_data()

        # 第2步：清洗数据
        [logging.info](http://logging.info)('\n步骤2: 清洗数据...')
        df_clean = clean_data(df_raw)

        # 第3步：分析数据
        [logging.info](http://logging.info)('\n步骤3: 分析数据...')
        store_analysis = analyze_stores(df_clean)
        region_analysis = analyze_regions(df_clean)

        # 第4步：生成报告
        [logging.info](http://logging.info)('\n步骤4: 生成报告...')
        report_path = generate_excel_report(
            df_clean, 
            store_analysis, 
            region_analysis
        )

        [logging.info](http://logging.info)(f'\n报告已生成: {report_path}')
        [logging.info](http://logging.info)('\n分析完成！')
        [logging.info](http://logging.info)('='*60)

    except Exception as e:
        logging.error(f'程序执行出错: {str(e)}', exc_info=True)
        raise

if __name__ == '__main__':
    main()

现在，运行整个分析流程只需要一个命令：

python [main.py](http://main.py)

第四步：添加进度显示

为了让程序运行时更直观，我们可以添加进度条：

安装进度条库

pip install tqdm

在数据读取中使用进度条

from tqdm import tqdm

def load_all_data():
    file_list = glob.glob(f'{DATA_FOLDER}/*.xlsx')

    df_list = []
    # 添加进度条
    for file_path in tqdm(file_list, desc='读取文件'):
        df = [pd.read](http://pd.read)_excel(file_path)
        # ... 处理逻辑
        df_list.append(df)

    return pd.concat(df_list, ignore_index=True)

运行时会看到：

读取文件: 100%|██████████████████| 150/150 [00:45<00:00,  3.31it/s]

第五步：异常处理与日志记录

为什么需要日志？

想象一下，程序在凌晨2点自动运行时出错了，第二天早上你怎么知道发生了什么？

日志就是程序的黑匣子。

完善的异常处理示例

def load_all_data():
    file_list = glob.glob(f'{DATA_FOLDER}/*.xlsx')

    success_count = 0
    fail_count = 0
    df_list = []

    for file_path in file_list:
        try:
            df = [pd.read](http://pd.read)_excel(file_path)
            df_list.append(df)
            success_count += 1

        except PermissionError:
            logging.error(f'文件被占用: {file_path}（可能正在被Excel打开）')
            fail_count += 1

        except pd.errors.EmptyDataError:
            logging.error(f'文件为空: {file_path}')
            fail_count += 1

        except Exception as e:
            logging.error(f'读取失败: {file_path}, 错误: {str(e)}')
            fail_count += 1

    [logging.info](http://logging.info)(f'读取完成: 成功{success_count}个, 失败{fail_count}个')

    if fail_count > 0:
        logging.warning(f'有{fail_count}个文件读取失败，请检查日志')

    return pd.concat(df_list, ignore_index=True)

第六步：参数化配置

让脚本更灵活

有时候，我们可能只想分析某个月的数据，或者只分析某个战区。这时候，参数化配置就很有用：

import argparse

def main():
    # 创建参数解析器
    parser = argparse.ArgumentParser(description='汽车售后数据分析系统')
    parser.add_argument('--month', type=int, help='指定分析月份（10-12）')
    parser.add_argument('--region', type=str, help='指定分析战区')
    parser.add_argument('--store', type=str, help='指定分析门店')

    args = parser.parse_args()

    # 读取数据
    df = load_all_data()

    # 根据参数过滤
    if args.month:
        df = df[df['月'] == args.month]
        [logging.info](http://logging.info)(f'只分析{args.month}月数据')

    if args.region:
        df = df[df['战区'] == args.region]
        [logging.info](http://logging.info)(f'只分析{args.region}数据')

    # 继续分析...

现在可以这样运行：

# 只分析10月数据
python [main.py](http://main.py) --month 10

# 只分析华东战区
python [main.py](http://main.py) --region 华东战区

# 只分析某个门店
python [main.py](http://main.py) --store 上海浦东店

关键收获

通过本节学习，你掌握了：

✅ 模块化设计：让代码结构清晰，易于维护

✅ 配置文件：集中管理参数，方便修改

✅ 主控脚本：一键运行完整流程

✅ 进度显示：让程序运行更直观

✅ 异常处理：让程序更健壮

✅ 日志记录：方便问题追踪

✅ 参数化：让脚本更灵活

最重要的是：你现在可以实现"一键分析"了！

真实案例：王明的自动化之路

还记得开篇提到的王明吗？他是这样实现自动化的：

第1周： 手动分析，每周工作2天

第2周： 学习Python和Pandas

第3周： 写出基本的数据清洗脚本，耗时从2天缩短到半天

第4周： 完善分析脚本，耗时缩短到2小时

第5周： 模块化改造，实现一键运行，耗时缩短到20秒

第6周： 添加定时任务，每周一早上自动运行

现在，他每周一早上9点打开电脑，报告已经生成好了。

他节省下来的时间用来做什么？

• 深度思考业务问题
• 和门店店长沟通改进措施
• 学习更多数据分析技能

结果：3个月后，他升职为大区总监。

下一节（Day 38-8），我们将学习如何为分析结果添加精美的可视化图表，让数据更有说服力。

准备好了吗？让我们继续！ ?