保姆级教程：用Python + m3u8库自动录制B站学习区直播（附防断流技巧）
·
Python自动化录制B站学习直播的实战指南
直播学习已经成为当代知识获取的重要方式，但直播的即时性往往让学习者难以反复消化内容。本文将深入探讨如何利用Python构建一个稳定可靠的B站学习直播录制系统，涵盖从直播流获取到本地存储的全流程解决方案。
1. 直播录制技术基础与准备工作
在开始构建自动化录制系统之前，我们需要理解几个核心技术概念。M3U8作为一种基于HTTP的流媒体传输协议，已经成为各大直播平台的标准配置。它通过将视频流分割为多个小片段(TS文件)来实现动态加载，既保证了流畅性又便于网络传输。
环境准备清单：
- Python 3.7+ (推荐3.9版本)
- m3u8库 (0.9.0以上版本)
- requests库 (网络请求)
- lxml或BeautifulSoup (HTML解析)
- FFmpeg (可选，用于后期处理)
安装核心依赖：
pip install m3u8 requests lxml
提示：建议在虚拟环境中进行开发，避免依赖冲突。可使用venv或conda创建隔离环境。
直播录制的基本原理是通过解析M3U8播放列表获取视频片段(TS文件)，然后按顺序下载这些片段并合并为完整视频。B站的直播流通常采用双层M3U8结构：
- 外层M3U8提供不同清晰度的播放列表
- 内层M3U8包含实际的TS文件地址
2. 智能获取直播流地址
获取直播流地址是整个系统的关键第一步。B站的API设计相对复杂，需要通过多个步骤才能获取到有效的M3U8地址。
直播流获取流程：
- 通过分区API获取直播间ID列表
- 选择目标直播间
- 调用播放地址接口获取外层M3U8
- 解析外层M3U8获取内层地址
def get_live_stream(room_id, quality=150):
"""获取B站直播M3U8地址"""
api_url = "https://api.live.bilibili.com/xlive/web-room/v1/playUrl/playUrl"
params = {
'cid': room_id,
'qn': quality, # 清晰度选择
'platform': 'h5',
'ptype': 16
}
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.get(api_url, headers=headers, params=params)
if response.status_code != 200:
raise ConnectionError("API请求失败")
data = response.json()
return data['data']['durl'][-1]['url'] # 获取最高清的流地址
清晰度参数对照表：
| qn值 | 分辨率 | 码率 |
|---|---|---|
| 80 | 360p | 约800kbps |
| 150 | 480p | 约1500kbps |
| 250 | 720p | 约2500kbps |
| 400 | 1080p | 约4000kbps |
注意：高清晰度会显著增加文件大小和带宽消耗，建议根据实际需求选择。
3. 构建稳定录制系统
直播录制面临的最大挑战是网络不稳定和地址失效问题。我们需要设计一套健壮的机制来应对这些情况。
核心稳定性策略：
- 定时刷新M3U8地址（约每30分钟）
- 断线自动重连机制
- 分段存储策略
- 网络异常处理
class LiveRecorder:
def __init__(self, room_id, output_dir="recordings"):
self.room_id = room_id
self.output_dir = output_dir
self.session = requests.Session()
self.current_sequence = 0
os.makedirs(output_dir, exist_ok=True)
def refresh_stream_url(self):
"""刷新M3U8地址"""
outer_url = get_live_stream(self.room_id)
playlist = m3u8.load(outer_url)
return playlist.playlists[0].absolute_uri
def record_segment(self, segment_url, filename):
"""录制单个TS片段"""
try:
response = self.session.get(segment_url, timeout=10)
response.raise_for_status()
with open(filename, 'wb') as f:
f.write(response.content)
return True
except Exception as e:
print(f"片段下载失败: {e}")
return False
def start_recording(self, duration=3600):
"""主录制循环"""
start_time = time.time()
stream_url = self.refresh_stream_url()
last_refresh = time.time()
while time.time() - start_time < duration:
# 每30分钟刷新一次地址
if time.time() - last_refresh > 1800:
stream_url = self.refresh_stream_url()
last_refresh = time.time()
try:
playlist = m3u8.load(stream_url)
for segment in playlist.segments:
segment_id = int(segment.uri.split('.')[0][1:])
if segment_id > self.current_sequence:
filename = f"{self.output_dir}/seg_{segment_id:08d}.ts"
if self.record_segment(segment.absolute_uri, filename):
self.current_sequence = segment_id
time.sleep(2) # 避免请求过于频繁
except Exception as e:
print(f"播放列表获取失败: {e}")
time.sleep(5)
continue
异常处理策略：
| 异常类型 | 处理方式 | 重试间隔 |
|---|---|---|
| 网络超时 | 自动重试 | 5秒 |
| 地址失效 | 刷新URL | 立即 |
| 404错误 | 跳过片段 | 无 |
| 连续失败 | 暂停并报警 | 30秒 |
4. 高级功能与优化技巧
基础录制功能实现后，我们可以添加一些增强功能来提升系统的实用性和用户体验。
4.1 智能直播筛选
通过分析直播间标题和分区信息，自动筛选符合条件的学习类直播：
def filter_learning_streams(min_viewers=1000):
"""筛选学习类直播"""
area_api = "https://api.live.bilibili.com/room/v1/Area/getList"
response = requests.get(area_api)
data = response.json()
learning_areas = [
area for area in data['data']
if '学习' in area['name'] or '教育' in area['name']
]
streams = []
for area in learning_areas:
room_api = f"https://api.live.bilibili.com/xlive/web-interface/v1/second/getList?platform=web&parent_area_id={area['id']}"
room_data = requests.get(room_api).json()
for room in room_data['data']['list']:
if room['online'] > min_viewers and any(
kw in room['title'] for kw in ['教程', '课程', '教学', '学习']
):
streams.append({
'room_id': room['roomid'],
'title': room['title'],
'author': room['uname'],
'viewers': room['online']
})
return sorted(streams, key=lambda x: x['viewers'], reverse=True)
4.2 录制后处理
原始TS文件可以直接播放，但进行一些后处理能获得更好的体验：
def post_process(output_dir, final_name="output.mp4"):
"""合并TS文件并转换为MP4"""
ts_files = sorted([
f for f in os.listdir(output_dir)
if f.endswith('.ts') and f.startswith('seg_')
])
# 生成文件列表
with open(f"{output_dir}/filelist.txt", 'w') as f:
for ts in ts_files:
f.write(f"file '{ts}'\n")
# 使用FFmpeg合并
subprocess.run([
'ffmpeg', '-f', 'concat', '-i', f"{output_dir}/filelist.txt",
'-c', 'copy', '-bsf:a', 'aac_adtstoasc', final_name
], check=True)
print(f"视频已合并为 {final_name}")
4.3 定时录制与自动化
结合任务调度实现全自动录制：
import schedule
import time
def job():
streams = filter_learning_streams()
if streams:
recorder = LiveRecorder(streams[0]['room_id'])
recorder.start_recording(duration=7200) # 录制2小时
# 每天9点和19点各执行一次
schedule.every().day.at("09:00").do(job)
schedule.every().day.at("19:00").do(job)
while True:
schedule.run_pending()
time.sleep(60)
5. 实战经验与避坑指南
在实际部署过程中，会遇到各种预料之外的问题。以下是一些关键经验：
网络优化技巧：
- 使用CDN加速：B站直播流分布在不同CDN节点，可通过修改域名前缀尝试不同节点
- 设置合理的超时时间：建议连接超时10秒，读取超时30秒
- 启用HTTP持久连接：减少TCP握手开销
存储优化方案：
| 方案 | 优点 | 缺点 |
|---|---|---|
| 原始TS存储 | 简单直接 | 占用空间大 |
| 实时转码 | 节省空间 | CPU占用高 |
| 分段压缩 | 平衡方案 | 增加复杂度 |
常见问题排查：
- 地址频繁失效 ：B站的M3U8地址通常有效期为30-60分钟，需要定时刷新
- TS文件损坏 ：可能是网络波动导致，建议增加重试机制
- 时间戳不连续 ：部分直播存在跳帧现象，属于正常情况
- 录制文件不同步 ：确保按正确顺序合并TS文件
# 增强版片段下载函数
def robust_download(url, max_retries=3):
for attempt in range(max_retries):
try:
response = requests.get(url, timeout=(10, 30))
response.raise_for_status()
# 验证TS文件完整性
if len(response.content) < 1024: # 小于1KB可能是无效文件
raise ValueError("文件过小，可能无效")
return response.content
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # 指数退避
在实际项目中，建议添加日志记录和监控功能，便于追踪录制状态和及时发现问题。可以记录以下关键指标：
- 片段下载成功率
- 网络延迟情况
- 文件大小变化趋势
- 地址刷新次数
更多推荐
9
0- 0
已为社区贡献31条内容
所有评论(0)