高德poi之关键字查询

xiexuhuai

专业相关，试着手搓了一个。因为部分借用了chatGTP，不算完全原创了。
1.调用高德api之“搜索服务-关键字查询”。
（1）用params传递参数。

[Python] 纯文本查看 复制代码

import csvimport requests

# 定义列表转字符串函数
def list_to_str(i):
    if i == []:
    # 如果为空数组，则将其设为 None 或者某个默认值
        return None
    else:
    # 如果不为空，则继续使用原来的值
        separator = '，'
        result = separator.join(i)
        return result
    
def keyword_search(input_file=None, keywords=None, city=None, types=None, output_file='poi.csv'):
    url = 'https://restapi.amap.com/v3/place/text'
    pois = []
    if input_file is not None:
        # 读取包含地名的csv文件
        with open(input_file, 'r', encoding='utf-8-sig') as f:
            reader = csv.DictReader(f)
            for row in reader:
                # 构造请求参数
                page = 1
                while True:
                    params = {
                        'key': 'YOUR KEY',
                        'keywords': row['名称'],
                        'city': row['城市'],
                        'citylimit':True,
                        'output': 'JSON',
                        'offset': 20,  # 每页输出的POI数量，默认为20
                        'page': page,
                        'extensions':'all'
                    }
                    # 发送请求
                    response = requests.get(url, params=params)
                    data = response.json()#data是一个字典
                    print(response.text) # 打印出服务器的响应文本
                    if data["status"] == "1":
                        if not data["pois"]:#data["pois"]是一个列表
                            break
                        pois.extend(data['pois'])
                    else:
                        print(f"Error: {data['info']}")
                        break
                    page += 1
    
    
                
        # 将结果保存到csv文件中
        with open(output_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['id','名称','类型','类型编码','地址', '经度','纬度',
                             '电话','邮编','网站','邮箱','省份编码','省份名称',
                             '城市编码','城市名称','区域编码','区域名称'])#writer.writerow只能写入传递格式为列表或者元组的参数
            for poi in pois:
                location = poi['location'].split(',')
                writer.writerow([poi['id'],poi['name'],poi['type'],poi['typecode'],
                                poi['address'],location[0],location[1],poi['tel'],
                                list_to_str(poi['postcode']),list_to_str(poi['website']),list_to_str(poi['email']),
                                poi['pcode'],poi['pname'],
                                poi['citycode'],poi['cityname'],
                                poi['adcode'],poi['adname']])#poi['postcode'],poi['website'],poi['email']均是列表
    
    elif city is not None or types is not None:
        # 构造请求参数
        page = 1
        while True:
            params = {
                'key': 'YOUR KEY',
                'city': city,
                'types': types,
                'citylimit':True,
                'output': 'JSON',
                'offset': '10',  # 每页输出的POI数量，默认为20
                'page': page,
                'extensions':'all'
            }
            #发送请求
            response = requests.get(url,params=params)
            data = response.json()
            if data["status"] == "1":
                if not data["pois"]:
                    break
                pois.extend(data["pois"])
            else:
                print(f"Error: {data['info']}")
                break
            page += 1
        
        # 将结果保存到csv文件中
        with open(output_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['id','名称','类型','类型编码','地址', '经度','纬度',
                             '电话','邮编','网站','邮箱','省份编码','省份名称',
                             '城市编码','城市名称','区域编码','区域名称'])#writer.writerow只能写入传递格式为列表或者元组的参数
            for poi in pois:
                location = poi['location'].split(',')
                writer.writerow([poi['id'],poi['name'],poi['type'],poi['typecode'],
                                poi['address'],location[0],location[1],poi['tel'],
                                list_to_str(poi['postcode']),list_to_str(poi['website']),list_to_str(poi['email']),
                                poi['pcode'],poi['pname'],
                                poi['citycode'],poi['cityname'],
                                poi['adcode'],poi['adname']])
            
#keyword_search(input_file='./测试.csv', output_file='地名_POI.csv')
keyword_search(city='310000',types='120300',output_file='上海_住宅区.csv')

（2）url直接传递参数。

[Python] 纯文本查看 复制代码

import csv
import requests

# 定义列表转字符串函数
def list_to_str(i):
    if i == []:
    # 如果为空数组，则将其设为 None 或者某个默认值
        return None
    else:
    # 如果不为空，则继续使用原来的值
        separator = '，'
        result = separator.join(i)
        return result
    
