简介：这套项目源码将高德地图 API 的批量地理编码流程打包成可直接运行的完整示例，适合需要把 Excel 地址数据转换为经纬度的 Python 开发者、数据分析和 GIS 相关人员使用。压缩包共 9 个文件，其中 6 个 py 脚本覆盖输入文件生成、地理编码请求、坐标转换与结果导出等环节，配合 1 个 xlsx 地址模板、1 个 HTML 可视化预览页面和 1 个 inscode 项目配置文件，整体仅 15KB，结构紧凑易于修改。资源已有 135 人学习。通过源码可掌握高德开放平台应用注册与 Key 申请、requests 调用地理编码接口、不同坐标系转换，以及基于 pandas 批量读写 Excel 的完整落地方式；脚本中针对文件权限与请求异常给出了可复用的处理思路，既能直接用于批量地址转坐标任务，也可作为 Python 调用地图 API 的工程化范本。

1. 高德API批量获取经纬度：地图服务商不提供的批量能力，靠脚本补齐

月初接到一份两千多条门店地址的表格，要转成经纬度做区域热力分布。逐个去网页地图搜一次、复制一次坐标，两条地址就能让人烦躁到怀疑人生。高德API本身接受的是单条地址解析，官方并没有直接给一个"上传Excel批量返回坐标"的界面，批量这件事必须自己写脚本完成。这份源码包做的事情并不神秘：读取地址清单，逐条调用高德地理编码服务接口，把返回的经纬度、匹配等级和行政区划回填到表格里，同时处理限流、重试、异常和增量更新。适合手里有几百到几十万条地址、需要快速标点或空间分析的数据从业者，也适合第一次接触Web API的Python学习者——边界在于，它不负责坐标之外的业务逻辑，也不解决地址本身的质量问题。

2. 地理编码选型与接口规则：正向接口、坐标体系与返回字段解析

2.1 正向地理编码为什么是批量场景的唯一正解

高德的Web服务里有两类坐标转换接口：地址转坐标叫正向地理编码，坐标转地址叫逆地理编码。批量拿经纬度这份需求，本质上是地址到坐标的映射，对应的是正向地理编码。很多从数据分析转过来的人容易绕弯路——手里已经有一批坐标，却先调逆地理编码去核验地址，再把核验后的地址反向匹配坐标，这一来一回不仅浪费配额，还把问题复杂度抬高了一截。正确做法很直接：读地址列表，循环调用地理编码接口，拿location字段填表。

正向地理编码的接口路径常规是/v3/geocode/geo，支持GET请求。核心参数就几个：address传入结构化地址，city限定行政区，key作为身份凭证，output指定返回格式。它对单次请求只处理一个address，意味着批量场景必须在代码里自己循环，而循环时要克服的限制主要有三个：单Key的每秒请求配额、模糊地址的误配、以及返回坐标的偏移风险。这三个问题在后面的章节里单独展开，这里先把接口方向选对。

我一般会把正向和逆向来一张对比表明确边界，免得团队里有人用错方向：

选正向还有一个隐性原因：返回的level字段能告诉你匹配到什么粒度。同一套循环代码，看到level是"门牌号"或"道路"就可以放心入库，看到"区县"甚至"乡镇"就要标记出来人工复核。逆地理编码接口虽然也能给出坐标周边的结构化地址，但它按坐标半径去做反向匹配，并不会帮你修正原始地址的精度问题。

2.2 请求参数、返回解析与坐标体系：看懂JSON才能拿对数据

先看一次标准请求的参数构成：

params = { "address": "某市某区学院路1号", "city": "某市", "key": "你的Key", "output": "json", }

address必须是完整度尽量高的结构化文本，带省市区、街道、门牌号最佳。city参数经常被人忽略，但它的作用非常实际：当address里的道路名在全国多个省市都存在时，city能直接圈定候选范围，避免"学院路"这种名称匹配到外省市去。output固定用json，xml解析起来多一道工序，没有必要。

返回的JSON长这样：

{ "status": "1", "info": "OK", "geocodes": [ { "formatted_address": "某市某区学院路1号", "location": "116.310003,39.991957", "level": "门牌号", "city": "某市", "adcode": "110105" } ] }

这里有两处极容易踩雷。第一，location是字符串而不是数组，逗号前是经度、逗号后是纬度，直接用split(",")拆，不做类型转换的话，后面存数据库会拿到一个文本型的假坐标。第二，geocodes可能为空数组，说明这次地址匹配失败，但HTTP状态码可能依然是200，只盯着请求层状态会漏掉这类失败。我通常会先看status是否是"1"，再看geocodes是否非空，两层判断都过了才写入结果。

