Python爬取酷我音乐排行榜实战：从CSRF Token到MP3下载全流程解析

Python爬取酷我音乐排行榜实战：从CSRF Token到MP3下载全流程解析

当你想批量获取酷我音乐排行榜的热门歌曲时，手动下载显然效率太低。这时候Python爬虫就能大显身手了。但酷我音乐的反爬机制可不是吃素的，特别是那个恼人的CSRF Token验证，让不少开发者栽了跟头。今天我们就来彻底解决这个问题，从获取歌曲列表到最终下载MP3文件，手把手带你走通全流程。

1. 准备工作与环境搭建

在开始编写爬虫之前，我们需要确保开发环境已经准备就绪。这里推荐使用Python 3.8+版本，因为它在异步处理和网络请求方面有更好的性能表现。

首先安装必要的依赖库：

pip install requests beautifulsoup4 fake-useragent

requests库用于发送HTTP请求，beautifulsoup4用于解析HTML页面（虽然酷我音乐主要使用JSON数据交互，但保留这个库以备不时之需），fake-useragent则可以帮助我们生成随机的User-Agent。

提示：建议使用虚拟环境来管理项目依赖，避免污染全局Python环境。

接下来，我们需要分析酷我音乐排行榜页面的结构。打开酷我音乐官网的排行榜页面（如http://www.kuwo.cn/rankList），按F12打开开发者工具，切换到Network选项卡，刷新页面观察请求。

你会发现几个关键点：

• 排行榜数据是通过AJAX请求获取的
• 每个请求都携带了特定的headers
• 服务器会验证CSRF Token

2. 破解CSRF Token验证机制

CSRF（Cross-Site Request Forgery）Token是网站用来防止跨站请求伪造的安全机制。酷我音乐在API请求中强制验证这个Token，如果没有正确的Token，服务器会返回{"success":false,"message":"CSRF Token Not Found!"}。

通过分析网络请求，我们发现：

• CSRF Token存储在Cookie中的kw_token字段
• 需要在请求头中设置csrf字段，其值等于kw_token

以下是获取和设置CSRF Token的Python代码示例：

import requests from fake_useragent import UserAgent # 初始化session session = requests.Session() ua = UserAgent() # 获取初始Cookie和CSRF Token init_url = 'http://www.kuwo.cn/rankList' headers = { 'User-Agent': ua.random } response = session.get(init_url, headers=headers) # 从Cookie中提取kw_token kw_token = session.cookies.get('kw_token') if not kw_token: raise ValueError("Failed to get CSRF token from cookies") # 设置后续请求的headers api_headers = { 'User-Agent': ua.random, 'Referer': 'http://www.kuwo.cn/rankList', 'csrf': kw_token, 'Cookie': f'kw_token={kw_token}' }

注意：每次会话开始时都需要重新获取CSRF Token，因为Token会过期。此外，建议使用随机User-Agent来模拟不同浏览器访问。

3. 获取排行榜歌曲列表

酷我音乐的排行榜数据是通过API接口返回的JSON格式数据。通过分析，我们找到了获取排行榜数据的API端点：

http://www.kuwo.cn/api/www/bang/bang/musicList?bangId=93&pn=1&rn=30

参数说明：

• bangId: 排行榜ID（93代表酷我飙升榜）
• pn: 页码
• rn: 每页记录数

以下是获取排行榜歌曲列表并解析出歌曲ID和名称的代码：

