SHOUTcastストリームを保存してみる

SHOUTcastストリームはhttpで流れてくるので、curlとwgetで保存できる。これを応用して、1時間ごとにファイルをローテーション（1時間区切りでファイルを保存し続ける）するようなスクリプトを作ってみた。（有料記事ですが無料で最後まで読めます）

pythonのインストール

pythonで動作するので、Windowsの場合はpythonのインストールが必要。

Download Python

また、requestsモジュールが必要なので、pythonをインストールした後に追加する必要がある。

pip install requests

ソースコード

以下のコードを適当なフォルダに「shoutrec.py」等の分かりやすい名前で保存する。この時、main()の中の以下の2つの変数を、自身の環境に合わせて書き換える。
stream_url
SHOUTcastストリームのURL　http://url:8000/;　等
save_dir
ファイルの保存場所　c:\temp　等

import requests
import os
from datetime import datetime, timedelta
import time

class StreamRecorder:
    def __init__(self, stream_url, save_dir):
        """
        初期化
        :param stream_url: ShoutcastストリームのURL
        :param save_dir: 録音ファイルの保存先ディレクトリ
        """
        self.stream_url = stream_url
        self.save_dir = save_dir
        self.ensure_save_dir()
        
    def ensure_save_dir(self):
        """保存先ディレクトリが存在しない場合は作成"""
        if not os.path.exists(self.save_dir):
            os.makedirs(self.save_dir)
            
    def get_wait_time_until_next_hour(self):
        """次の時間までの待機秒数を計算"""
        now = datetime.now()
        next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
        return (next_hour - now).total_seconds()
    
    def get_previous_hour_filename(self):
        """前の時間の録音ファイル名を生成"""
        previous_hour = datetime.now() - timedelta(hours=1)
        return previous_hour.strftime("%Y%m%d-%H%M") + ".mp3"
    
    def record_stream(self):
        """ストリームの録音を開始"""
        try:
            while True:
                # 一時ファイルと最終的な保存ファイルのパスを設定
                temp_file = os.path.join(self.save_dir, "current.mp3")
                target_file = os.path.join(self.save_dir, self.get_previous_hour_filename())
                
                # ストリームを受信して一時ファイルに保存
                with requests.get(self.stream_url, stream=True) as response:
                    response.raise_for_status()
                    
                    with open(temp_file, 'wb') as f:
                        # 次の時間までの待機時間を計算
                        wait_time = self.get_wait_time_until_next_hour()
                        end_time = time.time() + wait_time
                        
                        # ストリームを読み込んでファイルに書き込み
                        for chunk in response.iter_content(chunk_size=8192):
                            if time.time() >= end_time:
                                break
                            if chunk:
                                f.write(chunk)
                                f.flush()
                
                # ファイル名を変更
                if os.path.exists(temp_file):
                    if os.path.exists(target_file):
                        os.remove(target_file)
                    os.rename(temp_file, target_file)
                
                # 少し待機してから次のループを開始
                time.sleep(2)
                
        except requests.RequestException as e:
            print(f"ストリーム接続エラー: {e}")
        except Exception as e:
            print(f"エラーが発生しました: {e}")
        finally:
            # 一時ファイルが残っている場合は削除
            if os.path.exists(temp_file):
                os.remove(temp_file)

def main():
    # 設定
    stream_url = "http://your-shoutcast-stream-url:port"
    save_dir = r"C:\Recording"  # Windowsの場合
    
    # レコーダーを初期化して開始
    recorder = StreamRecorder(stream_url, save_dir)
    recorder.record_stream()

if __name__ == "__main__":
    main()

実行

powershellまたはコマンドプロンプトでスクリプトが保存されているフォルダに移動して以下のコマンドを実行する。

python shoutrec.py

特に何も表示されないが、バックグラウンドで保存が始まっており、save_dirに指定したフォルダにファイルが保存されてることがわかる。

保存されている一時ファイル

current.mp3という名前のファイルは、毎正時に、録音を開始した日付時刻のファイル名に書き換えられ、新たなcurrent.mp3を生成し保存し続けるようになっている。

録音を停止する際は[CTRL]＋[C]で停止させる。今回公開しているコードではcurrentは消去されてしまう仕様になっているが、そのうち「止めるまでに保存した部分は保存」できるように改良する予定。

まとめ

stream_urlを変更すれば、同様の仕組みで配信されているインターネト配信を「保存」することが可能になると思うので、今後アップデートしていきたい。