import requests
import csv
import time
# 高德API配置
API_KEY = '' # 替换为你的高德开放平台Web服务API Key
CITY = '南昌'
KEYWORDS = '韵达快递'
OUTPUT_FILE = 'nanchang_ydexpress_poi——1.csv'
MAX_PAGE_SIZE = 50 # 高德API每页最大返回数量
def get_poi_data(page):
url = "https://restapi.amap.com/v3/place/text"
params = {
'key': API_KEY,
'keywords': KEYWORDS,
'city': CITY,
'offset': MAX_PAGE_SIZE,
'page': page,
'extensions': 'all' # 获取全部信息
}
try:
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
if data['status'] == '1':
return data['pois']
else:
print(f"请求失败: {data.get('info', '未知错误')}")
return None
except Exception as e:
print(f"请求异常: {str(e)}")
return None
def parse_poi(pois):
results = []
for poi in pois:
result = {
'名称': poi.get('name', ''),
'地址': poi.get('address', ''),
'经度': poi.get('location', '').split(',')[0] if poi.get('location') else '',
'纬度': poi.get('location', '').split(',')[1] if poi.get('location') else '',
'联系电话': poi.get('tel', ''),
'省份': poi.get('pname', ''),
'城市': poi.get('cityname', ''),
'区域': poi.get('adname', ''),
'ID': poi.get('id', '')
}
results.append(result)
return results
def save_to_csv(data):
if not data:
return
with open(OUTPUT_FILE, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
def main():
all_pois = []
page = 1
while True:
print(f"正在获取第 {page} 页数据...")
pois = get_poi_data(page)
if not pois:
break
parsed_data = parse_poi(pois)
all_pois.extend(parsed_data)
# 检查是否还有更多数据
if len(pois) < MAX_PAGE_SIZE:
break
page += 1
time.sleep(0.5) # 控制请求频率避免触发限流
save_to_csv(all_pois)
print(f"数据已保存至 {OUTPUT_FILE}，共获取 {len(all_pois)} 条记录")
if __name__ == '__main__':
main()
