使用 import streamlink 模块或通过 runnong subprocess 或 os.system 命令编写？

Question

我看到一些可用的流链接 API，但几乎看不到任何信息，例如有关如何使用它的示例。

我实际上想做的是从特定的 youtube 频道捕获实时视频。

我只关心音频，而不关心视频。因此，我的脚本的一部分将在我使用 streamlink 捕获 .ts 视频文件之后，使用 ffmpeg 将其转换为 .mp3。因为我不认为streamlink可以将直播流捕获为mp3？

将文件转换为 mp3 的最佳方法是什么？是否会在捕获视频后运行 ffmpeg 命令，以便脚本等待进一步运行，直到转换完成？

我认为可能更好的方法是将 ffmpeg 命令作为单独的线程或与 streamlink 分开的线程运行。这样，streamlink 可以继续捕获/下载，而 ffmpeg 可以并行运行，同时将过去的 .ts 文件转换为 mp3。

通过 os.system 或 subprocess 运行 streamlink 似乎要容易得多，只需将 streamlink 作为传递所需参数的命令运行。

使用streamlink模块运行有什么好处吗？

我在这里找到了这个代码示例：Handle stream as individual frames using streamlink，但我对我可以省略哪些与视频相关的部分有疑问（因为我不关心流的视频部分）？

import numpy as np
import subprocess as sp
import threading
import cv2
import ffmpeg

#stream_url = 'https://www.nimo.tv/v/v-1712291636586087045'
stream_url = 'https://www.twitch.tv/esl_csgo'

# Assume video resolution is known.
width, height = 1920, 1080


# Writer thread (read from streamlink and write to FFmpeg in chunks of 1024 bytes).
def writer(streamlink_proc, ffmpeg_proc):
    while (not streamlink_proc.poll()) and (not ffmpeg_proc.poll()):
        try:
            chunk = streamlink_proc.stdout.read(1024)
            ffmpeg_proc.stdin.write(chunk)
        except (BrokenPipeError, OSError) as e:
            pass


streamlink_args = [r'c:\Program Files (x86)\Streamlink\bin\streamlink.exe', stream_url, "best", "-O"]  # Windows executable downloaded from: https://github.com/streamlink/streamlink/releases/tag/2.4.0
streamlink_process = sp.Popen(streamlink_args, stdout=sp.PIPE)  # Execute streamlink as sub-process


# Execute FFmpeg sub-process with URL as input and raw (BGR) output format.
ffmpeg_process = (
    ffmpeg
    .input('pipe:')
    .video
    .output('pipe:', format='rawvideo', pix_fmt='bgr24')
    .run_async(pipe_stdin=True, pipe_stdout=True) # In case ffmpeg in not in executable path, add cmd=fullpath like: .run_async(pipe_stdout=True, cmd=r'c:\FFmpeg\bin\ffmpeg.exe')
)


thread = threading.Thread(target=writer, args=(streamlink_process, ffmpeg_process))
thread.start()


# Read decoded video (frame by frame), and display each frame (using cv2.imshow)
while True:
    # Read raw video frame from stdout as bytes array.
    in_bytes = ffmpeg_process.stdout.read(width * height * 3)

    if not in_bytes:
        break

    # Transform the byte read into a NumPy array
    frame = np.frombuffer(in_bytes, np.uint8).reshape([height, width, 3])

    # Display the frame
    cv2.imshow('frame', frame)

    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

ffmpeg_process.stdout.close()
ffmpeg_process.wait()
#streamlink_process.stdin.close()
streamlink_process.kill()
cv2.destroyAllWindows()

由于我对视频质量不感兴趣，我不确定是否需要 cv2、numpy 或“视频块”或是否有帮助？

使用视频块下载的目的是什么？对流式下载的可靠性有帮助吗？

我需要帮助了解如何将我的脚本组合在一起以及哪些部分/模块会有所帮助？

Answer 1

如果streamlink没有断开，您可以直接将链接传递给FFmpeg。
建议的解决方案在下载时录制音频。

• 使用Streamlink模块从WEB地址获取直接视频流URL（如m3u8）。
• 将视频流 URL 传递给 FFmpeg。

---

决定何时以及如何终止录制可能有点棘手，因为我们希望优雅地关闭 FFmpeg。
我使用pynput 来检查是否按下了 Esc 键。

这是一个完整的代码示例：

from streamlink import Streamlink
import ffmpeg
import signal
from pynput.keyboard import Key, Listener
from threading import Thread
import time

web_url = 'https://www.twitch.tv/esl_csgo'

esc_key_pressed = False  # Global variable.


def stream_to_url(url, quality='audio_only'):
    # The "audio_only" quality may be invalid for some streams (check).
    session = Streamlink()
    streams = session.streams(url)
    return streams[quality].to_url()

 
# Wait for Esc key to be pressed (and released).
def wait_for_esc_key():
    global esc_key_pressed
    # Collect events until released https://stackoverflow.com/questions/24072790/how-to-detect-key-presses
    with Listener(on_press=None, on_release=lambda key: (False if key == Key.esc else True)) as listener:
        listener.join()
        esc_key_pressed = True    


# Get the stream URL from the WEB site URL:
stream_url = stream_to_url(web_url)

# Read the input stream from the link, and write the audio to audio.mp3 file.
ffmpeg_process = (
    ffmpeg
    .input(stream_url)
    .audio
    .output('audio.mp3')
    .overwrite_output()
    .run_async() # In case FFmpeg in not in executable path, add cmd=fullpath like: .run_async(cmd=r'c:\FFmpeg\bin\ffmpeg.exe')
)


print('Press Esc to end recording')


# Wait for Esc key in a thread (non-blocking wait).
wait_for_esc_key_thread = Thread(target=wait_for_esc_key)
wait_for_esc_key_thread.start()


# Wait for Escape key pressed or FFmpeg finished.
while (not esc_key_pressed) and (not ffmpeg_process.poll()):
    time.sleep(0.01)


if esc_key_pressed:
    # Close FFmpeg gracefully
    # https://stackoverflow.com/questions/67276793/output-always-corrupt-from-ffmpeg-using-selenium-python
    ffmpeg_process.send_signal(signal.CTRL_C_EVENT)  # Is signal.CTRL_C_EVENT Windows only?