在 python 中录制流媒体和保存网络广播

Posted 2023-03-27

【中文标题】在 python 中录制流媒体和保存网络广播

【英文标题】：Record streaming and saving internet radio in python

【发布时间】：2011-05-13 22:22:53

【问题描述】：

我正在寻找一个 python sn-p 来读取互联网广播流（.asx、.pls 等）并将其保存到文件中。

最后一个项目是 cron'ed 脚本，它将录制一两个小时的互联网广播，然后将其传输到我的手机上，以便在我上下班途中播放。 （3g 在我的通勤路上有点参差不齐）

欢迎任何片段或指针。

【参考方案1】：

以下内容对我使用 requests 库来处理 http 请求很有用。

import requests

stream_url = 'http://your-stream-source.com/stream'

r = requests.get(stream_url, stream=True)

with open('stream.mp3', 'wb') as f:
    try:
        for block in r.iter_content(1024):
            f.write(block)
    except KeyboardInterrupt:
        pass

这会将流保存到stream.mp3 文件，直到您用ctrl+C 中断它。

【讨论】：

它确实有帮助，但我如何丢弃文件中重复的块？

requests 似乎不支持 SHOUTcast 协议（某些收音机），它会引发BadStatusLine('ICY 200 OK\r\n',)) 的错误

在下面查看我的解决方案，@Doomsday - 我花了一段时间才找到最好的方法，但我忽略了它所做的状态检查并告诉它 ICY 200 OK 真的是 HTTP/1.0 200 OK这是最干净的方式来克服这一点并获得更多有趣的东西。希望对您有所帮助！

对于那些正在寻找流 url 的人来说，在许多网站上，在哪里找到它并不明显。但是很多流媒体网站都提供了一个 .pls（播放列表）文件供下载，如果你在文本编辑器中打开 .pls 文件，它就会有流的 url。

我想知道是否有办法获取“块”中的音频数据并将其处理为死区——这样您就可以决定是否写入该块。

【参考方案2】：

所以在修补和使用它之后，我发现Streamripper 工作得最好。这是我使用的命令

streamripper http://yp.shoutcast.com/sbin/tunein-station.pls?id=1377200 -d ./streams -l 10800 -a tb$FNAME

【参考方案3】：

如果您发现 Python 3 中的 requests 或 urllib.request 调用未能保存流，因为您收到“ICY 200 OK”而不是“HTTP/1.0 200 OK”标头，您需要告诉底层功能 ICY 200 OK 就OK了！

您可以有效地做的是在处理标头之前拦截处理打开流后读取状态的例程。

只需将这样的例程放在您的流打开代码上方即可。

def NiceToICY(self):
    class InterceptedHTTPResponse():
        pass
    import io
    line = self.fp.readline().replace(b"ICY 200 OK\r\n", b"HTTP/1.0 200 OK\r\n")
    InterceptedSelf = InterceptedHTTPResponse()
    InterceptedSelf.fp = io.BufferedReader(io.BytesIO(line))
    InterceptedSelf.debuglevel = self.debuglevel
    InterceptedSelf._close_conn = self._close_conn
    return ORIGINAL_HTTP_CLIENT_READ_STATUS(InterceptedSelf)

然后将这些行放在主程序的开头，然后再打开 URL。

ORIGINAL_HTTP_CLIENT_READ_STATUS = urllib.request.http.client.HTTPResponse._read_status
urllib.request.http.client.HTTPResponse._read_status = NiceToICY

他们将覆盖标准例程（仅此一次）并在打开流时运行 NiceToICY 函数来代替正常的状态检查。 NiceToICY 替换无法识别的状态响应，然后复制“真实”_read_status 函数所需的原始响应的相关位。最后调用原始数据并将其中的值传递回调用者，其他一切都照常进行。

我发现这是解决导致错误的状态消息问题的最简单方法。希望对你也有用。

【参考方案4】：

我知道这是一岁了，但这仍然是一个可行的问题，我最近一直在摆弄。

大多数互联网广播电台都会为您提供下载类型选项，我选择 MP3 版本，然后从原始套接字读取信息并将其写入文件。诀窍是计算出下载速度与播放歌曲相比的速度，以便您可以在读/写大小上取得平衡。这将在您的缓冲区定义中。

现在您有了文件，只需将其留在驱动器（记录）上就可以了，但是大多数播放器会从文件中删除已播放的块，并在流停止时将文件从驱动器和内存中清除。

我使用了一些来自文件存档的代码 sn-ps 没有压缩应用程序来处理大量文件文件处理、播放、缓冲魔术。流程的流程非常相似。如果您编写一些 sudo 代码（我强烈推荐），您可以看到相似之处。

【讨论】：

真的有必要担心它在缓冲方面的下载速度吗？您可以简单地让您的代码在音频块到达时保存它们。这对于 Python 来说是非常自动的，使用类似的东西：req=urllib.request.Request(URL); resp=urllib.request.urlopen(req); headers=resp.getheaders(); with open(outfile, 'wb') as f: shutil.copyfileobj(resp, f)

【参考方案5】：

我只熟悉广播流的工作原理（您提到的 .pls 文件）：

您下载 pls 文件，它只是一个播放列表。它的格式相当简单，因为它只是一个指向真实流所在位置的文本文件。

您可以连接到该流，因为它只是 HTTP，流 MP3 或 AAC。为了您的使用，只需将您获得的每个字节保存到一个文件中，您将获得一个 MP3 或 AAC 文件，您可以将其传输到您的 mp3 播放器。

Shoutcast 有一个可选的附加功能：元数据。你可以找到它是如何工作的 here，但这并不是真正需要的。

如果您想要一个执行此操作的示例应用程序，请告诉我，我稍后会补充。

【讨论】：

谢谢，我去试试