0 Asynchronous Bulk Image Downloader for Pexels API.md

Asynchronous Bulk Image Downloader for Pexels API (aiohttp + asyncio)

1. Configurable Pexels search and download control Uses environment-based API key and settings (query, orientation, per-page limit, max pages, max downloads, concurrency, timeout, save directory) to control how many and what kind of images are downloaded.

2. Asynchronous HTTP requests and file I/O Leverages aiohttp for non-blocking API calls and image downloads, and aiofiles for async file writes, with a semaphore to cap concurrent downloads.

3. Robust file handling and deduplication Automatically infers file extensions from URLs, skips already-downloaded or duplicate photos by ID, and cleans up zero-byte partial files on errors, reporting progress and final saved count.

main_a.py

# requirements:
#   pip install aiohttp aiofiles
import os
import asyncio
import aiohttp
import aiofiles
import pathlib
from urllib.parse import urlsplit

# ===== Settings =====
PEXELS_API_KEY = os.environ.get("PEXELS_API_KEY") or "YOUR_API_KEY_HERE"
QUERY = "cat"
ORIENTATION = "landscape"   # landscape (horizontal) / portrait (vertical) / square
PER_PAGE = 80               # 1–80 (API limit)
MAX_PAGES = 50              # Page scan upper bound (stops earlier if MAX_DOWNLOADS is reached)
MAX_DOWNLOADS = 10          # ★ Maximum number of files to download
CONCURRENCY = 1             # Number of concurrent downloads
TIMEOUT_SEC = 5

SAVE_DIR = pathlib.Path("downloads/pexels_cat_landscape")
# ====================

SEARCH_ENDPOINT = "https://api.pexels.com/v1/search"
SAVE_DIR.mkdir(parents=True, exist_ok=True)

def guess_ext_from_url(url: str) -> str:
    path = urlsplit(url).path
    ext = pathlib.Path(path).suffix.lower()
    return ext if ext in {".jpg", ".jpeg", ".png", ".webp"} else ".jpg"

def out_path(photo_id: int, original_url: str) -> pathlib.Path:
    return SAVE_DIR / f"pexels_{photo_id}{guess_ext_from_url(original_url)}"

async def fetch_json(session: aiohttp.ClientSession, url: str, params: dict):
    async with session.get(url, params=params, timeout=TIMEOUT_SEC) as r:
        r.raise_for_status()
        return await r.json()

async def download_one(session: aiohttp.ClientSession, url: str, dest: pathlib.Path, sem: asyncio.Semaphore):
    if dest.exists():
        print(f"skip (exists): {dest.name}")
        return False
    async with sem:
        try:
            async with session.get(url, timeout=TIMEOUT_SEC) as r:
                r.raise_for_status()
                # Save in chunks
                async with aiofiles.open(dest, "wb") as f:
                    async for chunk in r.content.iter_chunked(64 * 1024):
                        await f.write(chunk)
            print(f"saved: {dest.name}")
            return True
        except Exception as e:
            print(f"error: {dest.name}: {e}")
            # Clean up incomplete files
            try:
                if dest.exists() and dest.stat().st_size == 0:
                    dest.unlink(missing_ok=True)
            except Exception:
                pass
            return False

async def main():
    headers = {"Authorization": PEXELS_API_KEY}
    connector = aiohttp.TCPConnector(limit=CONCURRENCY * 2)
    timeout = aiohttp.ClientTimeout(total=None, sock_read=TIMEOUT_SEC)
    saved = 0
    seen_ids = set()

    async with aiohttp.ClientSession(headers=headers, connector=connector, timeout=timeout) as session:
        sem = asyncio.Semaphore(CONCURRENCY)

        for page in range(1, MAX_PAGES + 1):
            if saved >= MAX_DOWNLOADS:
                break

            per_page = min(PER_PAGE, MAX_DOWNLOADS - saved) or 1
            params = {
                "query": QUERY,
                "orientation": ORIENTATION,
                "per_page": per_page,
                "page": page,
                "people_count": 0,
            }
            data = await fetch_json(session, SEARCH_ENDPOINT, params)
            photos = data.get("photos", [])
            if not photos:
                print("No more results.")
                break

            tasks = []
            for p in photos:
                if saved + len(tasks) >= MAX_DOWNLOADS:
                    break
                pid = p.get("id")
                if pid in seen_ids:
                    continue
                seen_ids.add(pid)

                url = p["src"]["original"]
                dest = out_path(pid, url)
                # Skip immediately if file already exists (don't create a task)
                if dest.exists():
                    print(f"skip (exists): {dest.name}")
                    continue
                tasks.append(asyncio.create_task(download_one(session, url, dest, sem)))

            if tasks:
                results = await asyncio.gather(*tasks)
                saved += sum(1 for ok in results if ok)

    print(f"done. saved {saved} file(s) -> {SAVE_DIR.resolve()}")

