如何在 Python 中获取 Google 搜索结果（2026）

import os

import requests

API = "https://serpcrypto.com/api/v1/search"
KEY = os.environ["SERPCRYPTO_API_KEY"]

resp = requests.get(
    API,
    headers={"Authorization": "Bearer " + KEY},
    params={"engine": "google", "q": "best hiking boots", "gl": "us", "hl": "en"},
    timeout=60,
)
resp.raise_for_status()
results = resp.json()

for item in results.get("organic_results", []):
    print(item["position"], item["title"])
    print("   ", item["link"])

for q in results.get("people_also_ask", []):
    print("PAA:", q["question"])

import os
from urllib.parse import urlparse

import pandas as pd
import requests

API = "https://serpcrypto.com/api/v1/search"
HEADERS = {"Authorization": "Bearer " + os.environ["SERPCRYPTO_API_KEY"]}


def organic_rows(query, gl="us", hl="en"):
    params = {"engine": "google", "q": query, "gl": gl, "hl": hl}
    resp = requests.get(API, headers=HEADERS, params=params, timeout=60)
    resp.raise_for_status()
    for r in resp.json().get("organic_results", []):
        yield {
            "query": query,
            "position": r.get("position"),
            "title": r.get("title"),
            "domain": urlparse(r.get("link", "")).netloc,
            "link": r.get("link"),
            "snippet": r.get("snippet"),
        }


queries = ["crm for small business", "invoice software", "time tracking app"]
df = pd.DataFrame([row for q in queries for row in organic_rows(q)])
print(df.groupby("domain").size().sort_values(ascending=False).head(10))
df.to_csv("serp_results.csv", index=False)

import asyncio
import os

import httpx

API = "https://serpcrypto.com/api/v1/search"
HEADERS = {"Authorization": "Bearer " + os.environ["SERPCRYPTO_API_KEY"]}
LIMIT = asyncio.Semaphore(5)


async def search(client, query):
    async with LIMIT:
        resp = await client.get(API, params={"engine": "google", "q": query, "gl": "us", "hl": "en"})
        resp.raise_for_status()
        data = resp.json()
        top = data.get("organic_results", [])[:3]
        return query, [r["link"] for r in top]


async def main(queries):
    async with httpx.AsyncClient(headers=HEADERS, timeout=60) as client:
        for query, links in await asyncio.gather(*(search(client, q) for q in queries)):
            print(query, links)


if __name__ == "__main__":
    asyncio.run(main(["python web framework", "python orm", "python task queue"]))

import json
import os

import requests

API = "https://serpcrypto.com/api/v1/search"
HEADERS = {"Authorization": "Bearer " + os.environ["SERPCRYPTO_API_KEY"]}


def fetch_pages(query, pages=3, path="serp.jsonl"):
    with open(path, "a", encoding="utf-8") as out:
        for page in range(1, pages + 1):
            params = {"engine": "google", "q": query, "gl": "us", "hl": "en", "page": page}
            resp = requests.get(API, headers=HEADERS, params=params, timeout=60)
            resp.raise_for_status()
            data = resp.json()
            out.write(json.dumps(data, ensure_ascii=False) + "\n")
            if not data.get("organic_results"):
                break


fetch_pages("electric cargo bike", pages=3)