简介:这是一份面向Python爬虫初学者与数据分析入门者的实战项目包,围绕某音乐平台华语歌手信息采集展开,覆盖从网络请求、HTML解析到本地数据库存储、歌手热门歌曲与评论数抓取,再到可视化呈现的完整链路。包内共18个文件,以6个py脚本为核心,配合5个xml配置、1个html页面、1个ipynb笔记本及pyc缓存等辅助文件,压缩包约35KB,结构紧凑便于按模块阅读。项目涉及requests、BeautifulSoup或lxml解析、sqlite3数据库操作,以及通过歌手ID获取热门50首歌曲并按评论数筛选最受欢迎曲目,最终用Matplotlib等库完成可视化展示。已有227人学习下载,适合想打通爬虫采集、存储、分析与展示全流程的读者参考,也能帮助理解网页结构变化与反爬机制下的代码健壮性设计。
1. 爬音乐平台这件事,难点从来不在请求本身
很多人第一次写 Python 爬虫,都是从音乐平台下手的。理由很朴素:页面结构看着简单,歌曲信息一目了然,感觉一个 requests 加个循环就能搞定。但真正动手之后你会发现,翻车的地方往往不在“怎么发请求”,而在“请求发出去之后拿回来的东西对不对”。我见过太多人卡在三个地方:列表页拿到的歌曲 ID 拼不出可播放地址、请求头少一个字段直接返回空数据、批量下载跑到一半被限速封 IP。这篇笔记就围绕一个模拟的音乐平台场景,把从列表抓取、详情解析、音频地址还原到批量落盘的完整链路拆开讲。适合已经会写基础 requests 但没做过完整媒体类爬虫的开发者,也适合想搞清楚“为什么我的爬虫拿不到真实数据”的熟手。下面所有代码和参数都可以直接抄去改,平台域名和接口路径我会用占位符代替,你替换成自己目标站点的真实值即可。
2. 先搞清楚数据从哪来:列表页、详情页与音频地址的三层结构
2.1 为什么直接请求播放页往往拿不到音频链接
音乐平台和普通图文站点最大的区别在于,音频文件几乎不会以静态<audio src="...">的形式出现在 HTML 里。常见做法是:页面加载时先由 JavaScript 发起一个 XHR 请求,向服务端换取一个带时效签名的播放地址,这个地址通常还有 Referer 和 Cookie 校验。所以你用 requests 直接 GET 播放页,拿到的 HTML 里往往只有一个空的容器标签,真正的地址藏在后续的接口响应里。
我一般会先打开浏览器开发者工具的 Network 面板,筛选 XHR 或 Fetch 请求,然后刷新页面,观察哪几个请求返回了 JSON 且里面包含歌曲名、歌手、时长这类字段。找到这个接口,才是整个爬虫真正的入口。这一步没有捷径,必须手动看一遍,因为不同平台的接口路径和参数命名差异很大。
2.2 列表接口与详情接口的字段对应关系
假设我们面对的是一个模拟音乐平台,它的列表接口返回结构大致如下:
{ "code": 200, "data": { "total": 120, "songs": [ {"song_id": "1001", "name": "示例歌曲A", "artist": "示例歌手", "duration": 245}, {"song_id": "1002", "name": "示例歌曲B", "artist": "示例歌手", "duration": 198} ] } }列表接口通常只给 song_id、歌名、歌手、时长这些基础字段,不会直接给音频地址。真正的播放地址需要拿 song_id 去请求另一个详情或播放接口。这个设计本身就是一道门槛:你必须先拿到 ID,再二次请求,才能拼出完整链路。
| 层级 | 接口作用 | 关键字段 | 是否直接给音频地址 |
|---|---|---|---|
| 列表接口 | 分页返回歌曲概要 | song_id、name、artist | 否 |
| 详情接口 | 返回单曲完整信息 | song_id、album、bitrate | 否 |
| 播放接口 | 换取带签名播放地址 | song_id、quality、url | 是,但有时效 |
这张表是我在多个类似场景里总结出来的通用结构。你拿到目标站点后,先按这三层去对应,基本不会跑偏。
2.3 请求头里哪几个字段是必须带的
很多人翻车的第一个点就是请求头。音乐平台的接口通常校验 Referer 和 User-Agent,部分还会校验 Origin。少一个字段,返回的可能是{"code": 403, "msg": "forbidden"}或者干脆返回空列表。
import requests HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ), "Referer": "https://example-music-site.com/", "Origin": "https://example-music-site.com", "Accept": "application/json, text/plain, */*", } def fetch_list(page: int = 1, page_size: int = 20) -> dict: url = "https://example-music-site.com/api/song/list" params = {"page": page, "size": page_size} resp = requests.get(url, headers=HEADERS, params=params, timeout=10) resp.raise_for_status() return resp.json()这段代码里,Referer和Origin是最容易被忽略的两个字段。timeout=10也是必须的,否则遇到慢响应会一直挂住。raise_for_status()帮你把 4xx 和 5xx 直接抛出来,避免拿到错误页面还继续解析。参数page和size按目标站点的实际命名改,有的平台用offset和limit,有的用pageNo和pageSize,这个只能看接口文档或抓包确认。
3. 从 song_id 到可播放地址:签名参数与二次请求的完整实现
3.1 播放接口的签名是怎么算出来的
这是整个爬虫里最硬的一块。播放接口为了防止被直接调用,通常会要求带一个签名参数,常见形式是sign或token。签名的算法各家不同,但套路无非几种:把 song_id、时间戳、固定盐值拼成一个字符串,再做 MD5 或 SHA1。
我遇到过的模拟场景里,签名逻辑是这样的:sign = md5(song_id + timestamp + salt),其中 salt 是写在前端 JS 里的一个固定字符串。你需要去翻页面的 JS 文件,搜索sign或md5关键字,找到拼接顺序和盐值。
import hashlib import time SALT = "example_salt_2024" # 从目标站点前端 JS 中提取 def build_sign(song_id: str, ts: int) -> str: raw = f"{song_id}{ts}{SALT}" return hashlib.md5(raw.encode("utf-8")).hexdigest() def fetch_play_url(song_id: str, quality: str = "320") -> str: ts = int(time.time() * 1000) sign = build_sign(song_id, ts) url = "https://example-music-site.com/api/song/play" params = { "song_id": song_id, "quality": quality, "ts": ts, "sign": sign, } resp = requests.get(url, headers=HEADERS, params=params, timeout=10) data = resp.json() if data.get("code") != 200: raise RuntimeError(f"play url failed: {data}") return data["data"]["url"]这里有几个参数需要你重点确认:ts是毫秒时间戳还是秒级时间戳,quality的取值有哪些(常见是 128、192、320 或 flac),以及签名拼接时 song_id 和 ts 的先后顺序。顺序错了,签名就是错的,接口会返回 403。我一般会先用一个固定的 song_id 和时间戳,在浏览器控制台里手动算一遍,和接口返回的签名对比,确认算法无误后再写进代码。
3.2 批量抓取时的分页与去重策略
拿到列表接口后,下一步是翻页。这里有两个坑:一是总数字段可能不准,二是翻到后面可能返回重复数据。我的做法是用一个 set 记录已经抓到的 song_id,每次新数据进来先做差集。
def crawl_all_songs(max_pages: int = 50) -> list: seen = set() result = [] for page in range(1, max_pages + 1): data = fetch_list(page=page, page_size=20) songs = data.get("data", {}).get("songs", []) if not songs: break for item in songs: sid = item["song_id"] if sid in seen: continue seen.add(sid) result.append(item) time.sleep(0.5) # 控制频率,避免触发限流 return resulttime.sleep(0.5)这个间隔不是随便写的。太快会被限流,太慢效率低。我一般从 0.3 秒起步,如果遇到 429 状态码就翻倍。max_pages设一个上限是防止接口异常时无限循环。去重用的 set 在数据量大时内存占用可控,因为 song_id 是短字符串。
3.3 音频文件落盘:文件名清洗与断点续传
拿到播放地址后,最后一步是下载。文件名里经常有斜杠、冒号、问号这些不能做文件名的字符,必须清洗。另外,批量下载最怕跑到一半断了,所以我会加一个简单的断点续传逻辑:下载前先检查文件是否已存在且大小合理。
import os import re def safe_filename(name: str) -> str: name = re.sub(r'[\\/:*?"<>|]', "_", name) return name.strip()[:80] def download_audio(url: str, save_path: str, chunk_size: int = 8192) -> None: if os.path.exists(save_path) and os.path.getsize(save_path) > 1024: print(f"skip existing: {save_path}") return with requests.get(url, headers=HEADERS, stream=True, timeout=30) as r: r.raise_for_status() tmp_path = save_path + ".part" with open(tmp_path, "wb") as f: for chunk in r.iter_content(chunk_size=chunk_size): if chunk: f.write(chunk) os.rename(tmp_path, save_path)stream=True是必须的,否则 requests 会把整个音频读进内存。先写.part临时文件再重命名,是为了避免下载中断留下一个看似完整实则损坏的文件。chunk_size=8192是通用值,网络差可以调到 4096,网络好可以调到 65536。检查文件大小大于 1024 字节才跳过,是为了防止之前失败留下的空文件被误判为已完成。
4. 避坑与排查:那些让我重跑整晚的细节
4.1 返回 200 但 data 为空
现象:请求状态码是 200,但 JSON 里data字段是 null 或空数组。原因通常是请求头缺少 Referer,或者 Cookie 过期。解决方法是补全请求头,并把浏览器里当前有效的 Cookie 复制到 headers 里。如果 Cookie 有时效,就需要定期更新,或者改用登录接口自动获取。
4.2 签名正确但播放地址 403
现象:签名算法确认无误,但拿到的播放地址请求时返回 403。原因是播放地址本身有 Referer 校验,下载时也必须带上和接口请求一致的 Referer。解决方法是下载函数里复用同一套 HEADERS,不要图省事只带 User-Agent。
4.3 翻页到第 N 页后返回重复数据
现象:前几页正常,后面开始出现和前面完全相同的 song_id。原因是服务端分页逻辑在数据变动时不稳定,或者接口本身有缓存。解决方法是用 set 去重,并且在连续两页没有新数据时提前终止,不要死板地翻到 max_pages。
4.4 下载速度突然降到几 KB
现象:前几个文件下载很快,后面越来越慢。原因是触发了服务端的单 IP 限速。解决方法是降低并发(如果用了多线程就改回单线程),增大请求间隔,或者分批分时段跑。我一般不会在爬虫里做多线程下载音频,单线程加 0.5 秒间隔已经够用,稳定性比速度重要。
4.5 文件名乱码或超长导致写入失败
现象:Windows 下文件名包含特殊字符报错,或者文件名超过 255 字符被截断。原因是歌名里带了平台特有的符号。解决方法是统一用正则清洗,并限制长度在 80 字符以内。上面safe_filename函数里的[:80]就是干这个的。
5. 把爬虫跑稳的几个进阶习惯
5.1 用会话对象复用连接与 Cookie
每次请求都新建连接在批量场景下开销不小。用requests.Session()可以复用 TCP 连接,也能自动保持 Cookie。我一般会在 Session 初始化时把 HEADERS 设好,后续所有请求都走这个 session。
session = requests.Session() session.headers.update(HEADERS) def fetch_with_session(url: str, params: dict = None) -> dict: resp = session.get(url, params=params, timeout=10) resp.raise_for_status() return resp.json()这样改完之后,代码里不需要每次传 headers,也不用手动管理 Cookie。对于需要登录的场景,先调登录接口,Cookie 会自动存进 session,后续请求直接复用。
5.2 把中间结果落成 JSONL 方便断点重跑
爬虫跑到一半挂了,最怕前面抓的数据全丢。我的习惯是每抓完一页就往 JSONL 文件里追加一行,而不是全部攒在内存里最后一次性写。
import json def append_jsonl(path: str, record: dict) -> None: with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") def load_done_ids(path: str) -> set: if not os.path.exists(path): return set() done = set() with open(path, "r", encoding="utf-8") as f: for line in f: try: done.add(json.loads(line)["song_id"]) except (json.JSONDecodeError, KeyError): continue return doneJSONL 的好处是每行独立,即使最后一行写坏了也不影响前面数据。load_done_ids在重跑时先加载已完成 ID,跳过已抓部分。ensure_ascii=False保证中文歌名正常显示,不用转成\uXXXX。
5.3 验证抓取完整性的三个检查点
跑完之后怎么确认没漏?我一般看三个数:列表接口返回的 total、去重后的 song_id 数量、实际下载成功的文件数量。这三个数应该接近但不一定完全相等,因为可能有部分歌曲无版权或下架。如果去重数量远小于 total,说明翻页逻辑有问题;如果下载数量远小于去重数量,说明播放地址获取或下载环节有失败。把失败 ID 单独记一个日志文件,方便后续补跑。
| 检查点 | 预期关系 | 异常时的排查方向 |
|---|---|---|
| total 与去重数 | 去重数 ≥ total 的 90% | 翻页终止条件、接口分页参数 |
| 去重数与播放地址数 | 播放地址数 ≥ 去重数的 85% | 签名算法、quality 参数 |
| 播放地址数与落盘数 | 落盘数 ≥ 播放地址数的 95% | 网络超时、文件名清洗 |
这张表是我每次跑完批量任务后必对的。数字对不上不一定是代码错,也可能是平台本身有部分资源不可用,但至少能帮你快速定位是哪一层出了问题。
5.4 控制频率比换 IP 更有效
很多人一遇到限流就想着换 IP,但在音乐平台这类场景里,请求频率才是触发限流的主因。把间隔从 0.1 秒调到 0.5 秒,往往比折腾代理更省事。我自己的习惯是:单线程、固定间隔、每抓 100 条休息 10 秒。这套节奏跑了很久,基本没被封过。真遇到硬限流,先降频,再考虑分时段,最后才想别的办法。
最后说一个我踩过的坑:有次为了图快,把下载和解析放在同一个循环里,结果一个文件下载卡住,整个抓取流程全停。后来改成解析和下载分离,解析结果先落 JSONL,下载单独跑一个脚本读 JSONL,两边互不影响。这个习惯帮我省了很多重跑的时间。希望帮到你。
本文还有配套的精品资源,点击获取