chuanqi/cpp_reversed/http_downloader.py

#!/usr/bin/env python3
"""
996传奇客户端 httpDownloader 接口还原 + 等价 Python 实现
来源: 二进制 httpDownloader:AsyncDownloadEx 等 17 个 tolua 方法
      + 公开 curl_easy_setopt 文档对照

职责: libcurl 异步下载封装, 用于热更/资源下载/补丁包
典型场景:
  1. 启动期: 下载 manifest.txt / project.manifest, 对比本地版本
  2. 增量期: 下载 .zip 补丁包 (签名校验)
  3. 资源期: 下载 .png/.plist/.map 等 (单文件)

与原二进制差异 (Python 等价实现, 非直接对应):
  - 异步 → 同步 + threading (Python 3.13 之前原生 asyncio 不普及)
  - 进度回调: callback(downloaded, total) → 用闭包实现
  - 限速: 模拟 (token bucket)
"""
from __future__ import annotations
import os
import sys
import time
import hashlib
import threading
import urllib.request
import urllib.error
from typing import Optional, Callable
from dataclasses import dataclass, field


# ============== 二进制接口契约 (17 个 tolua 方法) ==============

# 单例
#   httpDownloader:Inst() -> httpDownloader*
#   (Python 等价: 模块级全局 state)

# 配置 (setter)
#   SetDownloadURL(url)        - 基础 URL
#   SetDownloadDir(dir)        - 保存目录
#   SetTimeOut(sec)            - 超时
#   SetSpeedLimit(bytes_per_sec) - 限速
#   SetResourceVer(ver)        - 资源版本号
#   EnableDownload(flag)       - 启用/禁用开关
#   SetDownloadProgress(cb)    - 进度回调 Lua 函数

# 状态 (getter)
#   GetLastCurlCode()    - 最近一次 curl 错误码
#   GetLastHttpCode()    - 最近一次 HTTP 状态码
#   IsLastDownloadOK()   - 是否成功
#   IsEnableDownaload()  - 启用状态 (typo 是原二进制写法)

# 操作
#   AsyncDownloadEasy(url)        - 简单下载 (默认设置)
#   AsyncDownloadEx(url, save_as) - 完整下载 (指定保存名)
#   CleanupAllDownload()          - 取消所有下载
#   Destroy()                      - 销毁单例

# ============== 数据结构 ==============

@dataclass
class DownloadTask:
    url: str
    save_as: str                     # 相对 SetDownloadDir 的路径
    total_bytes: int = 0
    downloaded: int = 0
    started_at: float = 0.0
    finished_at: float = 0.0
    http_code: int = 0
    error: Optional[str] = None
    sha1: str = ""                   # 期望 sha1 (manifest 提供)


@dataclass
class DownloadState:
    """httpDownloader 全局状态 (Python 等价)"""
    base_url: str = ""
    download_dir: str = ""
    timeout: float = 30.0
    speed_limit_bps: int = 0         # 0 = 无限制
    resource_ver: str = ""
    enabled: bool = True
    last_http_code: int = 0
    last_curl_code: int = 0
    last_ok: bool = False
    progress_cb: Optional[Callable] = None
    _tasks: dict = field(default_factory=dict)
    _lock: threading.Lock = field(default_factory=threading.Lock)


# ============== 单例 ==============

_STATE: Optional[DownloadState] = None


def inst() -> DownloadState:
    """httpDownloader:Inst() 等价"""
    global _STATE
    if _STATE is None:
        _STATE = DownloadState()
    return _STATE


def destroy() -> None:
    """httpDownloader:Destroy() 等价"""
    global _STATE
    _STATE = None


# ============== Setter ==============

