跳到正文
DEV Community· Atlas Data·· 3 小時前AI 評分56

如何排程監測 Google Trends 並處理 429 錯誤

Monitor Google Trends on a schedule (and stop fighting 429s)

AI 導讀

Atlas Data 示範用 Apify Actor 排程監測 Google Trends，擷取搜尋熱度及相關查詢，並標示上升詞項與 Breakout 查詢。範例每日 07:00 UTC 執行，涵蓋 4 個搜尋詞及過去 7 天，每次執行的費用上限為 $0.25。

正文

If you've tried polling Google Trends from a script, you know how it usually goes. pytrends works for a few requests, then you start getting 429 Too Many Requests, and the cron job writes empty files for a week before anyone notices.

This post sets up a Google Trends monitor that:

  • runs every morning without a machine of yours being on,
  • fetches interest over time and related queries for a list of terms,
  • flags terms that are up against their weekly average, plus queries Google marks Breakout,
  • and tells you explicitly when something didn't work, instead of returning nothing.

Disclosure: the scraper is an Apify Actor published by Atlas Data, the author of this post: atlasdata/google-trends-scraper-reliable. It is pay-per-result, and the task below caps the cost of every run. Docs and examples in other languages are in the GitHub repo.

Why Trends scrapers get blocked

Google Trends has no official API. The web app calls internal endpoints that are rate-limited per IP address and per cookie session, and the limits are tight: a handful of quick requests from one IP is often enough to trigger 429s, and datacenter IPs are throttled sooner. The usual DIY fixes are time.sleep() and retries. They help, but on a schedule they still fail eventually, because the IP is the same every run.

The Actor does what you would end up building yourself:

  • A fresh cookie session per attempt, so a throttled session isn't reused.
  • Exponential backoff with a new IP on every retry. maxRetries defaults to 8 (range 0–20).
  • Automatic proxy fallback (proxyFallback, on by default). If the current connection keeps getting blocked, it moves from your proxy to Apify datacenter, then to Apify residential if your plan includes it, then to a direct connection.
  • A per-term status report in the run's OUTPUT record (SUCCEEDED / PARTIAL / NO_DATA / FAILED), so an empty result is never ambiguous.

None of this is magic. Google can still withhold some data. For example, related topics are currently not returned to any automated session, and city-level regions are often missing. When that happens, the summary says so.

Setup 

pip install apify-client   # tested with apify-client 3.4
export APIFY_TOKEN=...      # https://console.apify.com/settings/integrations

1. Define what to watch 

ACTOR = "atlasdata/google-trends-scraper-reliable"
TASK_NAME = "trends-watch-ai-tools"
TRENDS_INPUT = {
    "searchTerms": ["chatgpt", "claude ai", "gemini ai", "perplexity"],
    "geo": "US",
    "timeRange": "now 7-d",
    "includeInterestOverTime": True,
    "includeInterestByRegion": False,
    "includeRelatedQueries": True,
    "includeRelatedTopics": False,
    "outputFormat": "flat",
    "maxRetries": 8,  # retries with fresh sessions/proxies when Google returns 429
}

The inputs worth knowing:

  • geo: empty for worldwide, a country ("US") or a sub-region ("US-CA").
  • timeRange: now 1-H, now 4-H, now 1-d, now 7-d (hourly points), today 1-m, today 3-m, today 12-m, today 5-y, all, or custom with startDate/endDate.
  • compareTerms: by default every term is scaled 0–100 on its own. Set it to true to query terms together (up to 5 per group), like the compare view on the website. Do that if you want to say "A is searched more than B". Leave it off if you only care about each term's own trend, as we do here.
  • outputFormat: "flat" returns one row per data point. Each row has a recordType of interestOverTime, interestByRegion, relatedQuery or relatedTopic, which is easy to filter in Python.

2. Save it as a task and schedule it

An Apify task is a saved input for an Actor. A schedule runs it on a cron expression in Apify's cloud. Both are created with the API client:

def setup() -> None:
    task = client.tasks().create(
        actor_id=ACTOR,
        name=TASK_NAME,
        task_input=TRENDS_INPUT,
        max_total_charge_usd=Decimal("0.25"),  # cost cap for every run of this task
    )
    schedule = client.schedules().create(
        name=f"{TASK_NAME}-daily",
        cron_expression="0 7 * * *",  # every day at 07:00
        timezone="UTC",
        is_enabled=True,
        is_exclusive=True,  # skip a tick if the previous run is still going
        actions=[{"type": "RUN_ACTOR_TASK", "actorTaskId": task.id}],
    )
    print(f"task {task.id} and schedule {schedule.id} created")

