手把手教你用Python批量下载全国省市区县GeoJSON地图数据(2023最新版)
你是否曾经为了一个数据可视化项目,需要一张中国地图的轮廓,却发现自己卡在了数据获取的第一步?网上零散的下载链接要么失效,要么格式不统一,手动一个个下载省、市、区的边界数据,工作量简直让人望而却步。对于数据分析师、前端开发者,或者任何需要将业务数据与地理空间结合的朋友来说,一份准确、完整、格式规范的行政区划GeoJSON数据,往往是项目启动的“临门一脚”。
今天,我们就来彻底解决这个问题。我将带你从零开始,用Python写一个自动化脚本,一次性获取全国从省级到区县级的最新GeoJSON地图数据。整个过程不需要你事先精通爬虫,我会把每一步的原理、可能遇到的“坑”以及对应的解决方案都掰开揉碎讲清楚。我们的目标不仅仅是拿到数据,更是让你理解背后的逻辑,以后遇到类似的数据抓取需求,也能举一反三。
1. 环境准备与数据源探秘
在动手写代码之前,搭建一个稳定、清晰的工作环境至关重要。这能避免很多因环境混乱导致的诡异错误。
1.1 Python环境与必要库
首先,确保你的电脑上安装了Python。我强烈推荐使用Python 3.7或更高版本。你可以打开终端(Windows上是命令提示符或PowerShell,macOS/Linux上是Terminal),输入python --version来检查。
接下来,我们需要安装几个核心的Python库。打开你的终端,执行以下命令:
pip install requests是的,只需要这一个。我们将使用requests库来替代原始文章中使用的urllib,因为它更简洁、更强大,错误处理也更友好。至于处理JSON和操作文件,Python的标准库json和os已经足够强大,无需额外安装。
提示:如果你在安装过程中遇到网络问题,可以考虑为pip命令添加镜像源,例如
pip install requests -i https://pypi.tuna.tsinghua.edu.cn/simple。
1.2 理解我们的数据源:阿里云DataV
工欲善其事,必先利其器。我们得先搞清楚数据从哪来,是什么结构。本次教程使用的数据源是阿里云DataV的地理小工具。这是一个公开的、维护相对及时的地理信息数据服务。
它的数据组织方式非常清晰:
- 数据以行政区划代码(adcode)作为唯一标识。
- 提供了两种关键的几何类型:
- 标准边界:仅包含该行政区自身的轮廓。
- 完整边界:包含其下所有子级行政区的轮廓集合(例如,“广东省_full”包含了广东省内所有城市的地块)。
为了后续编程,我们需要先获取一个“目录”或“索引”文件,这个文件里列出了所有可用的行政区划单元及其元信息(如名称、层级、中心点坐标)。这个索引文件的地址就是我们脚本的起点。
| 数据层级 | 说明 | 示例adcode |
|---|---|---|
| country | 国家 | 100000 (中国) |
| province | 省级(省、直辖市、自治区) | 440000 (广东省) |
| city | 市级(地级市、地区) | 440300 (深圳市) |
| district | 区县级(区、县、县级市) | 440304 (福田区) |
理解这个结构后,我们的脚本逻辑就呼之欲出了:先获取索引,再根据索引中的列表,循环下载每一个行政区划的GeoJSON文件,并按照层级和类型分类保存。
2. 构建核心爬取脚本
现在,让我们进入核心环节——编写Python脚本。我会将代码分成几个功能模块来讲解,确保每一部分你都知其然,也知其所以然。
2.1 获取行政区划索引
脚本的第一步是获取那个至关重要的“目录”。我们使用requests库来发送一个HTTP GET请求。
import requests import json import os # 定义数据源URL INDEX_URL = "https://geo.datav.aliyun.com/areas_v2/bound/infos.json" def fetch_area_index(): """ 获取全国行政区划的索引信息。 返回一个字典,键为adcode,值为包含名称、层级、中心点等信息的字典。 """ try: # 发送GET请求,设置超时时间避免无限等待 response = requests.get(INDEX_URL, timeout=10) response.raise_for_status() # 如果状态码不是200,抛出HTTPError异常 area_index = response.json() # 直接将响应内容解析为JSON print(f"成功获取索引,共包含 {len(area_index)} 个行政区划单元。") return area_index except requests.exceptions.RequestException as e: print(f"获取索引失败: {e}") return None except json.JSONDecodeError as e: print(f"解析索引JSON失败: {e}") return None # 测试一下这个函数 if __name__ == "__main__": index_data = fetch_area_index() if index_data: # 查看第一个条目(字典是无序的,这里用next(iter())获取第一个键值对) first_key, first_value = next(iter(index_data.items())) print(f"示例条目 - Adcode: {first_key}, 名称: {first_value.get('name')}, 层级: {first_value.get('level')}")这段代码做了几件关键事:
- 异常处理:使用
try...except包裹网络请求和JSON解析,这是编写健壮爬虫的基本功。网络不稳定、服务器错误、数据格式变化都可能导致程序崩溃,良好的异常处理能让程序优雅地报告错误而非直接退出。 - 超时设置:
timeout=10意味着如果10秒内没有收到响应,就认为请求失败。这防止了程序因网络卡顿而无限期挂起。 - 状态码检查:
response.raise_for_status()会在服务器返回404、500等错误状态码时抛出异常,让我们能及时知晓请求未成功。
2.2 设计高效的文件存储结构
数据下载下来,乱糟糟地堆在一个文件夹里可不是好主意。我们需要一个清晰、自动化的目录结构。参考数据源的特点,我设计了如下结构:
geojson_data_2023/ ├── index.json # 保存的行政区划索引 ├── country/ │ ├── 100000.json # 中国轮廓 │ └── 100000_full.json # 中国包含所有省份的完整版 ├── province/ │ ├── 110000.json # 北京市轮廓 │ ├── 110000_full.json # 北京市包含所有区县的完整版 │ ├── 440000.json # 广东省轮廓 │ └── ... ├── city/ │ ├── 110100.json # 北京市(直辖市级,特殊处理) │ ├── 440300.json # 深圳市轮廓 │ └── ... └── district/ ├── 110101.json # 东城区轮廓 ├── 440304.json # 福田区轮廓 └── ...实现这个结构的代码如下:
def ensure_directory_structure(base_dir="geojson_data"): """ 确保基础目录和各级子目录存在。 """ directories = [ base_dir, os.path.join(base_dir, "country"), os.path.join(base_dir, "province"), os.path.join(base_dir, "city"), os.path.join(base_dir, "district"), ] for dir_path in directories: os.makedirs(dir_path, exist_ok=True) # exist_ok=True 表示如果目录已存在也不报错 print(f"目录结构已在 '{base_dir}' 下准备就绪。") return base_dir2.3 实现数据下载与保存函数
有了索引和目录,现在可以编写下载单个GeoJSON文件和保存文件的函数了。
def download_geojson(adcode, is_full=False): """ 根据行政区划代码下载GeoJSON数据。 :param adcode: 行政区划代码,字符串 :param is_full: 是否下载包含子区域的完整版 :return: 下载成功的GeoJSON字典,失败则返回None """ # 构建下载URL suffix = "_full" if is_full else "" url = f"https://geo.datav.aliyun.com/areas_v2/bound/{adcode}{suffix}.json" try: response = requests.get(url, timeout=15) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"下载 {adcode}{suffix} 失败: {e}") return None except json.JSONDecodeError as e: print(f"解析 {adcode}{suffix} 的JSON失败: {e}") return None def save_geojson(data, adcode, level, base_dir, is_full=False): """ 将GeoJSON数据保存到对应的目录和文件中。 :param data: GeoJSON字典数据 :param adcode: 行政区划代码 :param level: 层级 ('country', 'province', 'city', 'district') :param base_dir: 基础目录 :param is_full: 是否为完整版 """ if data is None: return False # 确定文件名和目录 suffix = "_full" if is_full else "" # 对于‘country_full’这种,我们仍然保存在country目录下,只是文件名不同 save_dir = os.path.join(base_dir, level) filename = f"{adcode}{suffix}.json" filepath = os.path.join(save_dir, filename) try: with open(filepath, 'w', encoding='utf-8') as f: # ensure_ascii=False 保证中文正常显示,indent=2 让JSON文件更美观易读 json.dump(data, f, ensure_ascii=False, indent=2) # print(f"已保存: {filepath}") # 下载时太多输出,这里先注释,最后统一报告 return True except IOError as e: print(f"保存文件 {filepath} 失败: {e}") return False注意:
json.dump中的indent=2参数会让生成的JSON文件带有缩进,便于人类阅读,但会略微增加文件大小。如果纯粹用于程序读取,可以去掉这个参数。
3. 整合与执行:主流程控制
将上面的模块像拼图一样组合起来,就形成了我们的主程序。主程序需要控制下载流程,处理可能出现的错误,并给用户清晰的进度反馈。
3.1 主函数逻辑拆解
主函数的逻辑流如下:
- 创建目录。
- 获取索引并保存。
- 遍历索引中的每一个行政区划。
- 对于每个行政区划,下载其“标准边界”文件并保存。
- 如果该行政区划不是最底层的“district”,则再下载其“完整边界”文件并保存。
- 处理下载中的错误,记录成功与失败的数量。
def main(): BASE_DIR = "geojson_data_2023" # 你可以修改为你喜欢的文件夹名 ensure_directory_structure(BASE_DIR) # 1. 获取并保存索引 print("步骤1/4: 正在获取行政区划索引...") area_index = fetch_area_index() if not area_index: print("无法获取索引,程序终止。") return index_file = os.path.join(BASE_DIR, "index.json") with open(index_file, 'w', encoding='utf-8') as f: json.dump(area_index, f, ensure_ascii=False, indent=2) print(f"索引已保存至: {index_file}") total_areas = len(area_index) success_count = 0 fail_count = 0 fail_list = [] print(f"步骤2/4: 开始下载 {total_areas} 个行政区划的数据...") # 2. 遍历下载 for i, (adcode, info) in enumerate(area_index.items(), 1): level = info.get('level') name = info.get('name', '未知') # 进度显示 if i % 50 == 0 or i == total_areas: print(f" 进度: {i}/{total_areas} - 正在处理 [{name}]({adcode})...") # 下载标准版 standard_data = download_geojson(adcode, is_full=False) if save_geojson(standard_data, adcode, level, BASE_DIR, is_full=False): success_count += 1 else: fail_count += 1 fail_list.append(f"{adcode}({name})-标准版") # 如果不是区县级,下载完整版 if level != 'district': full_data = download_geojson(adcode, is_full=True) if save_geojson(full_data, adcode, level, BASE_DIR, is_full=True): success_count += 1 else: fail_count += 1 fail_list.append(f"{adcode}({name})-完整版") # 3. 输出总结报告 print("\n" + "="*50) print("下载任务完成!") print(f"总计处理单元: {total_areas}") print(f"成功保存文件数: {success_count}") print(f"失败文件数: {fail_count}") if fail_list: print("\n失败列表(可能由于网络或数据源暂时缺失):") for item in fail_list: print(f" - {item}") print("建议:可以稍后重新运行脚本,单独尝试下载失败的项目。") print(f"\n所有数据已保存至 '{BASE_DIR}' 目录。") print("="*50) if __name__ == "__main__": main()3.2 处理网络波动与重试机制
在实际运行中,网络波动可能导致个别文件下载失败。一个健壮的脚本应该具备简单的重试能力。我们可以改进download_geojson函数:
def download_geojson_with_retry(adcode, is_full=False, max_retries=2): """ 带重试机制的下载函数。 """ for attempt in range(max_retries + 1): # 尝试 max_retries + 1 次 data = download_geojson(adcode, is_full) if data is not None: return data elif attempt < max_retries: wait_time = 2 ** attempt # 指数退避:1秒,2秒,4秒... print(f" 第{attempt+1}次下载失败,{wait_time}秒后重试...") time.sleep(wait_time) else: print(f" 下载 {adcode}{'_full' if is_full else ''} 失败,已达最大重试次数。") return None记得在文件开头import time。然后在主循环中,用download_geojson_with_retry替换download_geojson。这个小技巧能显著提高在不太稳定网络环境下的成功率。
4. 数据验证、应用与拓展
脚本运行完毕,数据到手了,但这还不是终点。我们还需要知道如何验证数据的完整性,以及如何将这些数据用起来。
4.1 如何验证下载的数据
直接打开成百上千个JSON文件检查是不现实的。我们可以写一个简单的校验脚本:
import os import json def validate_data_integrity(base_dir="geojson_data_2023"): """ 快速校验数据完整性:检查各目录下文件数量是否大致合理,并随机抽样检查文件格式。 """ level_dirs = ["country", "province", "city", "district"] expected_min_counts = { # 基于常识的预期最小数量 "country": 1, "province": 30, # 23省+5自治区+4直辖市+2特别行政区 "city": 300, "district": 2000, } print("开始数据完整性校验...") for level in level_dirs: dir_path = os.path.join(base_dir, level) if not os.path.exists(dir_path): print(f" [错误] 目录缺失: {dir_path}") continue files = [f for f in os.listdir(dir_path) if f.endswith('.json')] file_count = len(files) expected = expected_min_counts.get(level, 0) status = "正常" if file_count >= expected * 0.8 else "警告" # 允许20%的偏差 print(f" [{status}] {level} 层级: 找到 {file_count} 个文件 (预期约>{expected})") # 随机抽样检查一个文件是否能被正确解析 if files: import random sample_file = random.choice(files) sample_path = os.path.join(dir_path, sample_file) try: with open(sample_path, 'r', encoding='utf-8') as f: data = json.load(f) # 检查GeoJSON基本结构 if data.get("type") == "FeatureCollection" and "features" in data: pass # 结构基本正确 else: print(f" [警告] 文件 {sample_file} 的GeoJSON结构可能异常。") except Exception as e: print(f" [错误] 无法解析文件 {sample_file}: {e}") # 检查索引文件 index_file = os.path.join(base_dir, "index.json") if os.path.exists(index_file): try: with open(index_file, 'r', encoding='utf-8') as f: index_data = json.load(f) print(f" [正常] 索引文件包含 {len(index_data)} 条记录。") except Exception as e: print(f" [错误] 索引文件损坏: {e}") else: print(f" [错误] 索引文件缺失。") print("校验完成。") # 运行校验 if __name__ == "__main__": validate_data_integrity()这个校验脚本能快速告诉你数据下载是否“大体完整”,并揪出那些损坏的、无法解析的文件。
4.2 在ECharts中的应用示例
获取GeoJSON的最终目的是为了可视化。这里给出一个最简短的ECharts 5使用示例,将我们下载的数据用起来。
假设你的项目目录结构如下:
my_project/ ├── geojson_data_2023/ # 我们下载的数据 │ └── province/ │ └── 440000.json # 广东省轮廓 └── echarts_demo.html那么echarts_demo.html可以这样写:
<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>广东省地图示例</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> </head> <body> <div id="mapChart" style="width: 800px; height: 600px;"></div> <script> // 初始化ECharts实例 const chartDom = document.getElementById('mapChart'); const myChart = echarts.init(chartDom); // 使用Fetch API异步加载本地的GeoJSON文件 fetch('./geojson_data_2023/province/440000.json') .then(response => { if (!response.ok) { throw new Error(`HTTP error! status: ${response.status}`); } return response.json(); }) .then(guangdongGeoJSON => { // 关键步骤:注册GeoJSON数据 echarts.registerMap('GuangDong', guangdongGeoJSON); // 配置项 const option = { title: { text: '广东省地图', left: 'center' }, tooltip: { trigger: 'item', formatter: '{b}' // 显示区域名称 }, visualMap: { // 可以配合假数据做颜色映射 min: 0, max: 100, text: ['高', '低'], realtime: false, calculable: true, inRange: { color: ['#e0f3f8', '#abd9e9', '#74add1', '#4575b4', '#313695'] } }, series: [ { name: '广东', type: 'map', map: 'GuangDong', // 使用注册的地图名 roam: true, // 允许缩放和平移 label: { show: true // 显示地名标签 }, // 这里可以添加你的业务数据,格式如:[{name: '深圳市', value: 90}, ...] data: [] } ] }; // 使用配置项和数据显示图表 myChart.setOption(option); }) .catch(error => { console.error('加载GeoJSON数据失败:', error); chartDom.innerHTML = `<p style="color:red;">加载地图数据失败,请检查控制台日志。</p>`; }); // 响应窗口大小变化 window.addEventListener('resize', function() { myChart.resize(); }); </script> </body> </html>用浏览器打开这个HTML文件,你就能看到一个可交互的广东省地图。通过修改fetch的路径和registerMap的名称,你可以轻松切换到其他省份、城市甚至全国地图。
4.3 拓展:数据更新与其它数据源
数据是有时效性的。行政区划可能会调整(虽然不频繁),数据源也可能更新。我们的脚本如何适应?
- 定期运行:最简单的办法是定期(如每季度或每半年)重新运行一次脚本,覆盖旧数据。
- 增量更新:可以修改脚本,先读取本地已保存的索引文件,与远程索引对比,只下载那些新增或adcode发生变化的区域。这需要更复杂的逻辑,但能节省带宽和时间。
- 备用数据源:正如原始资料提及,国家地理信息公共服务平台(天地图)是更权威的数据来源。其数据可能更精确,但API调用通常需要注册申请密钥,且有调用频率限制。对于学习和小型项目,阿里云DataV的免费接口更方便;对于正式、高精度的商业项目,则应该考虑使用天地图等官方来源。
最后,我想分享一个实际项目中的小经验:下载完数据后,我习惯用jq这个命令行工具(或Python的json模块)快速查看一下某个文件的概要信息,比如jq '.features | length' geojson_data_2023/province/440000.json可以立刻告诉我广东省由多少个地理多边形(MultiPolygon)构成。这种快速验证能让你对数据的粒度有一个直观感受。