Python爬取酷我音乐排行榜实战从CSRF Token到MP3下载全流程解析当你想批量获取酷我音乐排行榜的热门歌曲时手动下载显然效率太低。这时候Python爬虫就能大显身手了。但酷我音乐的反爬机制可不是吃素的特别是那个恼人的CSRF Token验证让不少开发者栽了跟头。今天我们就来彻底解决这个问题从获取歌曲列表到最终下载MP3文件手把手带你走通全流程。1. 准备工作与环境搭建在开始编写爬虫之前我们需要确保开发环境已经准备就绪。这里推荐使用Python 3.8版本因为它在异步处理和网络请求方面有更好的性能表现。首先安装必要的依赖库pip install requests beautifulsoup4 fake-useragentrequests库用于发送HTTP请求beautifulsoup4用于解析HTML页面虽然酷我音乐主要使用JSON数据交互但保留这个库以备不时之需fake-useragent则可以帮助我们生成随机的User-Agent。提示建议使用虚拟环境来管理项目依赖避免污染全局Python环境。接下来我们需要分析酷我音乐排行榜页面的结构。打开酷我音乐官网的排行榜页面如http://www.kuwo.cn/rankList按F12打开开发者工具切换到Network选项卡刷新页面观察请求。你会发现几个关键点排行榜数据是通过AJAX请求获取的每个请求都携带了特定的headers服务器会验证CSRF Token2. 破解CSRF Token验证机制CSRFCross-Site Request ForgeryToken是网站用来防止跨站请求伪造的安全机制。酷我音乐在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, headersheaders) # 从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: fkw_token{kw_token} }注意每次会话开始时都需要重新获取CSRF Token因为Token会过期。此外建议使用随机User-Agent来模拟不同浏览器访问。3. 获取排行榜歌曲列表酷我音乐的排行榜数据是通过API接口返回的JSON格式数据。通过分析我们找到了获取排行榜数据的API端点http://www.kuwo.cn/api/www/bang/bang/musicList?bangId93pn1rn30参数说明bangId: 排行榜ID93代表酷我飙升榜pn: 页码rn: 每页记录数以下是获取排行榜歌曲列表并解析出歌曲ID和名称的代码def get_rank_list(bang_id93, page1, page_size30): 获取排行榜歌曲列表 api_url fhttp://www.kuwo.cn/api/www/bang/bang/musicList?bangId{bang_id}pn{page}rn{page_size} try: response session.get(api_url, headersapi_headers) response.raise_for_status() data response.json() if not data.get(success): raise ValueError(fAPI 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?formatmp3rid{rid}responseurltypeconvert_url3其中{rid}需要替换为实际的歌曲ID。这个接口返回的JSON数据中包含url字段就是我们要的MP3地址。以下是获取歌曲播放地址的函数实现def get_music_url(rid): 获取歌曲播放地址 play_url fhttp://www.kuwo.cn/url?formatmp3rid{rid}responseurltypeconvert_url3 try: response session.get(play_url, headersapi_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地址后最后一步就是下载文件了。这里我们需要注意几个问题设置合适的请求头避免被服务器拒绝处理大文件下载时的内存问题添加适当的错误处理和重试机制以下是改进后的MP3下载函数def download_music(url, save_path, max_retry3): 下载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, headersheaders, streamTrue) as response: response.raise_for_status() with open(save_path, wb) as f: for chunk in response.iter_content(chunk_size8192): 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这个函数使用流式下载可以有效地处理大文件而不会占用过多内存。streamTrue参数让requests库以流的方式获取数据iter_content方法则允许我们分块处理数据。6. 完整爬虫实现与优化现在我们把所有功能整合起来创建一个完整的爬虫程序。为了提高效率和稳定性我们还需要添加以下功能多线程下载进度显示错误日志记录断点续传支持以下是完整实现import os import time import threading from queue import Queue from datetime import datetime class KuWoMusicSpider: def __init__(self, save_dirdownloads, max_workers5): 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_okTrue) # 初始化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, headersheaders) 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: fkw_token{self.kw_token} } def get_rank_list(self, bang_id93, page1, page_size30): 获取排行榜歌曲列表 api_url fhttp://www.kuwo.cn/api/www/bang/bang/musicList?bangId{bang_id}pn{page}rn{page_size} try: response self.session.get(api_url, headersself.api_headers) response.raise_for_status() data response.json() if not data.get(success): raise ValueError(fAPI 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 fhttp://www.kuwo.cn/url?formatmp3rid{rid}responseurltypeconvert_url3 try: response self.session.get(play_url, headersself.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_retry3): 下载文件实现 headers { User-Agent: self.ua.random, Referer: http://www.kuwo.cn/ } for attempt in range(max_retry): try: with self.session.get(url, headersheaders, streamTrue) as response: response.raise_for_status() with open(save_path, wb) as f: for chunk in response.iter_content(chunk_size8192): 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_id93, page1, page_size30): 下载排行榜歌曲 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(targetself.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_dirkuwo_music, max_workers5) spider.download_rank_list(bang_id93, page_size50)这个完整的爬虫类包含了我们之前讨论的所有功能并添加了多线程支持可以显著提高下载速度。使用时只需要创建一个KuWoMusicSpider实例然后调用download_rank_list方法即可。7. 常见问题与解决方案在实际使用中你可能会遇到以下问题问题1CSRF Token失效现象请求返回CSRF Token Not Found或CSRF Token验证失败解决方案重新初始化session和CSRF Token确保headers中的csrf字段和Cookie中的kw_token一致检查请求间隔避免过于频繁问题2IP被封禁现象请求返回403错误或连接超时解决方案使用代理IP轮换降低请求频率设置随机延迟问题3获取的MP3地址无效现象下载时返回404错误解决方案立即使用获取的MP3地址不要缓存实现自动重试机制检查地址是否包含特殊字符需要编码问题4下载速度慢现象单线程下载耗时过长解决方案使用多线程/多进程增加超时时间使用更快的网络连接以下是一些优化建议的代码实现# 代理设置示例 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_retries3, delay1): 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 decorator8. 高级技巧与扩展思路对于想要进一步优化爬虫的开发者这里提供几个高级技巧使用异步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())实现断点续传对于大文件下载可以实现断点续传功能def download_with_resume(url, save_path): headers {} if os.path.exists(save_path): headers[Range] fbytes{os.path.getsize(save_path)}- with requests.get(url, headersheaders, streamTrue) as r: mode ab if Range in headers else wb with open(save_path, mode) as f: for chunk in r.iter_content(chunk_size8192): if chunk: f.write(chunk)使用Headless浏览器处理复杂情况对于需要执行JavaScript的页面可以使用Selenium等工具from selenium import webdriver options webdriver.ChromeOptions() options.add_argument(--headless) driver webdriver.Chrome(optionsoptions) driver.get(http://www.kuwo.cn/rankList) cookies driver.get_cookies() driver.quit()分布式爬虫架构对于大规模爬取需求可以考虑使用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数据存储优化除了保存为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 智能延迟 自动重试。这种组合能在保证稳定性的同时最大化爬取效率。另外记得定期检查爬虫是否正常工作因为网站的反爬策略可能会更新。
Python爬取酷我音乐排行榜实战：从CSRF Token到MP3下载全流程解析
Python爬取酷我音乐排行榜实战从CSRF Token到MP3下载全流程解析当你想批量获取酷我音乐排行榜的热门歌曲时手动下载显然效率太低。这时候Python爬虫就能大显身手了。但酷我音乐的反爬机制可不是吃素的特别是那个恼人的CSRF Token验证让不少开发者栽了跟头。今天我们就来彻底解决这个问题从获取歌曲列表到最终下载MP3文件手把手带你走通全流程。1. 准备工作与环境搭建在开始编写爬虫之前我们需要确保开发环境已经准备就绪。这里推荐使用Python 3.8版本因为它在异步处理和网络请求方面有更好的性能表现。首先安装必要的依赖库pip install requests beautifulsoup4 fake-useragentrequests库用于发送HTTP请求beautifulsoup4用于解析HTML页面虽然酷我音乐主要使用JSON数据交互但保留这个库以备不时之需fake-useragent则可以帮助我们生成随机的User-Agent。提示建议使用虚拟环境来管理项目依赖避免污染全局Python环境。接下来我们需要分析酷我音乐排行榜页面的结构。打开酷我音乐官网的排行榜页面如http://www.kuwo.cn/rankList按F12打开开发者工具切换到Network选项卡刷新页面观察请求。你会发现几个关键点排行榜数据是通过AJAX请求获取的每个请求都携带了特定的headers服务器会验证CSRF Token2. 破解CSRF Token验证机制CSRFCross-Site Request ForgeryToken是网站用来防止跨站请求伪造的安全机制。酷我音乐在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, headersheaders) # 从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: fkw_token{kw_token} }注意每次会话开始时都需要重新获取CSRF Token因为Token会过期。此外建议使用随机User-Agent来模拟不同浏览器访问。3. 获取排行榜歌曲列表酷我音乐的排行榜数据是通过API接口返回的JSON格式数据。通过分析我们找到了获取排行榜数据的API端点http://www.kuwo.cn/api/www/bang/bang/musicList?bangId93pn1rn30参数说明bangId: 排行榜ID93代表酷我飙升榜pn: 页码rn: 每页记录数以下是获取排行榜歌曲列表并解析出歌曲ID和名称的代码def get_rank_list(bang_id93, page1, page_size30): 获取排行榜歌曲列表 api_url fhttp://www.kuwo.cn/api/www/bang/bang/musicList?bangId{bang_id}pn{page}rn{page_size} try: response session.get(api_url, headersapi_headers) response.raise_for_status() data response.json() if not data.get(success): raise ValueError(fAPI 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?formatmp3rid{rid}responseurltypeconvert_url3其中{rid}需要替换为实际的歌曲ID。这个接口返回的JSON数据中包含url字段就是我们要的MP3地址。以下是获取歌曲播放地址的函数实现def get_music_url(rid): 获取歌曲播放地址 play_url fhttp://www.kuwo.cn/url?formatmp3rid{rid}responseurltypeconvert_url3 try: response session.get(play_url, headersapi_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地址后最后一步就是下载文件了。这里我们需要注意几个问题设置合适的请求头避免被服务器拒绝处理大文件下载时的内存问题添加适当的错误处理和重试机制以下是改进后的MP3下载函数def download_music(url, save_path, max_retry3): 下载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, headersheaders, streamTrue) as response: response.raise_for_status() with open(save_path, wb) as f: for chunk in response.iter_content(chunk_size8192): 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这个函数使用流式下载可以有效地处理大文件而不会占用过多内存。streamTrue参数让requests库以流的方式获取数据iter_content方法则允许我们分块处理数据。6. 完整爬虫实现与优化现在我们把所有功能整合起来创建一个完整的爬虫程序。为了提高效率和稳定性我们还需要添加以下功能多线程下载进度显示错误日志记录断点续传支持以下是完整实现import os import time import threading from queue import Queue from datetime import datetime class KuWoMusicSpider: def __init__(self, save_dirdownloads, max_workers5): 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_okTrue) # 初始化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, headersheaders) 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: fkw_token{self.kw_token} } def get_rank_list(self, bang_id93, page1, page_size30): 获取排行榜歌曲列表 api_url fhttp://www.kuwo.cn/api/www/bang/bang/musicList?bangId{bang_id}pn{page}rn{page_size} try: response self.session.get(api_url, headersself.api_headers) response.raise_for_status() data response.json() if not data.get(success): raise ValueError(fAPI 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 fhttp://www.kuwo.cn/url?formatmp3rid{rid}responseurltypeconvert_url3 try: response self.session.get(play_url, headersself.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_retry3): 下载文件实现 headers { User-Agent: self.ua.random, Referer: http://www.kuwo.cn/ } for attempt in range(max_retry): try: with self.session.get(url, headersheaders, streamTrue) as response: response.raise_for_status() with open(save_path, wb) as f: for chunk in response.iter_content(chunk_size8192): 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_id93, page1, page_size30): 下载排行榜歌曲 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(targetself.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_dirkuwo_music, max_workers5) spider.download_rank_list(bang_id93, page_size50)这个完整的爬虫类包含了我们之前讨论的所有功能并添加了多线程支持可以显著提高下载速度。使用时只需要创建一个KuWoMusicSpider实例然后调用download_rank_list方法即可。7. 常见问题与解决方案在实际使用中你可能会遇到以下问题问题1CSRF Token失效现象请求返回CSRF Token Not Found或CSRF Token验证失败解决方案重新初始化session和CSRF Token确保headers中的csrf字段和Cookie中的kw_token一致检查请求间隔避免过于频繁问题2IP被封禁现象请求返回403错误或连接超时解决方案使用代理IP轮换降低请求频率设置随机延迟问题3获取的MP3地址无效现象下载时返回404错误解决方案立即使用获取的MP3地址不要缓存实现自动重试机制检查地址是否包含特殊字符需要编码问题4下载速度慢现象单线程下载耗时过长解决方案使用多线程/多进程增加超时时间使用更快的网络连接以下是一些优化建议的代码实现# 代理设置示例 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_retries3, delay1): 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 decorator8. 高级技巧与扩展思路对于想要进一步优化爬虫的开发者这里提供几个高级技巧使用异步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())实现断点续传对于大文件下载可以实现断点续传功能def download_with_resume(url, save_path): headers {} if os.path.exists(save_path): headers[Range] fbytes{os.path.getsize(save_path)}- with requests.get(url, headersheaders, streamTrue) as r: mode ab if Range in headers else wb with open(save_path, mode) as f: for chunk in r.iter_content(chunk_size8192): if chunk: f.write(chunk)使用Headless浏览器处理复杂情况对于需要执行JavaScript的页面可以使用Selenium等工具from selenium import webdriver options webdriver.ChromeOptions() options.add_argument(--headless) driver webdriver.Chrome(optionsoptions) driver.get(http://www.kuwo.cn/rankList) cookies driver.get_cookies() driver.quit()分布式爬虫架构对于大规模爬取需求可以考虑使用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数据存储优化除了保存为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 智能延迟 自动重试。这种组合能在保证稳定性的同时最大化爬取效率。另外记得定期检查爬虫是否正常工作因为网站的反爬策略可能会更新。
