0 A Simple Tool to Search and Download Photos from Unsplash.md

A Simple Tool to Search and Download Photos from Unsplash

This Python script is a handy command-line tool that searches Unsplash for photos and downloads a specified number of images to a local folder. It focuses on reliability (skipping files you already have, retrying failed downloads) and speed (using asynchronous, concurrent requests).

What the Script Does

1. Searches Unsplash It calls Unsplash’s public search endpoint (the same one the website uses) with your query, page number, and page size.

2. Collects Photo Items From the JSON response, it filters results to keep only items where asset_type is "photo".

3. Downloads Images Concurrently Using curl_cffi’s AsyncSession and asyncio, it fetches multiple images in parallel. You can control concurrency with a flag.

4. Determines File Extensions Safely It tries to infer the correct extension from the Content-Type header via a HEAD request; if needed, it falls back to a GET and a small mapping (e.g., image/jpeg → .jpg), or to .bin when unknown.

5. Skips Files You Already Have Before downloading, it checks for any existing file matching the Unsplash photo ID (e.g., abc123.*) and skips it to avoid duplicates.

6. Keeps Going Across Pages If one page doesn’t contain enough new photos to reach your target count, it automatically advances to the next page until the count is met or there are no more results.

7. Optional JSON Save You can save the first page’s raw JSON to a file for inspection or debugging.

Key Components at a Glance

• Asynchronous engine: asyncio + curl_cffi.requests.AsyncSession
• Search URL template: SEARCH_URL_TMPL builds the Unsplash search request.
• Extension inference: guess_ext_from_content_type and a small CONTENT_TYPE_EXT_FALLBACK map.
• Concurrency control: bounded_gather wraps tasks with a semaphore.
• Robust downloading: Retries with exponential backoff and clear log messages.
• De-duplication: find_existing_by_id and list_existing_ids prevent re-downloading.

How to Run It

Install the dependency:

pip install "curl_cffi>=0.7.1"

Run the script with your query and output directory. For example, to download up to 50 cat photos:

python unsplash_fetch_and_download.py "cat" out_dir --max-count 50 --page 1 --per-page 20 --concurrency 8 --save-json result.json

Important Flags

• --max-count (required): Total number of images to download (existing files are skipped).
• --page: Starting page (default: 1).
• --per-page: Items per page fetched from Unsplash (default: 20).
• --concurrency: Number of simultaneous downloads (default: 8).
• --save-json: Save the first page’s search JSON to a file (optional).

Why This Design Works Well

• Fast and efficient: Async I/O and concurrency make large downloads much quicker than serial requests.
• Resilient: Retries, backoff, and extension detection via headers/GETs reduce failures and bad file saves.
• Practical: It avoids duplicates by tracking file stems based on Unsplash IDs.
• Clear logging: You can see progress ([OK], [SKIP], [RETRY], [ERROR]) and final stats at the end.

Caveats and Tips

• API/Terms: The script calls Unsplash’s web API endpoints. Be mindful of Unsplash’s terms of service and usage limits.
• Network variability: If you see repeated retries, lower --concurrency or increase timeouts.
• Storage: Images can be large (urls.full), so ensure adequate disk space.

In short, this script is a compact, production-friendly utility for fetching a curated set of Unsplash photos quickly and reliably—great for prototyping datasets, mood boards, or local inspiration libraries.

main.py

#!/usr/bin/env python3
# pip install "curl_cffi>=0.7.1"
# Usage:
#   python unsplash_fetch_and_download.py "cat" out_dir --max-count 50 \
#       [--page 1] [--per-page 20] [--concurrency 8] [--save-json result.json] \
#       [--api napi|official] [--access-key XXXX]
#
# --api:
#   napi     ... Use the unofficial https://unsplash.com/napi/... (default)
#   official ... Use the official https://api.unsplash.com/... (requires access key)
#
# In official mode, we comply with the API rules: call links.download_location
# to register the download, then fetch the actual file from the returned URL.

