Downloads

download

File download utilities with aria2c acceleration and httpx fallback.

Replaces the PowerShell Save-File function from UmeAiRTUtils.psm1. Adds SHA256 checksum verification (a security improvement over the original).

verify_checksum(file_path, expected_sha256)

Verify the SHA256 checksum of a downloaded file.

Parameters:

Name	Type	Description	Default
file_path	Path	Path to the file to verify.	required
expected_sha256	str	Expected SHA256 hex digest (lowercase).	required

Returns:

Type	Description
bool	True if checksum matches, False otherwise.

Source code in src/utils/download.py

def verify_checksum(file_path: Path, expected_sha256: str) -> bool:
    """
    Verify the SHA256 checksum of a downloaded file.

    Args:
        file_path: Path to the file to verify.
        expected_sha256: Expected SHA256 hex digest (lowercase).

    Returns:
        True if checksum matches, False otherwise.
    """
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            sha256.update(chunk)
    return sha256.hexdigest().lower() == expected_sha256.lower()

download_file(url, dest, *, checksum=None, force=False, quiet=True, aria2c_hint=None, log=None)

Download a file from one or more URLs to a destination path.

When url is a list, each URL is tried in order — the first successful download wins. This enables transparent HuggingFace → ModelScope failover.

Tries aria2c first for speed, then falls back to httpx. Optionally verifies SHA256 checksum after download.

Parameters:

Name	Type	Description	Default
url	str | list[str]	Source URL(s) to download from. Pass a list for multi-source fallback (e.g. [hf_url, ms_url]).	required
dest	Path | str	Destination file path.	required
checksum	str | None	Optional SHA256 hex digest for verification.	None
force	bool	If True, re-download even if file exists.	False
quiet	bool	If True (default), suppress aria2c console output.	True
aria2c_hint	Path | None	Optional directory where aria2c may be found (e.g. install_path / "scripts" / "aria2").	None

Returns:

Type	Description
Path	Path to the downloaded file.

Raises:

Type	Description
RuntimeError	If all URLs fail or checksum doesn't match.

Source code in src/utils/download.py

def download_file(
    url: str | list[str],
    dest: Path | str,
    *,
    checksum: str | None = None,
    force: bool = False,
    quiet: bool = True,
    aria2c_hint: Path | None = None,
    log: InstallerLogger | None = None,
) -> Path:
    """
    Download a file from one or more URLs to a destination path.

    When *url* is a list, each URL is tried in order — the first successful
    download wins.  This enables transparent HuggingFace → ModelScope failover.

    Tries aria2c first for speed, then falls back to httpx.
    Optionally verifies SHA256 checksum after download.

    Args:
        url: Source URL(s) to download from. Pass a list for multi-source
            fallback (e.g. ``[hf_url, ms_url]``).
        dest: Destination file path.
        checksum: Optional SHA256 hex digest for verification.
        force: If True, re-download even if file exists.
        quiet: If True (default), suppress aria2c console output.
        aria2c_hint: Optional directory where aria2c may be found
            (e.g. ``install_path / "scripts" / "aria2"``).

    Returns:
        Path to the downloaded file.

    Raises:
        RuntimeError: If all URLs fail or checksum doesn't match.
    """
    dest = Path(dest)
    if log is None:
        log = get_logger()

    # Normalise to list
    urls: list[str] = [url] if isinstance(url, str) else list(url)
    if not urls:
        raise RuntimeError("download_file: no URL provided.")

    # Skip if already exists (and no checksum to verify or checksum matches)
    aria2_control = dest.with_suffix(dest.suffix + ".aria2")
    if dest.exists() and not force:
        # If an aria2 control file exists, the previous download was interrupted
        if aria2_control.exists():
            log.warning(f"Incomplete download detected for '{dest.name}', resuming...", level=2)
        elif checksum and not verify_checksum(dest, checksum):
            log.warning(f"Checksum mismatch for existing '{dest.name}', re-downloading...", level=2)
        else:
            log.sub(f"'{dest.name}' already exists, skipping.", style="success")
            return dest

    dest.parent.mkdir(parents=True, exist_ok=True)

    # Try each URL in order
    last_error: Exception | None = None
    for i, source_url in enumerate(urls):
        source_label = source_url.split("/")[2] if "/" in source_url else "unknown"
        if i > 0:
            log.info(f"Trying fallback source ({source_label})...")

        log.sub(f"Downloading \"{source_url.split('/')[-1]}\"", style="debug")

        try:
            # Try aria2c first
            aria2c = _find_aria2c(aria2c_hint)
            downloaded = False

            if aria2c:
                downloaded = _download_with_aria2c(source_url, dest, aria2c, quiet=quiet)
                if downloaded:
                    log.info("Download successful (aria2c).")

            # Fallback to httpx
            if not downloaded:
                if aria2c:
                    log.info("aria2c failed, falling back to httpx...")
                _download_with_httpx(source_url, dest)
                log.info("Download successful (httpx).")

            # If we got here, download succeeded — break out
            last_error = None
            break

        except (httpx.HTTPError, OSError, RuntimeError) as e:
            last_error = e
            log.info(f"Source {source_label} failed: {e}")
            # Clean up partial file before trying next source
            dest.unlink(missing_ok=True)
            continue

    if last_error is not None:
        raise RuntimeError(
            f"Download failed for '{dest.name}' from all {len(urls)} source(s): {last_error}"
        ) from last_error

    # Verify checksum
    if checksum:
        if not verify_checksum(dest, checksum):
            dest.unlink(missing_ok=True)
            raise RuntimeError(
                f"Checksum verification failed for '{dest.name}'. "
                f"Expected: {checksum[:16]}... File has been deleted."
            )
        log.info("Checksum verified ✓")

    return dest