|
import math # 新增这行，导入math模块
import requests
import os
from PIL import Image
from urllib.parse import quote
import time
# 高德瓦片下载配置
class AmapTileDownloader:
def __init__(self, save_path='amap_tiles', zoom_range=(1, 18)):
"""
初始化下载器
:param save_path: 瓦片保存根路径
:param zoom_range: 缩放级别范围（高德瓦片缩放级1-18）
"""
self.save_path = save_path
self.zoom_range = zoom_range
# 高德瓦片服务地址（矢量底图，可替换为卫星图）
# 矢量底图：http://webrd0{1-4}.is.autonavi.com/appmaptile?lang=zh_cn&size=1&scale=1&style=8&x={x}&y={y}&z={z}
# 卫星底图：http://webst0{1-4}.is.autonavi.com/appmaptile?style=6&x={x}&y={y}&z={z}
self.tile_url_template = "http://webrd0{server}.is.autonavi.com/appmaptile?lang=zh_cn&size=1&scale=1&style=8&x={x}&y={y}&z={z}"
# 服务节点（避免单节点请求频繁被限制）
self.servers = [1, 2, 3, 4]
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
def mercator2tile(self, lon, lat, zoom):
"""
经纬度转瓦片坐标（墨卡托投影）
:param lon: 经度
:param lat: 纬度
:param zoom: 缩放级别
:return: x, y 瓦片坐标
"""
lat_rad = lat * 3.141592653589793 / 180.0
n = 2.0 ** zoom
x = int((lon + 180.0) / 360.0 * n)
y = int((1.0 - (math.log(math.tan(lat_rad) + (1 / math.cos(lat_rad)))) / math.pi) / 2.0 * n)
return x, y
def download_tile(self, x, y, z, server=1):
"""
下载单个瓦片
:param x: 瓦片x坐标
:param y: 瓦片y坐标
:param z: 缩放级别
:param server: 服务节点（1-4）
:return: 是否下载成功
"""
# 拼接瓦片URL
tile_url = self.tile_url_template.format(server=server, x=x, y=y, z=z)
# 构建保存路径：根路径/缩放级别/x坐标/y坐标.png
tile_save_path = os.path.join(self.save_path, str(z), str(x))
os.makedirs(tile_save_path, exist_ok=True)
tile_file = os.path.join(tile_save_path, f"{y}.png")
# 避免重复下载
if os.path.exists(tile_file):
print(f"瓦片 {z}/{x}/{y} 已存在，跳过")
return True
try:
# 发送请求（添加超时和重试）
response = requests.get(tile_url, headers=self.headers, timeout=10)
response.raise_for_status() # 抛出HTTP错误
# 保存瓦片
with open(tile_file, 'wb') as f:
f.write(response.content)
# 校验图片是否有效
try:
img = Image.open(tile_file)
img.verify()
print(f"成功下载：{z}/{x}/{y}")
return True
except:
os.remove(tile_file)
print(f"瓦片无效，已删除：{z}/{x}/{y}")
return False
except Exception as e:
print(f"下载失败 {z}/{x}/{y}：{str(e)}")
return False
def download_area(self, min_lon, min_lat, max_lon, max_lat):
"""
下载指定经纬度范围的瓦片
:param min_lon: 最小经度
:param max_lon: 最大经度
:param min_lat: 最小纬度
:param max_lat: 最大纬度
"""
import math # 延迟导入，避免初始化时依赖
print(f"开始下载范围：[{min_lon},{min_lat}] - [{max_lon},{max_lat}]")
print(f"缩放级别范围：{self.zoom_range[0]} - {self.zoom_range[1]}")
for z in range(self.zoom_range[0], self.zoom_range[1] + 1):
# 计算范围对应的瓦片坐标
min_x, min_y = self.mercator2tile(min_lon, max_lat, z) # 注意lat反转
max_x, max_y = self.mercator2tile(max_lon, min_lat, z)
print(f"\n缩放级别 {z}：瓦片范围 x[{min_x}-{max_x}], y[{min_y}-{max_y}]")
server_idx = 0 # 轮询服务节点
# 遍历所有瓦片坐标
for x in range(min_x, max_x + 1):
for y in range(min_y, max_y + 1):
# 轮询服务节点
server = self.servers[server_idx % len(self.servers)]
server_idx += 1
# 下载瓦片（添加小延迟，避免请求过快被限制）
self.download_tile(x, y, z, server)
time.sleep(0.1) # 可根据实际情况调整
print("\n所有瓦片下载完成！")
# ------------------- 示例使用 -------------------
if __name__ == "__main__":
# 初始化下载器：保存路径+缩放级别（建议先测试1-10级，级别越高瓦片越多）
downloader = AmapTileDownloader(save_path='amap_tiles', zoom_range=(1, 10))
# 下载指定区域（示例：北京市中心经纬度范围，可替换为自己需要的区域）
# 经纬度可从高德地图拾取：https://lbs.amap.com/console/show/picker
min_lon, min_lat = 116.30, 39.80 # 左下角经纬度
max_lon, max_lat = 116.50, 40.00 # 右上角经纬度
downloader.download_area(min_lon, min_lat, max_lon, max_lat)
|
9
0
已为社区贡献2条内容
所有评论(0)