Python脚本抓取百度FM歌曲的完整实现指南

Python脚本抓取百度FM歌曲的完整实现指南

在音乐内容数字化浪潮中,如何通过编程手段高效获取百度FM平台上的歌曲资源,成为开发者关注的焦点。本文将从技术实现角度出发,系统讲解如何使用Python脚本完成数据抓取,同时强调合规性与技术优化。

一、技术实现前的合规性确认

在启动开发前,必须明确以下法律边界:

  1. 平台协议审查:仔细阅读百度FM的《用户服务协议》和《隐私政策》,确认是否存在禁止自动化访问的条款。部分平台会明确限制爬虫行为,违反协议可能导致IP封禁或法律追责。
  2. 版权合规性:确保抓取行为仅用于个人学习或研究目的,避免商业用途。根据《信息网络传播权保护条例》,未经授权的批量下载可能构成侵权。
  3. Robots协议检查:通过访问https://fm.baidu.com/robots.txt,确认哪些路径允许/禁止爬取。例如,若发现Disallow: /api/song/,则需规避相关接口。

二、核心抓取流程实现

1. 环境准备与依赖安装

  1. pip install requests beautifulsoup4 fake_useragent

关键库说明:

  • requests:处理HTTP请求
  • BeautifulSoup:解析HTML/XML文档
  • fake_useragent:生成随机User-Agent模拟浏览器访问

2. 请求头伪装技术

为避免被反爬机制识别,需构造完整的请求头:

  1. from fake_useragent import UserAgent
  2. headers = {
  3. 'User-Agent': UserAgent().random,
  4. 'Referer': 'https://fm.baidu.com/',
  5. 'Accept-Language': 'zh-CN,zh;q=0.9'
  6. }

3. 歌曲列表获取实现

通过分析百度FM的API接口,发现其歌曲数据通常通过以下方式返回:

  1. import requests
  2. def get_song_list(channel_id, page=1):
  3. url = f"https://fm.baidu.com/data/music/songList/{channel_id}?page={page}&size=20"
  4. try:
  5. response = requests.get(url, headers=headers, timeout=10)
  6. if response.status_code == 200:
  7. return response.json() # 假设返回JSON格式数据
  8. else:
  9. print(f"请求失败,状态码:{response.status_code}")
  10. return None
  11. except Exception as e:
  12. print(f"请求异常:{str(e)}")
  13. return None

4. 歌曲详情解析

获取到歌曲列表后,需解析关键字段:

  1. def parse_song_info(song_data):
  2. songs = []
  3. for item in song_data.get('songList', []):
  4. song_info = {
  5. 'song_id': item.get('songId'),
  6. 'title': item.get('title'),
  7. 'artist': item.get('artist'),
  8. 'duration': item.get('duration'),
  9. 'play_url': item.get('playUrl') # 实际可能需进一步解析
  10. }
  11. songs.append(song_info)
  12. return songs

三、高级技术优化

1. 反反爬策略应对

  • IP轮换:使用代理池切换IP地址
    ```python
    import random

proxies = [
{‘http’: ‘http://10.10.1.10:3128'},
{‘http’: ‘http://10.10.1.11:8080'}
]

response = requests.get(url, headers=headers, proxies=random.choice(proxies))

  1. - **请求频率控制**:通过`time.sleep()`实现随机延迟
  2. ```python
  3. import time
  4. import random
  5. def safe_request(url):
  6. delay = random.uniform(1, 3)
  7. time.sleep(delay)
  8. return requests.get(url, headers=headers)

