Spotify API Python脚本令牌刷新与卡顿问题求助

一、脚本卡顿无错误的可能原因 #

• 未设置请求超时：默认情况下requests等HTTP库无超时限制，若Spotify API因网络波动、服务器负载高等未及时响应，脚本会无限等待，不会主动抛出错误。
• 令牌逻辑阻塞：添加的缓存令牌代码可能存在死循环、同步锁未释放，或刷新令牌的请求本身卡住（如网络故障），导致后续流程停滞但未触发异常捕获。
• 静默速率限制：Spotify的速率限制有时不会返回明确的429错误，而是让请求挂起等待配额重置，这种情况控制台无错误输出。
• 文件IO阻塞：写入CSV时若未用with语句管理文件句柄，可能因系统资源占用导致写入操作阻塞，脚本停滞。

二、实现可靠的令牌自动刷新（Client Credentials模式） #

Client Credentials模式下令牌有效期为1小时，正确逻辑需在每次API请求前检查令牌是否过期，或在请求返回401（令牌失效）时自动刷新，同时给所有请求添加超时避免阻塞。

示例代码实现 #

import requests
import time
import csv
from typing import Dict

# Spotify API配置
CLIENT_ID = "你的客户端ID"
CLIENT_SECRET = "你的客户端密钥"
BASE_URL = "https://api.spotify.com/v1"

# 令牌缓存
token_info: Dict[str, any] = {
    "access_token": None,
    "expires_at": 0  # 过期时间戳
}

def get_access_token() -> str:
    """获取或刷新访问令牌"""
    global token_info
    current_time = time.time()
    
    # 提前60秒刷新，避免临界时间点失效
    if token_info["access_token"] and current_time < token_info["expires_at"] - 60:
        return token_info["access_token"]
    
    # 请求新令牌
    auth_url = "https://accounts.spotify.com/api/token"
    auth_data = {"grant_type": "client_credentials"}
    auth_headers = {
        "Authorization": f"Basic {requests.auth._basic_auth_str(CLIENT_ID, CLIENT_SECRET)}"
    }
    
    try:
        # 添加超时，避免请求阻塞
        response = requests.post(auth_url, data=auth_data, headers=auth_headers, timeout=10)
        response.raise_for_status()
        token_data = response.json()
        token_info["access_token"] = token_data["access_token"]
        token_info["expires_at"] = current_time + token_data["expires_in"]
        return token_info["access_token"]
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"获取令牌失败: {str(e)}")

def fetch_playlist_tracks(playlist_id: str) -> list:
    """获取播放列表所有音轨，自动处理分页和令牌刷新"""
    tracks = []
    next_url = f"{BASE_URL}/playlists/{playlist_id}/tracks"
    
    while next_url:
        token = get_access_token()
        headers = {"Authorization": f"Bearer {token}"}
        
        try:
            # 添加超时防止请求卡住
            response = requests.get(next_url, headers=headers, timeout=15)
            if response.status_code == 401:
                # 令牌失效，强制刷新后重试
                token_info["access_token"] = None
                token = get_access_token()
                response = requests.get(next_url, headers={"Authorization": f"Bearer {token}"}, timeout=15)
                response.raise_for_status()
            elif response.status_code == 429:
                # 处理速率限制，等待后重试
                retry_after = int(response.headers.get("Retry-After", 10))
                time.sleep(retry_after)
                continue
            else:
                response.raise_for_status()
            
            data = response.json()
            tracks.extend(data["items"])
            next_url = data["next"]
        except requests.exceptions.RequestException as e:
            print(f"获取播放列表数据失败: {str(e)}")
            time.sleep(5)
            continue
    return tracks

def extract_track_info(track_item: dict) -> dict:
    """提取音轨、艺人、专辑信息"""
    track = track_item["track"]
    return {
        "track_name": track["name"],
        "artist_name": ", ".join([artist["name"] for artist in track["artists"]]),
        "album_name": track["album"]["name"],
        "album_release_date": track["album"]["release_date"]
    }

def save_to_csv(tracks_info: list, filename: str):
    """保存信息到CSV文件"""
    if not tracks_info:
        return
    keys = tracks_info[0].keys()
    # 使用with语句确保文件正确关闭，避免IO阻塞
    with open(filename, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=keys)
        writer.writeheader()
        writer.writerows(tracks_info)

if __name__ == "__main__":
    playlist_id = "37i9dQZF1DXcBWIGoYBM5M"  # 示例：官方"New Music Friday"播放列表
    try:
        tracks = fetch_playlist_tracks(playlist_id)
        tracks_info = [extract_track_info(item) for item in tracks]
        save_to_csv(tracks_info, "spotify_new_tracks.csv")
        print("数据已成功保存到CSV文件")
    except Exception as e:
        print(f"脚本执行失败: {str(e)}")

关键注意事项 #

• 所有请求加超时：给requests.get/post设置timeout参数（10-15秒），避免网络问题导致无限等待。
• 主动检查令牌有效期：提前60秒刷新令牌，避免请求过程中令牌刚好过期。
• 针对性处理HTTP错误：捕获RequestException，对401（令牌失效）、429（速率限制）做专属处理。
• 用with管理文件IO：确保文件句柄自动关闭，避免资源占用导致阻塞。
• 添加重试逻辑：网络请求失败时短暂等待后重试，提升脚本稳定性。

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