关于坐标体系，高德返回的是GCJ-02，也就是大家常说的"火星坐标"。这本身不是错误，是国内地图通行的加密坐标体系。如果后续要把这批坐标投到GPS底图或其他标准底图上，需要做GCJ-02到WGS-84的偏移转换，偏移量在几十米到几百米不等。如果直接拿火星坐标叠加GPS轨迹，视觉效果上会整体漂移一块。这个转换在源码包里单独封装了一个函数，原因很实际：不转换，数据分析阶段的"点对不上"会变成常态。

还有一个容易被忽略的身份校验维度：Key是否开启了签名校验或IP白名单。开启签名后，请求里必须追加sigm参数和当前时间戳，sigm由请求参数按字典序拼接后做MD5生成。这类问题特征很隐蔽，本地调试的好好的，部署到服务器上就报"10020签名错误"，多半是服务器出口IP不在白名单里，或者签名参数没跟上。第一次配置Key时，建议先看控制台的安全设置，把这两个选项的开关状态记录下来。

3. Python批量实现：从单条编码请求到并发全量落库

3.1 环境准备与API Key配置：每个字段的含义和边界

批量脚本依赖不多，requests发请求，标准库json解析，pandas或openpyxl读表格。虚拟环境里执行一遍安装就能跑：

pip install requests pandas

Key的配置值得单独说。硬编码在脚本里是团队协作时的隐患，代码一旦提交到仓库，Key就跟着泄露了。我习惯用环境变量传入：

import os AMAP_KEY = os.getenv("AMAP_KEY") if not AMAP_KEY: raise RuntimeError("请先设置环境变量 AMAP_KEY，例如: export AMAP_KEY=your_key")

设置环境变量的方式在Windows和macOS/Linux略有差异，但脚本里不需要区分，只要保证运行时环境变量存在即可。Key的权限边界也要先确认：是否绑定了IP白名单，是否开启了域名校验，是否开通了地理编码服务的调用权限。白名单模式下，服务器出口IP不在名单里会直接报10005错误，这类问题在本地调试时最容易被误判成代码bug。

提示：Key的QPS限制以服务商控制台展示为准，不同认证等级差异很大。首次跑批不要把并发参数设成超出自己预期的值。

3.2 单条请求封装：requests调用与JSON解析

把单次地理编码封装成一个独立函数，是批量处理的基石。函数内部做三件事：拼接参数、发起请求、解析并归一化结果。

import requests def geocode(address: str, city: str = "") -> dict: url = "https://restapi.amap.com/v3/geocode/geo" params = { "address": address, "key": AMAP_KEY, "city": city, "output": "json", } try: resp = requests.get(url, params=params, timeout=5) data = resp.json() except requests.Timeout: return {"address": address, "status": "timeout", "error": "请求超时"} except ValueError: return {"address": address, "status": "invalid_json", "error": "响应不是合法JSON"} if data.get("status") != "1": return { "address": address, "status": "api_error", "error": f"infocode={data.get('infocode')}, info={data.get('info')}", } geocodes = data.get("geocodes") or [] if not geocodes: return {"address": address, "status": "no_match", "lng": None, "lat": None, "level": None} loc = geocodes[0].get("location", "").split(",") return { "address": address, "status": "ok", "lng": float(loc[0]) if len(loc) > 0 else None, "lat": float(loc[1]) if len(loc) > 1 else None, "level": geocodes[0].get("level"), "city": geocodes[0].get("city"), "adcode": geocodes[0].get("adcode"), }

这里的逻辑讲清楚一点。第一，每次请求都设置timeout=5秒，防止某个地址拖住整个循环。第二，城市参数city默认空字符串，由调用方按数据情况决定要不要传入。第三，状态判断分为两层：HTTP层响应是200不代表业务成功，status字段非1要返回错误码；geocodes为空是另一种失败，说明这次地址没匹配到任何位置信息。第四，location拆出来后必须转float，很多后续计算要求数值型坐标。

timeout值可以根据网络环境调整，局域网代理环境下3秒够用，走公网建议5到10秒。city参数如果确定所有地址都在同一个城市，可以写死在配置里，但地址来源复杂时，我建议改为从地址数据里按省份城市列动态传入。

3.3 批量任务拆分：并发度、批次大小与进度追踪

单线程循环是最保险的写法，但几千条地址串行调接口，耗时是并发方案的几倍。适合用ThreadPoolExecutor做轻量并发，原因是这个场景是IO密集型的，线程切换开销远小于等待网络响应的时间。

