项目实训——大数据租房推荐智能体——地图通勤评分(二)
·
本次博客主要记录map_service.py的编写,创建地图服务类MapService:负责处理所有与地图相关的功能,包括:初始化数据库、搜索周边、计算通勤
1.模块设计思路:数据存储与服务封装
在编写 map_service.py 时,我主要遵循了两个技术原则:
- 数据本地化:由于以往学校课设项目的习惯,首先考虑了数据库的建立用于储存搜索到数据。
- 职责单一化:
MapService类不仅要负责与高德API交互,还要负责数据的清洗和本地存储,为上层业务提供干净的数据接口。
2.核心实现:数据库架构与建表
这里我设计了 poi_data 和 commute_data 两张表。poi_data 用于存储房源周边的设施信息(如距离、类型),而 commute_data 则用于记录从房源到关键地点的通勤成本。使用 IF NOT EXISTS 语句是为了保证代码的幂等性,即无论运行多少次,只要表存在就不会报错,增强了程序的健壮性。
def create_tables(self):
cursor = self.conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS poi_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
type TEXT,
distance REAL,
address TEXT
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS commute_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
origin TEXT NOT NULL,
destination TEXT NOT NULL,
duration REAL,
distance REAL
)
''')
self.conn.commit()
3.核心功能:高德API对接与POI搜索
实现逻辑分析:
- 参数封装:将中心点坐标、搜索半径、关键词(如“地铁站”)封装成字典。这里需要注意高德API要求的参数格式。
- 请求发送:使用
requests.get发起HTTP请求。 - 数据清洗与入库:这是最关键的一步。API返回的是复杂的JSON结构,我通过遍历
data["pois"]提取出关键字段(名称、类型、距离),并立即调用save_poi_to_db方法将其存入本地数据库。这实现了“搜索即存储”的闭环。def search_poi_around(self, location, keyword="地铁站", radius=2000): url = "https://restapi.amap.com/v3/place/around" params = { "key": AMAP_KEY, "location": location, "keywords": keyword, "radius": radius, "output": "json" } try: response = requests.get(url, params=params) data = response.json() results = [] if data["status"] == "1": for poi in data["pois"]: # 提取核心信息 poi_info = { "name": poi.get("name", "未知"), "type": poi.get("type", "未知"), "distance": poi.get("distance", 0), "address": poi.get("address", "") } results.append(poi_info) # 【关键步骤】将API获取的数据同步写入本地数据库 self.save_poi_to_db(**poi_info) else: print("搜索失败:", data.get("info")) return results except Exception as e: print(f"请求异常: {e}") return []
4.通勤计算:获取距离,估算时间
实现逻辑设想与3类似,传坐标 -> 调API -> 解析JSON -> 存库 -> 返回数据,由于想实现更加真实的通勤时间估算,我试图获取驾车模式下的精准耗时和距离。尝试高德地图最新的 Web服务API V5版本(https://restapi.amap.com/v5/direction/driving),在路径规划上理论上更精准。
同样的,在get到response过后,提取最短路径对应的距离和时间,转为json格式并存到数据库中
def calculate_commute(self, origin, destination):
url = "https://restapi.amap.com/v5/direction/driving"
params = {
"key": AMAP_KEY,
"origin": origin,
"destination": destination
}
response = requests.get(url, params=params)
data = response.json()
if data["code"] == "10000" and data["route"]["paths"]:
path = data["route"]["paths"][0]
duration_seconds = float(path["duration"])
distance_meters = float(path["distance"])
duration_min = duration_seconds / 60.0
distance_km = distance_meters / 1000.0
# 将计算结果存入数据库
cursor = self.conn.cursor()
cursor.execute('''
INSERT INTO commute_data (origin, destination, duration, distance)
VALUES (?, ?, ?, ?)
''', (origin, destination, duration_min, distance_km))
self.conn.commit()
return {"duration_min": round(duration_min, 1), "distance_km": round(distance_km, 2)}
else:
print("通勤计算失败:", data.get("info"))
return None
5.测试与问题
采用山东大学软件园经纬度数据测试py文件,两个主函数分别弹出USERKEY_PLAT_NOMATCH和KeyError: 'code'错误

下次博客再分析原因
更多推荐
所有评论(0)