#!/usr/bin/env python3 """Capture a preregistered Google Trends matrix through SerpApi.""" from __future__ import annotations import argparse import csv import hashlib import json import math import os import re import statistics import time import urllib.error import urllib.parse import urllib.request from datetime import UTC, datetime from pathlib import Path from typing import Any, NamedTuple SERPAPI_ENDPOINT = "https://serpapi.com/search.json" SERPAPI_DOCS = "https://serpapi.com/google-trends-api" class Job(NamedTuple): stage: str group: dict[str, Any] period_id: str period: str item_index: int | None = None class CollectionError(RuntimeError): """Raised when a provider response cannot be safely collected.""" def build_jobs(spec: dict[str, Any]) -> list[Job]: """Expand the preregistered matrix into provider requests.""" jobs: list[Job] = [] for group in spec["groups"]: for period_id, period in spec["periods"].items(): jobs.append(Job("timeseries", group, period_id, period)) if period_id != "five_years": continue if group.get("fetch_geography"): jobs.append(Job("geography", group, period_id, period)) for item_index in group.get("related_indices", []): jobs.append( Job("related_queries", group, period_id, period, item_index) ) return jobs def redact_secret(value: Any, secret: str) -> Any: """Remove API-key fields and redact the key if a provider echoes it.""" if isinstance(value, dict): return { key: redact_secret(item, secret) for key, item in value.items() if str(key).lower() not in {"api_key", "serpapi_api_key"} } if isinstance(value, list): return [redact_secret(item, secret) for item in value] if isinstance(value, str) and secret: return value.replace(secret, "[REDACTED]") return value def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) def write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) def slug(value: str) -> str: normalized = re.sub(r"[^a-zA-Z0-9]+", "-", value).strip("-").lower() return normalized or "item" def _entries_by_query( values: list[dict[str, Any]], expected_queries: set[str] ) -> dict[str, dict[str, Any]]: entries: dict[str, dict[str, Any]] = {} for entry in values: query = str(entry.get("query", "")) if query in entries: raise CollectionError(f"duplicate provider query: {query!r}") entries[query] = entry if set(entries) != expected_queries: raise CollectionError( "provider query set mismatch: " f"expected={sorted(expected_queries)!r} " f"actual={sorted(entries)!r}" ) return entries def derive_timeseries( payload: dict[str, Any], group: dict[str, Any] ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Create long-form observations and per-term summaries.""" section = payload.get("interest_over_time", {}) timeline = section.get("timeline_data", []) items = group["items"] rows: list[dict[str, Any]] = [] values_by_item: list[list[tuple[str, int]]] = [[] for _ in items] expected_queries = {item["keyword"] for item in items} for point in timeline: entries = _entries_by_query(point.get("values", []), expected_queries) is_partial = bool(point.get("partial_data", False)) timestamp = datetime.fromtimestamp(int(point["timestamp"]), UTC).isoformat() for index, item in enumerate(items): entry = entries[item["keyword"]] value = int(entry.get("extracted_value", 0) or 0) has_data = str(entry.get("value", "")) != "" row = { "timestamp_utc": timestamp, "period_label": point.get("date", ""), "item": item["label"], "kind": item["kind"], "value": value, "has_data": "true" if has_data else "unknown", "is_partial": str(is_partial).lower(), } rows.append(row) if not is_partial: values_by_item[index].append((timestamp, value)) provider_averages = { str(entry.get("query", "")): entry.get("value", "") for entry in section.get("averages", []) } summaries: list[dict[str, Any]] = [] for index, item in enumerate(items): observations = values_by_item[index] numeric = [value for _, value in observations] peak_time, peak_value = ( max(observations, key=lambda pair: pair[1]) if observations else ("", 0) ) recent = numeric[-13:] previous = numeric[-26:-13] recent_mean = statistics.fmean(recent) if recent else math.nan previous_mean = statistics.fmean(previous) if previous else math.nan summaries.append( { "item": item["label"], "kind": item["kind"], "google_average_index": provider_averages.get(item["keyword"], ""), "computed_mean_index": round(statistics.fmean(numeric), 3) if numeric else "", "median_index": round(statistics.median(numeric), 3) if numeric else "", "latest_complete_index": numeric[-1] if numeric else "", "peak_index": peak_value, "peak_timestamp_utc": peak_time, "nonzero_share": round( sum(value > 0 for value in numeric) / len(numeric), 4 ) if numeric else "", "last_13_points_mean": round(recent_mean, 3) if recent else "", "previous_13_points_mean": round(previous_mean, 3) if previous else "", "last_vs_previous_13_pct": round( (recent_mean / previous_mean - 1) * 100, 2 ) if previous and previous_mean > 0 else "", "complete_points": len(numeric), } ) return rows, summaries def derive_geography( payload: dict[str, Any], group: dict[str, Any] ) -> list[dict[str, Any]]: """Create a long-form country comparison.""" rows: list[dict[str, Any]] = [] expected_queries = {item["keyword"] for item in group["items"]} for entry in payload.get("compared_breakdown_by_region", []): entries = _entries_by_query(entry.get("values", []), expected_queries) for item in group["items"]: value_entry = entries[item["keyword"]] value = int(value_entry.get("extracted_value", 0) or 0) has_data = str(value_entry.get("value", "")) != "" rows.append( { "geo_code": entry.get("geo", ""), "geo_name": entry.get("location", ""), "item": item["label"], "value": value, "has_data": "true" if has_data else "unknown", } ) return rows def derive_related_queries( payload: dict[str, Any], item_label: str ) -> list[dict[str, Any]]: """Create top and rising related-query rows.""" rows: list[dict[str, Any]] = [] section = payload.get("related_queries", {}) for metric in ("top", "rising"): for rank, entry in enumerate(section.get(metric, []), start=1): rows.append( { "source_item": item_label, "metric": metric, "rank": rank, "query": entry.get("query", ""), "value": entry.get("extracted_value", ""), "formatted_value": entry.get("value", ""), "link": entry.get("link", ""), } ) return rows def job_params(job: Job, spec: dict[str, Any]) -> dict[str, str]: group = job.group items = group["items"] selected_items = [items[job.item_index]] if job.item_index is not None else items params = { "engine": "google_trends", "q": ",".join(item["keyword"] for item in selected_items), "date": job.period, "tz": str(spec["timezone_offset_minutes"]), "hl": str(group["hl"]).split("-", maxsplit=1)[0], "cat": str(spec["category"]), "no_cache": "true", "data_type": { "timeseries": "TIMESERIES", "geography": "GEO_MAP", "related_queries": "RELATED_QUERIES", }[job.stage], } if group.get("geo"): params["geo"] = group["geo"] if job.stage == "geography": params["region"] = "COUNTRY" if spec.get("search_property"): params["gprop"] = spec["search_property"] return params def public_request_url(params: dict[str, str]) -> str: return SERPAPI_ENDPOINT + "?" + urllib.parse.urlencode(params) def classify_payload(payload: dict[str, Any]) -> str: """Classify a provider payload without conflating no-data with failure.""" status = payload.get("search_metadata", {}).get("status") error = str(payload.get("error", "") or "") normalized_error = error.lower() if status == "Success" and ( "hasn't returned any results" in normalized_error or "has not returned any results" in normalized_error ): return "ok_no_data" if status == "Success" and not error: return "ok_with_data" raise CollectionError(f"provider status={status!r} error={error!r}") class SerpApiClient: def __init__( self, api_key: str, timeout_seconds: float = 45, attempts: int = 3 ) -> None: self.api_key = api_key self.timeout_seconds = timeout_seconds self.attempts = attempts def fetch(self, params: dict[str, str]) -> tuple[dict[str, Any], dict[str, Any]]: private_params = {**params, "api_key": self.api_key} url = SERPAPI_ENDPOINT + "?" + urllib.parse.urlencode(private_params) last_status: int | None = None for attempt in range(1, self.attempts + 1): try: request = urllib.request.Request( url, headers={ "Accept": "application/json", "User-Agent": "global-validation-trends-collector/1.0", }, ) with urllib.request.urlopen( request, timeout=self.timeout_seconds ) as response: body = response.read() payload = json.loads(body.decode("utf-8")) if not isinstance(payload, dict): raise CollectionError("provider returned non-object JSON") payload = redact_secret(payload, self.api_key) terminal_status = classify_payload(payload) fetched_at = datetime.now(UTC).isoformat() return payload, { "params": params, "public_request_url": public_request_url(params), "fetched_at_utc": fetched_at, "terminal_status": terminal_status, "search_metadata": payload.get("search_metadata", {}), } except urllib.error.HTTPError as error: last_status = error.code if error.code not in {429, 500, 502, 503, 504}: raise CollectionError( f"provider HTTP status {error.code}" ) from None retry_after = error.headers.get("Retry-After") delay = ( min(float(retry_after), 30) if retry_after and retry_after.isdigit() else min(2 ** (attempt - 1), 8) ) except urllib.error.URLError as error: if attempt == self.attempts: reason = type(error.reason).__name__ raise CollectionError( f"provider connection failed: {reason}" ) from None delay = min(2 ** (attempt - 1), 8) except (json.JSONDecodeError, TimeoutError) as error: if attempt == self.attempts: raise CollectionError( f"provider response failed: {type(error).__name__}" ) from None delay = min(2 ** (attempt - 1), 8) if attempt < self.attempts: time.sleep(delay) raise CollectionError( f"provider failed after {self.attempts} attempts" + (f" with HTTP {last_status}" if last_status else "") ) def _artifact(path: Path, root: Path, **metadata: Any) -> dict[str, Any]: return { "path": str(path.relative_to(root)), "sha256": sha256_bytes(path.read_bytes()), **metadata, } def collect( spec_path: Path, output_root: Path, api_key: str ) -> tuple[Path, dict[str, Any]]: """Run the matrix and write raw, derived, and manifest artifacts.""" spec = json.loads(spec_path.read_text(encoding="utf-8")) started = datetime.now(UTC) run_id = started.strftime("%Y%m%dT%H%M%SZ") + "-serpapi" run_dir = output_root / "runs" / run_id raw_dir = run_dir / "raw" derived_dir = run_dir / "derived" run_dir.mkdir(parents=True, exist_ok=False) errors_path = run_dir / "errors.jsonl" errors_path.touch() spec_snapshot = run_dir / "query_spec.json" write_json(spec_snapshot, spec) artifacts = [_artifact(spec_snapshot, output_root)] failures: list[dict[str, Any]] = [] requests: list[dict[str, Any]] = [] related_rows: dict[tuple[str, str], list[dict[str, Any]]] = {} client = SerpApiClient(api_key) for job in build_jobs(spec): group_id = job.group["id"] params = job_params(job, spec) item = ( job.group["items"][job.item_index] if job.item_index is not None else None ) request_context = { "stage": job.stage, "group": group_id, "period": job.period_id, **({"item": item["label"]} if item else {}), } try: payload, request_record = client.fetch(params) requests.append({**request_context, **request_record}) base_raw = raw_dir / group_id / job.period_id base_derived = derived_dir / group_id / job.period_id if job.stage == "timeseries": raw_path = base_raw / "timeseries.json" rows, summaries = derive_timeseries(payload, job.group) observations_path = base_derived / "interest_over_time.csv" summary_path = base_derived / "summary.csv" write_csv( observations_path, [ "timestamp_utc", "period_label", "item", "kind", "value", "has_data", "is_partial", ], rows, ) write_csv( summary_path, [ "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, ) artifacts.extend( [ _artifact(observations_path, output_root), _artifact(summary_path, output_root), ] ) elif job.stage == "geography": raw_path = base_raw / "geography_compared.json" geography_path = base_derived / "geography_compared.csv" write_csv( geography_path, ["geo_code", "geo_name", "item", "value", "has_data"], derive_geography(payload, job.group), ) artifacts.append(_artifact(geography_path, output_root)) else: assert item is not None and job.item_index is not None raw_path = ( base_raw / "related_queries" / f"{job.item_index}-{slug(item['label'])}.json" ) related_rows.setdefault((group_id, job.period_id), []).extend( derive_related_queries(payload, item["label"]) ) write_json( raw_path, { "provider": "SerpApi Google Trends API", "provider_docs": SERPAPI_DOCS, "request": params, "fetched_at_utc": request_record["fetched_at_utc"], "response": payload, }, ) artifacts.append( _artifact( raw_path, output_root, source_url=request_record["public_request_url"], fetched_at_utc=request_record["fetched_at_utc"], ) ) except CollectionError as error: failure = {**request_context, "error": str(error)} failures.append(failure) with errors_path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(failure, ensure_ascii=False) + "\n") for (group_id, period_id), rows in related_rows.items(): path = derived_dir / group_id / period_id / "related_queries.csv" write_csv( path, [ "source_item", "metric", "rank", "query", "value", "formatted_value", "link", ], rows, ) artifacts.append(_artifact(path, output_root)) artifacts.append(_artifact(errors_path, output_root)) finished = datetime.now(UTC) status_counts = { status: sum(request["terminal_status"] == status for request in requests) for status in ("ok_with_data", "ok_no_data") } 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": spec["timezone_offset_minutes"], "source": "SerpApi Google Trends API", "source_classification": ( "Third-party parser of Google Trends UI data; not the " "invite-only official Google Trends API alpha." ), "normalization_warning": ( "Sampled 0-100 relative indexes, independently scaled per " "request. Zero means insufficient data or rounded index zero, " "not proven zero demand. SerpApi does not expose the original " "Google hasData flag; an explicit provider value including '0' " "is marked available, while an empty or absent value is unknown." ), "planned_job_count": len(build_jobs(spec)), "completed_job_count": len(requests), "successful_job_count": len(requests), "terminal_status_counts": status_counts, "failure_count": len(failures), "failures": failures, "requests": requests, "artifacts": artifacts, } manifest_path = run_dir / "manifest.json" write_json(manifest_path, manifest) if not failures and len(requests) == len(build_jobs(spec)): (output_root / "latest_run.txt").write_text(run_id + "\n", encoding="utf-8") return run_dir, manifest def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--spec", required=True, type=Path) parser.add_argument("--output-root", required=True, type=Path) return parser.parse_args() def main() -> int: args = parse_args() api_key = os.environ.get("SERPAPI_API_KEY", "") if not api_key: print("SERPAPI_API_KEY is required") return 2 run_dir, manifest = collect(args.spec, args.output_root, api_key) print(run_dir) print( f"planned={manifest['planned_job_count']} " f"completed={manifest['completed_job_count']} " f"failures={manifest['failure_count']}" ) return 0 if not manifest["failures"] else 2 if __name__ == "__main__": raise SystemExit(main())