def set_download_url(url: str) -> None: inst().base_url = url
def set_download_dir(d: str) -> None: inst().download_dir = d
def set_timeout(sec: float) -> None: inst().timeout = sec
def set_speed_limit(bps: int) -> None: inst().speed_limit_bps = bps
def set_resource_ver(ver: str) -> None: inst().resource_ver = ver
def enable_download(flag: bool) -> None: inst().enabled = flag
def set_download_progress(cb: Callable) -> None: inst().progress_cb = cb


# ============== Getter ==============

def get_last_http_code() -> int: return inst().last_http_code
def get_last_curl_code() -> int: return inst().last_curl_code
def is_last_download_ok() -> bool: return inst().last_ok
def is_enable_download() -> bool: return inst().enabled


# ============== 核心: 同步下载 (AsyncDownloadEx 的同步等价) ==============

def _check_disk_space(save_path: str, expected_size: int) -> None:
    """下载前检查磁盘空间"""
    need_bytes = expected_size + 4096
    free = os.statvfs(os.path.dirname(save_path) or ".").f_bavail * os.statvfs(
        os.path.dirname(save_path) or ".").f_frsize
    if free < need_bytes:
        raise IOError(f"磁盘空间不足: need={need_bytes} free={free}")


def _throttle(downloaded: int, started_at: float, bps: int) -> None:
    """限速: 令牌桶简化版, 让已下载量/已用时间 >= bps/1"""
    if bps <= 0:
        return
    elapsed = time.time() - started_at
    expected_min = downloaded / bps
    if expected_min > elapsed:
        time.sleep(expected_min - elapsed)


def _download_blocking(task: DownloadTask, state: DownloadState) -> bool:
    """单次同步下载 (AsyncDownloadEx 的同步等价实现)"""
    full_url = task.url if task.url.startswith("http") else state.base_url + task.url
    save_path = os.path.join(state.download_dir, task.save_as)
    os.makedirs(os.path.dirname(save_path), exist_ok=True)

    task.started_at = time.time()
    state.last_http_code = 0
    state.last_curl_code = 0
    state.last_ok = False

    try:
        req = urllib.request.Request(full_url)
        with urllib.request.urlopen(req, timeout=state.timeout) as resp:
            task.http_code = resp.getcode()
            state.last_http_code = task.http_code
            if task.http_code != 200:
                task.error = f"HTTP {task.http_code}"
                return False

            total_hdr = resp.headers.get("Content-Length")
            if total_hdr:
                task.total_bytes = int(total_hdr)
                _check_disk_space(save_path, task.total_bytes)

            with open(save_path, "wb") as f:
                while True:
                    chunk = resp.read(8192)
                    if not chunk:
                        break
                    f.write(chunk)
                    task.downloaded += len(chunk)
                    _throttle(task.downloaded, task.started_at, state.speed_limit_bps)
                    if state.progress_cb:
                        try:
                            state.progress_cb(task.downloaded, task.total_bytes)
                        except Exception:
                            pass  # 回调异常不阻断下载

        # sha1 校验
        if task.sha1:
            actual = hashlib.sha1(open(save_path, "rb").read()).hexdigest()
            if actual != task.sha1:
                task.error = f"sha1 mismatch: {actual[:8]}... != {task.sha1[:8]}..."
                return False

        task.finished_at = time.time()
        state.last_ok = True
        return True
    except urllib.error.HTTPError as e:
        task.http_code = e.code
        state.last_http_code = e.code
        task.error = f"HTTPError: {e.code} {e.reason}"
        return False
    except urllib.error.URLError as e:
        task.error = f"URLError: {e.reason}"
        return False
    except Exception as e:
        task.error = f"{type(e).__name__}: {e}"
        return False


def async_download_ex(url: str, save_as: str, sha1: str = "") -> DownloadTask:
    """
    httpDownloader:AsyncDownloadEx(url, save_as) 的 Python 等价
    :param url: 完整 URL 或相对于 base_url 的相对路径
    :param save_as: 相对 download_dir 的保存路径
    :param sha1: 期望 sha1 (空=不校验)
    :return: DownloadTask (可查 finished_at / error / downloaded)
    """
    state = inst()
    task = DownloadTask(url=url, save_as=save_as, sha1=sha1)
    if not state.enabled:
        task.error = "downloader disabled"
        return task

    with state._lock:
        state._tasks[save_as] = task
    _download_blocking(task, state)
    return task


