当前位置：首页 > news >正文

视频号直播视频录制

news 2025/11/16 5:55:24

视频号直播视频录制

为录制直播视频，首先应获取直播地址，你可以用Reqable抓包，Reqable的下载地址：https://reqable.com/zh-CN/,安装运行，打开直播，抓取视频号直播视频地址，这里仅是示例，具体根据情况分析地址。有了地址就简单了！
在这里插入图片描述

一、使用python录制

（以下Python程序，与GPT5交流的产物）

1、安装依赖

pip install requests

2、保存为 `stream_recorder.py`

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
stream_recorder.py
用纯 Python 持续录制直播直链（如 .flv），支持：
- 断线自动重连（默认 3 秒后重连）
- 限时录制（--duration 01:30:00 或 --duration 5400）
- 自定义保存文件名/目录（自动加时间戳）
- 自定义请求头、代理（支持 http/socks5）
- 实时速度/体积日志，Ctrl+C 安全退出用法示例见文件末尾或运行: python stream_recorder.py -h
"""
import argparse
import datetime as dt
import os
import sys
import time
import signal
import threading
from urllib.parse import urlparse
import requestsDEFAULT_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) ""AppleWebKit/537.36 (KHTML, like Gecko) ""Chrome/120.0.0.0 Safari/537.36")class GracefulKiller:def __init__(self):self.stop = threading.Event()signal.signal(signal.SIGINT, self._exit)if hasattr(signal, "SIGTERM"):signal.signal(signal.SIGTERM, self._exit)def _exit(self, *args):self.stop.set()def parse_duration_to_seconds(s: str | None) -> int | None:"""支持 'HH:MM:SS' 或 纯秒数；None 表示不限制时长"""if not s:return Nones = str(s).strip()if s.isdigit():return int(s)parts = s.split(":")if len(parts) == 3:h, m, sec = partsreturn int(h) * 3600 + int(m) * 60 + int(sec)if len(parts) == 2:  # MM:SSm, sec = partsreturn int(m) * 60 + int(sec)raise ValueError("无效的时长格式，请用秒数或 HH:MM:SS")def build_headers(hlist: list[str] | None) -> dict:"""--header 可多次传入 'Key: Value'"""headers = {"User-Agent": DEFAULT_UA,"Accept": "*/*","Connection": "keep-alive",}if not hlist:return headersfor item in hlist:if ":" not in item:continuek, v = item.split(":", 1)headers[k.strip()] = v.strip()return headersdef suggest_filename(url: str, out: str | None, outdir: str | None) -> str:ts = dt.datetime.now().strftime("%Y%m%d_%H%M%S")if out:fname = outelse:# 从 URL 推断一个简短基名path = urlparse(url).pathbase = os.path.basename(path) or "stream"if not base.lower().endswith(".flv"):base += ".flv"fname = f"{os.path.splitext(base)[0]}_{ts}.flv"if outdir:os.makedirs(outdir, exist_ok=True)fname = os.path.join(outdir, fname)return fnamedef human_bytes(n: int) -> str:units = ["B", "KB", "MB", "GB", "TB"]s = 0val = float(n)while val >= 1024 and s < len(units) - 1:val /= 1024.0s += 1return f"{val:.2f} {units[s]}"def recorder(url: str,output_path: str,duration_s: int | None = None,headers: dict | None = None,proxy: str | None = None,timeout: int = 15,chunk_size: int = 64 * 1024,reconnect_delay: int = 3,append: bool = True):"""核心录制逻辑：持续读流写文件；异常自动重连；到时自动收尾。"""killer = GracefulKiller()ses = requests.Session()proxies = Noneif proxy:# 既用于 http 也用于 https；支持 socks5/socks5h/http/httpsproxies = {"http": proxy, "https": proxy}written = 0start = time.time()last_log_t = startlast_log_written = 0mode = "ab" if append else "wb"os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)with open(output_path, mode) as f:print(f"[i] 开始录制：{url}")print(f"[i] 保存到：{output_path}")if duration_s:print(f"[i] 计划录制时长：{duration_s} 秒（{dt.timedelta(seconds=duration_s)}）")if proxies:print(f"[i] 使用代理：{proxy}")print("[i] Ctrl+C 可安全停止录制。")while not killer.stop.is_set():if duration_s is not None and (time.time() - start) >= duration_s:print("[i] 已达到预设时长，结束录制。")breaktry:resp = ses.get(url,headers=headers,stream=True,timeout=(10, timeout),allow_redirects=True,proxies=proxies,)resp.raise_for_status()# 部分服务会周期性断开；这里每次重连后继续 appendfor chunk in resp.iter_content(chunk_size=chunk_size):if killer.stop.is_set():breakif duration_s is not None and (time.time() - start) >= duration_s:breakif not chunk:continuef.write(chunk)written += len(chunk)# 每 3 秒打印一次速度/体积now = time.time()if now - last_log_t >= 3:delta_b = written - last_log_writtendelta_t = now - last_log_tspeed = delta_b / max(delta_t, 1e-6)print(f"[{dt.datetime.now().strftime('%H:%M:%S')}] "f"{human_bytes(written)} | {human_bytes(speed)}/s")last_log_t = nowlast_log_written = written# 流正常结束（直播结束）if not killer.stop.is_set() and duration_s is None:print("[i] 服务器关闭连接，可能是直播已结束。若仍在直播，将自动重连…")time.sleep(reconnect_delay)except requests.RequestException as e:print(f"[!] 网络异常：{e.__class__.__name__}: {e}")if killer.stop.is_set():breakprint(f"[i] {reconnect_delay} 秒后重连…")time.sleep(reconnect_delay)continueprint(f"[✓] 录制完成，总计写入：{human_bytes(written)}")return output_pathdef main():p = argparse.ArgumentParser(description="录制 .flv 等直播直链到本地文件（断线重连/限时录制/代理/自定义请求头）")p.add_argument("url", nargs="?", help="直播直链URL（例如以 .flv 结尾的链接）")p.add_argument("-o", "--output", help="输出文件名（默认自动基于URL+时间戳命名）")p.add_argument("--outdir", default="", help="输出目录（默认当前目录）")p.add_argument("--duration", help="录制时长，秒数或HH:MM:SS；不填则直到Ctrl+C或直播结束")p.add_argument("--proxy", help="代理，例如 http://127.0.0.1:7890 或 socks5h://127.0.0.1:10808")p.add_argument("--header", action="append",help="自定义请求头，可多次，例如 --header 'Referer: https://example.com'")p.add_argument("--timeout", type=int, default=15, help="单次请求读超时秒数（默认15）")p.add_argument("--chunk", type=int, default=64*1024, help="下载块大小（默认65536）")p.add_argument("--reconnect", type=int, default=3, help="断线重连间隔秒（默认3）")p.add_argument("--overwrite", action="store_true", help="若存在同名文件则覆盖（默认追加写入）")args = p.parse_args()url = args.urlif not url:# 允许启动后粘贴URLurl = input("请输入直播URL：").strip()duration_s = parse_duration_to_seconds(args.duration) if args.duration else Noneheaders = build_headers(args.header)out_path = suggest_filename(url, args.output, args.outdir)try:recorder(url=url,output_path=out_path,duration_s=duration_s,headers=headers,proxy=args.proxy,timeout=args.timeout,chunk_size=args.chunk,reconnect_delay=args.reconnect,append=not args.overwrite,)except Exception as e:print(f"[x] 录制失败：{e}")sys.exit(1)if __name__ == "__main__":main()

3、常用示例

最简单（直到你按 Ctrl+C 或直播结束）

python stream_recorder.py "粘贴你的.flv直播直链"

限定时长 2 小时，自动收尾：

python stream_recorder.py "URL" --duration 02:00:00

python stream_recorder.py "URL" --outdir D:\Streams --output my_show.flv

带代理（例如本机 10808 的 socks5）：

python stream_recorder.py "URL" --proxy socks5h://127.0.0.1:10808

添加请求头（若站点校验来源）：

python stream_recorder.py "URL" \--header "Referer: https://example.com" \--header "Origin: https://example.com"

运行中每隔 3 秒会打印累计体积与瞬时速度，方便观察是否在“不断下载”。

4、后台运行建议

Windows：
- 最小化后台：start /B python stream_recorder.py "URL" --duration 01:00:00 --outdir D:\Streams

二、使用`ffmpeg` 录制

https://www.gyan.dev/ffmpeg/builds/下载ffmpeg,解压，将FFmpeg\bin路径添加到系统变量path中，例如：我的ffmpeg放在E:\FFmpeg\bin,
在这里插入图片描述
打开PowerShell执行以下命令，替换命令中的直播url,退出ffmpeg请按q

ffmpeg -hide_banner -reconnect 1 -reconnect_streamed 1 -reconnect_at_eof 1 `-rw_timeout 15000000 -analyzeduration 100M -probesize 100M `-i "你的URL" -map 0 -c copy -f flv "out_$(Get-Date -Format yyyyMMdd_HHmmss).flv"