Spotipy的sp.track()调用过慢，批量处理1.8万首歌曲数据求助

1. 先确认批量请求的正确性 #

首先要确保你是用sp.tracks()（批量接口）而不是在循环里重复调用sp.track()（单条接口）。正确的批量调用逻辑应该是提取一批歌曲ID，一次性传给批量接口：

# 从URI中提取歌曲ID（示例URI格式：spotify:track:xxxxxx）
track_ids = [uri.split(":")[-1] for uri in batch_uris]
# 批量请求接口
response = sp.tracks(track_ids)

如果你的代码是在批量分组后仍循环调用单条接口，那和逐个处理效率没区别，这是最常见的坑。

2. 严格处理API速率限制 #

Spotify Web API默认限制是每30秒最多180次请求，批量请求不管传50个还是1个ID都算1次请求。18000首歌仅需360次批量请求，正常情况10分钟内就能完成。如果跑了数小时，大概率是没处理限流导致代码挂起，或者错误重试逻辑缺失。

可以加容错+重试逻辑：

import time
from requests.exceptions import ReadTimeout, ConnectionError

def safe_batch_request(track_ids):
    while True:
        try:
            return sp.tracks(track_ids)
        except (ReadTimeout, ConnectionError):
            # 网络波动，等待5秒重试
            time.sleep(5)
        except Exception as e:
            # 处理429限流，读取响应头的重试间隔
            if hasattr(e, 'headers') and 'Retry-After' in e.headers:
                retry_after = int(e.headers['Retry-After'])
                time.sleep(retry_after + 1)
            else:
                print(f"处理失败: {track_ids}, 错误: {str(e)}")
                return None

3. 优化DataFrame追加效率 #

不要每次循环用df.append()，这会频繁生成新DataFrame，效率极低。应该先把数据存在列表里，最后一次性转成DataFrame合并：

# 初始化空列表存储数据
track_data = []
batch_size = 50
all_track_uris = 你的歌曲URI列表

for i in range(0, len(all_track_uris), batch_size):
    batch_uris = all_track_uris[i:i+batch_size]
    batch_ids = [uri.split(":")[-1] for uri in batch_uris]
    response = safe_batch_request(batch_ids)
    
    if not response:
        continue
    
    # 解析响应数据
    for track in response['tracks']:
        if track:  # 跳过无效ID对应的空数据
            track_data.append({
                'uri': f"spotify:track:{track['id']}",
                'release_date': track['album']['release_date'],
                'explicit': track['explicit'],
                'popularity': track['popularity']
            })
    
    # 每批处理后加小间隔，避免触发限流
    time.sleep(0.1)
    # 打印进度，确认代码在运行
    print(f"已处理 {min(i+batch_size, len(all_track_uris))}/{len(all_track_uris)} 首歌曲")

# 最后合并到原DataFrame
new_df = pd.DataFrame(track_data)
final_df = pd.concat([你的原DataFrame, new_df], ignore_index=True)

4. 额外提速优化 #

• 去重预处理: 先过滤重复的URI，减少无效请求。
• 多线程异步处理: 用线程池控制并发数（最多10线程），进一步提速：

from concurrent.futures import ThreadPoolExecutor

def process_batch(batch_uris):
    batch_ids = [uri.split(":")[-1] for uri in batch_uris]
    response = safe_batch_request(batch_ids)
    if not response:
        return []
    data = []
    for track in response['tracks']:
        if track:
            data.append({
                'uri': f"spotify:track:{track['id']}",
                'release_date': track['album']['release_date'],
                'explicit': track['explicit'],
                'popularity': track['popularity']
            })
    return data

# 拆分所有URI为批次
batches = [all_track_uris[i:i+batch_size] for i in range(0, len(all_track_uris), batch_size)]

# 线程池批量处理
with ThreadPoolExecutor(max_workers=10) as executor:
    results = executor.map(process_batch, batches)

# 合并结果
track_data = []
for res in results:
    track_data.extend(res)

new_df = pd.DataFrame(track_data)
final_df = pd.concat([你的原DataFrame, new_df], ignore_index=True)

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