下载阿里云DataV.GeoAtlas的地图数据（到县级）

geoatlas_download.py

# python 3.11

import os
import json
from urllib import request
from urllib.error import HTTPError

def save_json(str,path):
    print("保存至\""+path+".json\"")
    open(path+".json","wb").write(str)

def download_json(id,path):
    print("下载 [%d]%s ..."%(id,path))
    if id%100==0:
        try:
            downloaded_str=request.urlopen("https://geo.datav.aliyun.com/areas_v3/bound/%d_full.json"%(id)).read()
            if id%10000!=0:
                save_json(downloaded_str,path)
            else:
                if not os.path.exists(path):
                    os.mkdir(path)
                gj=json.loads(downloaded_str)
                for f in gj["features"]:
                    #遍历ID和名称
                    #如果ID中间两位不是0
                    eachId=f["properties"]["adcode"]
                    eachName=f["properties"]["name"]
                    if eachName=="":
                        print("跳过空名称数据：adcode="+eachId)
                    else:
                        download_json(eachId,path+"/"+eachName)
        except HTTPError as e:#没有更详细的数据
            print("下载失败：",e.reason)
            downloaded_str=request.urlopen("https://geo.datav.aliyun.com/areas_v3/bound/%d.json"%(id)).read()
            save_json(downloaded_str,path)
    else:#已到县级
        downloaded_str=request.urlopen("https://geo.datav.aliyun.com/areas_v3/bound/%d.json"%(id)).read()
        save_json(downloaded_str,path)

download_json(100000,".")

改了一下，下载速度快很多

# python 3.11

import os
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib import request
from urllib.error import HTTPError

# 同时下载的线程数，可按网络情况调整
MAX_WORKERS = 8


def save_json(data, path):
    print(f'保存至"{path}.json"')
    with open(path + ".json", "wb") as f:
        f.write(data)


def download_children(features, base_path):
    """并行下载多个子地区（省、市等）。"""
    tasks = []
    for feature in features:
        each_id = feature["properties"]["adcode"]
        each_name = feature["properties"]["name"]
        if each_name == "":
            print(f"跳过空名称数据：adcode={each_id}")
            continue
        tasks.append((each_id, base_path + "/" + each_name))

    if not tasks:
        return

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {
            executor.submit(download_json, each_id, each_path): (each_id, each_path)
            for each_id, each_path in tasks
        }
        for future in as_completed(futures):
            each_id, each_path = futures[future]
            try:
                future.result()
            except Exception as e:
                print(f"下载失败 [{each_id}]{each_path}: {e}")


def download_json(area_id, path):
    print(f"下载 [{area_id}]{path} ...")
    if area_id % 100 == 0:
        try:
            url = f"https://geo.datav.aliyun.com/areas_v3/bound/{area_id}_full.json"
            downloaded_str = request.urlopen(url).read()
            if area_id % 10000 != 0:
                save_json(downloaded_str, path)
            else:
                os.makedirs(path, exist_ok=True)
                geojson = json.loads(downloaded_str)
                download_children(geojson["features"], path)
        except HTTPError as e:
            print(f"下载失败：{e.reason}")
            url = f"https://geo.datav.aliyun.com/areas_v3/bound/{area_id}.json"
            downloaded_str = request.urlopen(url).read()
            save_json(downloaded_str, path)
    else:
        url = f"https://geo.datav.aliyun.com/areas_v3/bound/{area_id}.json"
        downloaded_str = request.urlopen(url).read()
        save_json(downloaded_str, path)


if __name__ == "__main__":
    download_json(100000, "area")