from concurrent.futures import ThreadPoolExecutor, as_completed from threading import BoundedSemaphore sem = BoundedSemaphore(5) def bounded_geocode(address: str) -> dict: with sem: return geocode(address) def batch_geocode(addresses: list[str], max_workers: int = 5) -> list[dict]: results = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: future_map = {pool.submit(bounded_geocode, addr): addr for addr in addresses} for future in as_completed(future_map): addr = future_map[future] try: result = future.result() except Exception as exc: result = {"address": addr, "status": "exception", "error": str(exc)} results.append(result) return results

BoundedSemaphore在这里承担限流职责，max_workers控制线程数，信号量控制同时放行的请求数。两者分开调的好处是：线程数可以开到10，但真正打到接口的并发请求被信号量限制在5，这样既保住了Python侧的处理能力，又不会轻易触发服务端的QPS限制。初次跑批量任务，不知道Key的配额上限，建议从5开始试，跑几百条看日志里有没有10004或超时报错，再逐步调。

进度追踪不要省。批量任务跑半小时，没有进度输出会让人心里没底。我一般在as_completed循环里维护一个计数器，每完成50条打印一次：

counter = 0 for future in as_completed(future_map): counter += 1 if counter % 50 == 0: print(f"已完成 {counter}/{len(addresses)}")

输出到标准日志而不是print也可以，目的只有一个：中途挂掉时能根据最后一条进度日志定位到大致位置。

3.4 结果落库与增量更新：SQLite与CSV两种方案

批量结果不能只存在内存里，脚本一旦中断全部白跑。轻量场景我推荐SQLite，零配置文件级单表够用：

CREATE TABLE IF NOT EXISTS geo_result ( address TEXT PRIMARY KEY, lng REAL, lat REAL, level TEXT, city TEXT, status TEXT, updated_at TEXT DEFAULT (datetime('now', 'localtime')) );

address作为主键天然去重，重复执行脚本时用INSERT OR REPLACE语义来更新已有记录。增量场景的判断逻辑是：导入的新表里，只有那些在geo_result中查不到坐标的地址才需要重新请求。

import sqlite3 conn = sqlite3.connect("geocode.db") def save_result(item: dict): conn.execute( "INSERT OR REPLACE INTO geo_result(address, lng, lat, level, city, status) VALUES(?,?,?,?,?,?)", (item.get("address"), item.get("lng"), item.get("lat"), item.get("level"), item.get("city"), item.get("status")), ) conn.commit()

如果团队更习惯用表格交接，CSV追加写入也够用，注意打开文件时用append模式，并且每次写入后flush，防止进程被杀丢数据。SQLite方案的额外好处是后续可以直接写SQL做重编码、抽样评估，不需要把CSV重新load一遍。

4. 批量地理编码避坑指南：限流、偏移与地址清洗

4.1 现象：批量请求频繁TIME_OUT，日志里全是超时

现象很直接：跑了几百条之后，脚本开始连续报requests.Timeout，单条请求从原本的几百毫秒飙升到5秒超时上限。

原因通常是两类。第一，并发开太高，单Key的每秒请求配额被打满，服务端直接限流，表现就是超时和连接中断，而不是在JSON里返回错误码。第二，网络代理不稳定或者出口IP抖动。

解决方式：第一次跑任务时把并发压到最低跑通流程，比如max_workers=3、信号量=2，确认没有超时报错后再逐步加并发。同时把timeout从5秒改成8秒、并加入退避重试，见下：

def geocode_with_retry(address: str, retries: int = 3) -> dict: for attempt in range(retries): result = geocode(address) if result.get("status") not in ("timeout", "api_error"): return result if attempt < retries - 1: time.sleep(2 * (attempt + 1)) return result

这个重试逻辑里有个细节：退避间隔按attempt翻倍，第一次失败等2秒，第二次失败等4秒，第三次直接放弃并记录错误。如果连续出现大量重试，问题大概率不在代码，而是该降并发。

4.2 现象：地址正确但坐标与预期城市不符

一个很典型的例子：地址文本是"学院路1号"，省市区都不带。本地的人都默认它在某市，但返回的location却指向了另一个省份的"学院路1号"。

原因是address信息不够完整，服务端只能靠文本里的有限信息做全库匹配，同名道路全国到处都有，不限定城市就全凭运气。

解决方式分两级。第一级最简单，请求参数里带上city，用数据表里的城市列去填充：

result = geocode(row["address"], city=row.get("city", ""))

第二级是数据预处理层面的地址拼接，地址明细里可能省市区字段分散在不同列，先拼成规范的完整地址文本再请求：

def build_full_address(row): parts = [row.get("province", ""), row.get("city", ""), row.get("district", ""), row.get("street", "")] return "".join(p for p in parts if p)

