Python实现简易Icecast客户端连接失败问题求助

核心问题分析 #

• SSL兼容性冲突：urllib3 v2不再支持LibreSSL 2.8.3，HTTPS连接握手时会触发底层错误，直接导致连接中断。
• 音频格式与请求头不匹配：代码导出OGG格式音频，但请求头声明audio/opus，服务器无法识别该格式，直接断开连接。
• 音频推流逻辑错误：仅读取一次音频数据就进入循环重复发送，既没有持续采集新的麦克风输入，也不符合Icecast持续推流的要求。
• 头部参数冗余与矛盾：同时设置session.auth和Authorization头部，可能引发服务器认证冲突；ice-audio-info声明双声道，但实际采集的是单声道，参数不一致。
• URL端口存疑：需确认Caster.fm服务器的HTTPS推送端口，默认HTTPS端口为443，非标准端口需确保服务器确实监听该端口。

修复方案 #

1. 解决SSL兼容性问题 #

降级urllib3到兼容LibreSSL的版本：

pip install urllib3==1.26.15

2. 修正音频格式与头部匹配 #

将请求头Content-Type改为audio/ogg，与代码中导出的音频格式保持一致；若需使用Opus格式，需确保pydub依赖的ffmpeg支持该编码。

3. 修复音频采集与推流逻辑 #

在循环内持续读取新的音频数据，改用长连接推流（Icecast要求持续的单连接推送，而非反复创建新连接）。

4. 清理冗余头部参数 #

只保留一种认证方式，修正音频参数与实际采集配置一致。

修复后的完整代码 #

import pyaudio
from pydub import AudioSegment
import requests
import base64

# 音频采集配置：保持前后一致
FORMAT = pyaudio.paInt16
CHANNELS = 1
SAMPLERATE = 44100
CHUNK = 1024

# 初始化麦克风输入流
audio = pyaudio.PyAudio()
stream = audio.open(format=FORMAT, 
                    channels=CHANNELS,
                    rate=SAMPLERATE, 
                    input=True,
                    frames_per_buffer=CHUNK)

# Icecast服务器配置
ICECAST_HOST = "sapircast.caster.fm"
ICECAST_PORT = 15754  # 需确认该端口为HTTPS端口，若为HTTP则改用http://前缀
MOUNT_POINT = "/iCVJp"
USERNAME = "source"
PASSWORD = "m5auo6W0vP"

# 认证字符串生成
auth_str = base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode("utf-8")
stream_url = f"https://{ICECAST_HOST}:{ICECAST_PORT}{MOUNT_POINT}"

# 修正后的请求头：匹配实际音频参数，移除冗余配置
ice_headers = {
    "Authorization": f"Basic {auth_str}",
    "Content-Type": "audio/ogg",  # 与导出格式严格一致
    "icy-pub": "1",
    "icy-genre": "General",
    "ice-name": "Test Stream",
    "ice-description": "My Python Icecast Stream",
    "ice-url": f"https://{ICECAST_HOST}",
    "ice-bitrate": "96",
    "ice-audio-info": f"ice-channels={CHANNELS};ice-bitrate=96;ice-samplerate={SAMPLERATE};",
}

# 建立长连接并持续推流
session = requests.Session()
try:
    with session.put(stream_url, headers=ice_headers, stream=True) as response:
        response.raise_for_status()
        print("连接成功，开始推流...")
        while True:
            # 持续读取新的麦克风音频数据
            raw_audio_data = stream.read(CHUNK)
            # 转换为OGG格式
            audio_segment = AudioSegment(
                data=raw_audio_data,
                sample_width=audio.get_sample_size(FORMAT),
                frame_rate=SAMPLERATE,
                channels=CHANNELS
            )
            encoded_audio = audio_segment.export(format="ogg").read()
            # 向长连接写入音频数据
            response.connection.send(encoded_audio)
except requests.exceptions.RequestException as e:
    print(f"推流错误: {e}")
finally:
    # 清理资源
    stream.stop_stream()
    stream.close()
    audio.terminate()

额外注意事项 #

• 安装ffmpeg：pydub依赖ffmpeg进行音频格式转换，macOS可通过brew install ffmpeg安装，Windows需下载并配置环境变量。
• 验证服务器端口：登录Caster.fm后台确认推送端口的协议（HTTP/HTTPS），避免协议与端口不匹配。
• 测试音频采集：可先将采集的音频保存为本地文件，确认麦克风输入正常后再尝试推流。

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