想请教下想爬取ytb数据，但是显示访问超时

littleseal

想请教下，我用 ytbv3 官方 API 爬取数据时出现访问超时，已开启quanju代理，且环境测试脚本运行正常。请问有朋友能指导下如何解决吗？


爬取脚本

1. from googleapiclient.discovery import build
   
2. import pandas as pd
   
3. from datetime import datetime
   
4. import os
   
5. 
   
6. # -------------------------
   
7. # 1. 配置参数
   
8. # -------------------------
   
9. API_KEY = aaaaaaaaaaaa  # 替换成你的 YouTube Data API Key
   
10. PLAYLIST_ID = "aaaaaaaaaaaa"  # 替换成目标播放列表ID
    
11. OUTPUT_FILE = "youtube_playlist_stats.xlsx"
    
12. 
    
13. # -------------------------
    
14. # 2. 初始化 YouTube API
    
15. # -------------------------
    
16. youtube = build("youtube", "v3", developerKey=API_KEY)
    
17. 
    
18. # -------------------------
    
19. # 3. 获取播放列表中的视频ID
    
20. # -------------------------
    
21. video_ids = []
    
22. next_page_token = None
    
23. 
    
24. while True:
    
25.     playlist_request = youtube.playlistItems().list(
    
26.         part="contentDetails",
    
27.         playlistId=PLAYLIST_ID,
    
28.         maxResults=50,
    
29.         pageToken=next_page_token
    
30.     )
    
31.     playlist_response = playlist_request.execute()
    
32. 
    
33.     for item in playlist_response["items"]:
    
34.         video_ids.append(item["contentDetails"]["videoId"])
    
35. 
    
36.     next_page_token = playlist_response.get("nextPageToken")
    
37.     if not next_page_token:
    
38.         break
    
39. 
    
40. print(f"📋 播放列表共找到 {len(video_ids)} 个视频")
    
41. 
    
42. # -------------------------
    
43. # 4. 获取视频统计数据
    
44. # -------------------------
    
45. all_data = []
    
46. crawl_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
47. 
    
48. for i in range(0, len(video_ids), 50):
    
49.     batch = video_ids[i:i+50]
    
50.     video_request = youtube.videos().list(
    
51.         part="statistics,snippet",
    
52.         id=",".join(batch)
    
53.     )
    
54.     video_response = video_request.execute()
    
55. 
    
56.     for item in video_response["items"]:
    
57.         stats = item["statistics"]
    
58.         snippet = item["snippet"]
    
59. 
    
60.         all_data.append({
    
61.             "抓取时间": crawl_time,
    
62.             "视频标题": snippet["title"],
    
63.             "视频ID": item["id"],
    
64.             "播放量": int(stats.get("viewCount", 0)),
    
65.             "点赞数": int(stats.get("likeCount", 0)),
    
66.             "评论数": int(stats.get("commentCount", 0))
    
67.         })
    
68. 
    
69. # -------------------------
    
70. # 5. 保存到 Excel
    
71. # -------------------------
    
72. df_new = pd.DataFrame(all_data)
    
73. 
    
74. if os.path.exists(OUTPUT_FILE):
    
75.     df_old = pd.read_excel(OUTPUT_FILE)
    
76.     df_all = pd.concat([df_old, df_new], ignore_index=True)
    
77. else:
    
78.     df_all = df_new
    
79. 
    
80. df_all.to_excel(OUTPUT_FILE, index=False)
    
81. 
    
82. print(f"✅ 数据已保存到 {OUTPUT_FILE}")
    

复制代码

环境测试脚本a

1. import requests
   
2. 
   
3. try:
   
4.     # 访问 YouTube Data API 测试 URL
   
5.     r = requests.get("通过自己的apikey获取某个ytb视频的链接，目前还没有权限发链接", timeout=10)
   
6.     print(r.status_code, r.text[:200])
   
7. except Exception as e:
   
8.     print("访问失败：", e)

复制代码

FishC

您好！看到您在使用 YouTube Data API v3 时遇到访问超时问题，虽然已开启全局代理且环境测试正常，但爬取脚本仍超时。我来帮您分析一下可能的原因和解决方案。

首先，您的爬取脚本中有几处代码错误需要修复，这些错误可能导致请求无法正常发送。以下是修正后的代码关键部分：

1. 
   
2. # 初始化 video_ids 为空列表
   
3. video_ids = []
   
4. 
   
5. # 获取播放列表中的视频ID
   
6. next_page_token = None
   
