How to Download Images from URLs with Python

python3 -m venv .venv
source .venv/bin/activate
pip install requests tenacity

export PROXIESAPI_PROXY_URL="http://USER:PASS@gateway.proxiesapi.com:PORT"

from __future__ import annotations

import hashlib
import mimetypes
import os
import re
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse

import requests
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential

TIMEOUT = (10, 60)
PROXIESAPI_PROXY_URL = os.getenv("PROXIESAPI_PROXY_URL")

session = requests.Session()

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/123.0.0.0 Safari/537.36"
    ),
    "Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8",
}


def proxies_dict():
    if not PROXIESAPI_PROXY_URL:
        return None
    return {"http": PROXIESAPI_PROXY_URL, "https": PROXIESAPI_PROXY_URL}


def safe_filename(s: str, max_len: int = 120) -> str:
    s = re.sub(r"[^a-zA-Z0-9._-]+", "-", s).strip("-")
    return s[:max_len] if len(s) > max_len else s


def guess_ext(content_type: str | None, url: str) -> str:
    if content_type:
        ct = content_type.split(";")[0].strip().lower()
        ext = mimetypes.guess_extension(ct) or ""
        if ext in {".jpe", ".jpeg", ".jpg", ".png", ".webp", ".gif", ".bmp", ".tif", ".tiff"}:
            return ".jpg" if ext == ".jpeg" else ext

    path = urlparse(url).path
    _, ext = os.path.splitext(path)
    ext = ext.lower()
    if ext in {".jpg", ".jpeg", ".png", ".webp", ".gif"}:
        return ".jpg" if ext == ".jpeg" else ext

    return ".jpg"


def is_probably_image(content_type: str | None) -> bool:
    return bool(content_type and content_type.lower().startswith("image/"))


_lock = threading.Lock()


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=20),
    retry=retry_if_exception_type((requests.RequestException,)),
)
def download_one(url: str, out_dir: str) -> dict:
    os.makedirs(out_dir, exist_ok=True)

    r = session.get(
        url,
        headers=HEADERS,
        timeout=TIMEOUT,
        stream=True,
        proxies=proxies_dict(),
    )

    if r.status_code in (403, 429, 500, 502, 503, 504):
        raise requests.RequestException(f"HTTP {r.status_code} for {url}")

    r.raise_for_status()

    content_type = r.headers.get("Content-Type", "")
    if not is_probably_image(content_type):
        raise ValueError(f"Non-image content-type '{content_type}' for {url}")

    ext = guess_ext(content_type, url)
    parsed = urlparse(url)
    base = safe_filename(os.path.basename(parsed.path) or "image")
    if not base:
        base = "image"

    tmp_path = os.path.join(out_dir, f"{base}.partial{ext}")

    h = hashlib.sha256()
    size = 0

    with open(tmp_path, "wb") as f:
        for chunk in r.iter_content(chunk_size=1024 * 64):
            if not chunk:
                continue
            f.write(chunk)
            h.update(chunk)
            size += len(chunk)

    digest = h.hexdigest()
    final_path = os.path.join(out_dir, f"{digest}{ext}")

    with _lock:
        if os.path.exists(final_path):
            os.remove(tmp_path)
            return {"url": url, "path": final_path, "bytes": size, "sha256": digest, "deduped": True}
        os.replace(tmp_path, final_path)

    return {"url": url, "path": final_path, "bytes": size, "sha256": digest, "deduped": False}

def download_all(urls: list[str], out_dir: str = "images", workers: int = 12) -> list[dict]:
    results: list[dict] = []

    with ThreadPoolExecutor(max_workers=workers) as ex:
        futs = {ex.submit(download_one, u, out_dir): u for u in urls}

        for fut in as_completed(futs):
            url = futs[fut]
            try:
                res = fut.result()
                results.append(res)
                print("OK", url, res["path"], "deduped=" + str(res["deduped"]))
            except Exception as e:
                results.append({"url": url, "error": str(e)})
                print("ERR", url, str(e))

    return results


if __name__ == "__main__":
    urls = [
        "https://httpbin.org/image/jpeg",
        "https://httpbin.org/image/png",
    ]

    out = download_all(urls, out_dir="downloaded_images", workers=8)
    ok = [r for r in out if "error" not in r]
    print("done", "ok=", len(ok), "total=", len(out))

import csv


def write_manifest(rows: list[dict], path: str = "manifest.csv") -> None:
    with open(path, "w", newline="", encoding="utf-8") as fh:
        writer = csv.DictWriter(fh, fieldnames=["url", "path", "bytes", "sha256", "deduped", "error"])
        writer.writeheader()
        for row in rows:
            writer.writerow(row)