使用可追踪请求上下文对Google搜索结果进行分页

import json
import os
import uuid
from datetime import datetime, timezone
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import parse_qs, urlsplit
from urllib.request import Request, urlopen

ENDPOINT = 'https://api.scrapeless.com/api/v1/scraper/request'
OFFSETS = (0, 10, 20)
BASE = {'q': 'coffee', 'gl': 'us', 'hl': 'en', 'device': 'desktop'}


def now():
    return datetime.now(timezone.utc).isoformat()


def fetch(body, key):
    request = Request(ENDPOINT, data=json.dumps(body).encode(), method='POST',
                      headers={'Content-Type': 'application/json', 'x-api-token': key})
    try:
        with urlopen(request, timeout=60) as response:
            status, raw = response.status, response.read().decode('utf-8')
    except HTTPError as error:
        status, raw = error.code, error.read().decode('utf-8', errors='replace')
    except (URLError, TimeoutError) as error:
        return None, None, type(error).__name__
    try:
        return status, json.loads(raw), None
    except json.JSONDecodeError:
        return status, {'unparsed_body': raw}, 'response_not_json'


def next_offset(payload, current):
    pagination = payload.get('pagination')
    link = pagination.get('next') if isinstance(pagination, dict) else None
    if not isinstance(link, str) or not link:
        return None, 'next_link_unavailable'
    try:
        parts = urlsplit(link)
        if (parts.scheme != 'https' or parts.hostname not in ('google.com', 'www.google.com')
                or parts.username or parts.password or parts.port or parts.path != '/search'
                or parts.fragment or '...' in link or '…' in link):
            return None, 'next_link_needs_review'
        query = parse_qs(parts.query, keep_blank_values=True)
        if any(query.get(k) != [BASE[k]] for k in ('q', 'gl', 'hl')):
            return None, 'next_context_needs_review'
        values = query.get('start', [])
        if len(values) != 1 or not values[0].isascii() or not values[0].isdigit():
            return None, 'next_offset_needs_review'
        offset = int(values[0])
        if offset <= current or offset not in OFFSETS:
            return None, 'next_offset_outside_plan'
        return offset, None
    except ValueError:
        return None, 'next_link_needs_review'


def collect(directory, key):
    directory = Path(directory)
    directory.mkdir(parents=True, exist_ok=False)
    seen, visited, pages = set(), set(), []
    offset, stop = 0, None
    while offset in OFFSETS and offset not in visited:
        visited.add(offset)
        body = {'actor': 'scraper.google.search', 'input': dict(BASE, start=offset)}
        started = now()
        status, payload, error = fetch(body, key)
        run_id = uuid.uuid4().hex
        capture = {'run_id': run_id, 'requested_at': started, 'received_at': now(),
                   'request': body, 'http_status': status, 'response': payload, 'error': error}
        filename = f'{offset}-{run_id}.json'
        (directory / filename).write_text(json.dumps(capture, ensure_ascii=False, indent=2), encoding='utf-8')
        page = {'start': offset, 'capture': filename, 'new_exact_urls': None}
        pages.append(page)
        if status == 201:
            stop = 'pending'
            break
        if status != 200 or error:
            stop = 'collection_error'
            break
        rows = payload.get('organic_results') if isinstance(payload, dict) else None
        if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows):
            stop = 'unmapped_organic_results'
            break
        links = {row['link'] for row in rows if isinstance(row.get('link'), str) and row['link']}
        page['new_exact_urls'] = len(links - seen)
        seen.update(links)
        if not rows:
            stop = 'empty_organic_slice'
            break
        if not links:
            stop = 'no_usable_link_strings'
            break
        if page['new_exact_urls'] == 0:
            stop = 'no_new_exact_urls'
            break
        if len(visited) == len(OFFSETS):
            stop = 'planned_page_limit'
            break
        offset, stop = next_offset(payload, offset)
        if stop:
            break
    summary = {'pages': pages, 'stop_reason': stop, 'unique_exact_url_strings': len(seen)}
    (directory / 'collection-summary.json').write_text(json.dumps(summary, indent=2), encoding='utf-8')
    return summary


if __name__ == '__main__':
    key = os.environ['SCRAPELESS_API_KEY']
    output = 'search-pages-' + uuid.uuid4().hex
    print(json.dumps(collect(output, key), indent=2))