7. while True:
   
8.     playlist_request = youtube.playlistItems().list(
   
9.         part="contentDetails",
   
10.         playlistId=PLAYLIST_ID,
    
11.         maxResults=50,
    
12.         pageToken=next_page_token
    
13.     )
    
14.     playlist_response = playlist_request.execute()
    
15.    
    
16.     # 正确提取 items
    
17.     items = playlist_response.get('items', [])
    
18.     for item in items:
    
19.         video_id = item['contentDetails']['videoId']
    
20.         video_ids.append(video_id)  # 添加视频ID到列表
    
21.    
    
22.     next_page_token = playlist_response.get('nextPageToken')
    
23.     if not next_page_token:
    
24.         break
    
25. 
    
26. print(f"找到 {len(video_ids)} 个视频")
    
27. 
    
28. # 获取视频统计数据
    
29. all_data = []
    
30. crawl_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
31. 
    
32. # 分批处理视频ID，每批50个
    
33. for i in range(0, len(video_ids), 50):
    
34.     batch = video_ids[i:i+50]  # 正确切片
    
35.     video_request = youtube.videos().list(
    
36.         part="statistics,snippet",
    
37.         id=",".join(batch)
    
38.     )
    
39.     video_response = video_request.execute()
    
40.     items = video_response.get('items', [])
    
41.    
    
42.     for item in items:
    
43.         stats = item.get('statistics', {})
    
44.         snippet = item.get('snippet', {})
    
45.         all_data.append({
    
46.             "抓取时间": crawl_time,
    
47.             "视频标题": snippet.get('title', 'N/A'),
    
48.             "视频ID": item.get('id', 'N/A'),
    
49.             "播放量": int(stats.get("viewCount", 0)),
    
50.             "点赞数": int(stats.get("likeCount", 0)),
    
51.             "评论数": int(stats.get("commentCount", 0))
    
52.         })
    
53. 
    
54. # 保存到Excel（修正concat部分）
    
55. if os.path.exists(OUTPUT_FILE):
    
56.     df_old = pd.read_excel(OUTPUT_FILE)
    
57.     df_all = pd.concat([df_old, df_new], ignore_index=True)  # 添加缺失参数
    
58. else:
    
59.     df_all = df_new
    
60. df_all.to_excel(OUTPUT_FILE, index=False)
    

复制代码

### 解决超时问题的建议：
1. 代理设置问题：虽然您开启了全局代理，但 googleapiclient 可能不会自动使用系统代理。建议在代码中显式设置代理。例如，使用 requests 库的会话对象来配置代理，然后传递给 build 函数。示例：
  

1. 
   
2.    import requests
   
3.    from googleapiclient.discovery import build
   
4.    from googleapiclient.http import HttpRequest
   
5. 
   
6.    # 创建带代理的 HTTP 对象
   
7.    proxies = {
   
8.        "http": "[url=http://your-proxy-address:port",]http://your-proxy-address:port",[/url]  # 替换为您的代理地址和端口
   
9.        "https": "[url=http://your-proxy-address:port"]http://your-proxy-address:port"[/url]
   
10.    }
    
11.    session = requests.Session()
    
12.    session.proxies = proxies
    
13.    http = HttpRequest(session=session)
    
14. 
    
15.    # 初始化 YouTube API 时传入 http 对象
    
16.    youtube = build("youtube", "v3", developerKey=API_KEY, http=http)
    
17.    

复制代码

   注意：将 your-proxy-address:port 替换为实际代理信息。

2. 添加超时和重试机制：API 请求可能因网络波动超时，建议在请求中添加超时参数和重试逻辑。例如：
  

1. 
   
2.    from tenacity import retry, stop_after_attempt, wait_exponential
   
3. 
   
4.    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
   
5.    def make_request(request):
   
6.        return request.execute()
   
7.    
   
8.    # 在调用 request.execute() 的地方使用 make_request(playlist_request)
   
9.    

复制代码

   首先安装 tenacity 库： pip install tenacity。

3. 检查 API 配额和限制：确保您的 API key 没有超过每日配额或速率限制。您可以在 Google Cloud Console 中查看使用情况。

4. 网络测试：运行环境测试脚本时，确认使用的是相同代理设置。如果测试脚本正常，但主脚本超时，可能是代码逻辑或代理未生效导致。

如果问题依旧，请检查代理稳定性或尝试更换代理节点。希望这能帮到您！如果还有其他问题，欢迎继续问。

以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。