Google Image Scraper: Public Image Results via API

# imgparse.py: read Google Images tiles and original image URLs from a rendered results page
import json, re
from urllib.parse import urlsplit

# A JavaScript expression run in the page: one object per result tile. Check the selectors on a live page first.
EXTRACT = """
[...document.querySelectorAll('a[href^="http"]')]
  .filter(a => !a.href.includes('google.') && a.querySelector('img'))
  .map((a, i) => {
    const img = a.querySelector('img');
    return {position: i + 1, link: a.href,
            title: (a.innerText || img.alt || '').trim().slice(0, 200),
            thumbnail: img.src.startsWith('http') ? img.src : (img.dataset.src || '')};
  })
"""

# Original images sit in the page's inline data as ["<url>",<height>,<width>]; thumbnails are on gstatic.com.
ORIGINAL = re.compile(r'\["(https?://[^"]+)",(\d+),(\d+)\]')

def originals(html):
    seen, out = set(), []
    for url, height, width in ORIGINAL.findall(html):
        url = json.loads(f'"{url}"')                     # undo escapes such as =
        if "gstatic.com" in url or url in seen:
            continue
        seen.add(url)
        out.append((url, int(width), int(height)))
    return out

def site(url):
    """example.com from img.example.com; shop.example.co.uk -> example.co.uk"""
    parts = (urlsplit(url).hostname or "").split(".")
    n = 3 if len(parts) > 2 and len(parts[-1]) == 2 and len(parts[-2]) <= 3 else 2
    return ".".join(parts[-n:])

def pair(tiles, full):
    """Give each tile the first unused original on the same site as its source page, or blanks."""
    left = list(full)
    for t in tiles:
        hit = next((o for o in left if site(o[0]) == site(t["link"])), None)
        if hit:
            left.remove(hit)
        yield t, hit or ("", "", "")

# scrape_images.py: one Images page per keyword from keywords.txt, one CSV row per image
import csv, random, time, uuid
from urllib.parse import urlencode, urlsplit
from playwright.sync_api import sync_playwright
from imgparse import EXTRACT, originals, pair

LOGIN, PASSWORD = "<login>", "<password>"
keywords = [k.strip() for k in open("keywords.txt") if k.strip()]

with sync_playwright() as p, open("images.csv", "w", newline="", encoding="utf-8") as f:
    out = csv.writer(f)
    out.writerow(["keyword", "position", "title", "source", "link", "thumbnail",
                  "original", "original_width", "original_height"])
    for kw in keywords:
        session = uuid.uuid4().hex[:8]                    # one sticky IP per keyword, held up to 10 min
        browser = p.chromium.launch(proxy={
            "server": "http://gate.proxyfield.io:7000",
            "username": f"{LOGIN}-country-us-session-{session}-ttl-600",
            "password": PASSWORD})
        page = browser.new_page(locale="en-US")
        page.route("**/*", lambda r: r.abort() if r.request.resource_type in ("image", "media", "font") else r.continue_())
        url = "https://www.google.com/search?" + urlencode({"q": kw, "udm": 2, "hl": "en", "gl": "us"})
        try:
            page.goto(url, wait_until="networkidle", timeout=90_000)
            for _ in range(3):                            # scroll for more tiles
                page.mouse.wheel(0, 4000)
                page.wait_for_timeout(1500)
            tiles = page.evaluate(EXTRACT)
            for t, (orig, width, height) in pair(tiles, originals(page.content())):
                out.writerow([kw, t["position"], t["title"], urlsplit(t["link"]).hostname, t["link"],
                              t["thumbnail"], orig, width, height])
            print(kw, len(tiles), "tiles")
        except Exception as e:                            # a timeout or a challenge page: log it, go on
            print(kw, "failed:", e)
        browser.close()
        time.sleep(random.uniform(5, 10))                # a few seconds between searches