求助：YouTube Data API v3配额超限致MP3转播放列表脚本中断

问题原因 #

YouTube Data API v3的配额规则中，search().list每次请求消耗100单位配额，playlistItems().insert每次请求消耗50单位配额。每处理一首歌曲需要1次搜索+1次插入，合计150单位配额。每日10000配额的上限下，最多能处理约66首歌，和你遇到的60-70首的情况完全匹配。

解决方案 #

1. 缓存搜索结果：对相同标题+艺术家的歌曲，缓存对应的video_id，避免重复搜索浪费配额。
2. 保存处理进度：将已处理的歌曲、剩余待处理歌曲及播放列表ID写入本地文件，次日续跑时无需从头开始。
3. 配额异常拦截：捕获配额耗尽的API错误，自动保存进度后退出，避免数据丢失。

修改后的代码 #

import time
import os
import json
import eyed3
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.errors import HttpError

# 本地配置文件路径
PROGRESS_FILE = "youtube_playlist_progress.json"
SEARCH_CACHE_FILE = "youtube_search_cache.json"

# OAuth 2.0 授权流程
secrets_file_path = "client_secrets3.json"
if not os.path.exists(secrets_file_path):
    secrets_file_path = input("Enter the path to client_secrets.json: ")
flow = InstalledAppFlow.from_client_secrets_file(secrets_file_path, scopes=["https://www.googleapis.com/auth/youtube"])
credentials = flow.run_local_server(port=0)

# 构建YouTube服务实例
youtube = build("youtube", "v3", credentials=credentials)

def get_song_info(file_path):
    try:
        audiofile = eyed3.load(file_path)
        if audiofile is None:
            raise ValueError("Failed to load audio file")
        title = audiofile.tag.title.strip() if audiofile.tag.title else None
        artist = audiofile.tag.artist.strip() if audiofile.tag.artist else None
        if not title or not artist:
            raise ValueError("Missing title or artist tag")
        return {"title": title, "artist": artist}
    except Exception as e:
        print(f"Error processing file {file_path}: {e}")
        return None

def process_folder(folder_path):
    songs = []
    not_identified_songs = []
    for filename in os.listdir(folder_path):
        if filename.endswith(".mp3"):
            file_path = os.path.join(folder_path, filename)
            song_info = get_song_info(file_path)
            if song_info:
                songs.append(song_info)
            else:
                not_identified_songs.append(file_path)
    return songs, not_identified_songs

def load_progress():
    if os.path.exists(PROGRESS_FILE):
        with open(PROGRESS_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {"playlist_id": None, "processed_songs": [], "remaining_songs": [], "failed_songs": []}

def save_progress(progress):
    with open(PROGRESS_FILE, "w", encoding="utf-8") as f:
        json.dump(progress, f, indent=2, ensure_ascii=False)

def load_search_cache():
    if os.path.exists(SEARCH_CACHE_FILE):
        with open(SEARCH_CACHE_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

def save_search_cache(cache):
    with open(SEARCH_CACHE_FILE, "w", encoding="utf-8") as f:
        json.dump(cache, f, indent=2, ensure_ascii=False)

def get_video_id(song, cache):
    key = f"{song['title']}||{song['artist']}"
    if key in cache:
        return cache[key]
    
    try:
        search_response = youtube.search().list(
            part="snippet",
            q=f"{song['title']} {song['artist']}",
            type="VIDEO",
            maxResults=1,
        ).execute()
        
        if search_response.get("items"):
            video_id = search_response["items"][0]["id"]["videoId"]
            cache[key] = video_id
            save_search_cache(cache)
            return video_id
        else:
            print(f"No video found for {song['title']} by {song['artist']}")
            return None
    except HttpError as e:
        if e.resp.status == 403 and "quotaExceeded" in str(e):
            print("Quota limit reached. Saving progress and exiting.")
            save_progress(progress)
            exit(1)
        else:
            print(f"Search error for {song['title']} by {song['artist']}: {e}")
            return None

# 主执行逻辑
folder_path = r':\My Drive\desktop\Camilo\Todas mis músicas'
progress = load_progress()
search_cache = load_search_cache()

# 首次运行时创建播放列表并加载所有歌曲
if not progress["playlist_id"]:
    print("Creating new playlist...")
    playlist_response = youtube.playlists().insert(
        part="snippet,status",
        body={
            "snippet": {"title": "mp3 files", "description": "yeaah"},
            "status": {"privacyStatus": "PUBLIC"},
        },
    ).execute()
    progress["playlist_id"] = playlist_response["id"]
    
    songs, not_identified_songs = process_folder(folder_path)
    progress["remaining_songs"] = songs
    save_progress(progress)
    
    if not_identified_songs:
        print("\nUnidentified songs:")
        for path in not_identified_songs:
            print(f"- {path}")

playlist_id = progress["playlist_id"]
remaining_songs = progress["remaining_songs"]
processed_songs = progress["processed_songs"]
failed_songs = progress["failed_songs"]

print(f"\nStarting with {len(remaining_songs)} songs remaining to process.")

for idx, song in enumerate(remaining_songs.copy()):
    try:
        video_id = get_video_id(song, search_cache)
        if not video_id:
            failed_songs.append(song)
            remaining_songs.remove(song)
            save_progress(progress)
            continue
        
        # 插入歌曲到播放列表
        youtube.playlistItems().insert(
            part="snippet",
            body={
                "snippet": {
                    "playlistId": playlist_id,
                    "resourceId": {"kind": "youtube#video", "videoId": video_id},
                }
            },
        ).execute()
        
        print(f"Added {song['title']} by {song['artist']} to the playlist.")
        processed_songs.append(song)
        remaining_songs.remove(song)
        
        # 每处理10首保存一次进度
        if (idx + 1) % 10 == 0:
            save_progress(progress)
            print("Progress saved.")
            
    except HttpError as e:
        if e.resp.status == 403 and "quotaExceeded" in str(e):
            print("Quota limit reached. Saving progress and exiting.")
            save_progress(progress)
            exit(1)
        else:
            print(f"Error adding {song['title']} by {song['artist']}: {e}")
            failed_songs.append(song)
            remaining_songs.remove(song)
            save_progress(progress)
    except Exception as e:
        print(f"Unexpected error for {song['title']} by {song['artist']}: {e}")
        failed_songs.append(song)
        remaining_songs.remove(song)
        save_progress(progress)

# 最终保存进度
save_progress(progress)
print("\nAll songs processed.")

if failed_songs:
    print("\nSongs that couldn't be added:")
    for song in failed_songs:
        print(f"- {song['title']} by {song['artist']}")

代码改进说明 #

• 缓存机制：通过youtube_search_cache.json存储已搜索的歌曲-video_id映射，避免重复搜索消耗配额。
• 进度持久化：youtube_playlist_progress.json记录任务状态，重启脚本时自动续跑未完成的部分。
• 配额防护：捕获配额耗尽的API错误，自动保存进度后退出，次日直接运行脚本即可继续。
• 路径兼容：给Windows路径添加r前缀，避免转义字符问题。
• 增量保存：每处理10首歌就保存一次进度，防止意外中断丢失数据。

内容的提问来源于stack exchange，提问作者S. Cow