#!/usr/bin/env python3 """Rate-limit-resilient Google Trends capture using a fresh cookie session per widget.""" from __future__ import annotations import json from datetime import UTC, datetime from pathlib import Path from typing import Any from fetch_google_trends import ( ANTI_XSSI, ROOT, SPEC_PATH, TrendsClient, TrendsFetchError, api_url, derive_geography, derive_related_queries, derive_timeseries, explore_url, sha256_bytes, slug, widget_data_url, write_csv, write_json, ) def parse_payload(body: bytes) -> dict[str, Any]: return json.loads(ANTI_XSSI.sub("", body.decode("utf-8"), count=1)) def request_payload(group: dict[str, Any], period: str, spec: dict[str, Any]) -> dict[str, Any]: return { "comparisonItem": [ {"keyword": item["keyword"], "geo": group["geo"], "time": period} for item in group["items"] ], "category": spec["category"], "property": spec["search_property"], } def fresh_widget( *, group: dict[str, Any], period: str, spec: dict[str, Any], widget_id: str, error_log: Path, request_log: list[dict[str, Any]], ) -> tuple[dict[str, Any], dict[str, Any], str]: client = TrendsClient(error_log, min_interval_seconds=0.35) client.bootstrap() explore_request = request_payload(group, period, spec) explore_api = api_url( "/trends/api/explore", { "hl": group["hl"], "tz": spec["timezone_offset_minutes"], "req": explore_request, }, ) explore_payload, explore_result = client.fetch_json(explore_api) widgets = {widget["id"]: widget for widget in explore_payload.get("widgets", [])} widget = widgets.get(widget_id) if widget is None: request_log.extend(client.request_log) raise TrendsFetchError(f"widget {widget_id} absent for {group['id']} {period}") widget_url = widget_data_url( widget, hl=group["hl"], tz=spec["timezone_offset_minutes"] ) result = client.fetch(widget_url, retries=1) request_log.extend(client.request_log) return parse_payload(result.body), explore_payload, explore_result.fetched_at_utc def main() -> int: spec = json.loads(SPEC_PATH.read_text(encoding="utf-8")) started = datetime.now(UTC) run_id = started.strftime("%Y%m%dT%H%M%SZ") + "-resilient" run_dir = ROOT / "runs" / run_id raw_dir = run_dir / "raw" derived_dir = run_dir / "derived" run_dir.mkdir(parents=True) error_log = run_dir / "errors.jsonl" error_log.touch() request_log: list[dict[str, Any]] = [] failures: list[dict[str, str]] = [] artifacts: list[dict[str, Any]] = [] autocomplete_client = TrendsClient(error_log, min_interval_seconds=0.35) autocomplete_client.bootstrap() for query in spec["autocomplete_checks"]: url = ( "https://trends.google.com/trends/api/autocomplete/" + __import__("urllib.parse").parse.quote(query, safe="") + "?hl=en-US" ) try: payload, result = autocomplete_client.fetch_json(url) path = raw_dir / "autocomplete" / f"{slug(query)}.json" write_json(path, {"query": query, "endpoint": url, "response": payload}) artifacts.append( { "path": str(path.relative_to(ROOT)), "source_url": result.url, "fetched_at_utc": result.fetched_at_utc, "sha256": sha256_bytes(path.read_bytes()), } ) except TrendsFetchError as exc: failures.append({"stage": "autocomplete", "query": query, "error": str(exc)}) request_log.extend(autocomplete_client.request_log) for group in spec["groups"]: for period_id, period in spec["periods"].items(): base_raw = raw_dir / group["id"] / period_id base_derived = derived_dir / group["id"] / period_id try: payload, explore_payload, fetched_at = fresh_widget( group=group, period=period, spec=spec, widget_id="TIMESERIES", error_log=error_log, request_log=request_log, ) explore_path = base_raw / "explore_response.json" write_json( explore_path, { "request": request_payload(group, period, spec), "human_explore_url": explore_url(group, period), "fetched_at_utc": fetched_at, "response": explore_payload, }, ) path = base_raw / "timeseries.json" write_json(path, {"response": payload}) rows, summaries = derive_timeseries(payload, group) write_csv( base_derived / "interest_over_time.csv", [ "timestamp_utc", "period_label", "item", "kind", "value", "has_data", "is_partial", ], rows, ) write_csv( base_derived / "summary.csv", [ "item", "kind", "google_average_index", "computed_mean_index", "median_index", "latest_complete_index", "peak_index", "peak_timestamp_utc", "nonzero_share", "last_13_points_mean", "previous_13_points_mean", "last_vs_previous_13_pct", "complete_points", ], summaries, ) for artifact_path in (explore_path, path): artifacts.append( { "path": str(artifact_path.relative_to(ROOT)), "sha256": sha256_bytes(artifact_path.read_bytes()), } ) except TrendsFetchError as exc: failures.append( { "stage": "timeseries", "group": group["id"], "period": period_id, "error": str(exc), } ) if period_id != "five_years": continue if group.get("fetch_geography"): try: payload, _, _ = fresh_widget( group=group, period=period, spec=spec, widget_id="GEO_MAP", error_log=error_log, request_log=request_log, ) path = base_raw / "geography_compared.json" write_json(path, {"response": payload}) write_csv( base_derived / "geography_compared.csv", ["geo_code", "geo_name", "item", "value", "has_data"], derive_geography(payload, group), ) artifacts.append( { "path": str(path.relative_to(ROOT)), "sha256": sha256_bytes(path.read_bytes()), } ) except TrendsFetchError as exc: failures.append( { "stage": "geography", "group": group["id"], "period": period_id, "error": str(exc), } ) related_rows: list[dict[str, Any]] = [] for index in group.get("related_indices", []): item = group["items"][index] try: payload, _, _ = fresh_widget( group=group, period=period, spec=spec, widget_id=f"RELATED_QUERIES_{index}", error_log=error_log, request_log=request_log, ) path = base_raw / "related_queries" / f"{index}-{slug(item['label'])}.json" write_json(path, {"response": payload}) related_rows.extend(derive_related_queries(payload, item["label"])) artifacts.append( { "path": str(path.relative_to(ROOT)), "sha256": sha256_bytes(path.read_bytes()), } ) except TrendsFetchError as exc: failures.append( { "stage": "related_queries", "group": group["id"], "period": period_id, "item": item["label"], "error": str(exc), } ) if group.get("related_indices"): write_csv( base_derived / "related_queries.csv", [ "source_item", "metric", "rank", "query", "value", "formatted_value", "link", ], related_rows, ) finished = datetime.now(UTC) manifest = { "schema_version": 1, "run_id": run_id, "started_at_utc": started.isoformat(), "finished_at_utc": finished.isoformat(), "local_timezone_context": "Asia/Ho_Chi_Minh", "trends_timezone_offset_minutes": 0, "source": "Official Google Trends Explore web endpoints at trends.google.com", "session_strategy": ( "Fresh Google cookie session per Explore widget; Google first returned " "429 while setting NID, then 200 on immediate cookie-preserving retry." ), "normalization_warning": ( "Sampled 0-100 relative indexes, independently scaled per request. " "Zero means insufficient data or rounded index zero, not proven zero demand." ), "request_count": len(request_log), "failure_count": len(failures), "failures": failures, "requests": request_log, "artifacts": artifacts, } write_json(run_dir / "manifest.json", manifest) (ROOT / "latest_run.txt").write_text(run_id + "\n", encoding="utf-8") print(run_dir) print(f"requests={len(request_log)} failures={len(failures)}") return 0 if not failures else 2 if __name__ == "__main__": raise SystemExit(main())