使用Spotipy调用Spotify API获取曲目URI时的请求限制解决方案

关于无需URI直接获取音频特征的说明 #

Spotify官方API没有提供直接通过歌曲名获取音频特征的端点——audio-features接口必须传入曲目ID或URI才能返回特征数据。但你可以把搜索曲目获取ID和请求音频特征合并为一个步骤，不需要单独生成URI列，减少中间环节的存储和额外请求。

批量处理5万行数据的优化策略 #

针对请求频繁触发休眠的问题，核心思路是减少请求次数、控制请求速率、复用已有结果，以下是具体落地方法：

1. 启用本地缓存，复用重复请求 #

Billboard榜单数据中必然存在大量重复曲目（同一首歌多次上榜），用「歌名+歌手」作为唯一键，把已获取的特征结果存在本地JSON文件或内存字典里，重复曲目直接读缓存，不用重复调用API。

2. 批量请求音频特征 #

audio-features接口支持一次传入最多100个曲目ID，把搜索到的ID攒成批次再请求，能把5万次请求压缩到500次以内，大幅降低触发速率限制的概率。

3. 主动控制请求速率 #

免费版Spotify API的速率限制是180次/分钟，批量请求时每完成一批（100个ID）就主动休眠30-40秒，避免触发Spotipy的自动休眠机制（自动休眠可能会暂停更久）。

4. 预处理数据，减少无效请求 #

先清洗数据集：

• 去除空的歌名/歌手列
• 统一文本格式（比如转小写、去掉特殊字符、清理多余空格）
• 过滤明显无效的条目（比如歌名过短、歌手名模糊）

5. 合并搜索与特征请求的代码示例 #

import spotipy
from spotipy.oauth2 import SpotifyClientCredentials
import time
import json
import pandas as pd

# 初始化Spotipy客户端
sp = spotipy.Spotify(
    client_credentials_manager=SpotifyClientCredentials(
        client_id="你的客户端ID",
        client_secret="你的客户端密钥"
    )
)

# 加载本地缓存（首次运行自动创建）
feature_cache = {}
CACHE_FILE = "spotify_feature_cache.json"
try:
    with open(CACHE_FILE, "r", encoding="utf-8") as f:
        feature_cache = json.load(f)
except FileNotFoundError:
    pass

def get_track_features(track_name, artist_name):
    """合并搜索曲目+获取音频特征的函数，带缓存"""
    # 生成唯一缓存键
    cache_key = f"{track_name.lower().strip()}||{artist_name.lower().strip()}"
    
    # 优先从缓存读取
    if cache_key in feature_cache:
        return feature_cache[cache_key]
    
    # 搜索曲目（限定歌手+歌名，提升匹配准确率）
    search_query = f"track:{track_name} artist:{artist_name}"
    try:
        search_result = sp.search(q=search_query, type="track", limit=1)
        if not search_result["tracks"]["items"]:
            return None
        
        # 提取ID并请求音频特征
        track_id = search_result["tracks"]["items"][0]["id"]
        audio_features = sp.audio_features([track_id])[0]
        
        # 保存到缓存，每100条批量写入一次
        feature_cache[cache_key] = audio_features
        if len(feature_cache) % 100 == 0:
            with open(CACHE_FILE, "w", encoding="utf-8") as f:
                json.dump(feature_cache, f, indent=2)
        
        return audio_features
    except Exception as e:
        print(f"请求失败 [{track_name} - {artist_name}]: {str(e)}")
        # 触发速率限制时主动休眠10秒后重试
        time.sleep(10)
        return get_track_features(track_name, artist_name)

# 处理你的Billboard数据集
df = pd.read_csv("billboard_data.csv")  # 替换成你的数据集路径

# 遍历数据，添加音频特征列
for idx, row in df.iterrows():
    features = get_track_features(row["track_name"], row["artist"])
    if features:
        df.loc[idx, "danceability"] = features["danceability"]
        df.loc[idx, "energy"] = features["energy"]
        df.loc[idx, "valence"] = features["valence"]
        # 按需添加其他音频特征列（tempo、loudness等）
    else:
        df.loc[idx, "danceability"] = None  # 标记无效条目

# 最后保存缓存和处理后的数据集
with open(CACHE_FILE, "w", encoding="utf-8") as f:
    json.dump(feature_cache, f, indent=2)
df.to_csv("billboard_with_features.csv", index=False, encoding="utf-8")

6. 进阶：批量搜索+批量请求 #

如果想进一步提升效率，可以先批量收集所有未缓存的曲目ID，再分批次请求特征：

# 先收集所有未缓存的曲目ID
track_ids = []
row_indices = []
cache_keys = []

for idx, row in df.iterrows():
    cache_key = f"{row['track_name'].lower().strip()}||{row['artist'].lower().strip()}"
    if cache_key not in feature_cache:
        search_query = f"track:{row['track_name']} artist:{row['artist']}"
        search_result = sp.search(q=search_query, type="track", limit=1)
        if search_result["tracks"]["items"]:
            track_ids.append(search_result["tracks"]["items"][0]["id"])
            row_indices.append(idx)
            cache_keys.append(cache_key)
    
    # 每攒100个ID就批量请求特征
    if len(track_ids) == 100:
        features_list = sp.audio_features(track_ids)
        for idx_row, cache_key, features in zip(row_indices, cache_keys, features_list):
            if features:
                df.loc[idx_row, "danceability"] = features["danceability"]
                # 其他特征列
                feature_cache[cache_key] = features
        
        # 保存缓存+休眠
        with open(CACHE_FILE, "w", encoding="utf-8") as f:
            json.dump(feature_cache, f, indent=2)
        time.sleep(35)
        # 重置批次
        track_ids = []
        row_indices = []
        cache_keys = []

# 处理剩余的ID
if track_ids:
    features_list = sp.audio_features(track_ids)
    for idx_row, cache_key, features in zip(row_indices, cache_keys, features_list):
        if features:
            df.loc[idx_row, "danceability"] = features["danceability"]
            feature_cache[cache_key] = features
    with open(CACHE_FILE, "w", encoding="utf-8") as f:
        json.dump(feature_cache, f, indent=2)

关键注意事项 #

• 搜索时必须同时指定track和artist，否则会返回大量同名无关曲目，导致特征数据错误
• 免费版API不要尝试高并发请求，容易被临时限制访问；如果是付费版，可以适当用concurrent.futures控制并发数（建议≤5）
• 定期保存缓存，避免程序中断后丢失已获取的数据

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