这套拼接逻辑在实际业务里非常常见。做完之后还要注意level字段，如果返回level是"区县"级别，说明地址匹配到了城区却没定位到具体街道，这类结果要单独标记成待人工确认，不能直接进最终坐标表。

4.3 现象：相同地址两次请求返回不同坐标

现象是同一个地址在同一天内调两次，两次location值可能差了几百米，看起来很不稳定。

原因是地理编码不是纯静态哈希映射，服务端会结合周边POI数据做语义匹配，地址表达中的细微差别、周边兴趣点的变化干扰，都会影响最终坐标。另一个常见诱因是请求参数不一致，比如一次带了city一次没带，解析结果自然不同。

解决方式：固定请求参数，业务代码统一走同一个封装函数，不搞两套拼接逻辑。对重复请求做缓存，第一次成功的结果直接存库，之后相同地址不再发起网络请求，既避免坐标漂移又省配额。我在第5章会给出缓存的具体设计。这里先记住一个原则：地理编码结果不是绝对精确的物理定位，它是地址语义匹配的近似结果，精度足够业务使用即可。

4.4 现象：高并发下Key被临时封禁，全部返回10004

现象是批跑到一半，后续请求全部返回"10004 ACCESS_TOO_FREQUENT"，之前成功的记录还在，新的请求大批失败。

原因是单Key的请求频率超过了服务端设定阈值，服务端对Key做临时访问限制，一般持续几十秒到几分钟，具体时长以再次请求的结果为准。

解决方式：先停掉当前线程池，让请求量降下来，等待限流窗口过后再用更低并发重启。关键是要把10004当成一种期待中的状态而不是bug来处理：

error_codes_to_retry = {"10004", "10003", "10001"} if result.get("status") == "api_error" and result.get("error") in error_codes_to_retry: time.sleep(10) return geocode(address)

这里sleep 10秒是给服务端留恢复窗口，比无脑重试有效。如果整个Key触发了封禁，只靠单机重试是没用的，需要去控制台确认配额是否耗尽。

4.5 现象：地址里的括号备注导致解析失败

现象是把表格里的原始地址原样丢进去，部分带"（大学城店）""（万达广场）"这类备注的地址全部解析失败或匹配到错误位置。

原因是地理编码接口对过于口语化的地址文本容忍度有限，括号里的业务备注会干扰分词和语义推断，把它误当成地址主体的一部分。

解决方式是在请求前做一轮地址清洗，把括号内容连同常见后缀词过滤掉：

import re def clean_address(addr: str) -> str: addr = re.sub(r"[（(].*?[）)]", "", addr) addr = re.sub(r"(店|分店|旗舰店|体验店)$", "", addr) return addr.strip()

清洗逻辑要克制，只删业务备注和门店后缀，不要动门牌号和路名。清洗后如果还是匹配失败，记录到异常表，等第二批次人工补录，而不是反复对同一条地址轰炸请求。

5. 性能优化与精度校准：并发参数、缓存与结果验证

5.1 均匀限速：把吞吐量从每秒3条提到每秒20条

批量任务中很常见的一个矛盾：并发压高了触发限流，压低了跑太久。均匀限速是比单纯发并发更稳定的做法。思路是控制每两次请求之间的最小间隔，让它整体形成一个匀速的请求流，而不是突发脉冲。我用令牌桶思想简化实现：

import time class RateLimiter: def __init__(self, max_per_sec: float): self.interval = 1.0 / max_per_sec self.next_time = time.time() def wait(self): now = time.time() if now < self.next_time: time.sleep(self.next_time - now) self.next_time = time.time() + self.interval

每次请求前调用limiter.wait()，把发射速率压到设定值。max_per_sec设成3，就是每秒最多3条；改到20，就是每秒20条。这个参数比线程数更直观，也更接近服务端限流的维度。实际调参时从5开始，观察日志中是否出现10004或超时，没有就按1.5倍往上加，出现限流就退回上一档。

均匀限速和并发是配合关系：max_workers控制在10以内，限速器负责统一节奏。两者一起用，既保留并发处理能力，又不会让请求波形出现尖刺。这一套组合通常能把稳定的吞吐量从单线程的每秒几条提到更高档位，具体上限取决于网络延迟和服务端响应速度，核心收益是压测不翻车。

5.2 本地缓存设计：避免重复请求消耗配额

批量坐标任务有一个隐蔽的浪费：脚本改了参数重新跑数据时，已经成功过的地址又会从头请求一遍。如果每天增量任务要跑3万条地址，其中2.5万条昨天已经解析过，重复请求意味着配额白白烧掉，还更容易触发限流。