Two settings matter for unattended jobs:

  • max_total_charge_usd on the task caps the cost of every scheduled run, so a typo in the input can't turn into a large bill.
  • is_exclusive=True stops runs from piling up if one is slow.

You can do the same in the Apify Console (Actor → Save as a new task → Schedule). The code version lives in your repo and is easy to recreate.

3. Read the latest run and flag what's moving 

def latest_items() -> list[dict]:
    task = next(t for t in client.tasks().list().items if t.name == TASK_NAME)
    run = client.task(task.id).last_run(status="SUCCEEDED").get()
    if run is None:
        raise SystemExit("No finished run yet - wait for the schedule or start the task once.")
    print(f"latest run {run.id} finished {run.finished_at:%Y-%m-%d %H:%M} UTC")
    return list(client.dataset(run.default_dataset_id).iterate_items())


def check() -> None:
    items = latest_items()

    # 1) Interest over time: compare the last day with the week's average.
    for term in TRENDS_INPUT["searchTerms"]:
        points = [i["value"] for i in items
                  if i["recordType"] == "interestOverTime" and i["searchTerm"] == term
                  and i.get("hasData", True) and not i.get("isPartial")]
        if len(points) >= 48:
            last_day = sum(points[-24:]) / 24
            week = sum(points) / len(points)
            flag = "  <-- up" if last_day > 1.3 * week else ""
            print(f"{term:12s} last 24h avg {last_day:5.1f} vs 7-day avg {week:5.1f}{flag}")

    # 2) Related queries that are rising fast or marked "Breakout".
    rising = [i for i in items if i["recordType"] == "relatedQuery" and i["rankType"] == "rising"]
    print("\nBreakout / fast-rising related queries:")
    for i in sorted(rising, key=lambda i: (not i.get("isBreakout"), -(i.get("value") or 0)))[:15]:
        label = "Breakout" if i.get("isBreakout") else f'+{i["value"]}%'
        print(f'  [{i["searchTerm"]}] {i["query"]}  ({label})')

A few details:

  • isPartial marks the newest data point, which Google is still filling in. Leave it out, or the last hour always looks like a drop.
  • For relatedQuery rows with rankType: "rising", value is the percentage increase. isBreakout is true when Google shows "Breakout" (an increase above 5000%).
  • last_run(status="SUCCEEDED") skips a run that failed, so a bad night doesn't wipe the report.

Running it 

python trends_monitor.py setup   # once
python trends_monitor.py check   # whenever you want the report

Output from a real run (US, past 7 days, 4 terms; the run took about 9 seconds):

latest run 47xnXFExBms4IJ2ba finished 2026-10-27 13:27 UTC
chatgpt      last 24h avg  79.3 vs 7-day avg  71.4
claude ai    last 24h avg  71.5 vs 7-day avg  67.2
gemini ai    last 24h avg  69.8 vs 7-day avg  66.5
perplexity   last 24h avg  46.5 vs 7-day avg  45.3

Breakout / fast-rising related queries:
  [claude ai] claude duet  (Breakout)
  [claude ai] anthropic claude duet ai agent  (Breakout)
  [claude ai] claude ai software vulnerability analysis  (Breakout)
  [claude ai] claude ai oblivion fps boost  (Breakout)
  [perplexity] perplexity ai on device mac  (+1800%)
  [claude ai] venice ai  (+50%)
  [perplexity] is perplexity better than chatgpt  (+50%)
  [claude ai] anthropic claude  (+40%)

Nothing crossed the +30% threshold that day. The related queries are where the news shows up: four Breakout queries for one term in a single week.

The run's OUTPUT summary for the same run reported terms: {total: 4, succeeded: 4, partial: 0, noData: 0, failed: 0}, with 676 interest-over-time rows and 101 related-query rows. That is the record to alert on if you want to know about failures as well as trends:

summary = client.key_value_store(run.default_key_value_store_id).get_record("OUTPUT")["value"]
if summary["terms"]["failed"] or summary["terms"]["noData"]:
    print("some terms returned nothing:", [t["searchTerm"] for t in summary["perTerm"] if t["status"] != "SUCCEEDED"])

Getting the report to you

