3个步骤搞定目前手机销量排行榜实战项目
代码跑不通？别慌。很多初学者卡在环境配置和报错堆栈上，其实只要理清数据流，问题就解决了一半。今天咱们不聊虚的，直接拆解一个实战项目：基于真实场景的“目前手机销量排行榜”系统。
这不仅仅是写几行查询语句，而是从数据清洗、聚合计算到前端展示的完整闭环。哪怕你只是刚接触后端开发，跟着这个实战项目走一遍，也能摸清从0到1搭建业务逻辑的脉络。
项目目标与场景还原
在做任何代码之前，先搞清楚我们要解决什么问题。真实的手机销量数据不是整齐划一的Excel表格，而是散落在各个渠道、格式混乱的原始日志。
我们的核心目标是构建一个轻量级服务，输入过去30天的销售流水，输出实时更新的Top 10销量排行榜。这里有一个关键难点：数据清洗。现实中，同一款手机可能有“iPhone 15 Pro”、“苹果15Pro”、“IP15P”等多种叫法。如果不去重，排行榜就会失去意义。
这个实战项目的核心价值在于模拟真实业务的“脏数据”处理能力。很多教程直接给你干净的数据集，导致你一旦接触真实生产环境，代码就崩。我们要做的，就是在一个受控环境中，复现这种“脏”与“乱”，并找到稳健的解决方案。
通过完成这个实战项目，你不仅学会了排序，更学会了如何处理不确定性数据。这是初级工程师和中级工程师的分水岭。
目录结构与技术选型
为了保持轻量，我们选择 Python 作为后端语言，FastAPI 作为框架，SQLite 作为存储。为什么选这套组合？因为启动快，部署简单，非常适合快速验证业务逻辑。
项目目录结构如下：
phone_ranking_project/
├── main.py # FastAPI 入口
├── database.py # 数据库连接与初始化
├── models.py # Pydantic 数据模型
├── services.py # 核心业务逻辑（清洗、聚合）
├── data/
│ └── raw_sales.json # 模拟原始脏数据
└── requirements.txt
这种结构清晰明了。services.py 是灵魂，所有的脏数据清洗逻辑都放在这里，避免在路由层写复杂逻辑。database.py 负责连接管理，确保资源释放。
注意，我们特意没有引入复杂的 ORM 框架，而是使用 SQLAlchemy 的基础 Core 模块。因为在处理大量聚合查询时，直接写 SQL 往往比 ORM 生成的动态查询更高效，也更容易调试。这也是很多资深工程师在性能敏感场景下的偏好。
核心代码实现与逐行解析
接下来进入硬骨头。我们将分三步实现核心逻辑：数据加载与清洗、聚合计算、API 暴露。
1. 模拟脏数据与清洗逻辑
首先，看 data/raw_sales.json 的一部分。这里故意制造了大小写不一致、空格多余、品牌名称不统一的问题。
[{"id": 1, "name": " iPhone 15 Pro ", "qty": 100, "date": "2023-10-01"},{"id": 2, "name": "apple 15pro", "qty": 50, "date": "2023-10-01"},{"id": 3, "name": "iPhone 15", "qty": 200, "date": "2023-10-02"}
]
在 services.py 中，我们实现一个清洗函数。这里的关键是标准化映射。
import re
from collections import defaultdictdef normalize_phone_name(raw_name: str) -> str:"""标准化手机名称1. 去除首尾空格2. 统一转小写3. 替换常见别名"""name = raw_name.strip().lower()# 简单规则：将 'apple' 替换为 'iphone' 的前缀处理逻辑# 实际项目中，这里应该查一个映射表if name.startswith('apple'):name = name.replace('apple', 'iphone')# 去除中间多余空格name = re.sub(r'\s+', ' ', name)return name
这段代码虽然简单，但体现了处理脏数据的通用思路：规范化（Normalization）。在实际的实战项目中，这个映射表可能来自数据库，由运营人员维护。
2. 聚合计算与排序
清洗完成后，我们需要按标准化后的名称聚合销量。
from datetime import datetime, timedeltadef get_top_rankings(limit: int = 10) -> list[dict]:"""获取过去30天的销量排行榜"""cutoff_date = datetime.now() - timedelta(days=30)# 1. 从数据库获取原始数据# 假设 db_session 已建立raw_records = db_session.query(SaleRecord).filter(SaleRecord.date >= cutoff_date).all()# 2. 内存中聚合（数据量小适用；数据量大需用SQL GROUP BY）sales_dict = defaultdict(int)for record in raw_records:std_name = normalize_phone_name(record.name)sales_dict[std_name] += record.qty# 3. 排序sorted_items = sorted(sales_dict.items(), key=lambda item: item[1], reverse=True)# 4. 截取 Top Nresult = []for i, (name, total) in enumerate(sorted_items[:limit], 1):result.append({"rank": i,"name": name.title(), # 转回标题格式展示"total_sales": total})return result
这里有一个常见的坑：时间过滤放在数据库层还是应用层？
在数据量小于10万条时，拉取到内存处理更灵活，方便做复杂的清洗。但如果数据量达到百万级，必须在 SQL 层完成 GROUP BY 和 SUM，然后再应用清洗逻辑。这是性能优化的关键转折点。
3. API 暴露
在 main.py 中，暴露接口非常简单。
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()@app.get("/api/rankings")
def get_rankings():try:rankings = get_top_rankings(limit=10)return {"code": 200, "data": rankings}except Exception as e:# 记录日志，不要直接抛堆栈给前端print(f"Error fetching rankings: {e}")raise HTTPException(status_code=500, detail="Internal Server Error")
注意异常处理。在生产环境中，永远不要把原始的 Exception 堆栈返回给前端，这会泄露服务器路径信息，造成安全隐患。
运行与测试：如何验证代码没写错
代码写完只是开始，跑通才是真本事。很多初学者卡在这里，因为环境依赖冲突。
1. 环境准备
创建一个虚拟环境，避免污染全局 Python 环境。
python -m venv venv
source venv/bin/activate # Windows 用户用 venv\Scripts\activate
pip install -r requirements.txt
requirements.txt 内容如下：
fastapi==0.100.0
uvicorn==0.23.1
sqlalchemy==2.0.19
2. 初始化数据库
运行 database.py 中的初始化脚本，确保表结构存在。
python database.py
3. 启动服务与测试
uvicorn main:app --reload
打开浏览器访问 http://127.0.0.1:8000/docs，这是 FastAPI 自带的 Swagger 文档。点击“Try it out”，输入参数，点击 Execute。
重点检查：
- 返回的 JSON 格式是否正确？
- 销量数字是否与手动计算一致？
- 名称是否被正确标准化？例如 “apple 15pro” 是否变成了 “Iphone 15 Pro”？
如果返回 500 错误，查看终端日志。90% 的错误都是字段名不匹配或数据类型转换失败。例如，日期字段在数据库中是 String，但在 Python 中是 datetime 对象，比较时会报错。
4. 单元测试
在 tests/test_services.py 中写一个简单的测试，确保清洗逻辑的稳定性。
from services import normalize_phone_namedef test_normalize():assert normalize_phone_name(" Apple 15 Pro ") == "apple 15 pro"assert normalize_phone_name("apple15pro") == "apple 15 pro" # 假设你有空格填充逻辑
使用 pytest 运行：pytest -v。绿钩才是安全的。
优化扩展：从 Demo 到生产级
目前的实现是同步阻塞的，适合小规模数据。如果要扩展到真实生产环境，有几个方向值得探索。
1. 引入缓存机制
排行榜是典型的“读多写少”场景。每次请求都查数据库，压力巨大。引入 Redis 缓存，TTL 设置为 60 秒。
import redisr = redis.Redis(host='localhost', port=6379, db=0)def get_top_rankings_cached():key = "rankings:top10"cached = r.get(key)if cached:return json.loads(cached)data = get_top_rankings(limit=10)r.setex(key, 60, json.dumps(data))return data
2. 异步化处理
FastAPI 支持异步。如果数据源是远程 API 而非本地数据库，应将 get_raw_data 改为 async def，使用 httpx 或 aiohttp 发起非阻塞请求。
3. 数据可视化
后端只返回 JSON 太单调。可以集成 ECharts，前端直接渲染柱状图。这一步能极大提升项目的展示效果，尤其是在面试展示或 GitHub 仓库中。
4. 监控与告警
在 services.py 中增加 Prometheus 指标埋点。例如，统计清洗失败的记录数。如果失败率突然飙升，说明上游数据源格式发生了变更，需要立即告警。
这些优化点，正是区分“学生作业”和“工程代码”的关键。面试官往往不关心你的算法有多复杂，而关心你是否考虑了容错、性能、可观测性。
小结与互动
回顾这个目前手机销量排行榜的实战项目，我们完成了从脏数据处理到 API 暴露的全流程。核心收获有三点：
- 数据清洗是基础：不要假设输入数据是干净的，防御性编程是后端的基本功。
- 分层架构的重要性：业务逻辑、数据访问、接口展示分离，代码才能维护。
- 性能思维：从小数据量到大数据量的切换点，决定了你的架构选型。
这个实战项目代码已整理好，结构清晰，注释详尽，适合初学者直接复现。你可以在此基础上，尝试增加“按品牌分类”、“按月份趋势”等新功能，进一步锻炼自己的能力。
技术栈的选择没有绝对的对错，关键在于解决实际问题。Python + FastAPI 的组合轻量高效，非常适合快速原型开发；如果你更倾向于类型安全，可以尝试 TypeScript + NestJS 实现同样的功能，逻辑是相通的。
你公司项目里是怎么处理这种“脏数据”清洗的？是写在 SQL 里，还是 Python 里？有没有遇到过因为数据格式变更导致线上事故的经历？欢迎在评论区分享你的踩坑经验，咱们一起避坑。
