项目实训——大数据租房推荐智能体——地图通勤评分（二）

本次博客主要记录map_service.py的编写创建地图服务类MapService负责处理所有与地图相关的功能包括初始化数据库、搜索周边、计算通勤1.创建并连接数据库调用建表方法class MapService: def __init__(self, db_namehouse_data.db): self.conn sqlite3.connect(db_name) self.create_tables()2.创建POI和通勤两个表POI表用于储存周边设施信息包括设施名称设施类型距房距离具体地址commute表用于存储通勤计算信息包括始末点时间距离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.两个函数实现搜索和保存POI信息def search_poi_around(self, location, keyword地铁站, radius2000): # 1. 准备高德地图 API 的网址 url https://restapi.amap.com/v3/place/around # 2. 准备请求参数就像填表单 params { key: AMAP_KEY, # 你的钥匙 location: location, # 中心点坐标 keywords: keyword, # 搜什么 radius: radius, # 搜多远 output: json # 返回格式 } # 3. 发送请求 # requests.get 相当于在浏览器输入网址并回车 response requests.get(url, paramsparams) # 4. 解析结果 # 服务器返回的是 JSON 格式我们把它转成 Python 字典 data response.json() results [] # 准备一个空列表用来存结果 # 5. 检查请求是否成功 # status 1 代表高德 API 返回成功 if data[status] 1: # data[pois] 是一个列表里面包含了很多个地点 for poi in data[pois]: # 提取我们需要的信息 name poi.get(name, 未知) poi_type poi.get(type, 未知) distance poi.get(distance, 0) # 距离 address poi.get(address, ) # 把这些信息打包成一个字典 poi_info { name: name, type: poi_type, distance: distance, address: address } results.append(poi_info) # 加入列表 # 6. 存入数据库 (任务2.3的核心要求) self.save_poi_to_db(name, poi_type, distance, address) else: print(搜索失败:, data.get(info)) return results def save_poi_to_db(self, name, poi_type, distance, address): 内部方法把单个 POI 数据写入数据库 cursor self.conn.cursor() cursor.execute( INSERT INTO poi_data (name, type, distance, address) VALUES (?, ?, ?, ?) , (name, poi_type, distance, address)) # 每次修改数据库后都要提交 self.conn.commit() ​4.通勤计算获取距离估算时间def calculate_commute(self, origin, destination): 计算通勤改用距离测量 API更稳定避免跨域报错 逻辑获取两点直线距离 - 乘以系数估算实际路程 - 估算时间 print(f正在计算从 {origin} 到 {destination} 的通勤...) # 1. 获取两地的经纬度 origin_loc self.get_location(origin) dest_loc self.get_location(destination) if not origin_loc or not dest_loc: print(无法获取坐标通勤计算失败) return 0 # 提取经纬度字符串 经度,纬度 origins_str f{origin_loc[lng]},{origin_loc[lat]} destination_str f{dest_loc[lng]},{dest_loc[lat]} # 2. 调用高德“距离测量 API” url https://restapi.amap.com/v3/distance params { key: AMAP_KEY, origins: origins_str, destination: destination_str, type: 1 # 1 表示直线距离0 表示驾车距离驾车距离有时候也会报错先用直线距离保底 } try: response requests.get(url, paramsparams) data response.json() # 3. 解析结果 # 正确的返回结构是 {status:1, results:[{distance:1234, ...}]} if data.get(status) 1 and data.get(results): distance_meters int(data[results][0][distance]) print(f测得直线距离{distance_meters} 米) # 4. 简单估算时间假设骑车/开车速度这里仅作演示 # 实际项目中你可以直接用距离作为评分依据距离越近分越高 # 比如每公里算作 3 分钟 estimated_time int((distance_meters / 1000) * 3) return estimated_time else: print(f测距失败{data.get(info)}) return 0 except Exception as e: print(f程序出错{e}) return 0def calculate_commute(self, origin, destination): 计算通勤改用距离测量 API更稳定避免跨域报错 逻辑获取两点直线距离 - 乘以系数估算实际路程 - 估算时间 print(f正在计算从 {origin} 到 {destination} 的通勤...) # 1. 获取两地的经纬度 origin_loc self.get_location(origin) dest_loc self.get_location(destination) if not origin_loc or not dest_loc: print(无法获取坐标通勤计算失败) return 0 # 提取经纬度字符串 经度,纬度 origins_str f{origin_loc[lng]},{origin_loc[lat]} destination_str f{dest_loc[lng]},{dest_loc[lat]} # 2. 调用高德“距离测量 API” url https://restapi.amap.com/v3/distance params { key: AMAP_KEY, origins: origins_str, destination: destination_str, type: 1 # 1 表示直线距离0 表示驾车距离驾车距离有时候也会报错先用直线距离保底 } try: response requests.get(url, paramsparams) data response.json() # 3. 解析结果 # 正确的返回结构是 {status:1, results:[{distance:1234, ...}]} if data.get(status) 1 and data.get(results): distance_meters int(data[results][0][distance]) print(f测得直线距离{distance_meters} 米) # 4. 简单估算时间假设骑车/开车速度这里仅作演示 # 实际项目中你可以直接用距离作为评分依据距离越近分越高 # 比如每公里算作 3 分钟 estimated_time int((distance_meters / 1000) * 3) return estimated_time else: print(f测距失败{data.get(info)}) return 0 except Exception as e: print(f程序出错{e}) return 05.测试与问题采用山东大学软件园经纬度数据测试py文件两个主函数分别弹出USERKEY_PLAT_NOMATCH和KeyError: code错误下次博客再分析原因