def get_rank_list(bang_id=93, page=1, page_size=30): """获取排行榜歌曲列表""" api_url = f'http://www.kuwo.cn/api/www/bang/bang/musicList?bangId={bang_id}&pn={page}&rn={page_size}' try: response = session.get(api_url, headers=api_headers) response.raise_for_status() data = response.json() if not data.get('success'): raise ValueError(f"API request failed: {data.get('message')}") music_list = [] for item in data['data']['musicList']: music_list.append({ 'rid': item['musicrid'].split('_')[1], # 提取纯数字ID 'name': item['name'], 'artist': item['artist'] }) return music_list except Exception as e: print(f"获取排行榜失败: {str(e)}") return []

这个函数会返回一个包含歌曲信息的字典列表，每个字典包含歌曲ID(rid)、歌曲名称(name)和艺术家(artist)信息。

4. 获取歌曲播放地址

获取到歌曲ID后，下一步是获取实际的MP3播放地址。通过分析，我们发现酷我音乐的MP3地址是通过另一个API生成的：

http://www.kuwo.cn/url?format=mp3&rid={rid}&response=url&type=convert_url3

其中{rid}需要替换为实际的歌曲ID。这个接口返回的JSON数据中包含url字段，就是我们要的MP3地址。

以下是获取歌曲播放地址的函数实现：

def get_music_url(rid): """获取歌曲播放地址""" play_url = f'http://www.kuwo.cn/url?format=mp3&rid={rid}&response=url&type=convert_url3' try: response = session.get(play_url, headers=api_headers) response.raise_for_status() data = response.json() if 'url' not in data: raise ValueError("Invalid response format") return data['url'] except Exception as e: print(f"获取歌曲播放地址失败(RID: {rid}): {str(e)}") return None

重要提示：获取到的MP3地址通常有有效期限制，建议在获取后立即使用，不要长时间缓存。

5. 下载MP3文件

有了MP3地址后，最后一步就是下载文件了。这里我们需要注意几个问题：

1. 设置合适的请求头，避免被服务器拒绝
2. 处理大文件下载时的内存问题
3. 添加适当的错误处理和重试机制

以下是改进后的MP3下载函数：

def download_music(url, save_path, max_retry=3): """下载MP3文件""" if not url: return False headers = { 'User-Agent': ua.random, 'Referer': 'http://www.kuwo.cn/' } for attempt in range(max_retry): try: with session.get(url, headers=headers, stream=True) as response: response.raise_for_status() with open(save_path, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) return True except Exception as e: print(f"下载失败(尝试 {attempt + 1}/{max_retry}): {str(e)}") if attempt == max_retry - 1: return False

这个函数使用流式下载，可以有效地处理大文件而不会占用过多内存。stream=True参数让requests库以流的方式获取数据，iter_content方法则允许我们分块处理数据。

6. 完整爬虫实现与优化

现在我们把所有功能整合起来，创建一个完整的爬虫程序。为了提高效率和稳定性，我们还需要添加以下功能：

• 多线程下载
• 进度显示
• 错误日志记录
• 断点续传支持

以下是完整实现：

import os import time import threading from queue import Queue from datetime import datetime class KuWoMusicSpider: def __init__(self, save_dir='downloads', max_workers=5): self.session = requests.Session() self.ua = UserAgent() self.save_dir = save_dir self.max_workers = max_workers self.task_queue = Queue() self.lock = threading.Lock() # 创建保存目录 os.makedirs(self.save_dir, exist_ok=True) # 初始化CSRF Token self._init_csrf_token() def _init_csrf_token(self): """初始化CSRF Token""" init_url = 'http://www.kuwo.cn/rankList' headers = {'User-Agent': self.ua.random} response = self.session.get(init_url, headers=headers) self.kw_token = self.session.cookies.get('kw_token') if not self.kw_token: raise ValueError("Failed to initialize CSRF token") self.api_headers = { 'User-Agent': self.ua.random, 'Referer': 'http://www.kuwo.cn/rankList', 'csrf': self.kw_token, 'Cookie': f'kw_token={self.kw_token}' } def get_rank_list(self, bang_id=93, page=1, page_size=30): """获取排行榜歌曲列表""" api_url = f'http://www.kuwo.cn/api/www/bang/bang/musicList?bangId={bang_id}&pn={page}&rn={page_size}' try: response = self.session.get(api_url, headers=self.api_headers) response.raise_for_status() data = response.json() if not data.get('success'): raise ValueError(f"API request failed: {data.get('message')}") return [ { 'rid': item['musicrid'].split('_')[1], 'name': item['name'], 'artist': item['artist'] } for item in data['data']['musicList'] ] except Exception as e: print(f"获取排行榜失败: {str(e)}") return [] def get_music_url(self, rid): """获取歌曲播放地址""" play_url = f'http://www.kuwo.cn/url?format=mp3&rid={rid}&response=url&type=convert_url3' try: response = self.session.get(play_url, headers=self.api_headers) response.raise_for_status() data = response.json() if 'url' not in data: raise ValueError("Invalid response format") return data['url'] except Exception as e: print(f"获取歌曲播放地址失败(RID: {rid}): {str(e)}") return None def download_worker(self): """下载工作线程""" while True: task = self.task_queue.get() if task is None: break rid, name, artist = task safe_name = ''.join(c for c in f"{artist} - {name}" if c.isalnum() or c in ' -_') save_path = os.path.join(self.save_dir, f"{safe_name}.mp3") if os.path.exists(save_path): with self.lock: print(f"文件已存在，跳过: {save_path}") self.task_queue.task_done() continue mp3_url = self.get_music_url(rid) if not mp3_url: self.task_queue.task_done() continue success = self._download_file(mp3_url, save_path) with self.lock: if success: print(f"下载成功: {save_path}") else: print(f"下载失败: {artist} - {name}") self.task_queue.task_done() def _download_file(self, url, save_path, max_retry=3): """下载文件实现""" headers = { 'User-Agent': self.ua.random, 'Referer': 'http://www.kuwo.cn/' } for attempt in range(max_retry): try: with self.session.get(url, headers=headers, stream=True) as response: response.raise_for_status() with open(save_path, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) return True except Exception as e: print(f"下载失败(尝试 {attempt + 1}/{max_retry}): {str(e)}") if attempt == max_retry - 1: return False time.sleep(1) def download_rank_list(self, bang_id=93, page=1, page_size=30): """下载排行榜歌曲""" music_list = self.get_rank_list(bang_id, page, page_size) if not music_list: print("没有获取到歌曲列表") return # 创建并启动工作线程 threads = [] for _ in range(self.max_workers): t = threading.Thread(target=self.download_worker) t.start() threads.append(t) # 添加任务到队列 for music in music_list: self.task_queue.put((music['rid'], music['name'], music['artist'])) # 等待所有任务完成 self.task_queue.join() # 停止工作线程 for _ in range(self.max_workers): self.task_queue.put(None) for t in threads: t.join() print("所有下载任务完成") if __name__ == '__main__': spider = KuWoMusicSpider(save_dir='kuwo_music', max_workers=5) spider.download_rank_list(bang_id=93, page_size=50)

这个完整的爬虫类包含了我们之前讨论的所有功能，并添加了多线程支持，可以显著提高下载速度。使用时只需要创建一个KuWoMusicSpider实例，然后调用download_rank_list方法即可。

7. 常见问题与解决方案

在实际使用中，你可能会遇到以下问题：

问题1：CSRF Token失效

• 现象：请求返回CSRF Token Not Found或CSRF Token验证失败
• 解决方案：
  1. 重新初始化session和CSRF Token
  2. 确保headers中的csrf字段和Cookie中的kw_token一致
  3. 检查请求间隔，避免过于频繁

问题2：IP被封禁

• 现象：请求返回403错误或连接超时
• 解决方案：
  1. 使用代理IP轮换
  2. 降低请求频率
  3. 设置随机延迟

问题3：获取的MP3地址无效

• 现象：下载时返回404错误
• 解决方案：
  1. 立即使用获取的MP3地址，不要缓存
  2. 实现自动重试机制
  3. 检查地址是否包含特殊字符需要编码

问题4：下载速度慢

• 现象：单线程下载耗时过长
• 解决方案：
  1. 使用多线程/多进程
  2. 增加超时时间
  3. 使用更快的网络连接

以下是一些优化建议的代码实现：

# 代理设置示例 proxies = { 'http': 'http://your.proxy.address:port', 'https': 'http://your.proxy.address:port' } # 随机延迟 import random time.sleep(random.uniform(0.5, 2.0)) # 请求重试装饰器 from functools import wraps def retry(max_retries=3, delay=1): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if attempt == max_retries - 1: raise time.sleep(delay * (attempt + 1)) return wrapper return decorator

8. 高级技巧与扩展思路

对于想要进一步优化爬虫的开发者，这里提供几个高级技巧：

1. 使用异步IO提高效率

   使用aiohttp代替requests可以实现真正的异步请求，大幅提高爬虫效率：

   import aiohttp import asyncio async def fetch(session, url): async with session.get(url) as response: return await response.json() async def main(): async with aiohttp.ClientSession() as session: tasks = [fetch(session, url) for url in url_list] return await asyncio.gather(*tasks) results = asyncio.run(main())

2. 实现断点续传

   对于大文件下载，可以实现断点续传功能：

   def download_with_resume(url, save_path): headers = {} if os.path.exists(save_path): headers['Range'] = f'bytes={os.path.getsize(save_path)}-' with requests.get(url, headers=headers, stream=True) as r: mode = 'ab' if 'Range' in headers else 'wb' with open(save_path, mode) as f: for chunk in r.iter_content(chunk_size=8192): if chunk: f.write(chunk)

3. 使用Headless浏览器处理复杂情况

   对于需要执行JavaScript的页面，可以使用Selenium等工具：

   from selenium import webdriver options = webdriver.ChromeOptions() options.add_argument('--headless') driver = webdriver.Chrome(options=options) driver.get('http://www.kuwo.cn/rankList') cookies = driver.get_cookies() driver.quit()

4. 分布式爬虫架构

   对于大规模爬取需求，可以考虑使用Scrapy框架结合Redis实现分布式爬虫：

   # scrapy示例 import scrapy from scrapy_redis.spiders import RedisSpider class KuWoSpider(RedisSpider): name = 'kuwo' redis_key = 'kuwo:start_urls' def parse(self, response): # 解析逻辑 pass

5. 数据存储优化

   除了保存为MP3文件，还可以考虑将歌曲信息存入数据库：

   import sqlite3 def init_db(): conn = sqlite3.connect('music.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS songs (id INTEGER PRIMARY KEY, rid TEXT, name TEXT, artist TEXT, path TEXT)''') conn.commit() conn.close() def save_to_db(song_info): conn = sqlite3.connect('music.db') c = conn.cursor() c.execute("INSERT INTO songs VALUES (NULL, ?, ?, ?, ?)", (song_info['rid'], song_info['name'], song_info['artist'], song_info['path'])) conn.commit() conn.close()

在实际项目中，我发现最有效的优化组合是：异步IO + 智能延迟 + 自动重试。这种组合能在保证稳定性的同时最大化爬取效率。另外，记得定期检查爬虫是否正常工作，因为网站的反爬策略可能会更新。