244 lines
8.5 KiB
Python
244 lines
8.5 KiB
Python
import requests
|
||
from typing import Dict, Any, Optional, List
|
||
|
||
|
||
class AmapAPI:
|
||
"""高德地图API调用类"""
|
||
|
||
def __init__(self, api_key: str):
|
||
"""
|
||
初始化高德地图API
|
||
|
||
Args:
|
||
api_key: 高德地图API密钥
|
||
"""
|
||
self.api_key = api_key
|
||
self.base_url = "https://restapi.amap.com"
|
||
|
||
def geocode(self, address: str) -> Dict[str, Any]:
|
||
"""
|
||
地理编码API - 将地址转换为经纬度坐标
|
||
|
||
Args:
|
||
address: 需要转换的地址
|
||
|
||
Returns:
|
||
Dict: 包含地理编码结果的字典
|
||
"""
|
||
endpoint = f"{self.base_url}/v3/geocode/geo"
|
||
params = {
|
||
"key": self.api_key,
|
||
"address": address
|
||
}
|
||
|
||
response = requests.get(endpoint, params=params)
|
||
response.raise_for_status() # 如果请求失败则抛出异常
|
||
|
||
result = response.json()
|
||
|
||
if result.get("status") != "1":
|
||
raise Exception(f"高德地图API调用失败: {result.get('info')}")
|
||
|
||
return result
|
||
|
||
def parse_location(self, geocode_result: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
解析地理编码结果，提取位置信息
|
||
|
||
Args:
|
||
geocode_result: 地理编码API返回的结果
|
||
|
||
Returns:
|
||
Optional[Dict]: 提取的位置信息，如果解析失败则返回None
|
||
"""
|
||
if not geocode_result or "geocodes" not in geocode_result or not geocode_result["geocodes"]:
|
||
return None
|
||
|
||
geocode = geocode_result["geocodes"][0]
|
||
|
||
# 解析经纬度
|
||
location = geocode.get("location", "")
|
||
longitude, latitude = location.split(",") if "," in location else ("", "")
|
||
|
||
location_info = {
|
||
"formatted_address": geocode.get("formatted_address", ""),
|
||
"country": geocode.get("country", ""),
|
||
"province": geocode.get("province", ""),
|
||
"city": geocode.get("city", ""),
|
||
"district": geocode.get("district", ""),
|
||
"adcode": geocode.get("adcode", ""),
|
||
"longitude": longitude,
|
||
"latitude": latitude,
|
||
"level": geocode.get("level", "")
|
||
}
|
||
|
||
return location_info
|
||
|
||
def search_around(self, location: str, keywords: str, radius: int = 2000,
|
||
types: str = "", page: int = 1, offset: int = 20) -> Dict[str, Any]:
|
||
"""
|
||
周边搜索API - 根据经纬度坐标搜索周边的POI信息
|
||
|
||
Args:
|
||
location: 中心点坐标，格式为"longitude,latitude"
|
||
keywords: 搜索关键词，如"加油站"
|
||
radius: 搜索半径，单位：米，默认2000米
|
||
types: POI类型，可选
|
||
page: 页码，默认1
|
||
offset: 每页记录数，默认20，最大25
|
||
|
||
Returns:
|
||
Dict: 包含POI搜索结果的字典
|
||
"""
|
||
endpoint = f"{self.base_url}/v3/place/around"
|
||
params = {
|
||
"key": self.api_key,
|
||
"location": location,
|
||
"keywords": keywords,
|
||
"radius": radius,
|
||
"page": page,
|
||
"offset": offset
|
||
}
|
||
|
||
# 如果指定了类型，添加到参数中
|
||
if types:
|
||
params["types"] = types
|
||
|
||
response = requests.get(endpoint, params=params)
|
||
response.raise_for_status() # 如果请求失败则抛出异常
|
||
|
||
result = response.json()
|
||
|
||
if result.get("status") != "1":
|
||
raise Exception(f"高德地图API调用失败: {result.get('info')}")
|
||
|
||
return result
|
||
|
||
def parse_pois(self, search_result: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||
"""
|
||
解析周边搜索结果，提取POI信息列表
|
||
|
||
Args:
|
||
search_result: 周边搜索API返回的结果
|
||
|
||
Returns:
|
||
List[Dict]: 提取的POI信息列表，如果解析失败则返回空列表
|
||
"""
|
||
if not search_result or "pois" not in search_result or not search_result["pois"]:
|
||
return []
|
||
|
||
pois_list = []
|
||
|
||
for poi in search_result["pois"]:
|
||
# 解析经纬度
|
||
location = poi.get("location", "")
|
||
longitude, latitude = location.split(",") if "," in location else ("", "")
|
||
|
||
# 提取评分信息
|
||
rating = ""
|
||
if "biz_ext" in poi and "rating" in poi["biz_ext"]:
|
||
rating = poi["biz_ext"]["rating"]
|
||
|
||
# 提取第一张照片URL（如果有）
|
||
photo_url = ""
|
||
if "photos" in poi and poi["photos"] and "url" in poi["photos"][0]:
|
||
photo_url = poi["photos"][0]["url"]
|
||
|
||
poi_info = {
|
||
"id": poi.get("id", ""),
|
||
"name": poi.get("name", ""),
|
||
"type": poi.get("type", ""),
|
||
"typecode": poi.get("typecode", ""),
|
||
"address": poi.get("address", ""),
|
||
"province": poi.get("pname", ""),
|
||
"city": poi.get("cityname", ""),
|
||
"district": poi.get("adname", ""),
|
||
"distance": poi.get("distance", ""),
|
||
"longitude": longitude,
|
||
"latitude": latitude,
|
||
"tel": poi.get("tel", ""),
|
||
"rating": rating,
|
||
"photo_url": photo_url
|
||
}
|
||
|
||
pois_list.append(poi_info)
|
||
|
||
return pois_list
|
||
|
||
|
||
# 使用高德地图API搜索周边设施
|
||
def search_nearby(address: str, keywords: str, radius: int = 1000, limit: int = 10) -> str:
|
||
"""
|
||
根据地址搜索周边设施
|
||
|
||
Args:
|
||
address: 搜索的中心地址，如"骏丰商务中心"
|
||
keywords: 搜索关键词，如"美食"、"加油站"、"酒店"等
|
||
radius: 搜索半径，单位：米，默认1000米
|
||
limit: 返回结果数量，默认10条
|
||
|
||
Returns:
|
||
str: 格式化的搜索结果字符串
|
||
"""
|
||
# 创建API实例
|
||
api_key = "ea26b6bf15cece38dc812d9e6c1a7c65"
|
||
amap = AmapAPI(api_key=api_key)
|
||
|
||
result = []
|
||
|
||
try:
|
||
# 1. 地理编码获取位置坐标
|
||
result.append(f"▶ 正在获取 '{address}' 的经纬度...")
|
||
geocode_result = amap.geocode(address)
|
||
location_info = amap.parse_location(geocode_result)
|
||
|
||
if not location_info:
|
||
return "无法获取位置信息，请检查地址是否正确"
|
||
|
||
result.append(f"✓ 位置信息: {location_info['formatted_address']}")
|
||
result.append(f"✓ 经纬度坐标: {location_info['longitude']},{location_info['latitude']}")
|
||
|
||
# 2. 使用获取到的坐标搜索周边设施
|
||
location = f"{location_info['longitude']},{location_info['latitude']}"
|
||
result.append(f"\n▶ 正在搜索 '{address}' 周边{radius}米范围内的{keywords}...")
|
||
|
||
search_result = amap.search_around(
|
||
location=location,
|
||
keywords=keywords,
|
||
radius=radius,
|
||
offset=limit
|
||
)
|
||
|
||
pois = amap.parse_pois(search_result)
|
||
|
||
if not pois:
|
||
return f"未找到 '{address}' 周边的{keywords}"
|
||
|
||
result.append(f"✓ 找到 {len(pois)} 个{keywords}地点:")
|
||
|
||
# 3. 格式化结果列表
|
||
for i, poi in enumerate(pois, 1):
|
||
poi_info = []
|
||
poi_info.append(f"\n{i}. {poi['name']}")
|
||
poi_info.append(f" 类型: {poi['type']}")
|
||
poi_info.append(f" 地址: {poi['address']}")
|
||
poi_info.append(f" 距离: {poi['distance']}米")
|
||
|
||
if poi['rating']:
|
||
poi_info.append(f" 评分: {poi['rating']}")
|
||
if poi['tel']:
|
||
poi_info.append(f" 电话: {poi['tel']}")
|
||
|
||
result.append("\n".join(poi_info))
|
||
|
||
return "\n".join(result)
|
||
|
||
except Exception as e:
|
||
return f"搜索过程中发生错误: {e}"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 执行示例
|
||
result = search_nearby("骏丰商务中心", "加油站")
|
||
print(result)
|