### 批量音乐爬取的最佳实践 为了高效且稳定地完成批量音乐爬取的任务，建议遵循以下最佳实践： - **设置合理的请求间隔**：频繁的HTTP请求可能会触发目标网站的安全机制。因此，在发送连续请求之间加入适当的时间延迟是非常必要的[^1]。 - **处理异常情况**：网络环境复杂多变，可能出现连接超时、页面加载失败等问题。编写健壮的错误捕捉逻辑能够提高程序稳定性并减少因临时性问题导致的整体流程中断风险。 - **尊重robots协议**：在开始抓取前查阅目标站点根目录下的`robots.txt`文件，了解哪些路径允许访问，确保行为合法合规。 - **使用Session对象保持会话状态**：对于某些需要登录验证或其他形式的身份认证才能获取的数据源来说，利用`requests.Session()`可以帮助维持同一浏览器上下文中的一系列交互操作。 下面给出一段基于上述原则编写的Python代码片段用于示范如何实现批量音乐下载功能： ```python import os import time from concurrent.futures import ThreadPoolExecutor, as_completed import requests def fetch_music_info(music_id): """模拟从API接口获取单首歌曲的信息""" api_url = f"https://example.com/api/music/{music_id}" try: response = session.get(api_url) if response.status_code != 200: raise Exception(f"Failed to get music info {response.status_code}") data = response.json() return { "id": music_id, "title": data['title'], "download_link": data['link'] } except Exception as e: print(e) return None def download_single_music(info_dict): """根据给定链接保存一首歌至本地磁盘""" title = info_dict["title"] url = info_dict["download_link"] file_path = os.path.join(output_dir, f"{title}.mp3") try: with open(file_path, 'wb') as fp: resp = session.get(url) fp.write(resp.content) print(f"'{title}' 下载成功.") return True except IOError as io_err: print(io_err) return False except Exception as ex: print(ex) return False if __name__ == "__main__": output_dir = "./musics" max_workers = 5 # 创建输出文件夹（如果不存在） if not os.path.exists(output_dir): os.makedirs(output_dir) # 初始化session实例 session = requests.Session() ids_to_fetch = range(1, 101) # 假设有编号为1~100的歌曲待下载 futures = [] results = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: for id_ in ids_to_fetch: future = executor.submit(fetch_music_info, id_) futures.append(future) for fut in as_completed(futures): result = fut.result() if result is not None: results.append(result) successful_downloads = set() with ThreadPoolExecutor(max_workers=max_workers) as executor: tasks = (executor.submit(download_single_music, item) for item in results) for task in as_completed(tasks): success = task.result() if success: song_title = next((item["title"] for item in results if download_single_music(item)), "") successful_downloads.add(song_title) print(f"\n总共尝试下载了 {len(results)} 首歌曲,\n其中成功的有 {len(successful_downloads)} 首.") ``` 这段脚本通过线程池并发执行多个任务来加速整个过程，并妥善管理资源分配与错误恢复策略。