调用美国人口普查局地理编码API 运行千余条后触发JSONDecodeError报错

你碰到的JSONDecodeError: Expecting value: line 1 column 1 (char 0)本质是代码尝试将非JSON格式的响应内容（通常是空字符串、429限流提示、网关返回的错误HTML页面）当成合法JSON解析时触发的解析异常。结合你批量向公共接口发请求的场景，最常见的触发原因是请求频率过高被美国人口普查局服务端临时拦截，返回了不符合JSON格式要求的拦截响应。

• 异常捕获逻辑覆盖不全：当前捕获范围仅包含RuntimeError, TypeError, NameError, IndexError，完全没有覆盖json.JSONDecodeError类，一旦触发JSON解析失败，程序会直接中断，不会跳过异常请求继续运行
• 存在变量拼写不一致的隐性bug：代码中获取响应文本时定义的变量是reponse（拼写缺字母s），但JSON解析时传入的是response（拼写带s），两个变量名不匹配本身就会触发NameError。虽然这个异常被现有捕获逻辑吞掉，但属于易引发调试问题的低级错误
• 请求链路缺少基础校验：requests.get(link)发出请求后，没有校验HTTP状态码、没有判断响应内容是否为空，直接将返回内容丢给JSON解析器，碰到服务端异常返回时极易触发解析错误
• 无请求限频机制：连续无间隔发送6万条请求到公共接口，很容易触发服务端的限流策略，短时间内返回非JSON格式的拦截响应

• 补全异常捕获范围：将json.JSONDecodeError、requests库的常见请求异常（连接超时、连接中断）加入捕获列表，保证单条请求失败不会中断整体流程
• 统一变量命名，消除拼写不一致问题，避免无意义的NameError
• 增加响应基础判断：仅当请求返回200状态码、响应内容非空时，才执行JSON解析逻辑
• 增加100-300ms的随机请求间隔，避免触发接口限流
• 将解析失败的地址单独追加写入日志文件，方便程序中断后批量补跑，避免数据遗漏

import json
import csv
import time
import random
import requests
# ... 前置初始化逻辑（ven_lst、repostory_lst、已有CSV结果读取逻辑保持不变）
def get_fips(df):  
    num = len(ven_lst)
    for i,e in df.itertuples(index=False):
        if e not in repostory_lst:
            try:
                num += 1
                address = i
                vendor = e
                link = "https://geocoding.geo.census.gov/geocoder/geographies/onelineaddress?address={0}&benchmark=Public_AR_Census2020&vintage=Census2020_Census2020&layers=10&format=json".format(address)
                # 增加10s超时配置，避免请求永久挂起
                response = requests.get(link, timeout=10)
                # 校验响应状态，异常时直接记录失败地址
                if response.status_code != 200 or not response.text.strip():
                    with open('failed_address.log', 'a') as err_f:
                        err_f.write(f"{address}\n")
                    time.sleep(random.uniform(0.1, 0.3))
                    continue
                response_1 = json.loads(response.text)
                x = response_1['result']['addressMatches'][0]['geographies']['Census Blocks'][0]['GEOID'][:11]
                fields = [num, e, x]
                # 保留原有逐行写CSV的设计，避免中断丢数据
                with open(r'fibs&ven.csv', 'a', newline='') as f:
                    writer = csv.writer(f)
                    writer.writerow(fields)
                # 随机休眠降低请求频率，避免被限流
                time.sleep(random.uniform(0.1, 0.3))
            # 补全所有需要捕获的异常类型
            except (RuntimeError, TypeError, NameError, IndexError, json.JSONDecodeError, 
                    requests.exceptions.ConnectionError, requests.exceptions.Timeout):
                with open('failed_address.log', 'a') as err_f:
                    err_f.write(f"{address}\n")
                pass
        elif e in repostory_lst:
            pass
    return None

原代码逐行追加写入CSV的设计非常适配大体量批量处理场景，修改后完整保留了该逻辑，同时补充的失败地址日志机制可以进一步降低中断后的重跑成本，跑完主流程后可直接读取日志文件批量补跑失败请求，不需要重复遍历已处理完成的地址。

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