如何使用aiohttp获取Icecast流媒体的元数据

错误原因 #

• 核心逻辑遗漏：ICY流的结构为【icy-metaint长度的音频数据 + 1字节元数据长度 + N字节元数据】循环，原requests代码会先读取指定长度的音频块再读取元数据相关内容，你的aiohttp版本直接跳过了音频块读取步骤，读取的meta_byte实际是音频流的开头字节，后续逻辑完全错位自然无法得到正确元数据。
• 自动解压干扰：aiohttp默认开启响应自动解压缩功能，如果流返回了gzip/deflate格式的压缩内容，解压操作会破坏原始ICY流的字节排布，导致icy-metaint指定的偏移位置失效。
• 编码匹配错误：ICY元数据默认使用latin-1编码而非UTF-8，直接用UTF-8解码会抛出invalid start byte错误。

修复后代码 #

import asyncio
import aiohttp

async def main():
    async with aiohttp.ClientSession() as session:
        # 关闭自动解压缩，保证原始流结构不变
        async with session.get(
            "http://myurl.com/mystream",
            headers={'Icy-MetaData': "1"},
            auto_decompress=False
        ) as response:
            headers = response.headers
            meta_int = headers.get('icy-metaint')
            audio_length = int(meta_int)
            # 加回原逻辑：先读取对应长度的音频数据
            await response.content.read(audio_length)
            meta_byte = await response.content.read(1)
            if meta_byte:
                meta_length = ord(meta_byte) * 16
                meta_data = await response.content.read(meta_length)
                # 用latin-1解码避免报错
                print(meta_data.decode('latin-1'))

# Python 3.7+ 推荐使用asyncio.run，兼容旧版本可换回原run_until_complete写法
asyncio.run(main())

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