如何编写Python脚本调用IGN地理编码API批量获取学校经纬度

需求概述 #

处理含32091所学校的CSV文件，通过IGN地理编码API批量获取经纬度并更新至原文件：

• CSV列对应：第6列=学校名称，第7列=城市代码
• API请求需携带校名（URL编码）、城市代码（postcode参数），指定index=poi、limit=1
• 从返回JSON的features[0].geometry.coordinates提取[经度, 纬度]

完整Python脚本 #

import pandas as pd
import requests
import time
from urllib.parse import quote

# API基础URL
API_BASE_URL = "https://wxs.ign.fr/essentiels/geoportail/geocodage/rest/0.1/search"

def get_coordinates(school_name, city_code):
    """调用IGN API获取经纬度，处理异常情况"""
    # 对学校名称进行URL编码
    encoded_name = quote(school_name)
    # 构造请求参数
    params = {
        "q": encoded_name,
        "index": "poi",
        "limit": 1,
        "returntruegeometry": "false",
        "postcode": city_code
    }
    
    try:
        # 发起GET请求，设置超时
        response = requests.get(API_BASE_URL, params=params, timeout=10)
        response.raise_for_status()  # 捕获HTTP错误
        data = response.json()
        
        # 提取经纬度
        if data.get("features") and len(data["features"]) > 0:
            coordinates = data["features"][0]["geometry"]["coordinates"]
            return coordinates[0], coordinates[1]  # 返回(经度, 纬度)
        else:
            return None, None  # 无匹配结果
    except Exception as e:
        print(f"请求失败：{school_name} - {str(e)}")
        return None, None

if __name__ == "__main__":
    # 读取CSV文件
    df = pd.read_csv("myfile.csv")
    
    # 初始化经度、纬度列（如果不存在则创建）
    if "经度" not in df.columns:
        df["经度"] = None
    if "纬度" not in df.columns:
        df["纬度"] = None
    
    # 遍历每行数据，批量获取经纬度
    for idx, row in df.iterrows():
        # 跳过已获取经纬度的行（可选，避免重复请求）
        if pd.notna(row["经度"]) and pd.notna(row["纬度"]):
            continue
        
        school_name = row.iloc[5]  # 第6列（索引从0开始）
        city_code = row.iloc[6]    # 第7列
        
        # 跳过空值
        if pd.isna(school_name) or pd.isna(city_code):
            print(f"第{idx+1}行缺少校名或城市代码，跳过")
            continue
        
        # 获取经纬度
        lon, lat = get_coordinates(school_name, city_code)
        # 更新DataFrame
        df.loc[idx, "经度"] = lon
        df.loc[idx, "纬度"] = lat
        
        # 每请求10次保存一次进度，避免数据丢失
        if (idx + 1) % 10 == 0:
            df.to_csv("myfile_updated.csv", index=False)
            print(f"已完成{idx+1}条数据，进度已保存")
        
        # 设置请求间隔，避免触发API限流（根据IGN规则调整）
        time.sleep(0.5)
    
    # 最终保存完整结果
    df.to_csv("myfile_updated.csv", index=False)
    print("所有数据处理完成，结果已保存至myfile_updated.csv")

关键说明 #

• URL编码：使用urllib.parse.quote对学校名称进行编码，避免特殊字符导致请求错误
• 异常处理：捕获HTTP错误、JSON解析错误等，避免脚本中断
• 进度保存：每处理10条数据保存一次文件，防止意外中断丢失数据
• 限流控制：添加time.sleep(0.5)控制请求频率，需根据IGN API的限流规则调整间隔
• 跳过重复：已获取经纬度的行将被跳过，可重复运行脚本续传未完成的数据

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