用 Python 可靠下载文件：流式读取、大小上限与 SHA-256 校验

python3 --version
mkdir stream-download-demo
cd stream-download-demo
curl -fS https://www.bobobk.com/downloads/stream-download/download_file.py -o download_file.py
curl -fS https://www.bobobk.com/downloads/stream-download/selftest.py -o selftest.py
python3 download_file.py --help

"""Download one known HTTP(S) file and verify SHA-256 before replacing it."""
import argparse
import hashlib
from http.client import HTTPException
import json
import math
import os
from pathlib import Path
import re
import sys
import tempfile
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def download(url, target, expected_sha256, max_bytes, timeout):
    parts = urlsplit(url)
    if parts.scheme not in {"http", "https"} or not parts.hostname:
        raise ValueError("URL must use http:// or https:// with a host")
    target = Path(target).absolute()
    temporary = None
    request = Request(url, headers={"Accept-Encoding": "identity"})
    try:
        with build_opener(NoRedirect()).open(request, timeout=timeout) as response:
            if response.status != 200:
                raise ValueError(f"expected HTTP 200, got {response.status}")
            if response.headers.get("Content-Encoding", "identity").lower() != "identity":
                raise ValueError("unexpected Content-Encoding")
            lengths = response.headers.get_all("Content-Length", [])
            if lengths and (len(lengths) != 1 or not re.fullmatch(r"[0-9]+", lengths[0])):
                raise ValueError("invalid Content-Length")
            declared = int(lengths[0]) if lengths else None
            if declared is not None and declared > max_bytes:
                raise ValueError("Content-Length exceeds --max-bytes")
            digest = hashlib.sha256()
            total = 0
            with tempfile.NamedTemporaryFile(
                dir=target.parent, prefix=f".{target.name}.",
                suffix=".part", delete=False,
            ) as output:
                temporary = Path(output.name)
                while True:
                    chunk = response.read(min(65536, max_bytes - total + 1))
                    if not chunk:
                        break
                    total += len(chunk)
                    if total > max_bytes:
                        raise ValueError("body exceeds --max-bytes")
                    output.write(chunk)
                    digest.update(chunk)
                if declared is not None and total != declared:
                    raise ValueError("Content-Length mismatch (truncated response)")
                actual = digest.hexdigest()
                if actual != expected_sha256:
                    raise ValueError("SHA-256 mismatch")
                output.flush()
                os.fsync(output.fileno())
        os.replace(temporary, target)
        temporary = None
        return {"file": str(target), "bytes": total, "sha256": actual}
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("url")
    parser.add_argument("output", type=Path)
    parser.add_argument("--sha256", required=True, help="trusted 64-digit SHA-256")
    parser.add_argument("--max-bytes", type=int, required=True)
    parser.add_argument("--timeout", type=float, default=10.0,
                        help="blocking socket operation timeout, seconds (default: 10)")
    args = parser.parse_args()
    if not re.fullmatch(r"[0-9a-fA-F]{64}", args.sha256):
        parser.error("--sha256 must have exactly 64 hexadecimal digits")
    if args.max_bytes < 1:
        parser.error("--max-bytes must be positive")
    if not math.isfinite(args.timeout) or args.timeout <= 0:
        parser.error("--timeout must be finite and positive")
    try:
        result = download(args.url, args.output, args.sha256.lower(),
                          args.max_bytes, args.timeout)
    except (OSError, ValueError, HTTPException) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

mkdir source
printf 'id,value\n1,42\n' > source/data.csv
python3 -m http.server 8765 --bind 127.0.0.1 --directory source

export no_proxy=127.0.0.1,localhost
export NO_PROXY=127.0.0.1,localhost
download_sha=$(python3 -c 'import hashlib; from pathlib import Path; print(hashlib.sha256(Path("source/data.csv").read_bytes()).hexdigest())')
python3 download_file.py http://127.0.0.1:8765/data.csv result.csv \
  --sha256 "$download_sha" --max-bytes 1048576 --timeout 3
python3 - <<'PY_CHECK'
from pathlib import Path
assert Path("result.csv").read_bytes() == b"id,value\n1,42\n"
assert not list(Path(".").glob(".result.csv.*.part"))
print("PASS: exact contents; no partial file")
PY_CHECK

python3 selftest.py

PASS: 2 successes, 8 failures preserve old output; no partial files
PASS: failed first download leaves no output; 3 invalid arguments rejected

python3 download_file.py http://127.0.0.1:8765/data.csv result.csv \
  --sha256 "$download_sha" --max-bytes 1048576 --timeout 3 \
  && python3 -c 'from pathlib import Path; print(Path("result.csv").read_text())'