import asyncio
import glob
import json
import mimetypes
import os
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple

from curl_cffi.requests import AsyncSession, Response

SEARCH_URL_TMPL_NAPI = (
    "https://unsplash.com/napi/search/photos?"
    "orientation=landscape&page={page}&per_page={per_page}&query={q}"
)

SEARCH_URL_TMPL_OFFICIAL = (
    "https://api.unsplash.com/search/photos?"
    "orientation=landscape&page={page}&per_page={per_page}&query={q}"
)

CONTENT_TYPE_EXT_FALLBACK = {
    "image/jpeg": ".jpg",
    "image/jpg": ".jpg",
    "image/png": ".png",
    "image/webp": ".webp",
    "image/gif": ".gif",
    "image/tiff": ".tiff",
    "image/bmp": ".bmp",
    "image/heic": ".heic",
    "image/heif": ".heif",
}

def guess_ext_from_content_type(content_type: Optional[str]) -> str:
    if not content_type:
        return ".bin"
    ct = content_type.split(";")[0].strip().lower()
    if ct in CONTENT_TYPE_EXT_FALLBACK:
        return CONTENT_TYPE_EXT_FALLBACK[ct]
    ext = mimetypes.guess_extension(ct)
    return ext or ".bin"

def collect_photo_items_from_napi(data: Dict[str, Any]) -> List[Dict[str, Any]]:
    results = data.get("results", [])
    photos = []
    for item in results:
        try:
            if item.get("asset_type") == "photo":
                photos.append(item)
        except Exception:
            continue
    return photos

def collect_photo_items_from_official(data: Dict[str, Any]) -> List[Dict[str, Any]]:
    # Official API: results is already an array of photo objects
    return list(data.get("results", []))

async def fetch_one(session: AsyncSession, url: str, out_path: Path,
                    timeout: float = 60.0, max_retries: int = 3, retry_backoff: float = 1.5) -> bool:
    if out_path.exists():
        print(f"[SKIP] already exists -> {out_path.name}")
        return True
    last_exc: Optional[Exception] = None
    for attempt in range(1, max_retries + 1):
        try:
            resp: Response = await session.get(url, timeout=timeout)
            resp.raise_for_status()
            data = resp.content
            out_path.write_bytes(data)
            print(f"[OK]   saved -> {out_path.name} (size={len(data)} bytes)")
            return True
        except Exception as e:
            last_exc = e
            if attempt < max_retries:
                print(f"[RETRY {attempt}/{max_retries}] {out_path.name}: {e}")
                await asyncio.sleep(retry_backoff ** (attempt - 1))
            else:
                break
    sys.stderr.write(f"[ERROR] download failed: {url} -> {out_path.name} ({last_exc})\n")
    return False

async def fetch_one_with_ext(session: AsyncSession, url: str, id_name: str, out_dir: Path,
                             timeout: float = 60.0, max_retries: int = 3, retry_backoff: float = 1.5) -> Tuple[bool, Optional[Path]]:
    """
    Return: (success, saved_path or None)
    """
    existing = find_existing_by_id(out_dir, id_name)
    if existing:
        print(f"[SKIP] {id_name} -> {existing.name} (already exists)")
        return True, existing

    ext = None
    try:
        head_resp: Response = await session.head(url, timeout=20.0)
        if head_resp.status_code < 400:
            ext = guess_ext_from_content_type(head_resp.headers.get("Content-Type"))
    except Exception:
        pass

    if not ext:
        try:
            get_resp: Response = await session.get(url, timeout=timeout)
            get_resp.raise_for_status()
            ext = guess_ext_from_content_type(get_resp.headers.get("Content-Type"))
            out_path = out_dir / f"{id_name}{ext}"
            if out_path.exists():
                print(f"[SKIP] {id_name} -> {out_path.name} (already exists)")
                return True, out_path
            data = get_resp.content
            out_path.write_bytes(data)
            print(f"[OK]   {id_name} -> {out_path.name} (size={len(data)} bytes)")
            return True, out_path
        except Exception as e:
            sys.stderr.write(f"[ERROR] {id_name} download failed: {e}\n")
            return False, None
    else:
        out_path = out_dir / f"{id_name}{ext}"
        ok = await fetch_one(session, url, out_path, timeout=timeout, max_retries=max_retries, retry_backoff=retry_backoff)
        return ok, (out_path if ok else None)

