企业官网建设流程全解析

在平时的数据分析工作中，我们经常会遇到这样的场景：拿到了一份脏乱的 CSV 数据（比如薪资明细），需要清洗规范后输出成一份规范的PDF文档在提交给相关人员。那能不能把处理逻辑封装起来，别人只需要在网页上点一点，上传文件就能直接下载报告？

所以用FastAPI，在本地电脑上搭建了一个“数据清洗与报告生成微服务”，全程不需要购买云服务器！

1..技术栈选择

• Web 框架：FastAPI
• 数据处理：Pandas（DataWrangling.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(ascending=False)) 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(ascending=False)) # ====== 中文字体 ====== 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", edgecolor="white", width=0.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)}", ha="center", va="bottom", fontsize=11, fontweight="bold") ax.set_title("各标准岗位平均薪资对比", fontsize=16, fontweight="bold", pad=15) ax.set_ylabel("平均薪资（元）", fontsize=12) ax.set_xlabel("标准岗位", fontsize=12) ax.set_ylim(0, result.max() * 1.15) plt.xticks(rotation=30, ha="right") plt.tight_layout() pdf.savefig(fig) plt.close(fig) # ---------- 第2页：饼图 ---------- fig2, ax2 = plt.subplots(figsize=(8, 8)) ax2.pie(result.values, labels=result.index, autopct="%1.1f%%", colors=plt.cm.Set3.colors[:len(result)], startangle=90) ax2.set_title("各岗位薪资占比分布", fontsize=16, fontweight="bold", pad=15) plt.tight_layout() pdf.savefig(fig2) plt.close(fig2) # ====== 把内存里的PDF转成字节流返回 ====== pdf_bytes = pdf_buffer.getvalue() pdf_buffer.close() return pdf_bytes

3.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="薪资数据清洗服务", version="1.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_code=400, detail="只支持 .csv 文件") file_bytes = await file.read() try: result = clean_csv(file_bytes) except Exception as e: raise HTTPException(status_code=422, detail=f"清洗失败: {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_code=400, 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_code=422, detail=f"生成PDF失败: {str(e)}") # 返回给浏览器（不管浏览器能不能下，本地肯定有了） return StreamingResponse( io.BytesIO(pdf_bytes), media_type="application/pdf", headers={"Content-Disposition": "attachment; filename=salary_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_type="application/pdf", headers={"Content-Disposition": "attachment; filename=salary_report.pdf"} )

启动main.py文档，程序运行会有个网页

点击report提交需要清洗数据的文档

提交后会有输出的文档

以上。