def async_download_easy(url: str) -> DownloadTask:
    """AsyncDownloadEasy(url) 等价: 用 URL basename 作为 save_as"""
    save_as = url.split("?")[0].split("/")[-1] or "download.bin"
    return async_download_ex(url, save_as)


def cleanup_all_download() -> None:
    """CleanupAllDownload 等价: 清空任务列表 (本实现是同步, 无活跃任务)"""
    state = inst()
    with state._lock:
        state._tasks.clear()


# ============== 高级: 完整热更流程 ==============

def download_manifest(manifest_url: str) -> dict:
    """
    下载并解析 manifest 文件
    manifest 格式 (与 cocos2d-x AssetsManagerEx 兼容):
      version: 1.0.0
      res_version: 1.0.1
      file_list:
        - path: scene/map/0.map
          md5: abc123...
          size: 6860052
    :return: dict
    """
    import json
    state = inst()
    state.base_url = manifest_url.rsplit("/", 1)[0] + "/"
    task = async_download_ex(manifest_url, "manifest.json")
    if not state.last_ok:
        raise IOError(f"manifest download failed: {task.error}")
    with open(os.path.join(state.download_dir, "manifest.json")) as f:
        return json.load(f)


def download_update_files(manifest: dict, only_changed: bool = True) -> list:
    """
    根据 manifest 下载需要更新的文件
    :param manifest: download_manifest() 返回值
    :param only_changed: True=对比本地 sha1, 只下变更; False=强制全量
    :return: 成功下载的 DownloadTask 列表
    """
    tasks = []
    for entry in manifest.get("file_list", []):
        path = entry["path"]
        expected_sha1 = entry.get("md5", "")
        local_path = os.path.join(inst().download_dir, path)
        if only_changed and os.path.exists(local_path):
            actual = hashlib.sha1(open(local_path, "rb").read()).hexdigest()
            if actual == expected_sha1:
                continue  # 本地一致, 跳过
        t = async_download_ex(path, path, expected_sha1)
        tasks.append(t)
    return tasks


# ============== 单元自测 ==============

if __name__ == "__main__":
    import sys
    # 设个 mock 状态演示
    set_download_dir("/tmp/httpdl_test")
    set_timeout(10)

    print("=== httpDownloader 接口自测 (mock) ===")
    print(f"Inst: {inst() is not None}")
    print(f"Inst 二次: {inst() is inst()} (应 True, 单例)")
    set_download_url("https://example.com/")
    print(f"Base URL: {inst().base_url}")
    print(f"Enabled: {is_enable_download()}")
    print(f"Last OK: {is_last_download_ok()} (应 False)")

    # 测一次真实下载 (小文件)
    print("\n=== 真实下载测试 (https://httpbin.org/bytes/1024) ===")
    set_download_dir("/tmp/httpdl_test")
    t = async_download_ex("https://httpbin.org/bytes/1024", "test.bin")
    print(f"  task.url = {t.url}")
    print(f"  task.save_as = {t.save_as}")
    print(f"  task.downloaded = {t.downloaded}")
    print(f"  task.total_bytes = {t.total_bytes}")
    print(f"  task.http_code = {t.http_code}")
    print(f"  task.error = {t.error}")
    print(f"  is_last_download_ok = {is_last_download_ok()}")
    if t.error:
        print(f"  [warn] 网络请求失败, 这是预期的(无网环境)")
    else:
        # 验证文件存在
        path = os.path.join("/tmp/httpdl_test", "test.bin")
        if os.path.exists(path):
            print(f"  saved {os.path.getsize(path)}B to {path}")