如何用Python向Icecast服务器发送音频流？现有代码无法播放

你的代码核心问题在于每次循环发起独立的HTTP PUT请求发送音频块，而Icecast要求通过单个持续的PUT连接传输完整的流数据。多次独立请求会导致服务器无法拼接出可播放的连续音频，这就是挂载点状态正常但无法收听的原因。此外send_audio方法中未初始化self.session，会触发AttributeError。

修正方案一：使用requests流式上传（适合文件流式传输） #

通过生成器将音频文件分块传入单个PUT请求，保持连接持续，让Icecast接收连续的流数据：

import requests
from base64 import b64encode

class IcecastClient:
    def __init__(self, host, port, mount, user, password, audio_info):
        self.host = host
        self.port = port
        self.mount = mount
        self.user = user
        self.password = password
        self.audio_info = audio_info
        self.stream_url = f"http://{host}:{port}{mount}"
        self.session = None

    def connect(self):
        # 初始化会话与认证头
        auth_header = b64encode(f"{self.user}:{self.password}".encode()).decode("ascii")
        headers = {
            'Authorization': f'Basic {auth_header}',
            'Content-Type': 'audio/mpeg',
            'Ice-Public': '1',
            'Ice-Name': 'Auralyra Stream',
            'Ice-Description': 'Streaming with Auralyra',
            'Ice-Genre': 'Various',
            'Ice-Audio-Info': self.audio_info
        }
        self.session = requests.Session()
        self.session.headers.update(headers)

    def stream_audio_file(self, file_path, chunk_size=4096):
        if not self.session:
            print("请先调用connect()建立连接")
            return

        # 生成器分块读取文件，实现流式上传
        def chunk_generator():
            with open(file_path, 'rb') as audio_file:
                while True:
                    chunk = audio_file.read(chunk_size)
                    if not chunk:
                        break
                    yield chunk

        try:
            # 单次PUT请求流式发送所有数据
            response = self.session.put(self.stream_url, data=chunk_generator(), stream=True)
            response.raise_for_status()
            print("流式传输完成")
        except requests.RequestException as e:
            print(f"传输失败: {e}")
        finally:
            self.session.close()

    def send_audio(self, audio_chunk):
        if not self.session:
            print("请先调用connect()建立连接")
            return
        # 注意：此方法仅适合在同一个PUT请求的流式传输中使用，若要实时发送需用底层HTTP连接（见方案二）
        print("警告：send_audio方法不适用于requests的单次流式请求，若需实时推流请使用方案二")

修正方案二：使用http.client底层连接（适合实时音频块推流） #

如果需要实时生成并发送音频块（而非预存文件），直接使用http.client维护持续的PUT连接，可多次发送音频块直到流结束：

import http.client
from base64 import b64encode

class IcecastClient:
    def __init__(self, host, port, mount, user, password, audio_info):
        self.host = host
        self.port = port
        self.mount = mount
        self.user = user
        self.password = password
        self.audio_info = audio_info
        self.conn = None

    def connect(self):
        # 建立HTTP连接并发送PUT请求头
        auth = b64encode(f"{self.user}:{self.password}".encode()).decode("ascii")
        self.conn = http.client.HTTPConnection(self.host, self.port)
        self.conn.putrequest('PUT', self.mount)
        self.conn.putheader('Authorization', f'Basic {auth}')
        self.conn.putheader('Content-Type', 'audio/mpeg')
        self.conn.putheader('Ice-Public', '1')
        self.conn.putheader('Ice-Name', 'Auralyra Stream')
        self.conn.putheader('Ice-Description', 'Streaming with Auralyra')
        self.conn.putheader('Ice-Genre', 'Various')
        self.conn.putheader('Ice-Audio-Info', self.audio_info)
        self.conn.endheaders()

    def stream_audio_file(self, file_path, chunk_size=4096):
        if not self.conn:
            print("请先调用connect()建立连接")
            return
        try:
            with open(file_path, 'rb') as audio_file:
                while True:
                    chunk = audio_file.read(chunk_size)
                    if not chunk:
                        break
                    self.conn.send(chunk)
            # 发送结束后获取服务器响应
            response = self.conn.getresponse()
            print(f"传输完成，服务器状态: {response.status} - {response.reason}")
        except Exception as e:
            print(f"传输失败: {e}")
        finally:
            self.conn.close()

    def send_audio(self, audio_chunk):
        if not self.conn:
            print("请先调用connect()建立连接")
            return
        try:
            self.conn.send(audio_chunk)
        except Exception as e:
            print(f"发送音频块失败: {e}")

关键说明 #

• Icecast的流传输要求单个PUT连接持续发送连续的音频数据，多次独立PUT请求会被服务器视为不同的流初始化，无法拼接成可播放的内容。
• 方案一适合预存音频文件的流式推送，方案二更适合实时生成音频块的场景（如麦克风采集、实时编码）。
• 确保音频文件格式与Content-Type一致（若为MP3则用audio/mpeg，若为OGG则用audio/ogg等），格式不匹配也会导致无法播放。

内容的提问来源于stack exchange，提问作者Polymood