async def bounded_gather(tasks, limit: int = 8):
    sem = asyncio.Semaphore(limit)
    async def _wrap(coro):
        async with sem:
            return await coro
    return await asyncio.gather(*[_wrap(t) for t in tasks])

async def search_unsplash(query: str, page: int, per_page: int, api_mode: str,
                          access_key: Optional[str]) -> Tuple[Dict[str, Any], Dict[str, str]]:
    """
    Return: (JSON body, important headers)
    """
    if api_mode == "official":
        if not access_key:
            raise RuntimeError("official mode requires --access-key or env UNSPLASH_ACCESS_KEY.")
        url = SEARCH_URL_TMPL_OFFICIAL.format(page=page, per_page=per_page, q=query)
        headers = {"Authorization": f"Client-ID {access_key}"}
        async with AsyncSession(impersonate="chrome", headers=headers) as s:
            r = await s.get(url, timeout=60.0)
            r.raise_for_status()
            important = {
                "X-Ratelimit-Remaining": r.headers.get("X-Ratelimit-Remaining", ""),
                "X-Ratelimit-Limit": r.headers.get("X-Ratelimit-Limit", ""),
            }
            return r.json(), important
    else:
        url = SEARCH_URL_TMPL_NAPI.format(page=page, per_page=per_page, q=query)
        async with AsyncSession(impersonate="chrome") as s:
            r = await s.get(url, timeout=60.0)
            r.raise_for_status()
            return r.json(), {}

async def register_and_get_download_url(session: AsyncSession, download_location: str) -> str:
    """
    Official API compliance: call download_location to register the download.
    Returns the actual file URL from the response.
    """
    r = await session.post(download_location, timeout=30.0)
    r.raise_for_status()
    data = r.json()
    # Docs specify { url: "..." }
    return data.get("url") or data.get("location") or download_location

def find_existing_by_id(out_dir: Path, photo_id: str) -> Optional[Path]:
    for path_str in glob.glob(str(out_dir / f"{photo_id}.*")):
        p = Path(path_str)
        if p.is_file():
            return p
    return None

def list_existing_ids(out_dir: Path) -> Set[str]:
    existed: Set[str] = set()
    for path_str in glob.glob(str(out_dir / "*.*")):
        p = Path(path_str)
        stem = p.stem
        if stem:
            existed.add(stem)
    return existed

