24 小时自动录制直播——用 FFmpeg + Python 实现无人值守直播录像
你关注的主播每晚 8 点开播，但你经常加班错过。或者你需要监控一个 24 小时摄像头流，每隔 1 小时存一个文件。直播录制和普通视频下载不一样——你永远不知道直播什么时候结束，流随时可能断开，文件大小不可预测。这篇文章教你用 FFmpeg 和 Python 搭建一套自动直播录制系统：定时检测开播 → 自动开始录制 → 按时间/大小分段存储 → 断开自动重连 → 录制完成通知。
TL;DR：FFmpeg 录制直播流核心命令：
ffmpeg -i "直播URL" -c copy -f segment -segment_time 3600 output_%03d.mp4。-c copy不重新编码（CPU 占用极低），-f segment按时间自动分段。Python 脚本负责：用yt-dlp检测直播状态 → 开播后启动 FFmpeg 录制 → 监控 FFmpeg 进程 → 断流自动重连 → 录制完成后触发通知。
目录
- 一、直播录制和普通下载的区别
- 二、FFmpeg 录制基础命令
- 三、自动分段：按时间/大小切割
- 四、Python 自动化脚本
- 五、断流重连与异常处理
- 六、监控多个直播间
- 七、部署到服务器 24 小时运行
- 八、合规与温馨提示
一、直播录制和普通下载的区别
普通下载（点播）：
文件大小已知 → Content-Length 有值
下载一次完成 → 支持断点续传
失败了重试就好 → 文件不会变
直播录制：
文件大小未知 → 不知道主播会播多久
需要持续写入 → 不能"断点续传"
断了就断了 → 流断了数据就丢了
实时性要求高 → 不能有大的缓冲延迟
直播的本质是一个永无止境的流。录制工具要做的就是：一直读，一直写，直到你让它停或者流断了。
二、FFmpeg 录制基础命令
2.1 最简录制
# 录制直播流，直接存为 MP4
ffmpeg -i "https://example.com/live/stream.m3u8" \
-c copy output.mp4
-c copy 是关键——它不重新编码视频和音频，直接复制原始数据。CPU 占用几乎为零，画质无损。录几个小时对系统几乎没有压力。
2.2 常用直播流格式
| 平台 | 直播流格式 | 获取方式 |
|---|---|---|
| B 站 | FLV/HTTP-TS | 网页抓包或 API 获取 |
| YouTube | HLS (m3u8) | yt-dlp 提取 |
| 抖音 | FLV | 抓包获取 |
| 快手 | FLV | 抓包获取 |
| Twitch | HLS | yt-dlp 提取 |
| RTMP 推流 | RTMP | 直接连接 RTMP 地址 |
| RTSP 摄像头 | RTSP | 摄像头 IP 地址 |
2.3 获取直播流地址
# 用 yt-dlp 获取 YouTube/B站 直播流地址
yt-dlp -g "https://www.youtube.com/watch?v=xxx"
# 输出：https://manifest.googlevideo.com/.../index.m3u8
# 然后把这个地址喂给 FFmpeg
ffmpeg -i "$(yt-dlp -g 'https://www.youtube.com/watch?v=xxx')" \
-c copy output.mp4
三、自动分段：按时间/大小切割
3.1 按时间分段（推荐）
# 每 30 分钟切一个文件
ffmpeg -i "直播URL" \
-c copy \
-f segment \
-segment_time 1800 \
-segment_format mp4 \
-reset_timestamps 1 \
-strftime 1 \
"live_%Y%m%d_%H%M%S.mp4"
参数说明：
| 参数 | 含义 |
|---|---|
-f segment |
分段输出模式 |
-segment_time 1800 |
每段 1800 秒（30 分钟） |
-segment_format mp4 |
分段文件格式 |
-reset_timestamps 1 |
每段的时间戳从 0 开始 |
-strftime 1 |
文件名使用时间格式 |
输出效果：
live_20260811_200000.mp4 ← 20:00 开始
live_20260811_203000.mp4 ← 20:30
live_20260811_210000.mp4 ← 21:00
...
3.2 按大小分段
# 每 500MB 切一个文件
ffmpeg -i "直播URL" \
-c copy \
-f segment \
-segment_size 500M \
-segment_format mp4 \
"live_part_%03d.mp4"
3.3 同时按时间和大小分段
# 每 30 分钟或 500MB 切一个文件（先到为准）
ffmpeg -i "直播URL" \
-c copy \
-f segment \
-segment_time 1800 \
-segment_size 500M \
-segment_format mp4 \
"live_%03d.mp4"
3.4 为什么推荐分段
1. 文件可控：一个 4 小时直播如果存成单文件，可能 8GB+，分段后每段 1-2GB
2. 不怕中断：FFmpeg 崩溃了，已经写完的分段不受影响
3. 方便处理：分段后可以并行转码、上传、分享
4. 边录边看：已经写完的分段可以直接播放，不用等录制结束
四、Python 自动化脚本
#!/usr/bin/env python3
"""直播自动录制脚本"""
import subprocess
import os
import time
import signal
import json
import requests
from datetime import datetime
from pathlib import Path
class LiveRecorder:
"""直播录制器"""
def __init__(self, output_dir='./recordings',
segment_time=1800, dingtalk_webhook=None):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.segment_time = segment_time
self.dingtalk_webhook = dingtalk_webhook
self.process = None
self.current_streamer = ''
def get_stream_url(self, url):
"""获取直播流地址"""
# 尝试用 yt-dlp 提取
try:
result = subprocess.run(
['yt-dlp', '-g', url],
capture_output=True, text=True, timeout=30
)
if result.returncode == 0:
stream_url = result.stdout.strip().split('\n')[0]
return stream_url
except Exception:
pass
# 如果 yt-dlp 失败，假设传入的已经是直播流地址
return url
def check_live_status(self, url):
"""检查是否在直播"""
try:
result = subprocess.run(
['yt-dlp', '--dump-json', '--playlist-end', '1', url],
capture_output=True, text=True, timeout=30
)
if result.returncode == 0 and result.stdout.strip():
info = json.loads(result.stdout)
live_status = info.get('live_status', '')
return live_status in ('is_live', 'is_upcoming')
except Exception:
pass
return False
def start_recording(self, stream_url, streamer_name='unknown'):
"""开始录制"""
self.current_streamer = streamer_name
# 生成文件名模板
date_str = datetime.now().strftime('%Y%m%d')
prefix = f"{streamer_name}_{date_str}"
output_template = str(
self.output_dir / f"{prefix}_%Y%m%d_%H%M%S.mp4"
)
cmd = [
'ffmpeg',
'-i', stream_url,
'-c', 'copy',
'-f', 'segment',
'-segment_time', str(self.segment_time),
'-segment_format', 'mp4',
'-reset_timestamps', '1',
'-strftime', '1',
'-loglevel', 'warning',
'-stats',
output_template,
]
print(f"[{datetime.now()}] 开始录制: {streamer_name}")
self.notify(f"开始录制: {streamer_name}")
self.process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
)
# 监控输出（捕获录制时长等信息）
start_time = time.time()
for line in self.process.stdout:
# FFmpeg 的进度信息在 stderr
pass
self.process.wait()
duration = time.time() - start_time
if self.process.returncode == 0 or self.process.returncode == 255:
# 0 = 正常退出, 255 = 被信号中断（正常）
print(f"[{datetime.now()}] 录制结束，持续 {duration/60:.0f} 分钟")
self.notify(
f"录制结束: {streamer_name}\n"
f"持续: {duration/60:.0f} 分钟"
)
else:
print(f"录制异常退出，返回码: {self.process.returncode}")
self.notify(f"录制异常: {streamer_name} (返回码 {self.process.returncode})")
def stop_recording(self):
"""停止录制"""
if self.process and self.process.poll() is None:
print("正在停止录制...")
self.process.send_signal(signal.SIGINT)
try:
self.process.wait(timeout=10)
except subprocess.TimeoutExpired:
self.process.kill()
def notify(self, message):
"""发送通知"""
if not self.dingtalk_webhook:
return
try:
requests.post(self.dingtalk_webhook, json={
'msgtype': 'text',
'text': {'content': f'[直播录制] {message}'}
}, timeout=5)
except Exception:
pass
def monitor_and_record(self, url, streamer_name, check_interval=60):
"""循环监控：检测到开播就录制，结束后继续检测"""
print(f"开始监控: {streamer_name} ({url})")
while True:
if self.check_live_status(url):
# 开播了，获取流地址并开始录制
try:
stream_url = self.get_stream_url(url)
self.start_recording(stream_url, streamer_name)
except Exception as e:
print(f"录制出错: {e}")
time.sleep(check_interval)
else:
print(f"[{datetime.now().strftime('%H:%M')}] 未开播，{check_interval}秒后重试...")
time.sleep(check_interval)
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='直播自动录制')
parser.add_argument('url', help='直播页面 URL 或直播流地址')
parser.add_argument('--name', default='live', help='主播/频道名称')
parser.add_argument('--output', default='./recordings', help='输出目录')
parser.add_argument('--segment', type=int, default=1800, help='分段时长（秒）')
parser.add_argument('--interval', type=int, default=60, help='检测间隔（秒）')
parser.add_argument('--webhook', help='钉钉 Webhook 通知地址')
args = parser.parse_args()
recorder = LiveRecorder(
output_dir=args.output,
segment_time=args.segment,
dingtalk_webhook=args.webhook,
)
recorder.monitor_and_record(
url=args.url,
streamer_name=args.name,
check_interval=args.interval,
)
五、断流重连与异常处理
直播流断开是常态——主播网络波动、CDN 切换、平台维护都可能导致断流。
5.1 FFmpeg 内置重连
ffmpeg \
-reconnect 1 \
-reconnect_streamed 1 \
-reconnect_delay_max 10 \
-i "直播URL" \
-c copy output.mp4
| 参数 | 含义 |
|---|---|
-reconnect 1 |
断开后自动重连 |
-reconnect_streamed 1 |
即使已经开始传输也重连 |
-reconnect_delay_max 10 |
最大重连等待 10 秒 |
5.2 Python 层面的重试
def record_with_retry(self, stream_url, streamer_name, max_retries=10):
"""带重试的录制"""
for attempt in range(max_retries):
try:
self.start_recording(stream_url, streamer_name)
break # 正常结束
except Exception as e:
print(f"录制中断 (第 {attempt+1}/{max_retries} 次): {e}")
if attempt < max_retries - 1:
wait = min(30, 2 ** attempt)
print(f"等待 {wait} 秒后重试...")
time.sleep(wait)
# 重新获取流地址（可能变了）
try:
stream_url = self.get_stream_url(stream_url)
except Exception:
pass
六、监控多个直播间
import threading
def monitor_multiple(config_file):
"""同时监控多个直播间"""
with open(config_file, 'r', encoding='utf-8') as f:
configs = json.load(f)
threads = []
for cfg in configs:
recorder = LiveRecorder(
output_dir=cfg.get('output', './recordings'),
dingtalk_webhook=cfg.get('webhook'),
)
t = threading.Thread(
target=recorder.monitor_and_record,
args=(cfg['url'], cfg['name']),
daemon=True,
)
t.start()
threads.append(t)
print(f"已启动监控: {cfg['name']}")
# 保持主线程运行
for t in threads:
t.join()
// config.json
[
{
"name": "技术主播A",
"url": "https://www.youtube.com/@streamerA/live",
"output": "./recordings/streamerA",
"webhook": "https://oapi.dingtalk.com/robot/send?access_token=xxx"
},
{
"name": "游戏主播B",
"url": "https://live.bilibili.com/12345",
"output": "./recordings/streamerB"
}
]
七、部署到服务器 24 小时运行
7.1 Docker 部署
FROM python:3.11-slim
RUN apt-get update && apt-get install -y ffmpeg && rm -rf /var/lib/apt/lists/*
WORKDIR /app
RUN pip install yt-dlp requests
COPY recorder.py .
CMD ["python", "recorder.py", "--config", "/config/config.json"]
# docker-compose.yml
services:
recorder:
build: .
restart: unless-stopped
volumes:
- ./recordings:/recordings
- ./config:/config
environment:
- TZ=Asia/Shanghai
7.2 资源消耗
录制一路 1080P 直播（-c copy 模式）：
CPU: ~2%（不重新编码）
内存: ~50MB
磁盘: ~1.5GB/小时
带宽: ~3Mbps（取决于直播源码率）
同时录制 10 路直播：
CPU: ~15%
内存: ~500MB
磁盘: ~15GB/小时
八、合规与温馨提示
- 录制直播内容仅限个人观看和存档，未经授权不得公开传播或商用
- 部分平台的服务条款禁止录制直播内容，请在使用前阅读相关条款
- 24 小时录制可能产生大量数据，注意磁盘空间管理
- 更多讨论见 下载视频算侵权吗？聊聊个人备份与版权的那条线
直播录制这件事，搭好之后就两个字：省心。我有个树莓派在家里跑了半年，每天自动录几个技术主播的直播，第二天通勤路上看回放。FFmpeg 的 -c copy 模式几乎不消耗 CPU，一个树莓派同时录 4-5 路 1080P 直播毫无压力。
本文由 VidDown 技术博客原创发布。VidDown 桌面客户端内置了直播录制功能——输入直播间链接，自动检测开播状态并开始录制，支持分段存储和断流重连。访问 VidDown 了解更多。