3个坑让你少走弯路：上海地铁票价查询实战避坑指南
刚学完 Python 语法，面对“上海地铁票价查询”这种真实需求，是不是脑子一片空白？很多人卡在“代码能跑，但项目搭不起来”的尴尬阶段。这篇避坑指南，直接带你从零搭建一个可复现、可部署的票价查询工具，专治“学会语法却不知怎么搭项目”的顽疾。
项目目标与需求拆解
做项目前，先别急着敲代码。我们要做的不是一个简单的计算器，而是一个可维护、可扩展的查询系统。
核心需求只有三点：
数据源：获取上海地铁最新线路及站点数据。
算法逻辑：根据起终点，计算最短路径或最少换乘方案。
票价规则：严格遵循上海地铁现行计费标准（6元起步，之后按里程分段累加）。
很多新手一上来就写 if-else 判断距离，结果代码越写越长，维护地狱就此诞生。我们的目标是构建一个图论模型，将地铁站点视为节点，轨道视为边，通过图搜索算法解决路径问题。
目录结构工程化
不要把所有代码塞进一个 main.py 里。工程化的第一步，是清晰的文件划分。建议采用如下结构：
shanghai-metro-query/
├── data/
│ └── stations.json # 站点坐标及连接关系数据
├── core/
│ ├── __init__.py
│ ├── graph.py # 图结构封装
│ └── pricing.py # 票价计算逻辑
├── utils/
│ ├── __init__.py
│ └── data_loader.py # 数据加载与清洗
├── main.py # 入口文件
└── requirements.txt # 依赖管理
这种结构的好处是：关注点分离。数据加载、图构建、票价计算互不干扰。当你需要更换数据源或调整票价规则时，只需修改对应模块，无需触碰核心逻辑。
核心代码实现与逐行讲解
1. 数据加载与图构建
首先，我们需要将 JSON 数据转化为内存中的图结构。这里使用 networkx 库来简化图操作，但为了体现工程化思维，我们手动封装一层。
# core/graph.py
import json
from typing import Dict, List, Tuple
class MetroGraph:
def __init__(self, data_path: str):
self.nodes: Dict[str, dict] = {}
self.edges: List[Tuple[str, str, float]] = []
self._load_data(data_path)
def _load_data(self, path: str):
从JSON加载站点和边数据
with open(path, 'r', encoding='utf-8') as f:
data = json.load(f)
for node in data['stations']:
self.nodes[node['id']] = node
for edge in data['routes']:
# 边权值为距离（公里），用于后续票价计算
self.edges.append((edge['start'], edge['end'], edge['distance']))
def get_neighbors(self, station_id: str) - List[str]:
获取相邻站点，BFS遍历的关键
neighbors = []
for start, end, dist in self.edges:
if start == station_id:
neighbors.append(end)
elif end == station_id:
neighbors.append(start)
return neighbors
避坑点：很多新手在加载数据时直接硬编码路径。务必将数据路径作为参数传入，这样在单元测试或不同环境部署时，才能灵活切换数据文件。
2. 票价计算逻辑
上海地铁票价规则看似简单，实则分段复杂。这是最容易出 Bug 的地方。
# core/pricing.py
def calculate_fare(distance_km: float) - float:
上海地铁票价规则：
6km以内(含) 4元
6-16km(含) 每增加10km加1元
16km以上 每增加20km加1元
if distance_km = 0:
return 0.0
if distance_km = 6:
return 4.0
elif distance_km = 16:
# 4元 + (超过6km的部分/10km) * 1元，向上取整
extra = int((distance_km - 6) / 10) + 1
return 4.0 + extra
else:
# 16km及以上，基础价7元（4+3），每增加20km加1元
# 注意：16km时是7元，16-36km是8元...
base_price = 7.0
extra = int((distance_km - 16) / 20)
return base_price + extra
关键细节：这里的 int() 截断逻辑需要仔细验证。建议编写单元测试，覆盖边界值（如 6.0, 6.1, 16.0, 36.0）。很多线上事故就出在边界条件处理不当上。
3. 路径搜索：BFS 实现最短换乘
我们使用广度优先搜索（BFS）来查找最少换乘次数的路径。
# core/graph.py 补充方法
from collections import deque
def find_shortest_path(self, start: str, end: str) - Tuple[float, List[str]]:
返回: (总距离, 站点列表)
注意：这里简化为按站点数最短，实际需结合距离权重
if start not in self.nodes or end not in self.nodes:
return -1, []
queue = deque([(start, [start])])
visited = {start}
while queue:
current_node, path = queue.popleft()
if current_node == end:
# 计算总距离
total_dist = 0
for i in range(len(path) - 1):
total_dist += self._get_edge_distance(path[i], path[i+1])
return total_dist, path
for neighbor in self.get_neighbors(current_node):
if neighbor not in visited:
visited.add(neighbor)
queue.append((neighbor, path + [neighbor]))
return -1, []
def _get_edge_distance(self, s1: str, s2: str) - float:
查找两站间距离
for start, end, dist in self.edges:
if (start == s1 and end == s2) or (start == s2 and end == s1):
return dist
return float('inf')
性能陷阱：如果站点数量极大，纯 BFS 可能较慢。在真实项目中，可考虑引入 A* 算法，以欧氏距离作为启发函数，加速搜索过程。
运行与测试：确保代码可靠
代码写完不等于项目完成。必须通过测试验证。
在 tests/ 目录下创建 test_pricing.py：
import unittest
from core.pricing import calculate_fare
class TestPricing(unittest.TestCase):
def test_base_fare(self):
self.assertEqual(calculate_fare(5), 4.0)
def test_mid_range_fare(self):
self.assertEqual(calculate_fare(10), 5.0) # 6-16km区间
self.assertEqual(calculate_fare(15), 6.0)
def test_high_range_fare(self):
self.assertEqual(calculate_fare(20), 8.0) # 16-36km区间
if __name__ == '__main__':
unittest.main()
实战建议：在 CI/CD 流程中集成单元测试。哪怕只是一个简单的脚本，也要保证每次提交后，核心逻辑不会崩。这是从“写代码”到“做工程”的分水岭。
优化扩展：从 Demo 到生产级
当基础功能跑通后，如何让它更像生产级项目？
数据缓存：
站点数据变化频率低，不应每次请求都读取 JSON。引入 lru_cache 或 Redis 缓存图结构，减少 I/O 开销。
API 封装：
使用 Flask 或 FastAPI 将查询功能封装为 REST API。
from fastapi import FastAPI
app = FastAPI()
@app.get(/fare)
def get_fare(start: str, end: str):
# 调用 graph 和 pricing 逻辑
return {fare: 4.0, path: [People's Square, Lujiazui]}
日志与监控：
添加 logging 模块，记录查询耗时、异常堆栈。生产环境中，没有日志的代码等于“盲飞”。
可信来源：在 GitHub 开源仓库中，许多成熟的交通规划项目（如 OpenStreetMap 相关工具）都采用了类似的图论+缓存架构。参考这些开源仓库的 Issue 讨论，能帮你提前规避大量潜在坑点。
小结与互动
从零搭建“上海地铁票价查询”项目，核心不在于算法多高深，而在于工程化思维：模块解耦、边界测试、数据缓存。
学会语法只是入场券，懂得如何组织代码、如何保证稳定性，才是工程师的护城河。
你公司项目里是怎么处理这种“规则复杂+数据静态”的场景的？是用硬编码规则引擎，还是配置化管理？欢迎在评论区分享你的避坑经验。 
长河君 · 2026/9/22 14:42:59 · 阅读 1.2k 3个坑让你少走弯路：上海地铁票价查询实战避坑指南
3个坑让你少走弯路：上海地铁票价查询实战避坑指南
刚学完 Python 语法，面对“上海地铁票价查询”这种真实需求，是不是脑子一片空白？很多人卡在“代码能跑，但项目搭不起来”的尴尬阶段。这篇避坑指南，直接带你从零搭建一个可复现、可部署的票价查询工具，专治“学会语法却不知怎么搭项目”的顽疾。
项目目标与需求拆解
做项目前，先别急着敲代码。我们要做的不是一个简单的计算器，而是一个可维护、可扩展的查询系统。
核心需求只有三点：
数据源：获取上海地铁最新线路及站点数据。
算法逻辑：根据起终点，计算最短路径或最少换乘方案。
票价规则：严格遵循上海地铁现行计费标准（6元起步，之后按里程分段累加）。
很多新手一上来就写 if-else 判断距离，结果代码越写越长，维护地狱就此诞生。我们的目标是构建一个图论模型，将地铁站点视为节点，轨道视为边，通过图搜索算法解决路径问题。
目录结构工程化
不要把所有代码塞进一个 main.py 里。工程化的第一步，是清晰的文件划分。建议采用如下结构：
shanghai-metro-query/
├── data/
│ └── stations.json # 站点坐标及连接关系数据
├── core/
│ ├── __init__.py
│ ├── graph.py # 图结构封装
│ └── pricing.py # 票价计算逻辑
├── utils/
│ ├── __init__.py
│ └── data_loader.py # 数据加载与清洗
├── main.py # 入口文件
└── requirements.txt # 依赖管理
这种结构的好处是：关注点分离。数据加载、图构建、票价计算互不干扰。当你需要更换数据源或调整票价规则时，只需修改对应模块，无需触碰核心逻辑。
核心代码实现与逐行讲解
1. 数据加载与图构建
首先，我们需要将 JSON 数据转化为内存中的图结构。这里使用 networkx 库来简化图操作，但为了体现工程化思维，我们手动封装一层。
# core/graph.py
import json
from typing import Dict, List, Tuple
class MetroGraph:
def __init__(self, data_path: str):
self.nodes: Dict[str, dict] = {}
self.edges: List[Tuple[str, str, float]] = []
self._load_data(data_path)
def _load_data(self, path: str):
从JSON加载站点和边数据
with open(path, 'r', encoding='utf-8') as f:
data = json.load(f)
for node in data['stations']:
self.nodes[node['id']] = node
for edge in data['routes']:
# 边权值为距离（公里），用于后续票价计算
self.edges.append((edge['start'], edge['end'], edge['distance']))
def get_neighbors(self, station_id: str) - List[str]:
获取相邻站点，BFS遍历的关键
neighbors = []
for start, end, dist in self.edges:
if start == station_id:
neighbors.append(end)
elif end == station_id:
neighbors.append(start)
return neighbors
避坑点：很多新手在加载数据时直接硬编码路径。务必将数据路径作为参数传入，这样在单元测试或不同环境部署时，才能灵活切换数据文件。
2. 票价计算逻辑
上海地铁票价规则看似简单，实则分段复杂。这是最容易出 Bug 的地方。
# core/pricing.py
def calculate_fare(distance_km: float) - float:
上海地铁票价规则：
6km以内(含) 4元
6-16km(含) 每增加10km加1元
16km以上 每增加20km加1元
if distance_km = 0:
return 0.0
if distance_km = 6:
return 4.0
elif distance_km = 16:
# 4元 + (超过6km的部分/10km) * 1元，向上取整
extra = int((distance_km - 6) / 10) + 1
return 4.0 + extra
else:
# 16km及以上，基础价7元（4+3），每增加20km加1元
# 注意：16km时是7元，16-36km是8元...
base_price = 7.0
extra = int((distance_km - 16) / 20)
return base_price + extra
关键细节：这里的 int() 截断逻辑需要仔细验证。建议编写单元测试，覆盖边界值（如 6.0, 6.1, 16.0, 36.0）。很多线上事故就出在边界条件处理不当上。
3. 路径搜索：BFS 实现最短换乘
我们使用广度优先搜索（BFS）来查找最少换乘次数的路径。
# core/graph.py 补充方法
from collections import deque
def find_shortest_path(self, start: str, end: str) - Tuple[float, List[str]]:
返回: (总距离, 站点列表)
注意：这里简化为按站点数最短，实际需结合距离权重
if start not in self.nodes or end not in self.nodes:
return -1, []
queue = deque([(start, [start])])
visited = {start}
while queue:
current_node, path = queue.popleft()
if current_node == end:
# 计算总距离
total_dist = 0
for i in range(len(path) - 1):
total_dist += self._get_edge_distance(path[i], path[i+1])
return total_dist, path
for neighbor in self.get_neighbors(current_node):
if neighbor not in visited:
visited.add(neighbor)
queue.append((neighbor, path + [neighbor]))
return -1, []
def _get_edge_distance(self, s1: str, s2: str) - float:
查找两站间距离
for start, end, dist in self.edges:
if (start == s1 and end == s2) or (start == s2 and end == s1):
return dist
return float('inf')
性能陷阱：如果站点数量极大，纯 BFS 可能较慢。在真实项目中，可考虑引入 A* 算法，以欧氏距离作为启发函数，加速搜索过程。
运行与测试：确保代码可靠
代码写完不等于项目完成。必须通过测试验证。
在 tests/ 目录下创建 test_pricing.py：
import unittest
from core.pricing import calculate_fare
class TestPricing(unittest.TestCase):
def test_base_fare(self):
self.assertEqual(calculate_fare(5), 4.0)
def test_mid_range_fare(self):
self.assertEqual(calculate_fare(10), 5.0) # 6-16km区间
self.assertEqual(calculate_fare(15), 6.0)
def test_high_range_fare(self):
self.assertEqual(calculate_fare(20), 8.0) # 16-36km区间
if __name__ == '__main__':
unittest.main()
实战建议：在 CI/CD 流程中集成单元测试。哪怕只是一个简单的脚本，也要保证每次提交后，核心逻辑不会崩。这是从“写代码”到“做工程”的分水岭。
优化扩展：从 Demo 到生产级
当基础功能跑通后，如何让它更像生产级项目？
数据缓存：
站点数据变化频率低，不应每次请求都读取 JSON。引入 lru_cache 或 Redis 缓存图结构，减少 I/O 开销。
API 封装：
使用 Flask 或 FastAPI 将查询功能封装为 REST API。
from fastapi import FastAPI
app = FastAPI()
@app.get(/fare)
def get_fare(start: str, end: str):
# 调用 graph 和 pricing 逻辑
return {fare: 4.0, path: [People's Square, Lujiazui]}
日志与监控：
添加 logging 模块，记录查询耗时、异常堆栈。生产环境中，没有日志的代码等于“盲飞”。
可信来源：在 GitHub 开源仓库中，许多成熟的交通规划项目（如 OpenStreetMap 相关工具）都采用了类似的图论+缓存架构。参考这些开源仓库的 Issue 讨论，能帮你提前规避大量潜在坑点。
小结与互动
从零搭建“上海地铁票价查询”项目，核心不在于算法多高深，而在于工程化思维：模块解耦、边界测试、数据缓存。
学会语法只是入场券，懂得如何组织代码、如何保证稳定性，才是工程师的护城河。
你公司项目里是怎么处理这种“规则复杂+数据静态”的场景的？是用硬编码规则引擎，还是配置化管理？欢迎在评论区分享你的避坑经验。