2. 数据存储方案

  • CSV存储:适合结构化数据
    ```python
    import csv

def save_to_csv(songs, filename=’songs.csv’):
with open(filename, ‘w’, newline=’’, encoding=’utf-8’) as f:
writer = csv.DictWriter(f, fieldnames=[‘song_id’, ‘title’, ‘artist’])
writer.writeheader()
writer.writerows(songs)

  1. - **数据库存储**:适合大规模数据
  2. ```python
  3. import sqlite3
  4. def init_db():
  5. conn = sqlite3.connect('fm_songs.db')
  6. cursor = conn.cursor()
  7. cursor.execute('''
  8. CREATE TABLE IF NOT EXISTS songs (
  9. id INTEGER PRIMARY KEY,
  10. song_id TEXT UNIQUE,
  11. title TEXT,
  12. artist TEXT
  13. )
  14. ''')
  15. conn.commit()
  16. conn.close()
  17. def save_to_db(songs):
  18. conn = sqlite3.connect('fm_songs.db')
  19. cursor = conn.cursor()
  20. for song in songs:
  21. try:
  22. cursor.execute('''
  23. INSERT INTO songs (song_id, title, artist)
  24. VALUES (?, ?, ?)
  25. ''', (song['song_id'], song['title'], song['artist']))
  26. except sqlite3.IntegrityError:
  27. print(f"歌曲{song['song_id']}已存在,跳过")
  28. conn.commit()
  29. conn.close()

四、完整实现示例

  1. import requests
  2. from fake_useragent import UserAgent
  3. import time
  4. import random
  5. import csv
  6. class BaiduFMCrawler:
  7. def __init__(self):
  8. self.headers = {
  9. 'User-Agent': UserAgent().random,
  10. 'Referer': 'https://fm.baidu.com/'
  11. }
  12. self.base_url = "https://fm.baidu.com/data/music/songList"
  13. def get_song_list(self, channel_id, page=1):
  14. url = f"{self.base_url}/{channel_id}?page={page}&size=20"
  15. try:
  16. response = requests.get(url, headers=self.headers, timeout=10)
  17. if response.status_code == 200:
  18. return response.json()
  19. else:
  20. print(f"请求失败,状态码:{response.status_code}")
  21. return None
  22. except Exception as e:
  23. print(f"请求异常:{str(e)}")
  24. return None
  25. def parse_songs(self, data):
  26. songs = []
  27. for item in data.get('songList', []):
  28. songs.append({
  29. 'song_id': item.get('songId'),
  30. 'title': item.get('title'),
  31. 'artist': item.get('artist')
  32. })
  33. return songs
  34. def save_songs(self, songs, filename='baidu_fm_songs.csv'):
  35. with open(filename, 'w', newline='', encoding='utf-8') as f:
  36. writer = csv.DictWriter(f, fieldnames=['song_id', 'title', 'artist'])
  37. writer.writeheader()
  38. writer.writerows(songs)
  39. def crawl(self, channel_id, max_pages=5):
  40. all_songs = []
  41. for page in range(1, max_pages + 1):
  42. print(f"正在抓取第{page}页...")
  43. data = self.get_song_list(channel_id, page)
  44. if not data:
  45. break
  46. songs = self.parse_songs(data)
  47. all_songs.extend(songs)
  48. # 随机延迟避免被封
  49. time.sleep(random.uniform(1, 3))
  50. if len(songs) < 20: # 假设每页20首,少于说明已到最后
  51. break
  52. self.save_songs(all_songs)
  53. print(f"共抓取{len(all_songs)}首歌曲")
  54. # 使用示例
  55. if __name__ == "__main__":
  56. crawler = BaiduFMCrawler()
  57. crawler.crawl(channel_id="public_tujing_normal", max_pages=3)

五、关键注意事项

  1. 频率控制:建议QPS不超过1次/秒,避免触发反爬机制
  2. 数据合法性:仅保存歌曲元数据(标题、艺术家等),不存储音频文件本身
  3. 异常处理:完善网络异常、数据解析异常的处理逻辑
  4. 版本兼容性:定期检查API接口是否变更,及时调整解析逻辑

六、技术延伸方向

  1. 结合百度智能云:可将抓取的数据存储至对象存储BOS,或使用函数计算FC实现无服务器架构
  2. 数据分析应用:对抓取的歌曲数据进行情感分析、流行度预测等二次开发
  3. 多平台整合:扩展支持其他音乐平台的抓取,实现跨平台音乐库管理

通过本文介绍的技术方案,开发者可以在合规框架内实现百度FM歌曲数据的高效抓取。实际开发中需持续关注平台规则变化,及时调整技术策略,确保项目的长期稳定性。