ArcGIS REST API 数据爬取实战：Python 脚本 500米网格获取东莞控规 36972 条记录
·
ArcGIS REST API 数据爬取实战：Python 脚本 500米网格获取东莞控规 36972 条记录
城市规划数据的获取与分析是城市研究与规划实践的基础。东莞作为中国制造业重镇，其控制性详细规划数据对研究者、规划师和开发商具有重要价值。本文将详细介绍如何利用 Python 自动化脚本从 ArcGIS REST API 接口中高效爬取空间数据，并通过 500 米网格划分方法获取东莞全市范围内的控规数据，最终获得 36972 条完整记录。
1. 技术准备与环境搭建
在开始爬取数据前，我们需要确保开发环境配置正确。推荐使用 Python 3.8 或更高版本，并安装以下关键库：
pip install requests geopandas pandas numpy shapely
各库的主要作用如下：
- requests ：处理 HTTP 请求，与 ArcGIS REST API 交互
- geopandas ：处理地理空间数据，支持 Shapefile 格式输出
- pandas ：数据处理与分析
- numpy ：数值计算，用于网格划分
- shapely ：几何对象操作
对于大规模数据爬取，建议配置以下环境参数：
import os
os.environ['USE_PYGEOS'] = '0' # 避免geopandas与shapely的兼容性问题
2. ArcGIS REST API 接口分析
东莞控规数据通过 ArcGIS Server 提供的 REST API 接口公开。通过浏览器开发者工具分析，我们确定了核心数据接口：
http://120.86.191.153:6080/arcgis/rest/services/KGCX/KGCX/MapServer/identify
该接口接收以下关键参数：
| 参数名 | 必选 | 说明 | 示例值 |
|---|---|---|---|
| geometry | 是 | 查询点坐标 | 12710257,2593523 |
| geometryType | 是 | 几何类型 | esriGeometryPoint |
| layers | 是 | 图层ID | visible:20 |
| tolerance | 是 | 容差 | 2 |
| mapExtent | 是 | 地图范围 | 12621272,2591650,12715908,2649830 |
| f | 是 | 返回格式 | json |
| token | 是 | 访问令牌 | smLb_QfiJj_0mBbVF32JNFpCvn2... |
关键发现 ：实际测试表明，只有 geometry 参数需要动态变化，其他参数可固定为最优值。这极大简化了爬虫设计。
3. 网格化爬取策略实现
3.1 东莞市范围确定与网格划分
通过多次测试，我们确定了东莞市的完整空间范围：
x_min, y_min = 12621272, 2591650 # 左下角坐标
x_max, y_max = 12715908, 2649830 # 右上角坐标
采用 500 米间隔的网格划分策略，计算网格中心点坐标：
import numpy as np
def generate_grid_centers(x1, y1, x2, y2, delta=500):
"""生成覆盖指定范围的网格中心点坐标"""
x = np.arange(x1, x2 + delta, delta)
y = np.arange(y1, y2 + delta, delta)
x_center = (x[:-1] + x[1:]) / 2
y_center = (y[:-1] + y[1:]) / 2
return [(x, y) for x in x_center for y in y_center]
3.2 稳健的数据爬取函数实现
爬取函数需要处理网络异常、数据解析等问题：
import requests
import time
import random
from shapely.geometry import Polygon
def fetch_parcel_data(point, max_retries=3):
"""从API获取单个点的地块数据"""
url_template = "http://120.86.191.153:6080/arcgis/rest/services/KGCX/KGCX/MapServer/identify?geometry={},{}&geometryType=esriGeometryPoint&layers=visible:20&tolerance=2&mapExtent=12621272,2591650,12715908,2649830&imageDisplay=1269,929,96&f=json&token=smLb_QfiJj_0mBbVF32JNFpCvn2PlHdRTRNRkwpRV_HB26u84puTMkLZWYjXTKbk8cucgJVlATcCUX9KvqM5Ew..&sr=3857"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
'Referer': 'http://120.86.191.153/'
}
for attempt in range(max_retries):
try:
response = requests.get(
url_template.format(point[0], point[1]),
headers=headers,
timeout=(3, 7)
)
data = response.json().get('results', [])
if data:
features = []
for item in data:
attrs = item['attributes']
rings = item['geometry']['rings']
geometry = Polygon(rings[0], rings[1:]) if len(rings) > 1 else Polygon(rings[0])
features.append({**attrs, 'geometry': geometry})
return features
time.sleep(random.uniform(0.1, 0.3))
return None
except Exception as e:
if attempt == max_retries - 1:
print(f"Failed after {max_retries} attempts for point {point}")
return None
time.sleep(random.randint(5, 10))
注意：实际应用中应遵守网站的使用政策，合理设置请求间隔，避免对服务器造成过大压力。
4. 数据存储与格式转换
4.1 内存中的数据积累
使用 pandas 和 geopandas 实时积累数据：
import pandas as pd
import geopandas as gpd
class DataCollector:
def __init__(self):
self.attributes = []
self.geometries = []
def add_features(self, features):
if not features:
return
for feature in features:
geom = feature.pop('geometry')
self.attributes.append(feature)
self.geometries.append({
'OBJECTID': feature['OBJECTID'],
'geometry': geom
})
def get_dataframes(self):
attr_df = pd.DataFrame(self.attributes).drop_duplicates()
geom_gdf = gpd.GeoDataFrame(
pd.DataFrame(self.geometries),
geometry='geometry',
crs='EPSG:3857'
).drop_duplicates()
return attr_df, geom_gdf
4.2 数据持久化存储
将最终数据保存为 CSV 和 Shapefile 格式：
def save_results(attr_df, geom_gdf, output_dir='output'):
os.makedirs(output_dir, exist_ok=True)
# 保存属性数据
attr_path = os.path.join(output_dir, 'dgkg_attributes.csv')
attr_df.to_csv(attr_path, index=False, encoding='utf-8-sig')
# 保存几何数据
shp_path = os.path.join(output_dir, 'dgkg_shapefile')
geom_gdf.to_file(shp_path, driver='ESRI Shapefile', encoding='utf-8')
print(f"数据已保存至 {output_dir} 目录")
5. 完整爬取流程与优化
5.1 主爬取流程实现
def main():
# 初始化数据收集器
collector = DataCollector()
# 生成网格中心点
grid_centers = generate_grid_centers(12621272, 2591650, 12715908, 2649830)
total_points = len(grid_centers)
# 遍历所有网格点
for idx, center in enumerate(grid_centers, 1):
print(f"处理进度: {idx}/{total_points} ({idx/total_points:.1%})")
features = fetch_parcel_data(center)
collector.add_features(features)
# 每处理100个点保存一次中间结果
if idx % 100 == 0:
attr_df, geom_gdf = collector.get_dataframes()
save_results(attr_df, geom_gdf, 'temp_output')
# 保存最终结果
attr_df, geom_gdf = collector.get_dataframes()
save_results(attr_df, geom_gdf)
print(f"共获取 {len(attr_df)} 条有效记录")
5.2 性能优化技巧
- 并行处理 ：使用
concurrent.futures实现多线程爬取
from concurrent.futures import ThreadPoolExecutor, as_completed
def parallel_fetch(centers, max_workers=4):
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(fetch_parcel_data, center): center for center in centers}
for future in as_completed(futures):
yield future.result()
- 断点续爬 ：记录已处理网格点，避免重复工作
import json
def load_progress(progress_file='progress.json'):
if os.path.exists(progress_file):
with open(progress_file) as f:
return set(json.load(f))
return set()
def save_progress(processed_points, progress_file='progress.json'):
with open(progress_file, 'w') as f:
json.dump(list(processed_points), f)
- 自适应间隔 ：根据服务器响应动态调整请求频率
class AdaptiveRequester:
def __init__(self, base_delay=0.2, max_delay=10):
self.base_delay = base_delay
self.max_delay = max_delay
self.current_delay = base_delay
def adjust_delay(self, success):
if success:
self.current_delay = max(self.base_delay, self.current_delay * 0.9)
else:
self.current_delay = min(self.max_delay, self.current_delay * 1.5)
time.sleep(self.current_delay)
在实际项目中，500米网格划分最终获取了36972条记录。测试发现某些区域数据密度较低，可以考虑以下改进：
- 对数据稀疏区域进行二次网格细分（如250米）
- 实现自适应网格算法，根据数据密度动态调整网格大小
- 增加数据验证步骤，确保覆盖完整性
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
10
0- 0
已为社区贡献10条内容
所有评论(0)