def keyword_search(input_file=None, keywords=None, city=None, types=None, output_file='poi.csv'):
    key = 'YOUR KEY'
    citylimit = 'true'
    offset = 50
    page = 1
    extensions = 'all'
    pois = []
    if input_file is not None:
        # 读取包含地名的csv文件
        with open(input_file, 'r', encoding='utf-8-sig') as f:
            reader = csv.DictReader(f)
            for row in reader:
                while True:
                    keywords = row['名称']
                    city = row['城市']
                    # url中的请求参数为空而不是为None
                    if keywords is None:
                        keywords=''
                    if city is None:
                        city=''
                    if types is None:
                        types=''
                    # 发送请求
                    url = f'https://restapi.amap.com/v3/place/text?key={key}&keywords={keywords}&types={types}&city={city}&citylimit={citylimit}&offset={offset}&page={page}&extensions={extensions}'
                    response = requests.get(url)
                    data = response.json()#data是一个字典
                    print(response.text) # 打印出服务器的响应文本
                    if data["status"] == "1":
                        if not data["pois"]:#data["pois"]是一个列表
                            break
                        pois.extend(data['pois'])
                    else:
                        print(f"Error: {data['info']}")
                        break
                    page += 1
    
    elif city is not None or types is not None:
        while True:
            # url中的请求参数为空而不是为None
            if keywords is None:
                keywords=''
            if city is None:
                city=''
            if types is None:
                types=''
            # 发送请求
            url = f'https://restapi.amap.com/v3/place/text?key={key}&keywords={keywords}&types={types}&city={city}&citylimit={citylimit}&offset={offset}&page={page}&extensions={extensions}'
            response = requests.get(url)
            data = response.json()#data是一个字典
            print(response.text) # 打印出服务器的响应文本
            if data["status"] == "1":
                if not data["pois"]:#data["pois"]是一个列表
                    break
                pois.extend(data['pois'])
            else:
                print(f"Error: {data['info']}")
                break
            page += 1
            
    # 将结果保存到csv文件中
    with open(output_file, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['id','名称','类型','类型编码','地址', '经度','纬度',
                         '电话','邮编','网站','邮箱','省份编码','省份名称',
                         '城市编码','城市名称','区域编码','区域名称'])#writer.writerow只能写入传递格式为列表或者元组的参数
        for poi in pois:
            location = poi['location'].split(',')
            writer.writerow([poi['id'],poi['name'],poi['type'],poi['typecode'],
                            poi['address'],location[0],location[1],poi['tel'],
                            list_to_str(poi['postcode']),list_to_str(poi['website']),list_to_str(poi['email']),
                            poi['pcode'],poi['pname'],
                            poi['citycode'],poi['cityname'],
                            poi['adcode'],poi['adname']])

#keyword_search(input_file='./测试.csv', output_file='地名_POI.csv')
keyword_search(city='310000',types='120300',output_file='上海市_住宅区.csv')

（3）两者没有什么区别，但是经过尝试，用params传递参数的话，offset好像没有用，不管设置多少，都是按默认的20条记录/页返回，我也不知道为什么，坛友有人知道的话可以在评论区回复一下。


2.调用高德api之“搜索服务-关键字查询2.0”。
高德api有两个搜索服务，个人开放者账号每天都有100次的免费访问次数，用完了v3再来用v5嘛。

[Python] 纯文本查看 复制代码

import csv
import requests

# 定义列表转字符串函数
def list_to_str(i):
    if i == []:
    # 如果为空数组，则将其设为 None 或者某个默认值
        return None
    else:
    # 如果不为空，则继续使用原来的值
        separator = '，'
        result = separator.join(i)
        return result
    
