在平时的数据分析工作中我们经常会遇到这样的场景拿到了一份脏乱的 CSV 数据比如薪资明细需要清洗规范后输出成一份规范的PDF文档在提交给相关人员。那能不能把处理逻辑封装起来别人只需要在网页上点一点上传文件就能直接下载报告所以用FastAPI在本地电脑上搭建了一个“数据清洗与报告生成微服务”全程不需要购买云服务器1..技术栈选择Web 框架FastAPI数据处理PandasDataWrangling.py里用的核心库报告生成Matplotlib 的PdfPages运行环境本地 Windows/Mac 电脑 断网都能跑2.项目结构设计project/├── main.py # FastAPI 主入口负责接收请求和返回响应├── DataWrangling.py # 核心业务逻辑清洗数据、生成PDF└── Data.csv # 用于测试的原始数据3.核心逻辑与代码实现3.1业务逻辑层部分代码实现def clean_csv(file_bytes: bytes) - dict: 接口1用返回JSON df pd.read_csv(io.BytesIO(file_bytes)) df _do_clean(df) result (df.groupby(standard_title)[salary] .mean() .sort_values(ascendingFalse)) report {k: round(v, 2) for k, v in result.items()} detail df[[raw_title, clean_title, standard_title, salary]].to_dict(records) return {total_rows: len(df), report: report, detail: detail} def generate_pdf(file_bytes: bytes) - bytes: 接口2用清洗数据 → 画图 → 返回PDF字节流 不写本地文件全部在内存中完成 # 清洗数据 df pd.read_csv(io.BytesIO(file_bytes)) df _do_clean(df) result (df.groupby(standard_title)[salary] .mean() .sort_values(ascendingFalse)) # 中文字体 sys_name platform.system() if sys_name Windows: plt.rcParams[font.sans-serif] [SimHei] else: plt.rcParams[font.sans-serif] [Arial Unicode MS] plt.rcParams[axes.unicode_minus] False # 在内存中生成PDF pdf_buffer io.BytesIO() # 内存中的文件 with PdfPages(pdf_buffer) as pdf: # ---------- 第1页柱状图 ---------- fig, ax plt.subplots(figsize(10, 6)) bars ax.bar(result.index, result.values, color#4C72B0, edgecolorwhite, width0.6) for bar, val in zip(bars, result.values): ax.text(bar.get_x() bar.get_width() / 2, bar.get_height() 300, f¥{int(val)}, hacenter, vabottom, fontsize11, fontweightbold) ax.set_title(各标准岗位平均薪资对比, fontsize16, fontweightbold, pad15) ax.set_ylabel(平均薪资元, fontsize12) ax.set_xlabel(标准岗位, fontsize12) ax.set_ylim(0, result.max() * 1.15) plt.xticks(rotation30, haright) plt.tight_layout() pdf.savefig(fig) plt.close(fig) # ---------- 第2页饼图 ---------- fig2, ax2 plt.subplots(figsize(8, 8)) ax2.pie(result.values, labelsresult.index, autopct%1.1f%%, colorsplt.cm.Set3.colors[:len(result)], startangle90) ax2.set_title(各岗位薪资占比分布, fontsize16, fontweightbold, pad15) plt.tight_layout() pdf.savefig(fig2) plt.close(fig2) # 把内存里的PDF转成字节流返回 pdf_bytes pdf_buffer.getvalue() pdf_buffer.close() return pdf_bytes3.2接口层import io from fastapi import FastAPI, UploadFile, File, HTTPException from fastapi.responses import StreamingResponse from fastapi.middleware.cors import CORSMiddleware from DataWrangling import clean_csv, generate_pdf from DataWrangling import RULES app FastAPI(title薪资数据清洗服务, version1.0) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) app.post(/api/clean) async def clean_data(file: UploadFile File(...)): 上传CSV返回清洗后的JSON数据 if not file.filename.endswith(.csv): raise HTTPException(status_code400, detail只支持 .csv 文件) file_bytes await file.read() try: result clean_csv(file_bytes) except Exception as e: raise HTTPException(status_code422, detailf清洗失败: {str(e)}) return {status: success, data: result} app.post(/api/report) async def download_report(file: UploadFile File(...)): 上传CSV → 清洗数据 → 生成PDF → 存到本地 返回给浏览器下载 if not file.filename.endswith(.csv): raise HTTPException(status_code400, detail只支持 .csv 文件) file_bytes await file.read() try: # 1. 调用你的函数生成 PDF 字节流 pdf_bytes generate_pdf(file_bytes) # 【排错关键】打印一下生成的数据类型和长度看看对不对 print(f 生成结果类型: {type(pdf_bytes)}, 长度: {len(pdf_bytes) if pdf_bytes else 0}) except Exception as e: # 如果这里报错了会在黑框框里打印完整的错误原因而不是被浏览器吞掉 print(f❌ 生成PDF彻底失败错误详情: {e}) import traceback traceback.print_exc() # 打印详细报错栈 raise HTTPException(status_code422, detailf生成PDF失败: {str(e)}) # 返回给浏览器不管浏览器能不能下本地肯定有了 return StreamingResponse( io.BytesIO(pdf_bytes), media_typeapplication/pdf, headers{Content-Disposition: attachment; filenamesalary_report.pdf} ) app.get(/api/rules) async def get_rules(): 查看当前的岗位映射规则 return {status: success, rules: RULES} app.get(/api/health) async def health(): return {status: ok, message: 服务运行中}这里有个问题就是电脑可能会遇到IDM出错你可以尝试按Alt快捷键如果无效你可以尝试Ctrl快捷键或者其他不过也没有关系可以加个直接写在本地pdf文件的功能或者自己在浏览器下载pdf文档# 新增强制写入本地文件夹 local_filename 本地生成的测试报告.pdf try: with open(local_filename, wb) as f: f.write(pdf_bytes) print(f✅ 成功保存到本地文件{local_filename}) except Exception as e: print(f⚠️ 保存本地文件失败: {e}) # # 返回给浏览器不管浏览器能不能下本地肯定有了 return StreamingResponse( io.BytesIO(pdf_bytes), media_typeapplication/pdf, headers{Content-Disposition: attachment; filenamesalary_report.pdf} )启动main.py文档程序运行会有个网页点击report提交需要清洗数据的文档提交后会有输出的文档以上。