flickr_scraper.py

# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license

# Generated by Glenn Jocher (glenn.jocher@ultralytics.com) for https://github.com/ultralytics

import argparse
import os
import time
from pathlib import Path

from flickrapi import FlickrAPI

from utils.general import download_uri

key = ""  # Flickr API key https://www.flickr.com/services/apps/create/apply
secret = ""


def build_photo_url(photo):
    """Return the Flickr-provided original image URL when available."""
    return photo.get("url_o")


def resolve_credentials(api_key=None, api_secret=None):
    """Resolve Flickr API credentials from arguments, environment variables, or source constants."""
    api_key = api_key or os.getenv("FLICKR_API_KEY") or key
    api_secret = api_secret or os.getenv("FLICKR_API_SECRET") or secret
    if not api_key or not api_secret:
        help_url = "https://www.flickr.com/services/apps/create/apply"
        raise ValueError(
            "Flickr API key and secret are required. Pass --key/--secret, set "
            f"FLICKR_API_KEY/FLICKR_API_SECRET, or edit flickr_scraper.py. To apply visit {help_url}"
        )
    return api_key, api_secret


def get_urls(search="honeybees on flowers", n=10, download=False, api_key=None, api_secret=None, page=1):
    """Fetch Flickr URLs for a search term, optionally downloading up to n images."""
    if n < 1:
        return []

    t = time.time()
    api_key, api_secret = resolve_credentials(api_key, api_secret)
    flickr = FlickrAPI(api_key, api_secret, format="parsed-json")
    licenses = ""  # https://www.flickr.com/services/api/explore/?method=flickr.photos.licenses.getInfo
    page = max(page, 1)

    if download:
        dir_path = Path.cwd() / "images" / search.replace(" ", "_")
        dir_path.mkdir(parents=True, exist_ok=True)

    urls = []
    while len(urls) < n:
        try:
            response = flickr.photos.search(
                text=search,  # https://www.flickr.com/services/api/flickr.photos.search.html
                extras="url_o",
                per_page=min(500, n - len(urls)),  # 1-500
                page=page,
                license=licenses,
                media="photos",
                sort="relevance",
            )
        except Exception as e:
            raise RuntimeError(f"Flickr API request failed: {e}") from e

        if response.get("stat") == "fail":
            code = response.get("code", "unknown")
            message = response.get("message", "unknown Flickr API error")
            raise RuntimeError(f"Flickr API error {code}: {message}")

        photos = response.get("photos", {})
        photo_list = photos.get("photo", [])
        if not photo_list:
            break

        for photo in photo_list:
            url = build_photo_url(photo)
            if url is None:
                print("skipped (missing original URL)")
                continue

            if url in urls:
                continue

            if download:
                download_uri(url, dir_path)

            urls.append(url)
            print(f"{len(urls)}/{n} {'downloaded' if download else 'found'}")

            if len(urls) >= n:
                break

        if page >= int(photos.get("pages", page)):
            break
        page += 1

    print(f"Done. ({time.time() - t:.1f}s)" + (f"\nAll images saved to {dir_path}" if download else ""))
    return urls


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--search", nargs="+", default=["honeybees on flowers"], help="flickr search term")
    parser.add_argument("--n", type=int, default=10, help="number of images")
    parser.add_argument("--download", action="store_true", help="download images")
    parser.add_argument("--page", type=int, default=1, help="Flickr results page to start from")
    parser.add_argument("--key", default=None, help="Flickr API key, overrides FLICKR_API_KEY")
    parser.add_argument("--secret", default=None, help="Flickr API secret, overrides FLICKR_API_SECRET")
    opt = parser.parse_args()

    try:
        for search in opt.search:
            get_urls(
                search=search,
                n=opt.n,
                download=opt.download,
                api_key=opt.key,
                api_secret=opt.secret,
                page=opt.page,
            )
    except (RuntimeError, ValueError) as e:
        parser.exit(1, f"{e}\n")