check() only reads stored data and never runs the scraper, so it costs nothing to call. Some ways to wire it up:

  • Run it from your own cron or a GitHub Actions schedule an hour after the Apify schedule.
  • Add a webhook on the task (Console → task → Integrations) that calls your endpoint when a run succeeds, and have the endpoint run check() and post to Slack.
  • Load the dataset into a sheet or BI tool. GET https://api.apify.com/v2/datasets/{datasetId}/items?format=csv also works.

Full script 

"""Watch Google Trends on a schedule and flag rising / breakout queries.

    python trends_monitor.py setup   # once: create a saved task + a daily schedule
    python trends_monitor.py check   # any time: read the latest scheduled run
"""
import os
import sys
from decimal import Decimal

from apify_client import ApifyClient

client = ApifyClient(os.environ["APIFY_TOKEN"])

ACTOR = "atlasdata/google-trends-scraper-reliable"
TASK_NAME = "trends-watch-ai-tools"
TRENDS_INPUT = {
    "searchTerms": ["chatgpt", "claude ai", "gemini ai", "perplexity"],
    "geo": "US",
    "timeRange": "now 7-d",
    "includeInterestOverTime": True,
    "includeInterestByRegion": False,
    "includeRelatedQueries": True,
    "includeRelatedTopics": False,
    "outputFormat": "flat",
    "maxRetries": 8,  # retries with fresh sessions/proxies when Google returns 429
}


def setup() -> None:
    task = client.tasks().create(
        actor_id=ACTOR,
        name=TASK_NAME,
        task_input=TRENDS_INPUT,
        max_total_charge_usd=Decimal("0.25"),  # cost cap for every run of this task
    )
    schedule = client.schedules().create(
        name=f"{TASK_NAME}-daily",
        cron_expression="0 7 * * *",  # every day at 07:00
        timezone="UTC",
        is_enabled=True,
        is_exclusive=True,  # skip a tick if the previous run is still going
        actions=[{"type": "RUN_ACTOR_TASK", "actorTaskId": task.id}],
    )
    print(f"task {task.id} and schedule {schedule.id} created")


def latest_items() -> list[dict]:
    task = next(t for t in client.tasks().list().items if t.name == TASK_NAME)
    run = client.task(task.id).last_run(status="SUCCEEDED").get()
    if run is None:
        raise SystemExit("No finished run yet - wait for the schedule or start the task once.")
    print(f"latest run {run.id} finished {run.finished_at:%Y-%m-%d %H:%M} UTC")
    return list(client.dataset(run.default_dataset_id).iterate_items())


def check() -> None:
    items = latest_items()

    # 1) Interest over time: compare the last day with the week's average.
    for term in TRENDS_INPUT["searchTerms"]:
        points = [i["value"] for i in items
                  if i["recordType"] == "interestOverTime" and i["searchTerm"] == term
                  and i.get("hasData", True) and not i.get("isPartial")]
        if len(points) >= 48:
            last_day = sum(points[-24:]) / 24
            week = sum(points) / len(points)
            flag = "  <-- up" if last_day > 1.3 * week else ""
            print(f"{term:12s} last 24h avg {last_day:5.1f} vs 7-day avg {week:5.1f}{flag}")

    # 2) Related queries that are rising fast or marked "Breakout".
    rising = [i for i in items if i["recordType"] == "relatedQuery" and i["rankType"] == "rising"]
    print("\nBreakout / fast-rising related queries:")
    for i in sorted(rising, key=lambda i: (not i.get("isBreakout"), -(i.get("value") or 0)))[:15]:
        label = "Breakout" if i.get("isBreakout") else f'+{i["value"]}%'
        print(f'  [{i["searchTerm"]}] {i["query"]}  ({label})')


if __name__ == "__main__":
    {"setup": setup, "check": check}[sys.argv[1] if len(sys.argv) > 1 else "check"]()

Variations

  • Seasonality: timeRange: "today 5-y" with includeInterestByRegion: True shows when and where demand peaks each year.
  • Content planning: feed the Breakout queries into your editorial backlog every Monday.
  • Other Google properties: property can be YouTube, News, Images or Shopping searches instead of web search.
  • Ask an assistant: the same Actor is available as an MCP tool for Claude, Cursor and other clients. Configs are in atlas-data-mcp.

Every input and output field is documented on the Store page and in the GitHub repo.

來源：DEV Community · dev.to

#教學/實踐#搜尋#部署/工程#Google