"""
从 OpenStreetMap Overpass API 获取中国长城数据，转换为 Shapefile
"""
import requests
import geopandas as gpd
from shapely.geometry import LineString
import os
OUTPUT_DIR = r"C:\Users\yl\Desktop\GIS"
def fetch_great_wall():
# Overpass API endpoints
urls = [
"https://overpass-api.de/api/interpreter",
"https://lz4.overpass-api.de/api/interpreter",
"https://overpass.kumi.systems/api/interpreter",
]
# 长城在 OSM 中的标签
bbox = "36,105,45,125"
# Overpass QL 查询
overpass_query = f"""
[out:json][timeout:300];
(
way["name"~"长城"]({bbox});
way["man_made"="wall"]({bbox});
way["barrier"="wall"]["historic"="yes"]({bbox});
relation["name"~"长城"]({bbox});
);
out body;
>;
out skel qt;
"""
print("正在查询 Overpass API...")
print(f"查询范围: {bbox}")
data = None
for url in urls:
try:
print(f"尝试: {url}")
response = requests.post(
url,
data=overpass_query.encode('utf-8'),
timeout=300,
headers={
'Content-Type': 'text/plain',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'User-Agent': 'GreatWallDownloader/1.0'
}
)
print(f"状态码: {
response.status_code}")
if response.status_code == 200:
data = response.json()
break
except Exception as e:
print(f"错误: {e}")
continue
if data is None:
print("所有 API 端点都失败了")
return False
elements = data.get('elements', [])
ways = [e for e in elements if e['type'] == 'way']
nodes = {n['id']: (n['lon'], n['lat']) for e in elements if e['type'] == 'node' for n in [e]}
relations = [e for e in elements if e['type'] == 'relation']
print(f"找到: {len(ways)} 条way, {len(nodes)} 个节点, {len(relations)} 个relation")
geometries = []
properties = []
# 处理 ways
for way in ways:
node_ids = [n['ref'] if isinstance(n, dict) else n for n in way.get('nodes', [])]
coords = []
for nid in node_ids:
if isinstance(nid, int) and nid in nodes:
coords
.append(nodes[nid])
elif isinstance(nid, dict) and 'lat' in nid and 'lon' in nid:
coords.append((nid['lon'], nid['lat']))
if len(coords) >= 2:
tags = way.get('tags', {})
name = tags.get('name', '')
geometries.append(LineString(coords))
properties.append({
'name': name,
'historic': tags.get('historic', ''),
'man_made': tags.get('man_made', ''),
'barrier': tags.get('barrier', ''),
'tourism': tags.get('tourism', ''),
'osm_id': way['id'],
})
# 处理 relations 中的 ways
for rel in relations:
tags = rel.get('tags', {})
name = tags.get('name', '')
print(f" Relation: {name} (type={tags.get('type','')})")
for member in rel.get('members', []):
if member.get('type') == 'way':
member_ref = member.get('ref')
way = next((w for w in ways if w['id'] == member_ref), None)
if way:
node_ids = [n['ref'] if isinstance(n, dict) else n for n in way.get('nodes', [])]
coords = []
for nid in node_ids:
if isinstance(nid, int) and nid in nodes:
coords.append(nodes[nid])
elif isinstance(nid, dict) and 'lat' in nid and 'lon' in nid:
coords.append((nid['lon'], nid['lat']))
if len(coords) >= 2:
way_tags = way.get('tags', {})
geometries.append(LineString(coords))
properties.append({
'name': name or way_tags.get('name', ''),
'relation_type': tags.get('type', ''),
'historic': way_tags.get('historic', ''),
'osm_id': way['id'],
})
if not geometries:
print("未找到长城数据")
return False
gdf = gpd.GeoDataFrame(properties, geometry=geometries, crs="EPSG:4326")
# 保存 GeoJSON
geojson_path = os.path.join(OUTPUT_DIR, "great_wall.geojson")
gdf.to_file(geojson_path, driver="GeoJSON", encoding="utf-8")
print(f"\n已保存 GeoJSON: {geojson_path}")
# 保存 Shapefile
shp_path = os.path.join(OUTPUT_DIR, "great_wall.shp")
gdf.to_file(shp_path, encoding="utf-8")
print(f"已保存 Shapefile: {shp_path}")
print(f"\n共 {len(gdf)} 条线段")
bounds
= gdf.total_bounds
print(f"范围: 经度 [{bounds[0]:.2f}, {bounds[2]:.2f}], 纬度 [{bounds[1]:.2f}, {bounds[3]:.2f}]")
# 显示前几条记录
print("\n前5条记录:")
print(gdf[['name', 'historic', 'man_made']].head(5).to_string())
return True
if __name__ == "__main__":
fetch_great_wall()