def keyword_search(input_file=None, keywords=None, region=None, types=None, output_file='poi.csv'):
    key = 'YOUR KEY'
    city_limit = 'true'
    page_size = 25
    page_num = 1
    pois = []
    if input_file is not None:
        # 读取包含地名的csv文件
        with open(input_file, 'r', encoding='utf-8-sig') as f:
            reader = csv.DictReader(f)
            for row in reader:
                while True:
                    keywords = row['名称']
                    region = row['城市']
                    # url中的请求参数为空而不是为None
                    if keywords is None:
                        keywords=''
                    if region is None:
                        region=''
                    if types is None:
                        types=''
                    # 发送请求
                    url = f'https://restapi.amap.com/v5/place/text?key={key}&keywords={keywords}&types={types}®ion={region}&city_limit={city_limit}&page_size={page_size}&page_num={page_num}'
                    response = requests.get(url)
                    data = response.json()#data是一个字典
                    print(response.text) # 打印出服务器的响应文本
                    if data["status"] == "1":
                        if not data["pois"]:#data["pois"]是一个列表
                            break
                        pois.extend(data['pois'])
                    else:
                        print(f"Error: {data['info']}")
                        break
                    page_num += 1
    
    elif region is not None or types is not None:
        while True:
            # url中的请求参数为空而不是为None
            if keywords is None:
                keywords=''
            if region is None:
                region=''
            if types is None:
                types=''
            # 发送请求
            url = f'https://restapi.amap.com/v5/place/text?key={key}&keywords={keywords}&types={types}®ion={region}&city_limit={city_limit}&page_size={page_size}&page_num={page_num}'
            response = requests.get(url)
            data = response.json()#data是一个字典
            print(response.text) # 打印出服务器的响应文本
            if data["status"] == "1":
                if not data["pois"]:#data["pois"]是一个列表
                    break
                pois.extend(data['pois'])
            else:
                print(f"Error: {data['info']}")
                break
            page_num += 1
            
    # 将结果保存到csv文件中
    with open(output_file, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['id','名称','类型','类型编码','地址', '经度','纬度','省份编码','省份名称',
                         '城市编码','城市名称','区域编码','区域名称'])#writer.writerow只能写入传递格式为列表或者元组的参数
        for poi in pois:
            location = poi['location'].split(',')
            writer.writerow([poi['id'],poi['name'],poi['type'],poi['typecode'],
                            poi['address'],location[0],location[1],
                            poi['pcode'],poi['pname'],
                            poi['citycode'],poi['cityname'],
                            poi['adcode'],poi['adname']])

#keyword_search(input_file='./测试.csv', output_file='地名_POI.csv')
keyword_search(region='310000',types='120302',output_file='上海市_住宅小区v5.csv')

3.使用方法。安装csv和request库，把代码中的key改为自己在高德开放平台申请的api就行了。
爬取的模式有两种，一种是根据csv来爬取，一种是自己定义搜索城市和搜索类型来爬取。前者适合自己得到地名爬取经纬度；后者适合爬取某区域的某种类型poi点。要选那种类型自己看代码的最后两行修改就可以了。


4.不足。
不知道是不是自己用的个人认证开发者账号从没充钱，关键字查询最多爬取200条数据，1.0还是2.0的接口都是这样，这应该是高德服务器端的设置。论坛中的帖子高德地图POI数据获取v1.1 2022.8.17更新的评论区也谈到了这个问题。上网检索，其他的帖子都说最多可以爬取1000条或者900条，但我只能爬取200条，或者说关键字查询只能爬取200条，而多边形查询能爬900-1000条？不知道有没有了解的坛友能热心回答一下。


5.吐槽。
地信专业，要爬取某行政区域内的所有居民点，结果高德只能爬取200条，大失所望；不知道百度地图的api爬取结果怎样，不说了，改百度地图代码去了，弄好再发出来看看能不能解决吧。

侃遍天下无二人

offset是偏移的意思，返回结果数应该是pageSize或者limit

xiexuhuai

侃遍天下无二人 发表于 2023-3-24 15:41
offset是偏移的意思，返回结果数应该是pageSize或者limit

高德官方文档是用的offset，2.0用的是page_size。

odmin

x 度的没有限制

[Python] 纯文本查看 复制代码

############# 程序功能 ####################
#    本程序用来获取百度地图某城市某种poi周围一定范围内的另一种类型的poi数据
#    如示例是获取武汉市中学周围500米内的网吧
#    生成的info.txt格式如下
#     1, xxx中学
#     1-1， xxx网吧
#     1-2， xxx网吧
#     2， xxx中学
#     2-1, xxx网吧
############################################

import requests, json
import time, sys

############### 自主设置区 ###############
ak = 'dASz7ubuSpHiWKuAK3weqwe'  # API key
keyword = "中学"
keyword2 = "网吧"
radius = 500
city = "武汉市"
baseBound = '29.972898,113.707695,31.367052,115.085775'  # 武汉市矩形框的左下角经纬度和右上角经纬度
############################################


# 构造header
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36',
    'Accept-Encoding': 'gzip, deflate',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3',
    'Host': 'api.map.baidu.com',
    'Upgrade-Insecure-Requests': '1',
    'Connection': 'keep-alive',
    'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7',
    'DNT': '1'
}

baseUrl = "http://api.map.baidu.com/place/v2/search?query={}&bounds={}&page_size=20&page_num={}&output=json&ak={}"
searchBaseUrl = "http://api.map.baidu.com/place/v2/search?query={}&location={}&radius={}&output=json&ak={}&page_size=20&page_num={}"


