Python调用API写入JSON触发KeyError错误求助

错误根源 #

1. KeyError直接原因：首次调用write_json时，写入的是单个电影字典（TMDB API返回结果是字典），文件内容为字典而非列表。第二次读取文件后data是字典结构，执行data[0]自然报错——字典没有索引键0。
2. 函数逻辑混乱：write_json函数硬覆盖传入的fname参数，且未定义newData变量，读取后追加数据的逻辑完全错误，每次调用都会覆盖之前的内容。
3. 未处理API异常：未判断请求是否成功（如404、500状态码），若API返回错误信息，后续处理会直接报错。

修复后的代码（推荐方案） #

该方案先收集所有电影数据到列表，最后一次性写入文件，逻辑清晰且效率更高：

import json
import requests

# 替换为你的TMDB API密钥
API_KEY = "****"
INPUT_FILE = "test.json"
OUTPUT_FILE = "NewdataTest.json"

def main():
    # 从test.json提取tmdb_id（需确保test.json结构符合推导式预期）
    tmdb_ids = []
    with open(INPUT_FILE, 'r') as json_fp:
        imdb_info = json.load(json_fp)
        tmdb_ids = [movie_info['tmdb_id'] 
                   for movies_chunk in imdb_info 
                   for movie_index, movie_info in movies_chunk.items()]

    # 收集所有电影数据
    all_movies = []
    for tmdb_id in tmdb_ids:
        api_url = f"https://api.themoviedb.org/3/movie/{tmdb_id}?api_key={API_KEY}"
        print(f"请求URL: {api_url}")
        try:
            response = requests.get(api_url)
            response.raise_for_status()  # 触发HTTP错误异常
            movie_data = response.json()
            all_movies.append(movie_data)
            print(f"成功获取ID {tmdb_id} 的数据")
        except requests.exceptions.RequestException as e:
            print(f"获取ID {tmdb_id} 数据失败: {str(e)}")
            continue

    # 将所有数据写入JSON文件
    with open(OUTPUT_FILE, 'w', encoding='utf-8') as f:
        json.dump(all_movies, f, indent=4, ensure_ascii=False)
    print(f"所有数据已写入 {OUTPUT_FILE}")

if __name__ == "__main__":
    main()

关键改动说明 #

• 逻辑重构：放弃逐次写入的错误逻辑，先批量收集数据再一次性写入，避免文件读写冲突。
• 异常处理：捕获请求过程中的网络错误、HTTP错误，避免程序中途崩溃，同时打印错误信息便于排查。
• 编码优化：写入时指定encoding='utf-8'和ensure_ascii=False，解决非ASCII字符乱码问题。
• 可读性提升：使用f-string拼接API URL，代码更简洁易懂。

可选：大数据量逐次写入方案 #

若需处理超大规模数据，可采用逐次追加写入的方式，避免占用过多内存：

import json
import requests

API_KEY = "****"
INPUT_FILE = "test.json"
OUTPUT_FILE = "NewdataTest.json"

def get_existing_data():
    """读取已存在的电影数据，文件不存在则返回空列表"""
    try:
        with open(OUTPUT_FILE, 'r', encoding='utf-8') as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return []

def main():
    tmdb_ids = []
    with open(INPUT_FILE, 'r') as json_fp:
        imdb_info = json.load(json_fp)
        tmdb_ids = [movie_info['tmdb_id'] 
                   for movies_chunk in imdb_info 
                   for movie_index, movie_info in movies_chunk.items()]

    existing_movies = get_existing_data()
    for tmdb_id in tmdb_ids:
        # 跳过已存在的ID，避免重复请求
        if any(movie['id'] == tmdb_id for movie in existing_movies):
            print(f"ID {tmdb_id} 已存在，跳过")
            continue
        
        api_url = f"https://api.themoviedb.org/3/movie/{tmdb_id}?api_key={API_KEY}"
        print(f"请求URL: {api_url}")
        try:
            response = requests.get(api_url)
            response.raise_for_status()
            movie_data = response.json()
            existing_movies.append(movie_data)
            # 写入文件
            with open(OUTPUT_FILE, 'w', encoding='utf-8') as f:
                json.dump(existing_movies, f, indent=4, ensure_ascii=False)
            print(f"ID {tmdb_id} 数据已写入文件")
        except requests.exceptions.RequestException as e:
            print(f"获取ID {tmdb_id} 数据失败: {str(e)}")
            continue
    print("处理完成")

if __name__ == "__main__":
    main()

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