Python使用高德地图API批量获取地址所属街道并写回Excel

pip install pandas openpyxl requests

key = "你的高德API_KEY"

df = pd.read_excel(input_file)
if '注册地址' not in df.columns:
    df['注册地址'] = df.iloc[:,16]
addresses = df['注册地址'].tolist()

wb = load_workbook(input_file)
ws = wb.active
ws.insert_cols(target_col)
ws.cell(row=header_row_index, column=target_col, value="街道")

def get_street_from_amap(address, retries=max_retries):
    if not isinstance(address, str) or not address.strip():
        return ""
    for attempt in range(1, retries+1):
        try:
            geo_resp = requests.get(
                "https://restapi.amap.com/v3/geocode/geo",
                params={"key": key, "address": address, "city": "上海"},
                timeout=15
            ).json()

            if not geo_resp.get("geocodes"):
                continue

            location = geo_resp["geocodes"][0]["location"]

            regeo_resp = requests.get(
                "https://restapi.amap.com/v3/geocode/regeo",
                params={"key": key, "location": location, "extensions": "base", "radius":500},
                timeout=15
            ).json()

            if regeo_resp.get("regeocode"):
                township = regeo_resp["regeocode"]["addressComponent"].get("township","") or ""
                return township
        except Exception as e:
            print(f"[尝试 {attempt}/{retries}] 地址查询失败: {address}, 错误: {e}")
        time.sleep(sleep_time + random.random()*0.5)
    return None

cache = {}
failed_addresses = []

for row_idx, addr in enumerate(addresses, start=header_row_index+1):
    if not isinstance(addr,str) or not addr.strip():
        ws.cell(row=row_idx, column=target_col, value="")
        continue
    if addr in cache:
        township = cache[addr]
    else:
        township = get_street_from_amap(addr)
        if township is None:
            failed_addresses.append((row_idx, addr))
            township = ""
        cache[addr] = township
        time.sleep(sleep_time + random.random()*0.5)
    ws.cell(row=row_idx, column=target_col, value=township)

if failed_addresses:
    print(f"第一次查询失败地址共 {len(failed_addresses)} 条，开始自动重查……")
    still_failed = []
    for row_idx, addr in failed_addresses:
        township = get_street_from_amap(addr)
        if township is None:
            still_failed.append((row_idx, addr))
            township = ""
        cache[addr] = township
        ws.cell(row=row_idx, column=target_col, value=township)
        time.sleep(sleep_time + random.random()*0.5)
    failed_addresses = still_failed

if failed_addresses:
    df_fail = pd.DataFrame([addr for _, addr in failed_addresses], columns=["地址"])
    df_fail.to_excel(failed_file, index=False)