如何用Python批量从坐标提取斯里兰卡地区信息？

一、优化Nominatim批量请求方案 #

你的原有代码存在每次请求创建新geolocator实例、未严格遵守API速率限制、无批量处理逻辑的问题，以下是优化后的实现：

from geopy.geocoders import Nominatim
from geopy.point import Point
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import pandas as pd

# 全局复用geolocator实例，设置合规的user_agent（建议替换成自己的标识）
geolocator = Nominatim(user_agent="sri-lanka-district-fetcher", timeout=10)
MAX_RETRIES = 3
REQUEST_DELAY = 1.2  # 严格遵守Nominatim 1秒/请求的限制，留0.2秒缓冲

def get_district(lat, longi):
    point = Point(lat, longi)
    retries = 0
    while retries < MAX_RETRIES:
        try:
            location = geolocator.reverse(point, exactly_one=True)
            # 容错处理：避免地址字段缺失
            address = location.raw.get('address', {}) if location else {}
            return address.get('state_district')
        except Exception as e:
            print(f"坐标({lat}, {longi})请求失败: {str(e)}")
            retries += 1
            time.sleep(REQUEST_DELAY)
    print(f"坐标({lat}, {longi})重试次数超限")
    return None

def batch_get_districts(coords_list):
    results = []
    # 用线程池控制并发，避免触发API反爬
    with ThreadPoolExecutor(max_workers=5) as executor:
        future_map = {executor.submit(get_district, lat, lon): (lat, lon) for lat, lon in coords_list}
        for future in as_completed(future_map):
            lat, lon = future_map[future]
            try:
                district = future.result()
                results.append((lat, lon, district))
            except Exception as e:
                print(f"坐标({lat}, {lon})处理异常: {str(e)}")
                results.append((lat, lon, None))
            time.sleep(REQUEST_DELAY)
    return results

# 调用示例
# coords = [(6.9271, 79.8612), (7.2906, 80.6337)]
# district_results = batch_get_districts(coords)
# print(district_results)

方案优缺点 #

• 优点：无需额外下载数据，代码改动幅度小，复用实例降低开销，线程池提升批量处理效率
• 缺点：受API速率限制，10000条数据需耗时3-4小时，网络波动仍可能导致部分请求失败

二、本地地理数据匹配方案（推荐） #

对于10000条级别的数据，本地空间匹配是效率最高的方案，完全无需依赖外部API。核心思路是用斯里兰卡行政区划的Shapefile，通过空间计算匹配坐标所属区域。

步骤与代码 #

1. 先安装依赖库：

pip install geopandas shapely

2. 下载斯里兰卡行政区划Shapefile：从GADM获取Level 2数据（对应District层级），解压后得到.shp格式文件。

3. 批量匹配代码：

import geopandas as gpd
from shapely.geometry import Point
import pandas as pd

def load_sl_districts(shapefile_path):
    # 加载Shapefile并转换为WGS84坐标系（经纬度通用格式）
    sl_gdf = gpd.read_file(shapefile_path)
    if sl_gdf.crs != "EPSG:4326":
        sl_gdf = sl_gdf.to_crs("EPSG:4326")
    # 保留所需字段（NAME_2是GADM数据中District的名称字段，需根据实际文件调整）
    return sl_gdf[['NAME_2', 'geometry']]

def batch_match_districts(coords_list, sl_gdf):
    # 将坐标转换为GeoDataFrame
    points_gdf = gpd.GeoDataFrame(
        geometry=[Point(lon, lat) for lat, lon in coords_list],
        crs="EPSG:4326"
    )
    # 空间连接：找出每个点所在的District
    joined_gdf = gpd.sjoin(points_gdf, sl_gdf, how="left", predicate="within")
    # 整理结果
    results = []
    for idx, row in joined_gdf.iterrows():
        lat, lon = coords_list[idx]
        district = row['NAME_2'] if not pd.isna(row['NAME_2']) else None
        results.append((lat, lon, district))
    return results

# 调用示例
# sl_districts = load_sl_districts("gadm41_LKA_2/gadm41_LKA_2.shp")
# coords = [(6.9271, 79.8612), (7.2906, 80.6337)]
# district_results = batch_match_districts(coords, sl_districts)
# print(district_results)

方案优缺点 #

• 优点：本地处理速度极快（10000条数据仅需几秒），不受网络和API限制，结果稳定可靠
• 缺点：需要下载几十MB的Shapefile，需了解基础的空间数据概念

三、第三方API备选方案 #

如果对处理时间要求不高且愿意承担少量成本，可以使用谷歌地图地理编码API（有免费额度），但不建议用于10000条以上的大规模数据：

import requests
import time

GOOGLE_API_KEY = "你的API密钥"
MAX_RETRIES = 3

def get_district_google(lat, lon):
    url = f"https://maps.googleapis.com/maps/api/geocode/json?latlng={lat},{lon}&key={GOOGLE_API_KEY}&region=lk"
    retries = 0
    while retries < MAX_RETRIES:
        try:
            response = requests.get(url)
            data = response.json()
            if data['status'] == 'OK':
                # 遍历地址组件找到District（对应administrative_area_level_2）
                for component in data['results'][0]['address_components']:
                    if 'administrative_area_level_2' in component['types']:
                        return component['long_name']
                return None
            else:
                print(f"坐标({lat}, {lon})请求失败: {data['status']}")
                retries += 1
                time.sleep(2)
        except Exception as e:
            print(f"坐标({lat}, {lon})请求异常: {str(e)}")
            retries += 1
            time.sleep(2)
    print(f"坐标({lat}, {lon})重试次数超限")
    return None

内容的提问来源于stack exchange，提问作者Minura Punchihewa