当前位置：首页 > news >正文

高德获取poi—基于python的多边形四分递归 POI 采集脚本

news 2026/7/29 19:40:59

如果仅仅只需要代码只需要看第一，第三部分即可，第二部分稍微看一下。

1. 基本准备

1.1获得key（高德开放平台：高德开放平台 | 高德地图API）

接下来按照这个操作进行获取key

创建应用和 Key-Web服务 API|高德地图APIhttps://lbs.amap.com/api/webservice/create-project-and-key最后就会看到这个页面

1.2拼接 HTTP 请求 URL

首先我们先看一怎么拼接：

因为我们是利用多边形区域搜索，所以按照以下进行操作

他这个下面提供了一个例子

首先你需要先输入

#https://restapi.amap.com/v5/place/polygon?parameters

这里面的parameters就是你需要获得的参数

有必须填的也可以不填的（key,polygon一定要填）

各种参数需要通过&（shift键+7）进行连接。注意ploygon的要求。

知道这些基本上就可以开始写代码了，还有需要注意的且比较重要的点我会再后续讲解中提到。

2. 代码调试

2.1 测试区域

2.2 请求url

首先拼接本次研究的使用的HTTP（本次举例使用的查询的poi是“120000商务住宅”）

https://restapi.amap.com/v5/place/polygon?key=********{your KEY}********&polygon=116.354706,39.943538|116.427546,39.902423&types=120000&page_size=25&page_num=1

你可以直接将这个网址复制到在浏览器就可以出现这个页面

这个是浏览器展示这段内容JSON 格式数据，是高德地图 Web API（v5 多边形 POI 检索接口）返回的响应报文。

因此需要导入request库模拟浏览器访问网址，发送 HTTP 请求、接收网页 / 接口返回的数据，并利用python中json模块进行解析数据。

import json,time from urllib import request

这个time库是为了控制请求次数，每秒内只能请求3次,从这里可以看到除了多变形搜索其他的月配额都是0，不知道能不能用我还没试过。

time.sleep(0.5)

所以每隔0.5秒请求一次。

2.3获取数据

可以参考B站上一个视频“【环哥】如何采集POI数据——以高德地图为例”，他讲的很好，但是后续提到了空间切分没有进行实现，我这里写了，在2.5。

接下来来看一下怎么获取数据

url = "https://restapi.amap.com/v5/place/polygon?key=******************&polygon=116.354706,39.943538|116.427546,39.902423&types=120000&page_size=25&page_num=1" html = request.urlopen(url).read() t = json.loads(html)