def req(url):
    # 访问url获取返回的数据读取为json
    errorNum = 0
    while True:
        try:
            res = requests.get(url, headers=headers)
            res.encoding = 'utf-8'
            jd = json.loads(res.text)
            return jd
        except Exception as e:
            errorNum += 1
            print("出错了！")
            print(e)
            if errorNum == 5:
                sys.exit(1)
            time.sleep(1)


def splitArea(bound):
    # 分割矩形区域的函数
    boundList = []
    row_num = 2  # 按照 2 X 2 进行分割
    bound = list(map(float, bound.split(',')))
    step_lat = (bound[2] - bound[0]) / row_num
    step_lon = (bound[3] - bound[1]) / row_num
    for i in range(0, row_num):
        for j in range(0, row_num):
            boundTemp = []
            boundTemp.append(bound[0] + step_lat * i)
            boundTemp.append(bound[1] + step_lon * j)
            boundTemp.append(bound[0] + step_lat * (i + 1))
            boundTemp.append(bound[1] + step_lon * (j + 1))
            boundTemp = ",".join(["%s" % x for x in boundTemp])
            # 是否要继续分割
            if check(boundTemp):
                boundList.append(boundTemp)
            else:
                boundList += splitArea(boundTemp)
    return boundList


def check(bound):
    # 检查一下当前矩形框是否需要再次分割
    checkUrl = baseUrl.format(keyword, bound, 0, ak)
    checkJd = req(checkUrl)
    if checkJd["total"] < 400:
        return True
    return False


def getPois(bound):
    # 获取一定范围内所有的想要的poi的信息
    poiListTemp = []
    pageNum = 0
    while True:
        getUrl = baseUrl.format(keyword, bound, pageNum, ak)
        getJd = req(getUrl)
        if getJd['results'] == []:  # 如果这一页没数据了就是结束啦，退出循环
            break
        for result in getJd['results']:
            if result["city"] != city:  # 在矩形框中，但是不是想要的城市的，就不要记录啦
                continue
            poiListTemp.append(result)
        pageNum += 1
    print("当前区域{}的poi已爬完，共有{}个".format(bound, len(poiListTemp)))
    return poiListTemp


def searchPois(poiInfo):
    # 获取某个poi周边的其他poi信息
    poiListTemp = []
    pageNum = 0
    while True:
        searchUrl = searchBaseUrl.format(keyword2,
                                         str(poiInfo['location']['lat']) + ',' + str(poiInfo['location']['lng']),
                                         radius, ak, pageNum)
        searchJd = req(searchUrl)
        if searchJd['results'] == []:
            break
        for result in searchJd['results']:
            poiListTemp.append(result["name"])
            print(result["name"], " 已添加")
        pageNum += 1
    print(poiInfo["name"], "周边poi已爬完")
    return [poiInfo["name"], poiListTemp]


if __name__ == '__main__':
    # 分割想要的区域
    boundList = splitArea(baseBound)
    print(boundList)

    # 获取范围内所有的poi信息
    poiList = []
    for bound in boundList:
        poiList += getPois(bound)
    print('已获得所有poi信息')

    # 对于获取到的每个poi进行周边检索
    infoList = []
    for poi in poiList:
        infoList.append(searchPois(poi))
    print('已获得所有周边poi信息')

    # 把获取到的信息都写入txt
    with open('info.txt', 'w') as f:
        for i in range(len(infoList)):
            f.write(str(i + 1) + ', ' + infoList[i][0] + '\n')
            for j in range(len(infoList[i][1])):
                f.write(str(i + 1) + '-' + str(j + 1) + ', ' + infoList[i][1][j] + '\n')
    print("所获得的信息已全部写入txt啦！")

giscloud

记得高德个人开发者也不止200条吧？又限制啦？

bennnz

giscloud 发表于 2023-4-9 01:17
记得高德个人开发者也不止200条吧？又限制啦？

早就限制了每天100

bennnz

odmin 发表于 2023-3-26 16:16
x 度的没有限制

[mw_shl_code=python,true]############# 程序功能 ####################

大佬，百度的个人开发者不是也限制每天100次poi请求吗

bennnz

腾讯个人用户数量10000/天，但是我用腾讯返回的数据经常会重复，或者大佬试一下腾讯的？

xiexuhuai

api请求次数是100次，但按照一次请求返回20条数据来算也应该有2000条，而不是200条

公主月月

大量求个好友位呀