简介基于Python实现百度地图POI数据抓取的实战代码包面向爬虫学习者和位置数据应用开发者解决城市级兴趣点批量获取与结构化存储问题既适合初学者理解爬虫框架也可作为中高级开发者快速搭建POI采集模块的参考。压缩包内共2个Python脚本文件整体仅3KB包含完整版城市级爬取脚本与多城市参数化处理脚本代码精简便于阅读和二次修改。已有1382人学习下载。围绕百度地图POI抓取场景资源完整覆盖requests网络请求、HTML/JSON解析、数据清洗、Session与Cookie会话模拟、多线程并发提速及CSV/JSON存储等关键环节并给出城市名动态传参的构建思路针对频繁访问触发反爬的情况代码也通过请求头设置与访问间隔作了基本规避处理。学习这份代码可快速掌握地图类爬虫的通用流程与常见反爬应对技巧进而根据业务需要扩展为商圈检索、客流分析或城市规划等数据采集工具。1. 城市级POI爬虫与其买数据不如自己抓一份百度的第一次做 POI 抓取时我在网上搜到的源码大多是“按矩形框切网格”的版本跑一次少则几十分钟多则几个小时还动不动被限流。后来我换成了直接按城市名调百度地图接口配合分页和并发限速整个城市几十万条 POI 也能在可接受的时间里落盘。这份源码包里完整版城市级.py和city_version.py就是干这件事的输入城市名、选中业态类型自动翻页抓取输出结构化数据。它不依赖 Selenium不碰渲染核心就是 requests JSON 解析 线程池。适合理清 POI 数据从请求、解析到存储的全链路也适合需要周期性更新城市商家点位做分析的人。2. POI请求层URL参数、坐标系与反爬前置2.1 先搞懂百度地图POI查询参数百度地图 Web API 的 POI 检索接口最常用的是place/v2/search。它的 HTTP 请求是 GET核心参数包括query关键词比如“餐厅”、region城市名、outputjson、ak开发者密钥、page_size、page_num。城市级抓取实际上就是在region里填入城市名然后遍历page_num。还有一个容易被忽略的参数是scope它决定返回详情的详细程度scope2能拿回更多字段。我整理了一份常用参数表写代码前建议先对照着看参数示例值作用抓取时的注意点query美食、酒店、学校POI 关键词一次只传一个词词太宽会触发配额限制region北京城市名直辖市、地级市都能用拼音不行outputjson返回格式别用 xml解析成本高ak你的密钥鉴权一个 key 每天有配额超了会被拒page_size20每页条数官方上限 20调大无效page_num0页码从 0 开始和 total 配合做循环scope2详情级别需要地址、电话时用 2这里有一个新手容易踩的坑page_size写大了之后接口不会报错但返回的results数量照样只有 20。所以分页逻辑必须按total除以 20 向上取整来计算页数而不是按名字猜测。2.2 requests 请求的完整写法拿到了参数表就可以写最基础的请求函数。我不喜欢把ak写死在代码里一般用环境变量读取这样不同环境切换时不用改业务代码。import os import time import requests def baidu_poi_request(query: str, region: str, page_num: int 0, page_size: int 20, ak: str None) - dict: if ak is None: ak os.getenv(BAIDU_AK) params { query: query, region: region, output: json, ak: ak, page_size: page_size, page_num: page_num, scope: 2, } resp requests.get( https://api.map.baidu.com/place/v2/search, paramsparams, timeout10, headers{User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64)} ) resp.raise_for_status() return resp.json()这段代码做了三件事把参数全部收进字典交给requests.get的params由它负责 URL 编码设置 10 秒超时防止接口卡死指定浏览器 User-Agent 减少被识别成脚本的概率。time模块暂时没用等后面做限速的时候会派上用场。注意这里的params中page_size保持 20不要自行调大。2.3 接口返回结构总数量与分页接口返回的是一个 JSON核心字段是total、results和status。我在实际抓取时发现status不等于 0 时results是空的。常见错误码里status1表示服务器错误status2表示请求参数错误status4是权限问题status302表示当天配额用尽。下面这段代码的作用就是根据total计算总页数再逐页拉数据。由于每次循环都要判断status我不会把分页逻辑合并进请求函数而是单独封装一个循环函数。def fetch_all_pois(query: str, region: str) - list: first baidu_poi_request(query, region, page_num0) total first.get(total, 0) if total 0: return [] all_pois first.get(results, []) page_count (total 19) // 20 for page in range(1, page_count): data baidu_poi_request(query, region, page_numpage) if data.get(status) 0: all_pois.extend(data.get(results, [])) time.sleep(0.3) return all_pois这段分页代码用(total 19) // 20做向上取整避免漏掉余数页。每次翻页之间我加了0.3秒的延迟虽然只是简单的time.sleep但对于限制并发频率已经足够。真正跑的时候你可以在fetch_all_pois外面包一层异常捕获防止单页失败导致整个城市的数据中断。3. 数据解析与落库把 JSON 变成可用的 CSV3.1 JSON 数据清洗与字段抽取百度返回的results数组里每个元素又是一个嵌套字典。字段常用的有name、location里面还有lng、lat、address、telephone、detail_url、province、city、area等。我见过很多人直接把resp.json()整个扔进 excel结果地址、电话列全是空的原因就是没有处理嵌套结构。清洗的目标是把嵌套结构拉平成一行记录。下面这个函数按白名单抽字段遇到缺失值统一补空字符串避免后续写 CSV 时出现列错位。def flatten_poi(item: dict) - dict: location item.get(location, {}) return { name: item.get(name, ), lng: location.get(lng, ), lat: location.get(lat, ), address: item.get(address, ), telephone: item.get(telephone, ), province: item.get(province, ), city: item.get(city, ), area: item.get(area, ), detail_url: item.get(detail_url, ), }我在实际项目里还遇到过address字段带\n和连续空格的情况处理方式是value.replace(\n, ).strip()。这条清理规则可以写在flatten_poi的 return 之前统一做一个字段级清洗。location如果不存在直接给lng和lat置空不要用.get(location, {}).get(lng, )以外的方式访问否则会抛AttributeError。3.2 城市级循环用 city_version 的思路批量抓取city_version.py的核心思路是按城市名做外层遍历城市名从列表或者配置文件里读取。这里有一个关键点不同的城市同一个query的总量差别巨大像北京、上海单业态可能过万小城市只有几百条。所以要在每个城市第一次请求后立即记录total再决定是否继续分页。我用的是读取城市的文件来驱动循环这样后续新增城市不用改代码def run_city_batch(cities: list, query: str): for city in cities: print(f[{city}] 开始抓取 {query}...) pois fetch_all_pois(query, city) save_to_csv(pois, fpoi_{city}_{query}.csv) print(f[{city}] 完成共 {len(pois)} 条)这段代码对应源码包里city_version.py的骨架外层遍历城市内层遍历分页。脚本在每次完成一个城市后打印行数方便人工核对是否抓全。注意我用了query作为文件名的一部分这样可以避免多个业态类型互相覆盖。如果你要抓多个业态可以再加一层for query in [美食, 酒店, 学校]的循环。3.3 存储格式选择CSV 还是 SQLite对于城市级 POI 数据我一般分两种情况处理。如果只是给业务方做一次性的数据整理直接写 CSV因为 Excel 和 BI 工具打开方便。但如果要做增量更新、多城市合并去重就必须用 SQLite。CSV 写起来简单可一旦超过十万行后续过滤、去重就只能依赖 pandas内存吃紧。下面是写 CSV 的完整逻辑用了newline参数避免 Windows 下出现空行。标题行先通过fieldnames固定字典缺失的字段会被csv.DictWriter自动填成空值。import csv def save_to_csv(pois: list, file_path: str) - None: if not pois: return fieldnames [name, lng, lat, address, telephone, province, city, area, detail_url] with open(file_path, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(pois)这里编码选了utf-8-sig就是为了让 Excel 直接打开时不含乱码。如果你用 pandas 后续分析可以改成encodingutf-8没什么影响。还有一点是写文件前先判断pois是否为空否则只写一个标题空文件对后续合并很麻烦。4. 并发与反爬多线程、重试和限速的平衡4.1 用 ThreadPoolExecutor 提速单独按城市循环跑最慢的就是单页请求的串行等待。一个城市一万条数据要请求 500 次每次 0.3 秒延迟加 0.2 秒响应就是 250 秒。把页与页之间的顺序请求拆成并发可以显著缩短整体时间。我用的是concurrent.futures.ThreadPoolExecutor线程数控制在 5 左右避免百度侧把任务太容易识别成并发攻击。from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_page_wrapper(args): query, region, page args return baidu_poi_request(query, region, page_numpage) def fetch_all_pois_parallel(query: str, region: str, workers: int 5) - list: first baidu_poi_request(query, region, page_num0) total first.get(total, 0) if total 0: return [] page_count (total 19) // 20 results first.get(results, []) with ThreadPoolExecutor(max_workersworkers) as executor: futures [ executor.submit(fetch_page_wrapper, (query, region, page)) for page in range(1, page_count) ] for future in as_completed(futures): data future.result() if data.get(status) 0: results.extend(data.get(results, [])) return results这段代码用page作为任务的最小单位而不是把整个城市作为一个任务这样线程池能更均匀地分配请求。as_completed的作用是拿到一个就先处理一个减少内存里堆积的 future 对象。你可能会看到网上有人用gather的半协程写法那种方案对代理和 DNS 解析有额外要求线程池更稳一些。4.2 限速与重试的细节并发一旦开启重试逻辑就不能只写一种time.sleep。我在代码里把重试拆成三层连接错误重试、状态码异常重试、配额超限重试。连接错误可以立即重试最多三次配额限制则需要等待一个相对长的时间比如 30 秒后再试。最简单的实现是用装饰器包装baidu_poi_request。import functools def retry_request(func, retries: int 3, wait: float 1.0): functools.wraps(func) def wrapper(*args, **kwargs): for attempt in range(retries): try: resp func(*args, **kwargs) if resp.get(status) 0: return resp if resp.get(status) 302: time.sleep(30) except requests.RequestException: time.sleep(wait * (attempt 1)) return {} return wrapper这个重试包装器在每次请求失败后都会指数退避第一次等待 1 秒第二次 2 秒第三次 3 秒。遇到 302 配额耗尽时直接睡 30 秒再试这比快速重刷更有效。实际上配额一旦耗尽等 30 秒不一定能恢复我一般会在外部判断返回值里的status如果连续 3 次都是 302就直接结束当前城市抓取留下进度记录下次用增量方式继续。4.3 遇到配额限制时的降级策略百度 POI 接口的一天配额和并发数有关如果你只是个人练习一个 key 的配额可能只够抓两三个中型城市。遇到配额不够时一个可用的降级策略是把 POI 拆成多个小网格每个网格单独请求。虽然请求次数更多但能避开单 key 的每日上限也方便断点续抓。我常用的降级流程是先在当前城市中心点生成一个矩形范围按 0.01 度间距切割成网格然后把每个网格的bounds参数传给请求。bounds的格式是lng1,lat1;lng2,lat2比如116.3,39.9;116.4,40.0。这样可以绕开 region 方式搜索时有 4000 条的返回上限代价是页面重复项变多需要按name加lng/lat做去重。5. 验证抓取结果并落成一个可复用命令5.1 校验 POI 数量与坐标偏移抓完一个城市后除了看打印的行数我还会做三个校验。第一是抽样几个detail_url确认链接能返回真实详情页第二是检查area字段是否覆盖了所有行政区比如北京的area应该包含朝阳区、海淀区等第三是坐标偏移检查百度返回的是 BD-09 坐标系我有时会随机抽几条数据和高德地图的 GCJ-02 坐标做对比如果经纬度差异超过 100 米就需要在后续做坐标转换。5.2 把脚本包装成接受城市参数的入口源码包里完整版城市级.py已经把主流程串通了我在本地使用时会再加一个argparse入口让脚本支持从命令行传城市名和业态关键词。这个改造很轻量但实际复用价值很高配合 cron 就能定时更新某个城市的 POI。import argparse if __name__ __main__: parser argparse.ArgumentParser(description城市POI抓取) parser.add_argument(--city, requiredTrue, help城市名如广州) parser.add_argument(--query, default美食, helpPOI关键词如美食、酒店) args parser.parse_args() pois fetch_all_pois_parallel(args.query, args.city, workers5) save_to_csv(pois, fpoi_{args.city}_{args.query}.csv) print(f保存完成共 {len(pois)} 条)命令行执行方式就是python city_version.py --city 广州 --query 美食。这里把城市名作为必填参数业态作为可选参数默认抓“美食”。跑完生成的文件名会带上城市和业态方便多个城市的结果同时存在一个目录里不混淆。如果你要定时更新就把这条命令写进 Linux 的 crontab每天凌晨跑一次再对抓到的 CSV 做增量去重。这样一套下来你不需要维护任何可视化界面只用一条命令就能拿到一份能直接送进数据分析流程的城市 POI 表。本文还有配套的精品资源点击获取 百度地图POI爬虫实战：城市级数据抓取与存储全流程
简介基于Python实现百度地图POI数据抓取的实战代码包面向爬虫学习者和位置数据应用开发者解决城市级兴趣点批量获取与结构化存储问题既适合初学者理解爬虫框架也可作为中高级开发者快速搭建POI采集模块的参考。压缩包内共2个Python脚本文件整体仅3KB包含完整版城市级爬取脚本与多城市参数化处理脚本代码精简便于阅读和二次修改。已有1382人学习下载。围绕百度地图POI抓取场景资源完整覆盖requests网络请求、HTML/JSON解析、数据清洗、Session与Cookie会话模拟、多线程并发提速及CSV/JSON存储等关键环节并给出城市名动态传参的构建思路针对频繁访问触发反爬的情况代码也通过请求头设置与访问间隔作了基本规避处理。学习这份代码可快速掌握地图类爬虫的通用流程与常见反爬应对技巧进而根据业务需要扩展为商圈检索、客流分析或城市规划等数据采集工具。1. 城市级POI爬虫与其买数据不如自己抓一份百度的第一次做 POI 抓取时我在网上搜到的源码大多是“按矩形框切网格”的版本跑一次少则几十分钟多则几个小时还动不动被限流。后来我换成了直接按城市名调百度地图接口配合分页和并发限速整个城市几十万条 POI 也能在可接受的时间里落盘。这份源码包里完整版城市级.py和city_version.py就是干这件事的输入城市名、选中业态类型自动翻页抓取输出结构化数据。它不依赖 Selenium不碰渲染核心就是 requests JSON 解析 线程池。适合理清 POI 数据从请求、解析到存储的全链路也适合需要周期性更新城市商家点位做分析的人。2. POI请求层URL参数、坐标系与反爬前置2.1 先搞懂百度地图POI查询参数百度地图 Web API 的 POI 检索接口最常用的是place/v2/search。它的 HTTP 请求是 GET核心参数包括query关键词比如“餐厅”、region城市名、outputjson、ak开发者密钥、page_size、page_num。城市级抓取实际上就是在region里填入城市名然后遍历page_num。还有一个容易被忽略的参数是scope它决定返回详情的详细程度scope2能拿回更多字段。我整理了一份常用参数表写代码前建议先对照着看参数示例值作用抓取时的注意点query美食、酒店、学校POI 关键词一次只传一个词词太宽会触发配额限制region北京城市名直辖市、地级市都能用拼音不行outputjson返回格式别用 xml解析成本高ak你的密钥鉴权一个 key 每天有配额超了会被拒page_size20每页条数官方上限 20调大无效page_num0页码从 0 开始和 total 配合做循环scope2详情级别需要地址、电话时用 2这里有一个新手容易踩的坑page_size写大了之后接口不会报错但返回的results数量照样只有 20。所以分页逻辑必须按total除以 20 向上取整来计算页数而不是按名字猜测。2.2 requests 请求的完整写法拿到了参数表就可以写最基础的请求函数。我不喜欢把ak写死在代码里一般用环境变量读取这样不同环境切换时不用改业务代码。import os import time import requests def baidu_poi_request(query: str, region: str, page_num: int 0, page_size: int 20, ak: str None) - dict: if ak is None: ak os.getenv(BAIDU_AK) params { query: query, region: region, output: json, ak: ak, page_size: page_size, page_num: page_num, scope: 2, } resp requests.get( https://api.map.baidu.com/place/v2/search, paramsparams, timeout10, headers{User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64)} ) resp.raise_for_status() return resp.json()这段代码做了三件事把参数全部收进字典交给requests.get的params由它负责 URL 编码设置 10 秒超时防止接口卡死指定浏览器 User-Agent 减少被识别成脚本的概率。time模块暂时没用等后面做限速的时候会派上用场。注意这里的params中page_size保持 20不要自行调大。2.3 接口返回结构总数量与分页接口返回的是一个 JSON核心字段是total、results和status。我在实际抓取时发现status不等于 0 时results是空的。常见错误码里status1表示服务器错误status2表示请求参数错误status4是权限问题status302表示当天配额用尽。下面这段代码的作用就是根据total计算总页数再逐页拉数据。由于每次循环都要判断status我不会把分页逻辑合并进请求函数而是单独封装一个循环函数。def fetch_all_pois(query: str, region: str) - list: first baidu_poi_request(query, region, page_num0) total first.get(total, 0) if total 0: return [] all_pois first.get(results, []) page_count (total 19) // 20 for page in range(1, page_count): data baidu_poi_request(query, region, page_numpage) if data.get(status) 0: all_pois.extend(data.get(results, [])) time.sleep(0.3) return all_pois这段分页代码用(total 19) // 20做向上取整避免漏掉余数页。每次翻页之间我加了0.3秒的延迟虽然只是简单的time.sleep但对于限制并发频率已经足够。真正跑的时候你可以在fetch_all_pois外面包一层异常捕获防止单页失败导致整个城市的数据中断。3. 数据解析与落库把 JSON 变成可用的 CSV3.1 JSON 数据清洗与字段抽取百度返回的results数组里每个元素又是一个嵌套字典。字段常用的有name、location里面还有lng、lat、address、telephone、detail_url、province、city、area等。我见过很多人直接把resp.json()整个扔进 excel结果地址、电话列全是空的原因就是没有处理嵌套结构。清洗的目标是把嵌套结构拉平成一行记录。下面这个函数按白名单抽字段遇到缺失值统一补空字符串避免后续写 CSV 时出现列错位。def flatten_poi(item: dict) - dict: location item.get(location, {}) return { name: item.get(name, ), lng: location.get(lng, ), lat: location.get(lat, ), address: item.get(address, ), telephone: item.get(telephone, ), province: item.get(province, ), city: item.get(city, ), area: item.get(area, ), detail_url: item.get(detail_url, ), }我在实际项目里还遇到过address字段带\n和连续空格的情况处理方式是value.replace(\n, ).strip()。这条清理规则可以写在flatten_poi的 return 之前统一做一个字段级清洗。location如果不存在直接给lng和lat置空不要用.get(location, {}).get(lng, )以外的方式访问否则会抛AttributeError。3.2 城市级循环用 city_version 的思路批量抓取city_version.py的核心思路是按城市名做外层遍历城市名从列表或者配置文件里读取。这里有一个关键点不同的城市同一个query的总量差别巨大像北京、上海单业态可能过万小城市只有几百条。所以要在每个城市第一次请求后立即记录total再决定是否继续分页。我用的是读取城市的文件来驱动循环这样后续新增城市不用改代码def run_city_batch(cities: list, query: str): for city in cities: print(f[{city}] 开始抓取 {query}...) pois fetch_all_pois(query, city) save_to_csv(pois, fpoi_{city}_{query}.csv) print(f[{city}] 完成共 {len(pois)} 条)这段代码对应源码包里city_version.py的骨架外层遍历城市内层遍历分页。脚本在每次完成一个城市后打印行数方便人工核对是否抓全。注意我用了query作为文件名的一部分这样可以避免多个业态类型互相覆盖。如果你要抓多个业态可以再加一层for query in [美食, 酒店, 学校]的循环。3.3 存储格式选择CSV 还是 SQLite对于城市级 POI 数据我一般分两种情况处理。如果只是给业务方做一次性的数据整理直接写 CSV因为 Excel 和 BI 工具打开方便。但如果要做增量更新、多城市合并去重就必须用 SQLite。CSV 写起来简单可一旦超过十万行后续过滤、去重就只能依赖 pandas内存吃紧。下面是写 CSV 的完整逻辑用了newline参数避免 Windows 下出现空行。标题行先通过fieldnames固定字典缺失的字段会被csv.DictWriter自动填成空值。import csv def save_to_csv(pois: list, file_path: str) - None: if not pois: return fieldnames [name, lng, lat, address, telephone, province, city, area, detail_url] with open(file_path, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(pois)这里编码选了utf-8-sig就是为了让 Excel 直接打开时不含乱码。如果你用 pandas 后续分析可以改成encodingutf-8没什么影响。还有一点是写文件前先判断pois是否为空否则只写一个标题空文件对后续合并很麻烦。4. 并发与反爬多线程、重试和限速的平衡4.1 用 ThreadPoolExecutor 提速单独按城市循环跑最慢的就是单页请求的串行等待。一个城市一万条数据要请求 500 次每次 0.3 秒延迟加 0.2 秒响应就是 250 秒。把页与页之间的顺序请求拆成并发可以显著缩短整体时间。我用的是concurrent.futures.ThreadPoolExecutor线程数控制在 5 左右避免百度侧把任务太容易识别成并发攻击。from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_page_wrapper(args): query, region, page args return baidu_poi_request(query, region, page_numpage) def fetch_all_pois_parallel(query: str, region: str, workers: int 5) - list: first baidu_poi_request(query, region, page_num0) total first.get(total, 0) if total 0: return [] page_count (total 19) // 20 results first.get(results, []) with ThreadPoolExecutor(max_workersworkers) as executor: futures [ executor.submit(fetch_page_wrapper, (query, region, page)) for page in range(1, page_count) ] for future in as_completed(futures): data future.result() if data.get(status) 0: results.extend(data.get(results, [])) return results这段代码用page作为任务的最小单位而不是把整个城市作为一个任务这样线程池能更均匀地分配请求。as_completed的作用是拿到一个就先处理一个减少内存里堆积的 future 对象。你可能会看到网上有人用gather的半协程写法那种方案对代理和 DNS 解析有额外要求线程池更稳一些。4.2 限速与重试的细节并发一旦开启重试逻辑就不能只写一种time.sleep。我在代码里把重试拆成三层连接错误重试、状态码异常重试、配额超限重试。连接错误可以立即重试最多三次配额限制则需要等待一个相对长的时间比如 30 秒后再试。最简单的实现是用装饰器包装baidu_poi_request。import functools def retry_request(func, retries: int 3, wait: float 1.0): functools.wraps(func) def wrapper(*args, **kwargs): for attempt in range(retries): try: resp func(*args, **kwargs) if resp.get(status) 0: return resp if resp.get(status) 302: time.sleep(30) except requests.RequestException: time.sleep(wait * (attempt 1)) return {} return wrapper这个重试包装器在每次请求失败后都会指数退避第一次等待 1 秒第二次 2 秒第三次 3 秒。遇到 302 配额耗尽时直接睡 30 秒再试这比快速重刷更有效。实际上配额一旦耗尽等 30 秒不一定能恢复我一般会在外部判断返回值里的status如果连续 3 次都是 302就直接结束当前城市抓取留下进度记录下次用增量方式继续。4.3 遇到配额限制时的降级策略百度 POI 接口的一天配额和并发数有关如果你只是个人练习一个 key 的配额可能只够抓两三个中型城市。遇到配额不够时一个可用的降级策略是把 POI 拆成多个小网格每个网格单独请求。虽然请求次数更多但能避开单 key 的每日上限也方便断点续抓。我常用的降级流程是先在当前城市中心点生成一个矩形范围按 0.01 度间距切割成网格然后把每个网格的bounds参数传给请求。bounds的格式是lng1,lat1;lng2,lat2比如116.3,39.9;116.4,40.0。这样可以绕开 region 方式搜索时有 4000 条的返回上限代价是页面重复项变多需要按name加lng/lat做去重。5. 验证抓取结果并落成一个可复用命令5.1 校验 POI 数量与坐标偏移抓完一个城市后除了看打印的行数我还会做三个校验。第一是抽样几个detail_url确认链接能返回真实详情页第二是检查area字段是否覆盖了所有行政区比如北京的area应该包含朝阳区、海淀区等第三是坐标偏移检查百度返回的是 BD-09 坐标系我有时会随机抽几条数据和高德地图的 GCJ-02 坐标做对比如果经纬度差异超过 100 米就需要在后续做坐标转换。5.2 把脚本包装成接受城市参数的入口源码包里完整版城市级.py已经把主流程串通了我在本地使用时会再加一个argparse入口让脚本支持从命令行传城市名和业态关键词。这个改造很轻量但实际复用价值很高配合 cron 就能定时更新某个城市的 POI。import argparse if __name__ __main__: parser argparse.ArgumentParser(description城市POI抓取) parser.add_argument(--city, requiredTrue, help城市名如广州) parser.add_argument(--query, default美食, helpPOI关键词如美食、酒店) args parser.parse_args() pois fetch_all_pois_parallel(args.query, args.city, workers5) save_to_csv(pois, fpoi_{args.city}_{args.query}.csv) print(f保存完成共 {len(pois)} 条)命令行执行方式就是python city_version.py --city 广州 --query 美食。这里把城市名作为必填参数业态作为可选参数默认抓“美食”。跑完生成的文件名会带上城市和业态方便多个城市的结果同时存在一个目录里不混淆。如果你要定时更新就把这条命令写进 Linux 的 crontab每天凌晨跑一次再对抓到的 CSV 做增量去重。这样一套下来你不需要维护任何可视化界面只用一条命令就能拿到一份能直接送进数据分析流程的城市 POI 表。本文还有配套的精品资源点击获取