打印一下结果是：{'count': '25', 'infocode': '10000', 'pois': [{'parent': '', 'address': '景山前街与大石作胡同交叉口东北约60米', 'distance': '', 'pcode': '110000', 'adcode': '110102', 'pname': '北京市', 'cityname': '北京市', 'type': '商务住宅;商务住宅相关;商务住宅相关', 'typecode': '120000', 'adname': '西城区', 'citycode': '010', 'name': '景山前街20号院', 'location': '116.392320,39.923384', 'id': 'B0FFL450RZ'},..............后面内容太多就不写了

这个一个页面最多可以获得25条，分别放到字典（<class 'dict'>）中，因为这是一个字典所以我们可以通过“键”获的我们需要的内容（pois）

我们先打印一下,看pois中有什么

print(t.keys()) 结果：dict_keys(['count', 'infocode', 'pois', 'status', 'info']) print(t["pois"]) 结果：[{'parent': '', 'address': '景山前街与大石作胡同交叉口东北约60米', 'distance': '', 'pcode': '110000', 'adcode': '110102', 'pname': '北京市', 'cityname': '北京市', 'type': '商务住宅;商务住宅相关;商务住宅相关', 'typecode': '120000', 'adname': '西城区', 'citycode': '010', 'name': '景山前街20号院', 'location': '116.392320,39.923384', 'id': 'B0FFL450RZ'}, {'parent': '', 'address': '景山西街与景山前街交叉口北140米', 'distance': '', 'pcode': '110000', 'adcode': '110102', 'pname': '北京市', 'cityname': '北京市', 'type': '商务住宅;住宅区;住宅小区', 'typecode': '120302', 'adname': '西城区', 'citycode': '010', 'name': '景山西街48号', 'location': '116.393968,39.924391', 'id': 'B000A9TKHG'}, {'parent': '', 'address': '文津街11号', 'distance': '', 'pcode': '110000', 'adcode': '110102', 'pname': '北京市', 'cityname': '北京市', 'type': '商务住宅;住宅区;住宅小区', 'typecode': '120302', 'adname': '西城区', 'citycode': '010', 'name': '文津街11号院', 'location': '116.383905,39.923572', 'id': 'B0FFG4BCRC'}，........]

可以看到pois中有：parent，address，distance，pcode，location等，选择需要的字段，后续用于提取内容。

本次使用字段如(fields)所示：

fields = ['id', 'name', 'citycode', 'adname', 'typecode', 'type', 'cityname', 'adcode', 'pcode', 'address']

使用这些字段，获取信息，操作如下：

pois = t["pois"][0]#这个就是获取pois中的第一条，类型是字典 fields = ['id', 'name', 'citycode', 'adname', 'typecode', 'type', 'cityname', 'adcode', 'pcode', 'address'] for f in fields:#对fields做一个循环把内容都取出来 print(pois[f],end=" ") 结果：B0FFL450RZ 景山前街20号院 010 西城区 120000 商务住宅;商务住宅相关;商务住宅相关 北京市 110102 110000 景山前街与大石作胡同交叉口东北约60米

如果要将一页25条都取出来的话再在外面做一个循环

k=[]#创建一个空列表用装数据 for i in range(25):#因为一页25条所以就循环25次，去除25条 pois = t["pois"][i] fields = ['id', 'name', 'citycode', 'adname', 'typecode', 'type', 'cityname', 'adcode', 'pcode', 'address'] m = [] for f in fields:#这个是poi内部的一个循环，是为了将每一条中的内容取出 m.append(pois[f]) k.append(m) print(k)

你框选的区域不可能只有一页所以还需要对页数进行一个循环，再取出每一页中的内容，此时这个page就不能写死了,写成这种page_num={page}。

url = f"https://restapi.amap.com/v5/place/polygon?key=******************&polygon=116.354706,39.943538|116.427546,39.902423&types=120000&page_size=25&page_num={page}"

import json,time from urllib import request k = []#注意这个不能写到循环里面要不然，你换一页他就会重写一遍导致最后只有25条 for page in range(1, 101): time.sleep(0.5) url = f"https://restapi.amap.com/v5/place/polygon?key=*****************&polygon=116.354706,39.943538|116.427546,39.902423&types=120000&page_size=25&page_num={page}" html = request.urlopen(url).read() t = json.loads(html) pois_list = t.get("pois", [])#避免因为没有数据报错 #⚠️⚠️⚠️没有数据可能是网络波动，或者请求速度太快可以用time.sleep(0.5) # 当前页没有数据，说明已经读取完毕 if not pois_list: print(f"第{page}页无数据，停止读取") continue for pois in pois_list: fields = [ 'id', 'name', 'citycode', 'adname', 'typecode', 'type', 'cityname', 'adcode', 'pcode', 'address' ] m = [] for f in fields: m.append(pois.get(f)) k.append(m) print(f"第{page}页，本页{len(pois_list)}条，累计{len(k)}条") 结果：第1页，本页25条，累计25条 第2页，本页25条，累计50条 第3页，本页25条，累计75条 第4页，本页25条，累计100条 第5页，本页25条，累计125条 第6页，本页25条，累计150条 第7页，本页25条，累计175条 第8页，本页25条，累计200条 第9页，本页25条，累计225条 第10页无数据，停止读取

这个结果一看（其实是经过很多次验证）就不对，首先数据是25的倍数，其次这么大的区域而且在北京不可能只有225条。

真正的原因就是：高德规定了一个区域最多可以获得200条左右的数据（我也是搜的），所以会有很多的点，被删除。如果是小区域的话这样肯定是可以的，但是应用价值就不大了。

上述代码仅仅是为了显示整个获取poi数据的一个过程，代码写的可能比较LOW，但是应该是比较容易看懂。从上面我们可以知道对于大范围区域这个代码就不能使用了，非要用的话就需要分割成很多小块去使用，那就没啥意义了。

那我们能不能直接利用代码的方式去帮我们分呢？———当然可以。就像下图所示

这个就是空间切分（四叉树/网格）的方法

2.5构架空间切分函数

def split_four(poly_str):#poly_str)只是一个形参站位用的后续会传入数据 left_top_str, right_down_str = poly_str.split("|")#location的格式都是"116.354706,39.943538|116.427546,39.902423"这样的，所以这条代码可以获取左上和右下的坐标 X1, Y1 = map(float, left_top_str.split(","))#准换后再将坐标字符串类型转化成浮点型，其中map(类型, 可迭代对象:列表，字典，元组，ABC..,range(n)等) X2, Y2 = map(float, right_down_str.split(",")) #限制四分区域的最小值 dx = abs(X2 - X1) dy = abs(Y1 - Y2) if dx < MIN_DEG and dy < MIN_DEG:#限制四分区域的时候小到这个阈值就不能再小了 return [] x_mid = (X1 + X2) / 2 y_mid = (Y1 + Y2) / 2 sub_blocks = [ f"{X1:.6f},{Y1:.6f}|{x_mid:.6f},{y_mid:.6f}",#{变量名 : 格式控制规则} f"{x_mid:.6f},{Y1:.6f}|{X2:.6f},{y_mid:.6f}", f"{X1:.6f},{y_mid:.6f}|{x_mid:.6f},{Y2:.6f}", f"{x_mid:.6f},{y_mid:.6f}|{X2:.6f},{Y2:.6f}", ]#格式化一下 return sub_blocks 代码中的一些都是一些全局变量，只是在这个函数中没有体现，总的代码中会出现。

写成函数的话后续就非常方便进行调用。

2.6数据获取函数

这一部分主要是将2.3获取数据中的逻辑定义成了一个函数，，没有用for循环用的是While，并添加一些抛出错误的代码。其中status（如下图）这个是JSON 数据的一部分，通常是程序调用 API 后返回的结果，而代码中t.get("status") != "1"就是不等于1，做一个判段，不等于1就直接结束循环了。

"status": "1"：表示请求成功
"info": "OK"：表示状态正常/成功

def query_region(poly_str): page = 1 consecutive_full = 0 need_split = False while True: time.sleep(0.5) try: url = (f"https://restapi.amap.com/v5/place/polygon" f"?key={KEY}&polygon={poly_str}&types={TYPES}" f"&page_size={page_size}&page_num={page}") html = request.urlopen(url, timeout=10).read()#读取网页，如果10秒无响应就直接抛出超时异常 t = json.loads(html)# if t.get("status") != "1": print(f"【接口返回失败】{poly_str} page={page} 响应：{t}") break pois = t.get("pois", [])# except Exception as e: print(f"【请求异常】区域:{poly_str}, page={page}, err:{str(e)}")#格式化输出那个区域，那一页常见的错误 break if not pois: break #四分网格存在边界重叠：同一个 POI 可能同时落在左右 / 上下两个网格里，多次被抓取。可以利用ID这个每个点独有标记进行去重 for item in pois: #pois里面有25条信息,item就是每一条 pid = item["id"]#将每一条的id复制给pid if pid not in poi_dict:#如果这个id不在poi_dict，这里注意字典用 in / not in 判断时，只会去匹配【键（key）】，不会匹配值 poi_dict[pid] = item#添加进去 if len(pois) == page_size: consecutive_full += 1 if consecutive_full >= TRIGGER_FULL_PAGE:# need_split = True break#这个虽然嵌套在if里面也是跳出while循环，和上面哪几个break一样 else: consecutive_full = 0 page += 1 if not need_split: return sub_regions = split_four(poly_str)在这里调用了四分函数 if len(sub_regions) == 0: print(f"【到达最小网格，停止继续拆分】{poly_str}") return for sub in sub_regions:#这个for循环是将四分后的四个小区域进行逐一调用数据获取函数将上面的流程再跑一变，如果还有需要四分的在四分，循环往复。 query_region(sub)这调用了数据获取函数，这个就是递归即在函数内部调用函数

下面这部分代码我单独拿出来，比较重要，这个就是用来判断什么时候需要进行开始四分，就是一个触发四分函数的一个判断按钮。我当时写的是连续6-7页都是25条的时候触发（你也可以改成总数是多少条触发），但是我问了一些AI说的是尽量用2-3页比价稳妥，具体原因可以问一下AI。【注】这里面也有些参数是定义在前面了。

if len(pois) == page_size: consecutive_full += 1 if consecutive_full >= TRIGGER_FULL_PAGE:# need_split = True break#这个虽然嵌套在if里面也是跳出while循环，和上面哪几个break一样 else: consecutive_full = 0

2.7数据导出

这个我就不解释了

if __name__ == "__main__": print("=====开始采集=====") query_region(init_poly) print(f"采集结束,去重后POI总数:{len(poi_dict)}") with open(out_path, "wb") as output: title = '\t'.join(fields) + "\tx\ty\n" output.write(title.encode("utf8")) for p in poi_dict.values(): row = [] for fd in fields: row.append(str(p.get(fd, ""))) lon, lat = p["location"].split(",") row.append(lon) row.append(lat) line = '\t'.join(row) + "\n" output.write(line.encode("utf8")) print("文件写入完成！")

3. 完整代码

需要改的地方：out_path,KEY,TYPES,init-poly。

import json, time from urllib import request out_path = r"C:\Users\61639\Desktop\1\test.txt" KEY = 写你的key TYPES = "120000" page_size = 25#每页大条数 init_poly = "116.354706,39.943538|116.427546,39.902423"#选择研究区域 MIN_DEG = 0.0008#最小单元格的长度 TRIGGER_FULL_PAGE = 2#连续两页都是25条 fields = ['id', 'name', 'citycode', 'adname', 'typecode', 'type', 'cityname', 'adcode', 'pcode', 'address']#这个是poi中的字段可通过pois.keys()获取 poi_dict = {}#创建一个空的字典 #定义四分递归函数 def split_four(poly_str): left_top_str, right_down_str = poly_str.split("|") X1, Y1 = map(float, left_top_str.split(","))#map(类型, 可迭代对象:列表，字典，元组，ABC..,range(n)等) X2, Y2 = map(float, right_down_str.split(",")) #限制四分区域的最小值 dx = abs(X2 - X1) dy = abs(Y1 - Y2) if dx < MIN_DEG and dy < MIN_DEG:#限制四分区域的时候小到这个阈值就不能再小了 return [] x_mid = (X1 + X2) / 2 y_mid = (Y1 + Y2) / 2 sub_blocks = [ f"{X1:.6f},{Y1:.6f}|{x_mid:.6f},{y_mid:.6f}",#{变量名 : 格式控制规则} f"{x_mid:.6f},{Y1:.6f}|{X2:.6f},{y_mid:.6f}", f"{X1:.6f},{y_mid:.6f}|{x_mid:.6f},{Y2:.6f}", f"{x_mid:.6f},{y_mid:.6f}|{X2:.6f},{Y2:.6f}", ] return sub_blocks def query_region(poly_str): page = 1 consecutive_full = 0 need_split = False while True: time.sleep(0.5) try: url = (f"https://restapi.amap.com/v5/place/polygon" f"?key={KEY}&polygon={poly_str}&types={TYPES}" f"&page_size={page_size}&page_num={page}") html = request.urlopen(url, timeout=10).read()#读取网页，如果10秒无响应就直接抛出超时异常 t = json.loads(html)# if t.get("status") != "1": print(f"【接口返回失败】{poly_str} page={page} 响应：{t}") break pois = t.get("pois", [])# except Exception as e: print(f"【请求异常】区域:{poly_str}, page={page}, err:{str(e)}")#格式化输出那个区域，那一页常见的错误 break if not pois: break #四分网格存在边界重叠：同一个 POI 可能同时落在左右 / 上下两个网格里，多次被抓取。可以利用ID这个每个点独有标记进行去重 for item in pois: #pois里面有25条信息,item就是每一条 pid = item["id"]#将每一条的id复制给pid if pid not in poi_dict:#如果这个id不在poi_dict，这里注意字典用 in / not in 判断时，只会去匹配【键（key）】，不会匹配值 poi_dict[pid] = item#添加进去 if len(pois) == page_size: consecutive_full += 1 if consecutive_full >= TRIGGER_FULL_PAGE: need_split = True break#这个虽然嵌套在if里面也是跳出while循环，和上面哪几个break一样 else: consecutive_full = 0 page += 1 if not need_split: return sub_regions = split_four(poly_str) if len(sub_regions) == 0: print(f"【到达最小网格，停止继续拆分】{poly_str}") return for sub in sub_regions: query_region(sub) if __name__ == "__main__": print("=====开始采集=====") query_region(init_poly) print(f"采集结束,去重后POI总数:{len(poi_dict)}") with open(out_path, "wb") as output: title = '\t'.join(fields) + "\tx\ty\n" output.write(title.encode("utf8")) for p in poi_dict.values(): row = [] for fd in fields: row.append(str(p.get(fd, ""))) lon, lat = p["location"].split(",") row.append(lon) row.append(lat) line = '\t'.join(row) + "\n" output.write(line.encode("utf8")) print("文件写入完成！")

【注】这个虽然是多边形搜索，不能根据行政边界搜索，但是你可以将矩形框住多边形边界，并根据cityname在excel中进行筛选，或导入arcgispro中进行裁剪。