2010-11-22 85 views
7

我正在尋找一個python片段來讀取互聯網廣播流(.asx,.pls等)並將其保存到文件中。在python中記錄流式並保存網絡收音機

最終的項目是cron'ed腳本,它會記錄一兩個小時的網絡收音機,然後將其傳輸到我的手機以便在通勤期間播放。 (3g是我通勤時的點狀)

任何snippits或指針是受歡迎的。

回答

4

所以經過修補和玩它後,我發現Streamripper工作最好。這是命令我使用

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

我只熟悉Shoutcast的流媒體是如何工作的(這將是.PLS文件,你提):

你下載的請文件,它只是一個播放列表。它的格式非常簡單,因爲它只是一個指向實際流的位置的文本文件。

您可以連接到該流,因爲它只是HTTP,可以流式傳輸MP3或AAC。爲了您的使用,只需將保存的每個字節保存到一個文件中,即可獲得一個MP3或AAC文件,您可以將其傳輸到您的MP3播放器。

Shoutcast有一個額外的選項:元數據。你可以找到如何工作here,但並不是真的需要。

如果你想要一個這樣的示例應用程序,請告訴我,稍後我會做出一些事情。

+0

謝謝,我會試一試 – madmaze 2010-11-23 01:01:33

2

我知道這是一歲,但是這仍然是一個可行的問題,這是我最近一直襬弄用。

大多數網絡電臺會給你一個下載類型的選項,我選擇MP3版本,然後從原始套接字讀取信息並將其寫入文件。訣竅是確定您的下載與播放歌曲的速度相比有多快,以便您可以在讀/寫大小上創建一個平衡點。這將在你的緩衝區def中。

現在你已經有了這個文件,把它放在你的驅動器(記錄)上是可以的,但是大多數播放器會從文件中刪除已經播放的塊,並且在流停止時將文件從驅動器和文件中清除。

我已經使用了一些沒有壓縮應用程序的文件歸檔代碼片段來處理大量的文件處理,播放,緩衝魔法。流程的流程非常相似。如果你寫了一些sudo代碼(我強烈推薦),你可以看到相似之處。

+0

是否真的需要擔心它下載的速度有多慢?你可以簡單地讓你的代碼在他們到達時保存音頻塊。這是Python自動使用類似於: `req = urllib.request.Request(URL); RESP = urllib.request.urlopen(REQ);標題= resp.getheaders();打開(outfile,'wb')作爲f:shutil.copyfileobj(resp,f)` – dingles 2016-12-27 02:15:11

3

以下對我使用請求庫來處理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 

,直到你ctrl+C打斷它那將流保存到文件stream.mp3

+0

它確實有幫助,但我該如何丟棄在文件中重複的塊? – 2016-06-23 18:09:33

2

如果您發現在Python 3中的請求或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函數所需的原始響應的相關位。最後,調用原始文件,並將其中的值傳回給調用者,其他所有內容都會像平常一樣繼續。

我發現這是解決導致錯誤的狀態消息問題的最簡單方法。希望它對你也有用。