if __name__ == "__main__":
    asyncio.run(main())

main_b.py

import asyncio
import os
import re
import argparse
from pathlib import Path
from typing import Optional

import aiohttp


# -------------------------------------------------------
# Load secret key from environment variable
# -------------------------------------------------------
PEXELS_SECRET_KEY = os.getenv("PEXELS_SECRET_KEY")
if PEXELS_SECRET_KEY is None:
    raise RuntimeError("PEXELS_SECRET_KEY is not set")


PEXELS_ENDPOINT = "https://www.pexels.com/en-us/api/v3/search/photos"


def sanitize_filename(name: str) -> str:
    name = name.replace("\n", " ").replace("\r", " ").replace("\t", " ")
    name = re.sub(r"\s+", " ", name).strip()
    name = re.sub(r'[\\/:*?"<>|]', "_", name)
    return name


async def fetch_json(
    session: aiohttp.ClientSession,
    query: str,
    page: int,
    per_page: int,
    people_count: int,
    orientation: str,
) -> dict:
    params = {
        "query": query,
        "page": page,
        "per_page": per_page,
        "people_count": people_count,
        "orientation": orientation,
        "seo_tags": "true",
    }
    headers = {
        "secret-key": PEXELS_SECRET_KEY,
    }

    async with session.get(PEXELS_ENDPOINT, params=params, headers=headers) as resp:
        resp.raise_for_status()
        return await resp.json()


async def download_image(
    session: aiohttp.ClientSession,
    url: str,
    fallback_url: Optional[str],
    dest_path: Path,
) -> bool:
    async def _try_download(download_url: str) -> bool:
        try:
            async with session.get(download_url) as resp:
                if resp.status != 200:
                    return False
                data = await resp.read()
        except Exception:
            return False

        dest_path.write_bytes(data)
        return True

    if await _try_download(url):
        return True

    if fallback_url:
        return await _try_download(fallback_url)

    return False


async def download_pexels_images(
    query: str,
    total_images: int,
    output_dir: str,
    people_count: int,
    orientation: str,
    per_page: int = 24,
):
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    downloaded_count = 0
    page = 1

    async with aiohttp.ClientSession() as session:
        while downloaded_count < total_images:
            json_data = await fetch_json(
                session=session,
                query=query,
                page=page,
                per_page=per_page,
                people_count=people_count,
                orientation=orientation,
            )

            data_list = json_data.get("data") or []
            if not data_list:
                print(f"No data on page {page}. Stopping.")
                break

            print(f"Processing page {page} ...")

            for item in data_list:
                if downloaded_count >= total_images:
                    break

                try:
                    # -------------------------------
                    # User-requested new structure
                    # -------------------------------
                    _id = item["id"]
                    slug = item["attributes"]["slug"]
                    download_link = item["attributes"]["image"]["download_link"]
                    fallback_url = item["attributes"]["image"]["large"]
                except Exception as e:
                    print(f"Invalid JSON structure, skipping entry: {e}")
                    continue

                filename_base = sanitize_filename(f"{_id} {slug}")

                ext = os.path.splitext(download_link.split("?")[0])[1] or ".jpg"
                filename = filename_base + ext
                dest = output_path / filename

                if dest.exists():
                    print(f"Skipping existing file: {dest.name}")
                    continue

                ok = await download_image(session, download_link, fallback_url, dest)
                if ok:
                    downloaded_count += 1
                    print(f"[{downloaded_count}/{total_images}] Saved: {dest.name}")
                else:
                    print(f"Download failed: {filename}")

            page += 1

    print(f"Finished. Downloaded {downloaded_count} images.")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Download images from the Pexels API using aiohttp."
    )

    parser.add_argument("--query", required=True, help="Search keyword.")
    parser.add_argument("--total_images", type=int, required=True, help="Number of images to download.")
    parser.add_argument("--people_count", type=int, default=0, help="people_count API parameter.")
    parser.add_argument("--orientation", default="landscape", help="orientation API parameter.")
    parser.add_argument("--output_dir", default="./downloads", help="Directory to save downloaded images.")

    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    asyncio.run(
        download_pexels_images(
            query=args.query,
            total_images=args.total_images,
            output_dir=args.output_dir,
            people_count=args.people_count,
            orientation=args.orientation,
        )
    )