2. HTTP固件下载

任务目标

实现一个基于HTTP协议的固件下载模块，能够从指定URL下载固件包到本地存储，并在下载过程中处理网络错误、HTTP错误状态码以及断点续传失败时的重试逻辑。本案例聚焦于下载链路的可靠性与错误恢复，不涉及固件校验、解压或写入Flash等后续步骤。

环境依赖

• Python 3.8+（仅用于本地模拟验证，实际嵌入式环境可移植到C/ESP-IDF等平台）
• 标准库：urllib.request、os、time、hashlib（用于验证文件完整性）

核心实现

import urllib.request
import urllib.error
import os
import time
import hashlib

class HTTPFirmwareDownloader:
    def __init__(self, url, dest_path, max_retries=3, timeout=10):
        self.url = url
        self.dest_path = dest_path
        self.max_retries = max_retries
        self.timeout = timeout
        self.retry_delay = 2  # 秒

    def _http_request(self, headers=None):
        """发起HTTP请求，返回响应对象"""
        req = urllib.request.Request(self.url, headers=headers or {})
        return urllib.request.urlopen(req, timeout=self.timeout)

    def download(self):
        """执行带重试的固件下载，返回(是否成功, 文件路径或错误信息)"""
        for attempt in range(1, self.max_retries + 1):
            try:
                # 检查本地是否已有部分下载（断点续传支持）
                existing_size = 0
                if os.path.exists(self.dest_path):
                    existing_size = os.path.getsize(self.dest_path)

                headers = {}
                if existing_size > 0:
                    headers['Range'] = f'bytes={existing_size}-'

                resp = self._http_request(headers)

                # 处理HTTP状态码
                if resp.status == 200:
                    # 服务器不支持断点续传，重新全量下载
                    mode = 'wb'
                    existing_size = 0
                elif resp.status == 206:
                    # 支持断点续传，追加写入
                    mode = 'ab'
                elif resp.status == 404:
                    return False, f"固件不存在(HTTP 404)，URL: {self.url}"
                elif resp.status == 403:
                    return False, "服务器拒绝访问(HTTP 403)"
                else:
                    return False, f"HTTP错误状态码: {resp.status}"

                # 流式写入文件
                with open(self.dest_path, mode) as f:
                    while True:
                        chunk = resp.read(4096)
                        if not chunk:
                            break
                        f.write(chunk)

                # 验证下载是否完整（通过Content-Length判断）
                total_size = resp.headers.get('Content-Length')
                if total_size and int(total_size) + existing_size != os.path.getsize(self.dest_path):
                    raise IOError("文件大小不匹配，下载不完整")

                return True, self.dest_path

            except urllib.error.HTTPError as e:
                # 服务器返回4xx/5xx错误
                if e.code == 416:  # Range请求超出范围，需重新下载
                    if os.path.exists(self.dest_path):
                        os.remove(self.dest_path)
                    continue
                return False, f"HTTP错误: {e.code}"
            except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
                # 网络层错误，进行重试
                if attempt < self.max_retries:
                    print(f"网络错误(第{attempt}次): {e}，{self.retry_delay}秒后重试...")
                    time.sleep(self.retry_delay)
                    continue
                return False, f"网络错误，重试{self.max_retries}次后仍失败: {e}"
            except IOError as e:
                # 文件写入错误，不重试（可能是磁盘问题）
                return False, f"文件写入错误: {e}"

        return False, "未知错误"

    def verify_sha256(self, expected_hash):
        """校验下载文件的SHA256哈希值"""
        if not os.path.exists(self.dest_path):
            return False
        sha256 = hashlib.sha256()
        with open(self.dest_path, 'rb') as f:
            for chunk in iter(lambda: f.read(4096), b''):
                sha256.update(chunk)
        return sha256.hexdigest() == expected_hash.lower()

# 使用示例
if __name__ == "__main__":
    # 配置下载参数
    FIRMWARE_URL = "http://example.com/firmware/v1.2.3.bin"
    LOCAL_PATH = "./firmware_download.bin"
    EXPECTED_SHA256 = "a3f5...（此处填写实际固件的SHA256值）"

    downloader = HTTPFirmwareDownloader(FIRMWARE_URL, LOCAL_PATH, max_retries=3)
    success, result = downloader.download()

    if success:
        print(f"固件下载成功: {result}")
        if EXPECTED_SHA256 and downloader.verify_sha256(EXPECTED_SHA256):
            print("SHA256校验通过")
        else:
            print("警告: SHA256校验失败或未提供预期哈希值")
    else:
        print(f"固件下载失败: {result}")

运行与验证

本地模拟测试：

# 1. 启动本地HTTP服务器模拟固件源
python -m http.server 8000 --directory ./firmware_dir

# 2. 将上述代码保存为 downloader.py，修改URL为：
#    FIRMWARE_URL = "http://127.0.0.1:8000/firmware.bin"

# 3. 运行下载
python downloader.py

验证通过条件：

• 下载成功后，本地文件大小与服务器端固件文件大小一致
• 使用sha256sum firmware_download.bin计算哈希，与服务器端原始文件哈希一致
• 模拟错误场景：在下载过程中断开网络（或使用无效URL），观察程序按预期重试3次后退出并返回错误信息
• 测试断点续传：下载到一半时终止程序，重新运行，观察日志确认发送了Range请求且服务器返回206

关键设计说明

• 错误分类处理：网络层错误（超时、连接断开）自动重试；HTTP 4xx错误（除416外）直接返回失败不重试，避免无效请求；416错误表示Range范围无效，删除本地残留后重新全量下载
• 断点续传：通过检测本地已有文件大小，自动添加Range头。服务器返回206时追加写入，返回200时覆盖重写
• 完整性验证：下载完成后通过Content-Length对比文件大小，并支持外部传入SHA256进行强校验