简介面向需要批量处理地理位置的开发者这套基于高德地图接口与Python语言的项目源码能够自动将地址转换为对应经纬度适用于数据采集、数据分析、地图可视化等位置服务场景。压缩包内共有9个文件其中6个为Python脚本分别负责调用高德接口发送地理编码请求、完成不同坐标系之间的转换、生成输入Excel表格以及读写输出结果另附带样例数据、网页展示页面和多个可运行的演示脚本整个包仅有15KB结构简单清晰方便按需摘取使用。目前已有135人学习。源码完整覆盖从读取地址列表、请求高德接口、处理返回结果到输出经纬度的主要流程也对文件权限等常见运行问题给出了解决思路。用户替换自己的密钥后即可直接运行也可以将相关模块嵌入更大型的数据处理项目中从而免去逐条手工查询降低出错风险明显提升地理编码效率。整体上是一套适合初学者上手、也能满足日常批量坐标查询需求的轻量型工具集。1. 高德API批量获取经纬度这个需求藏着的四道坎比接口本身更花时间在位置相关的业务里“手里有一批地址文本想批量换成经纬度”几乎是所有可视化项目的第一道工序。高德开放平台的Web服务地理编码接口能完成地址到坐标的转换但高德API批量获取经纬度真正让人头疼的不是请求本身而是输入地址的脏数据、并发频控和中断后的恢复。这个需求我做过不止一次有人用一个for循环就把三百条地址跑完了也有人因为没做缓存和断点项目上线后每天重复调用把配额烧光第二天全员干等。这篇文章面向要把几百、几千条地址落成坐标的开发者从单条请求开始逐步讲清楚Key准备、Python批量脚本、多线程节流和五个高频踩坑最后给一套坐标质量校验的思路。2. 接口选型与Key准备地理编码单次请求长什么样配额边界在哪里2.1 正向地理编码把地址变成经纬度的请求参数与返回结构所谓正向地理编码就是传一个地址文本接口返回对应的经纬度坐标。高德开放平台把这个能力封装在Web服务API里接口路径是/v3/geocode/geo请求方式为GET单次调用就能拿到一个或多个候选坐标。先看单条请求的完整结构。我用curl来验证这样不依赖任何编程语言浏览器也能直接打开# 用curl发起一次地理编码请求注意替换为自己的Key # address 参数一定要URL编码中文和特殊字符都要转义 curl -s -w \nHTTP_CODE:%{http_code}\n \ https://restapi.amap.com/v3/geocode/geo?address%E6%9F%90%E5%B8%82%E6%9F%90%E5%8C%BA%E6%9F%90%E8%B7%AF1%E5%8F%B7city%E6%9F%90%E5%B8%82key你的Keyoutputjson这段命令里-s表示静默模式不输出下载进度-w \nHTTP_CODE:%{http_code}\n会在响应末尾打印HTTP状态码用来区分“接口正常返回”和“网络层失败”。address是必填参数city选填但强烈建议带上output默认就是json不传也行。返回的JSON结构是嵌套的geocodes是一个数组因为一个地址可能命中了多个级别不同的候选结果。下面是精简后的结构{ status: 1, info: OK, infocode: 10000, count: 1, geocodes: [ { formatted_address: 某市某区某路1号, country: 中国, province: 某省, citycode: 0001, city: 某市, district: 某区, adcode: 110105, location: 116.486409,39.921489, level: 门牌号 } ] }代码逻辑说明status为1表示业务成功为0表示失败具体原因看info和infocode。真正有用的字段是geocodes[0].location它是字符串形式的“经度,纬度”用英文逗号分隔绝大多数情况下直接split(,)就能拆出两列。level字段标注了这次匹配的精度从“门牌号”到“道路”再到“区县”精度逐级下降。如果你看到count大于 1说明存在多个候选结果。批量脚本里如果只拿第一个就可能落到错误行政区这个坑在第五章会重点展开。2.2 Web服务Key的创建与权限边界白名单和配额怎么看走到批量这一步之前先把Key的权限边界搞清楚不然脚本跑一半会被各种鉴权错误打断。常见流程是在高德开放平台控制台里创建一个应用然后在应用下添加Key服务类型选择“Web服务”。这里最容易犯的错是把用于网页JS API的Key直接拿来请求Web服务接口Key类型不匹配接口会返回鉴权失败。Web服务Key还需要设置IP白名单只有白名单内的出口IP才能调用。提示如果跑脚本的机器是云服务器IP白名单填服务器的公网IP如果是本地电脑填当前宽带的出口IP。出口IP不是固定的换网络环境后要记得更新。配额方面控制台会明确展示每日调用量和并发QPS两个数字个人开发者和企业认证后的差别不小。批量获取经纬度是一次性任务还是长期任务决定你需不需要做企业认证。我的建议是先拿个人Key跑通小批量如果线上每天有几万条稳定需求再去处理认证的事不要一上来就追求高配额毕竟Key的权限越大泄露后的风险也越大。2.3 先用curl打一次真实请求确认链路再写程序很多批量脚本跑出全量空坐标问题不在代码而在Key或参数从第一步就错了。在写Python之前先手动执行一次curl确认三件事接口通不通、Key认不认、address参数解析正不正确。# 第二次curl换一个不带行政区前缀的地址看city参数的作用 curl -s https://restapi.amap.com/v3/geocode/geo?address%E6%9F%90%E8%B7%AF1%E5%8F%B7city%E6%9F%90%E5%B8%82key你的Keyoutputjson如果返回的count是0说明地址文本和高德的道路库对不上。这时做两个实验把完整地址带上行政区重新请求或者把city参数去掉再请求对比一下哪种返回的结果更准。这一步花五分钟能省掉后面写代码和排错的两小时。3. 最小批量脚本用Python把CSV地址批量转成经纬度并落盘3.1 地址预处理空行、空格、冗余后缀与上下文拼接批量场景第一步不是写请求而是先看输入数据。实际项目里拿到的地址往往长这样有全角空格、有换行符、有的带省市区分前缀、有的只写“某某路1号”、有的把“号”写成“.”甚至还有不可见字符。这些脏数据直接喂给接口匹配命中率会明显下降。我通常先做一个清洗函数把明显干扰项去掉但不过度处理import re # clean_address把脏地址收敛成比较规整的文本 # 核心思路只处理空白和全半角语义部分尽量保留交给高德去解析 def clean_address(raw: str) - str: if not raw or not isinstance(raw, str): return text raw.replace(, ,).replace(, ().replace(, )) text re.sub(r\s, , text) # 去掉所有空白包括全角空格和换行 text text.replace(, .).replace(., 号) # 兼容“x.”写成“x号” return text.strip()逻辑说明re.sub(r\s, , text)把地址里的空格、全角空格、换行符一次性删干净因为地址文本里本来就不该有空白。全角逗号替换成半角是为了避免后面写CSV时字段被逗号干扰。把.替换成号不是通用规则只有当你的数据源习惯把“某路1号”写成“某路1.”时才启用默认别开。参数说明地址清洗没有万能规则。我一般只处理空白、全半角和个别符号行政区前缀、道路名、门牌号这些语义信息尽量保留让高德的解析引擎去做分词。你替接口做太多语义判断反而容易引入错误。3.2 单线程主流程请求、解析、落盘的最小完整代码清洗完地址下面是最小可用的批量脚本。它不追求速度先把链路跑通import csv import time import requests GEOCODE_URL https://restapi.amap.com/v3/geocode/geo # 读取CSV里的地址列逐条请求结果写到一个新CSV def geocode_batch(in_csv: str, out_csv: str, key: str): with open(in_csv, r, encodingutf-8) as f: reader csv.DictReader(f) fieldnames reader.fieldnames [lng, lat, level, geocode_status] rows list(reader) with open(out_csv, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() for row in rows: address row.get(address, ) row[geocode_status] empty row[lng], row[lat], row[level] , , if not address: writer.writerow(row) continue params { address: address, city: row.get(city, ), key: key, output: json, } try: resp requests.get(GEOCODE_URL, paramsparams, timeout10) data resp.json() except requests.exceptions.Timeout: row[geocode_status] timeout writer.writerow(row) continue if data.get(status) ! 1 or not data.get(geocodes): row[geocode_status] data.get(info, unknown) writer.writerow(row) continue geo data[geocodes][0] loc geo.get(location, ) if loc: row[lng], row[lat] loc.split(,, 1) row[level] geo.get(level, ) row[geocode_status] ok writer.writerow(row) # 单线程限速每请求间隔0.1秒约10次/秒 time.sleep(0.1) if __name__ __main__: geocode_batch(addresses.csv, addresses_geo.csv, key你的Key)代码逻辑说明csv.DictWriter配合fieldnames可以在原CSV的每一行后面追加lng、lat、level和geocode_status四列。把geocode_status这种失败原因写进输出而不是直接抛异常是为了让批量任务跑完后可以针对失败行做二次处理。参数说明timeout10表示单次请求最多等10秒防止某条地址导致整个脚本卡死。time.sleep(0.1)是故意加的把请求频率压到每秒10次以内。如果你的Key并发配额只有每秒3次就把这个时间改成0.4秒宁可慢一点也不要触发频控。3.3 location字段与level精度拿到坐标后还要存什么跑完上面的脚本你手里会是这样一个CSV原始地址、lng、lat、level、status。这里的level一定要保留它是后面判断数据能不能用的关键。常见level取值和对应含义我用一张表来说明level取值含义适合的用途门牌号精确匹配到具体门牌精确定位、导航、落点道路匹配到某条道路近邻分析、聚合展示区县只匹配到区县中心不适合精确落点省市只匹配到省市级别只能用来做行政统计注意如果你拿到了level为“区县”的结果说明高德没有解析出门牌号location返回的是区县政府或区域中心点。这种坐标拿去算门店周边客流结果会严重失真。处理办法是在后续分析里把level不是“门牌号”的行过滤掉或者单独标记低精度不要混在一张表里做空间计算。4. 批量进阶线程池并发、本地缓存与断点续跑4.1 用ThreadPoolExecutor控制QPS线程数不是越大越好单线程脚本稳定但处理几千条地址要跑几分钟到十几分钟因为每条请求之间都sleep了0.1秒。批量场景要提速常见做法是引入线程池但线程数不能随便开。线程池的意义是减少网络IO等待而不是无限制增加并发。我一般会先写一个简单的限速器再套线程池import threading import time # RateLimiter固定速率限速器保证每秒最多放行 max_calls 次 class RateLimiter: def __init__(self, max_calls: float): self.interval 1.0 / max_calls self._lock threading.Lock() self._last 0.0 def wait(self): with self._lock: now time.monotonic() gap now - self._last if gap self.interval: time.sleep(self.interval - gap) self._last time.monotonic()然后在每个线程的任务函数里调用limiter.wait()再发请求from concurrent.futures import ThreadPoolExecutor import requests limiter RateLimiter(max_calls5) # 每秒最多5次 # 单条地址的地理编码请求带限速 def geocode_one(row, key): limiter.wait() params { address: row[address], city: row.get(city, ), key: key, output: json, } resp requests.get(GEOCODE_URL, paramsparams, timeout10) data resp.json() # 这里继续复用上一章的解析逻辑把结果写入row return row # 开5个线程配合上面的5QPS限速器实际上每秒就是5次 with ThreadPoolExecutor(max_workers5) as pool: futures [pool.submit(geocode_one, row, key) for row in rows] for future in futures: updated_row future.result() # 把updated_row写入最终结果文件逻辑说明RateLimiter里的threading.Lock保证多线程环境下time.sleep的累计频率不会超限。max_workers5和RateLimiter(max_calls5)配合就是“5个线程轮流跑每秒总计最多5次请求”。如果你的并发配额是10就都改成10。参数说明线程数不是越大越好。当线程数超过QPS限制时多余的线程只是在排队不会提高吞吐反而会占用本地连接资源。经验做法是把max_workers设成不超过控制台并发配额的二分之一留出余量给重试请求。4.2 加一道本地缓存相同地址不重复消耗配额批量转坐标的项目通常不是跑一次就结束。地址库每周有新数据如果把全部地址重新请求一遍前两天刚跑过的几千条就白消耗配额了。我习惯在脚本里加一层本地缓存以清洗后的地址文本为key把坐标存到JSON文件里import json import os CACHE_FILE geo_cache.json # 启动时加载已有缓存 def load_cache(): if os.path.exists(CACHE_FILE): with open(CACHE_FILE, r, encodingutf-8) as f: return json.load(f) return {} # 带缓存的查询入口 def geocode_with_cache(cache, address, city, key): cache_key f{address}|{city} if cache_key in cache: return cache[cache_key] data do_geocode(address, city, key) # 实际请求接口 if data: cache[cache_key] data # 缓存仅在批量任务结束时统一写入避免频繁IO return data # 批量任务跑完后调用save_cache def save_cache(cache): tmp_file CACHE_FILE .tmp with open(tmp_file, w, encodingutf-8) as f: json.dump(cache, f, ensure_asciiFalse) os.replace(tmp_file, CACHE_FILE)逻辑说明缓存key把address和city拼在一起是为了避免“某市某路1号”和“某县某路1号”互相覆盖。任务跑完再统一写盘比每条请求都写一次JSON高效得多。用os.replace写临时文件可以防止脚本中途崩溃把缓存文件写坏。参数说明这个缓存的粒度是整个地址如果你传的地址每次都有细微差异比如多了一个空格或者后缀缓存命中率就会很低。所以在第3.1节的清洗函数一定要在进缓存前调用保证同一地址的key始终一致。4.3 断点续跑与幂等中断后不要让结果归零批量脚本最大的风险是跑到一半进程被杀、网络断开或者配额耗尽。如果没有进度记录重跑一遍不仅浪费时间还会重复扣掉已经成功过的那些请求配额。我做断点续跑的方式很朴素准备一个done.txt每成功一条就把地址追加进去重新运行时先加载这个集合遇到已经存在的地址直接跳过。# 已完成的地址集合重复运行自动跳过 done_set set() if os.path.exists(done.txt): with open(done.txt, r, encodingutf-8) as f: done_set set(line.strip() for line in f) # 在任务函数里先判断再请求 def process_skip_done(row): if row[address] in done_set: return None # 已处理过跳过 updated geocode_one(row, key) if updated and updated.get(geocode_status) ok: with open(done.txt, a, encodingutf-8) as f: f.write(row[address] \n) return updated逻辑说明写done.txt用的是追加模式即使脚本在第1000条崩溃前999条记录都还在。重新运行时从第一条开始扫遇到已在done_set里的直接跳过不发起任何请求。这样脚本天然具备幂等性跑多少遍都不会产生重复扣费。提示done.txt本质上也是缓存的一种只不过它只记录“已完成”不存储具体坐标。两者可以共存缓存负责存坐标进度文件负责去重。5. 避坑实录高德批量地理编码的五个高频翻车现场5.1 现象请求返回status0排查半天发现是Key的问题现象脚本刚启动就大面积失败返回的info字段不是OK而是各种鉴权或参数异常也有一些情况是半小时后突然失败前面正常。原因最常见的有三种。第一Key类型用错了拿的是Web端JS API的Key去请求Web服务接口第二IP白名单没配置或配置的不是当前出口IP第三Key本身写错或过期。解决先去控制台确认Key的“服务类型”是不是Web服务再看“IP白名单”里有没有当前机器出口IP。可以用一个在线接口查出口IP也可以直接在服务器上用系统命令看公网地址。改完白名单要等一两分钟生效不要立刻重试。另外我习惯把Key放到环境变量里不在代码里写死方便不同环境切换也不容易误提交到仓库。5.2 现象前期请求正常跑到一半开始返回频控限制现象脚本开头几百条全部成功跑到中段开始连续报错info里出现类似“CUQPS_HAS_EXCEEDED_THE_LIMIT”的提示重试几次还是一样。原因这是触发了配额限制。可能是单日总配额用完也可能是并发QPS超了。日配额用完时当天后续所有请求都会被拒直到第二天的配额重置并发超限则是瞬时的等一小段时间能恢复。解决先区分是哪一种。看报错信息里的关键词或者去控制台查看今日剩余配额。如果是日配额用完只能等重置或者升级认证后提高配额如果是并发超限说明你的线程数或请求频率超过了Key允许的QPS调低max_workers和RateLimiter的速率即可。一个实用习惯是批量任务开始前先记录当前配额剩余值跑完后再看一眼差值就是你这次实际消耗的量这样能知道脚本离限制还有多少余量。5.3 现象地址明明存在坐标却偏出几个路口或者落在附近园区现象拿一条“某大厦B座”去请求返回的status是1坐标也是合法的经度纬度但拉到地图上一看位置偏到了旁边的路口甚至另一个园区。原因地理编码接口是按照结构化地址来解析的它擅长处理“省市区道路门牌号”这种文本。但像“某大厦”“某产业园”这种名称型地址本质上是POI地理编码的词典里不一定有只能靠相近的门牌号推断所以坐标就会偏。解决对这种存量数据不要死磕地理编码接口。把地址分成两类结构化地址走地理编码名称型地址走POI搜索接口。POI搜索的请求参数类似但接口不同返回的坐标精度要高得多。如果暂时不想换接口可以在address参数前面拼上区级前缀比如“某市某区某大厦”让高德先锁定行政区再在区内搜索命中率会明显上升。5.4 现象一开线程池就大量超时和连接重置比单线程还慢现象把max_workers从1改成10预期速度翻几倍结果大量请求超时部分连接被重置整个脚本跑下来耗时反而比单线程还长。原因这是一个典型的“并发超限雪崩”。线程数超过QPS配额后超出的请求直接被限流它们占着线程池里的槽位等待重试后续请求进不来整体吞吐反而下降。更糟的是重试请求又叠加在已经超限的流量上情况越来越糟。解决把线程数降回到配额以内并给请求加指数退避重试。重试逻辑通常是这样的第一次失败等2秒第二次等4秒第三次等8秒最多重试两次就放弃把这条地址标记为失败不拖累整个队列。比例上说宁可让线程池空着也不要让它满载跑。5.5 现象同一地址两次运行结果不同坐标相差不小现象同一份CSV跑完第一次再把结果删掉重跑发现部分地址的经纬度变了偏差甚至跨区。原因当geocodes返回多个候选时我们默认取geocodes[0]但这个顺序并不保证每次都是最优解。地址文本有歧义时比如“某路1号”在多个区都存在高德会返回多个候选第一次可能排在前面的是A区第二次排序权重变了排到前面的成了B区坐标自然就变了。解决批量落库时把count字段和level字段一起存下来。如果count大于1把整条geocodes数组打印到日志里人工抽查或写个小脚本对比几个候选的位置。同时通过city参数把搜索范围限定在正确城市绝大多数歧义都能消除。如果结果只是相差几百米且level都是“道路”级别那属于匹配多义性不一定要处理。6. 数据落地前的最后一道工序三条坐标质量校验规则批量结果拿到手先别急着建表或画图。我自己走过的弯路是脚本看起来全跑通了坐标格式也合法结果落到地图上才发现有几十条落在了完全错误的方向。后来我给自己定了一套三道校验流程每次批量结果落地前都走一遍。第一道是范围合法性校验。虽然坐标格式是数字就能解析但国内项目的坐标范围基本可以限定在一个合理边界内。我在代码里写死这个范围超出直接标记异常def is_valid_lnglat(lng, lat): return 73 lng 135 and 18 lat 54第二道是精度等级过滤。把所有level不是“门牌号”和“道路”的行单独提出来。这些低精度坐标不是不能用但不能和精确坐标混在一张表里做距离计算。我一般会加一列precision_flag标记为精确或模糊下游分析时按需过滤。第三道是随机抽检用逆地理编码验证。随机抽20条结果把坐标再拿给/v3/geocode/regeo接口让它反查地址再和原始地址的行政区做比对def reverse_check(lng, lat, key): url https://restapi.amap.com/v3/geocode/regeo params {location: f{lng},{lat}, key: key, output: json} resp requests.get(url, paramsparams, timeout10) data resp.json() if data.get(status) 1: regeo data.get(regeocode, {}) return regeo.get(addressComponent, {}).get(district, ) return 抽出原始地址里的区名和反查结果里的区名做对比偏差超过一定数量就说明输入数据存在系统性问题需要回到清洗环节排查而不是继续往下游推。这个三道校验的习惯帮我拦住过很多次“坐标数值正常但位置离谱”的脏数据比出了问题再洗数据省事得多。希望这篇高德API批量获取经纬度的实战笔记能帮你少走几步弯路希望帮到你。本文还有配套的精品资源点击获取