Googleデータを大規模に収集するBulk SERP API

# bulk.py  -  python bulk.py queue keywords.txt   then later   python bulk.py catch-up
import json, sys, time
import requests

API = "https://api.semscraper.com/v1/serp"
HEADERS = {"Authorization": "Bearer API_KEY"}
CALLBACK = "https://your-server.com/serp-callback"  # or None: catch-up only

def save(serp, path="serps.jsonl"):
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(serp, ensure_ascii=False) + "\n")

def queue(keywords):
    for i in range(0, len(keywords), 1000):  # 1,000 keywords per call
        batch = [{
            "search_engine": "google_search",
            "keyword": kw,  # a keyword must not contain a comma
            "device": "desktop",
            "location": "en",
            "language": "en",
            "depth": 1,
            **({"callback_url": CALLBACK} if CALLBACK else {}),
        } for kw in keywords[i:i + 1000]]
        for attempt in range(6):
            r = requests.post(API, json=batch, headers=HEADERS, timeout=60)
            if r.status_code != 429:  # 429: over 500 calls/minute, nothing was queued
                break
            time.sleep(10 * (attempt + 1))
        r.raise_for_status()
        print(r.json()["count"], "keywords queued")

# callback_server.py  -  flask --app callback_server run
import hashlib, hmac, json
from flask import Flask, request

SECRET = b"hmac_..."  # HMAC key from your API keys menu
app = Flask(__name__)

def save(serp, path="serps.jsonl"):
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(serp, ensure_ascii=False) + "\n")

@app.post("/serp-callback")
def serp_callback():
    raw = request.get_data()
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, request.headers.get("X-Signature", "")):
        return "Invalid signature", 401
    for serp in json.loads(raw):  # a list holding the finished SERP
        save(serp)
    return {"status": "ok"}       # anything but 200 leaves it unfetched

# bulk.py, continued: finished results that were never fetched
def catch_up():
    while True:
        r = requests.get(API, headers=HEADERS, timeout=60)  # up to 1,000 per call
        r.raise_for_status()
        ids = [p["id"] for p in r.json().get("data", [])]
        if not ids:
            break
        for i in range(0, len(ids), 100):
            r = requests.get(API, headers=HEADERS, timeout=60,
                             params={"ids": ",".join(ids[i:i + 100]), "output": "json"})
            r.raise_for_status()
            for serp in r.json()["data"]:
                save(serp)  # fetching marks it as fetched, and is never billed

if __name__ == "__main__":
    if sys.argv[1] == "queue":
        queue(open(sys.argv[2], encoding="utf-8").read().splitlines())
    else:
        catch_up()