缓存设计的要点是按归一化后的地址做键。地址清洗函数处理完的结果再作为key，这样"某市某区学院路1号"和"某市某区学院路1号(大学城店)"清洗后不冲突，也避免同一个地址两种写法分别占了两个缓存项。

cache = {} def get_or_geocode(address: str) -> dict: key = clean_address(address) if key in cache: return cache[key] result = geocode_with_retry(key) if result.get("status") == "ok": cache[key] = result return result

缓存放到进程内存里，简单且对单机任务足够。跨天运行的任务，SQLite表本身就是持久化缓存，判断逻辑改为先查库再请求。对已经入库的地址，直接读取，不再消耗配额。只有新地址或之前解析失败的地址才会真正发起请求，这个增量逻辑可以在几万条地址的日更新任务里把请求量打下来一大半，也让每次重跑时"为什么还在请求"的疑问少很多。

5.3 精度验证方法：抽样对比地图App的坐标结果

跑完批量编码，不等于可以无脑上线。任何第三方地理编码服务都有匹配精度波动，必须做一轮抽样验证，这也是对数据负责任的做法，我一般会抽50条做人工核验。

import random sample = random.sample(results, min(50, len(results)))

逐条把脚本返回的坐标粘到高德地图App的搜索框里，看定位点与实际门店位置是否一致。核对时记录两件事：一是坐标点是否落在正确街区，二是level字段是否与真实地点粒度匹配。门牌号级别的地址，定位偏差应该在几十米内；区县级别的地址，偏差到几公里都是正常现象。如果50条里超过3条定位到错误街区，就要怀疑地址清洗规则是不是过于激进，比如把门牌号码中的"1号"误删的情况。

还有一个校验技巧是把坐标逆地理编码回去，对比返回地址和原地址的语义距离。统计level的分布也能快速发现批次问题：一个地址库大多数记录停在"区县"级别，说明地址本身缺门牌号或街道后缀，这类数据建议先用数据治理手段补齐关键层次，而不是继续调接口白白耗配额。

6. 把批量地理编码做成命令行工具：一条命令跑完全量数据

批量处理好之后，最怕的是每次跑数据都翻出脚本改变量。所以我把这套流程收敛成一个命令行工具，输入一个CSV，输出一个带坐标的新CSV，中间过程用参数控制。用argparse做参数解析，各功能独立成函数。

import argparse def parse_args(): parser = argparse.ArgumentParser(description="批量地理编码工具") parser.add_argument("--file", required=True, help="输入CSV路径") parser.add_argument("--address-col", default="address", help="地址列名") parser.add_argument("--city-col", default="city", help="城市列名，可为空") parser.add_argument("--out", default="geocoded.csv", help="输出CSV路径") parser.add_argument("--workers", type=int, default=5, help="并发线程数") parser.add_argument("--qps", type=float, default=5.0, help="每秒最大请求数") return parser.parse_args()

参数中--qps是最值得调的项，它直接决定任务耗时和服务端限流风险。首次跑新地址库，先用默认5试十分钟，观察日志；稳定后就按1.5倍往上试探，找到当前网络环境下最合适的档位。主流程里有一个容易被忽略的细节：CSV编码不是UTF-8就会静默读到乱码。高德接口拿乱码地址去匹配必然失败，而且失败结果还不容易察觉。读取文件时显式指定编码并做异常捕获：

try: data = pd.read_csv(args.file, dtype=str) except UnicodeDecodeError: data = pd.read_csv(args.file, dtype=str, encoding="gbk")

用dtype=str强制所有列按文本读，避免门牌号"001"被pandas转成数字1，这类隐式类型转换是地址清洗里最隐蔽的坑之一。

命令行工具的骨架搭好后，整个批量任务从"打开Jupyter Notebook改代码"变成一句固定命令：

python batch_geocode.py --file store_addresses.csv --city-col city --out result.csv --workers 10 --qps 10

跑完之后检查两件事：第一看输出的status列里ok的占比，第二看level列的分布。这两项同时达标，数据才敢交给下游。日志输出到文件而不仅是控制台，我是用标准logging模块按天滚动，这样万一凌晨任务挂了，第二天能根据日志快速定位中断位置。这套工具沉淀下来之后，每天的增量任务都走同一个命令，因为SQLite缓存表的存在，第二次跑相同地址库时几乎不再发请求，全程几分钟收尾。

从那以后，我养成了一个习惯：不管多小的批量调用任务，都强制走一遍地址清洗、限速、抽样校准三步，少一步都不敢把结果交给下游。希望帮到你。

本文还有配套的精品资源，点击获取