async def run(query: str, out_dir: Path, start_page: int, per_page: int, concurrency: int,
              save_json: Optional[Path], max_count: int, api_mode: str, access_key: Optional[str]):
    out_dir.mkdir(parents=True, exist_ok=True)

    existed_ids = list_existing_ids(out_dir)
    downloaded_paths: List[Path] = []
    page = start_page
    remaining = max_count

    # Official API session headers
    base_headers = {}
    if api_mode == "official":
        base_headers["Authorization"] = f"Client-ID {access_key}"

    async with AsyncSession(impersonate="chrome", headers=base_headers) as session:
        while remaining > 0:
            data, headers = await search_unsplash(query, page=page, per_page=per_page, api_mode=api_mode, access_key=access_key)
            if save_json and page == start_page:
                save_json.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")

            if api_mode == "official":
                if headers:
                    rem = headers.get("X-Ratelimit-Remaining", "")
                    lim = headers.get("X-Ratelimit-Limit", "")
                    if rem:
                        print(f"[RATE] remaining={rem}/{lim}")
                items = collect_photo_items_from_official(data)
            else:
                items = collect_photo_items_from_napi(data)

            if not items:
                print(f"[PAGE] page={page}: no more results, stop.")
                break

            # nAPI/official both expose 'id'
            new_items = [it for it in items if str(it.get("id")) not in existed_ids]
            if not new_items:
                print(f"[PAGE] page={page}: all items already exist, go next.")
                page += 1
                total_pages = data.get("total_pages")
                if isinstance(total_pages, int) and page > total_pages:
                    print(f"[PAGE] reached last page (total_pages={total_pages}).")
                    break
                continue

            batch = new_items[:remaining]
            print(f"[PAGE] page={page}: queue {len(batch)} downloads (remaining target={remaining})")

            tasks = []
            for it in batch:
                _id = str(it["id"])
                if api_mode == "official":
                    # Official: register download then fetch from returned URL
                    dl_loc = it.get("links", {}).get("download_location")
                    if not dl_loc:
                        target_url = it.get("urls", {}).get("full") or it.get("urls", {}).get("raw")
                    else:
                        target_url = await register_and_get_download_url(session, dl_loc)
                else:
                    # Unofficial: direct download from urls.full
                    target_url = it["urls"]["full"]

                tasks.append(
                    fetch_one_with_ext(
                        session,
                        target_url,
                        _id,
                        out_dir,
                        timeout=90.0,
                        max_retries=3,
                        retry_backoff=1.8,
                    )
                )

            results: List[Tuple[bool, Optional[Path]]] = await bounded_gather(tasks, limit=concurrency)

            got = 0
            for ok, p in results:
                if ok and p is not None:
                    downloaded_paths.append(p)
                    existed_ids.add(p.stem)
                    got += 1
            remaining -= got

            page += 1
            total_pages = data.get("total_pages")
            if isinstance(total_pages, int) and page > total_pages:
                print(f"[PAGE] reached last page (total_pages={total_pages}).")
                break

    ok = len(downloaded_paths)
    need = max_count
    print(f"[DONE] query={query!r}, requested={need}, downloaded={ok}, saved_to={out_dir.resolve()}")

def parse_args(argv: List[str]):
    import argparse
    p = argparse.ArgumentParser(description="Unsplash search → multi-page download (skip existing, target count, verbose logs)")
    p.add_argument("query", help="Search query (e.g., 'cat')")
    p.add_argument("out_dir", help="Output directory")
    p.add_argument("--max-count", type=int, required=True, help="Total number of images to download (existing files are skipped)")
    p.add_argument("--page", type=int, default=1, help="Start page (default=1)")
    p.add_argument("--per-page", type=int, default=20, help="Results per page (per_page for the API)")
    p.add_argument("--concurrency", type=int, default=8, help="Concurrent downloads")
    p.add_argument("--save-json", type=str, default=None, help="Save the first page's raw JSON to this path (optional)")
    p.add_argument("--api", choices=["napi", "official"], default="napi", help="Select API to use (default=napi)")
    p.add_argument("--access-key", type=str, default=None, help="Unsplash access key for official mode (or set env UNSPLASH_ACCESS_KEY)")
    return p.parse_args(argv)

def main():
    args = parse_args(sys.argv[1:])
    out_dir = Path(args.out_dir)
    save_json = Path(args.save_json) if args.save_json else None

    if args.max_count <= 0:
        print("--max-count must be >= 1", file=sys.stderr)
        sys.exit(2)

    access_key = args.access_key or os.environ.get("UNSPLASH_ACCESS_KEY")

    try:
        asyncio.run(
            run(
                query=args.query,
                out_dir=out_dir,
                start_page=args.page,
                per_page=args.per_page,
                concurrency=args.concurrency,
                save_json=save_json,
                max_count=args.max_count,
                api_mode=args.api,
                access_key=access_key,
            )
        )
    except RuntimeError as e:
        print(str(e), file=sys.stderr)
        sys.exit(2)

if __name__ == "__main__":
    main()