背景痛点

最近在开发一个音频处理工具时，需要从网络下载AAC格式的音频文件。过程中遇到了几个典型问题：

• 部分服务器只支持流媒体传输，不提供完整文件下载
• 大文件下载时内存占用飙升导致程序崩溃
• 网络不稳定时经常需要从头开始重新下载

技术选型

对比了几个常用的HTTP库后，我最终选择了aiohttp，原因如下：

1. 异步支持：对于大量并发下载任务效率更高
2. 内存友好：原生支持流式传输(chunked)
3. 功能全面：自动处理连接池、重定向等

核心实现

分块下载实现

关键点在于HTTP Range头的使用：

headers = {'Range': f'bytes={start}-{end}'}
async with session.get(url, headers=headers) as resp:
    if resp.status == 206:  # Partial Content
        return await resp.read()

AAC帧识别

通过解析文件头确保数据有效性：

def is_valid_aac(data: bytes) -> bool:
    # ADTS头固定模式检查
    return len(data) > 7 and data[0] == 0xFF and (data[1] & 0xF0) == 0xF0

资源自动清理

使用contextlib确保文件句柄正确释放：

from contextlib import contextmanager

@contextmanager
def open_temp_file(path):
    try:
        f = open(path, 'wb')
        yield f
    finally:
        f.close()

完整代码示例

import aiohttp
import asyncio
from pathlib import Path

async def download_chunk(session, url, start, end, retries=3):
    for _ in range(retries):
        try:
            headers = {'Range': f'bytes={start}-{end}'}
            async with session.get(url, headers=headers) as resp:
                if resp.status == 206:
                    return await resp.read()
        except Exception as e:
            print(f"Retry {_+1} for chunk {start}-{end}: {str(e)}")
    return None

async def download_aac(url, output_path, chunk_size=1024*1024):
    async with aiohttp.ClientSession() as session:
        # 获取文件总大小
        async with session.head(url) as resp:
            total_size = int(resp.headers.get('content-length', 0))

        # 创建临时文件
        temp_path = f"{output_path}.temp"
        with open(temp_path, 'wb') as f:
            for start in range(0, total_size, chunk_size):
                end = min(start + chunk_size - 1, total_size - 1)
                data = await download_chunk(session, url, start, end)
                if data and is_valid_aac(data):
                    f.write(data)
                else:
                    raise ValueError("Invalid AAC data received")

        # 重命名临时文件
        Path(temp_path).rename(output_path)

生产环境考量

1. User-Agent规范：避免被服务器屏蔽

   headers = {'User-Agent': 'MyAudioTool/1.0'}

2. 超时设置：

   timeout = aiohttp.ClientTimeout(total=3600, sock_connect=30)

3. 存储检查：

   if not os.access(os.path.dirname(output_path), os.W_OK):
       raise PermissionError("No write permission")

常见问题解决

1. ID3标签混淆：

2. 解决方法：跳过文件头部的ID3标签区域

3. 采样率异常：

4. 检测方法：解析ADTS帧头的采样率索引

5. 文件损坏：

6. 预防措施：下载完成后做CRC校验

思考与延伸

目前方案主要针对简单的HTTP直连下载，如果要适配更复杂的HLS协议分片下载，应该如何调整架构？特别是遇到m3u8播放列表中的AAC分片时，如何保证下载顺序和合并的正确性？欢迎在评论区分享你的见解。