一：项目背景和需求

背景：某团队每月有一份 sales_raw.xlsx，字段为 日期、销售员、产品、数量、单价、地区，存在空值、重复行与负数量等脏数据。

需求：编写 Python 脚本，读取原始表，完成清洗与统计，输出 report.xlsx。要求：

1. 数据清洗：删除关键字段缺失的行，剔除数量<=0 的异常行，按全字段去重；

2. 统计维度：按销售员汇总销售额（数量×单价），按地区汇总销售额，并给出月度总额；

3. 输出 report.xlsx：含"明细清洗后""按销售员""按地区""总览"四个工作表；

4. 约束：不得修改原始文件；对缺失字段、空文件、无有效数据等情况需给出明确报错或提示。

验收标准：给定样例数据能一键运行生成正确报表；统计数值与人工核对一致；异常输入不崩溃且有可读提示。

二：项目文件结构

三：原始数据说明

四：核心代码

import os import pandas as pd def generate_sales_report(input_file: str = "sales_raw.xlsx", output_file: str = "report.xlsx"): key_columns = ["销售员", "数量", "单价", "地区"] all_columns = ["日期", "销售员", "产品", "数量", "单价", "地区"] # 1. 判断输入文件是否存在 if not os.path.exists(input_file): print(f"【错误】输入文件 {input_file} 不存在，请检查文件是否放在当前目录！") return try: df_raw = pd.read_excel(input_file) except Exception as e: print(f"【错误】读取Excel文件失败：{str(e)}") return # 判断原始文件是否为空 if df_raw.empty: print("【提示】原始Excel文件没有任何数据！") writer = pd.ExcelWriter(output_file, engine="openpyxl") pd.DataFrame().to_excel(writer, sheet_name="明细清洗后", index=False) pd.DataFrame().to_excel(writer, sheet_name="按销售员", index=False) pd.DataFrame().to_excel(writer, sheet_name="按地区", index=False) pd.DataFrame().to_excel(writer, sheet_name="总览", index=False) writer.close() return # 判断是否缺少关键字段 missing_cols = [c for c in key_columns if c not in df_raw.columns] if missing_cols: print(f"【错误】原始文件缺少关键字段：{','.join(missing_cols)}，必须包含{','.join(key_columns)}") return original_rows = df_raw.shape[0] # ========= 数据清洗 ========== df_clean = df_raw.copy() # 1 删除关键字段为空的行 df_clean = df_clean.dropna(subset=key_columns) # 2 过滤数量 <= 0 df_clean = df_clean[df_clean["数量"] > 0] # 3 全字段去重 df_clean = df_clean.drop_duplicates(subset=all_columns, keep="first") clean_rows = df_clean.shape[0] drop_rows = original_rows - clean_rows # 判断清洗完是否还有有效数据 if df_clean.empty: print(f"【警告】清洗完成后无有效业务数据！原始行数:{original_rows},丢弃行数:{drop_rows}") writer = pd.ExcelWriter(output_file, engine="openpyxl") pd.DataFrame().to_excel(writer, sheet_name="明细清洗后", index=False) pd.DataFrame().to_excel(writer, sheet_name="按销售员", index=False) pd.DataFrame().to_excel(writer, sheet_name="按地区", index=False) overview_df = pd.DataFrame([{ "原始总行数": original_rows, "清洗后有效行数": clean_rows, "丢弃行数": drop_rows, "月度总销售额": 0 }]) overview_df.to_excel(writer, sheet_name="总览", index=False) writer.close() return # 计算每行销售额 df_clean["销售额"] = df_clean["数量"] * df_clean["单价"] # ========== 统计聚合 ========== # 按销售员汇总 df_salesman = df_clean.groupby("销售员", as_index=False).agg({"销售额": "sum"}) df_salesman = df_salesman.sort_values("销售额", ascending=False) # 按地区汇总 df_area = df_clean.groupby("地区", as_index=False).agg({"销售额": "sum"}) df_area = df_area.sort_values("销售额", ascending=False) # 月度总览 total_sales = df_clean["销售额"].sum() overview_df = pd.DataFrame([{ "原始总行数": original_rows, "清洗后有效行数": clean_rows, "丢弃行数": drop_rows, "月度总销售额": total_sales }]) # ========== 写入多sheet Excel ========== with pd.ExcelWriter(output_file, engine="openpyxl") as writer: df_clean.to_excel(writer, sheet_name="明细清洗后", index=False) df_salesman.to_excel(writer, sheet_name="按销售员", index=False) df_area.to_excel(writer, sheet_name="按地区", index=False) overview_df.to_excel(writer, sheet_name="总览", index=False) print(f"【成功】报表已经生成 {output_file}") print(f"原始行数:{original_rows} | 清洗有效行数:{clean_rows} |丢弃:{drop_rows}|月度总销售额:{total_sales:.2f}") if __name__ == "__main__": generate_sales_report()五:运行测试与结果

1.正常数据运行

2.原始文件不存在

将sales_raw.xlsx移出项目文件夹，再次运行代码。

3. 全部为无效脏数据

修改sales_raw.xlsx，将所有数量改为负数，保存后运行

六：总结

1. 使用 pandas 完成 Excel 读取、数据清洗、分组统计，一行代码实现多 sheet 写入。
2. 增加异常捕获与边界判断，覆盖文件丢失、空数据、脏数据等异常场景，鲁棒性强。
3. 自动输出多维度销售报表，减少人工 Excel 统计，提升数据分析效率。
4. 项目结构规范，包含依赖声明、说明文档，方便他人复现运行。