bobseares-2/google-trends-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

deploy

Browse Source
This commit is contained in:
a2a-platform
2026-06-12 18:12:21 +00:00
commit 2684a2fdbc
7 changed files with 1136 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
.DS_Store vendored Normal file
View File

Binary file not shown.

198
.a2a/agent.dsl.json Normal file
View File

@@ -0,0 +1,198 @@
{
"schema_version": "2026-06-04",
"language": "python",
"name": "google-trends-agent",
"description": "Fetch Google Trends pages with SeleniumBase CDP Mode",
"version": "0.1.0",
"entrypoint": {
"module": "agent",
"class_name": "GoogleTrendsAgent",
"function": null,
"command": []
},
"skills": [
{
"name": "fetch",
"description": "Open Google Trends with SeleniumBase CDP Mode",
"handler": "fetch",
"tags": [],
"scopes": [],
"stream": false,
"policy": {
"timeout_seconds": null,
"idempotent": false,
"max_retries": 0,
"cost_class": null,
"allow_scope_expansion": false,
"grant_mode": null,
"grant_allow_patterns": [],
"grant_deny_patterns": [],
"grant_outputs_prefix": null,
"grant_write_prefixes": [],
"grant_ttl_seconds": null,
"grant_run_timeout_seconds": null,
"grant_approval_timeout_seconds": null,
"grant_scope_approval_timeout_seconds": null
},
"input_schema": {
"type": "object",
"properties": {
"task": {
"type": "string"
},
"date": {
"type": "string"
},
"geo": {
"type": "string"
},
"stealth_warmup": {
"type": "boolean"
}
},
"required": [],
"additionalProperties": false
},
"output_schema": {
"type": "string"
}
},
{
"name": "fetch_structured",
"description": "Open Google Trends, capture raw HTML/PDF, and use a DeepAgent to extract structured trend data.",
"handler": "fetch_structured",
"tags": [],
"scopes": [],
"stream": false,
"policy": {
"timeout_seconds": 240.0,
"idempotent": false,
"max_retries": 0,
"cost_class": null,
"allow_scope_expansion": false,
"grant_mode": null,
"grant_allow_patterns": [],
"grant_deny_patterns": [],
"grant_outputs_prefix": null,
"grant_write_prefixes": [],
"grant_ttl_seconds": null,
"grant_run_timeout_seconds": null,
"grant_approval_timeout_seconds": null,
"grant_scope_approval_timeout_seconds": null
},
"input_schema": {
"type": "object",
"properties": {
"task": {
"type": "string"
},
"date": {
"type": "string"
},
"geo": {
"type": "string"
},
"extraction_goal": {
"type": "string"
},
"refresh_once": {
"type": "boolean"
},
"stealth_warmup": {
"type": "boolean"
}
},
"required": [],
"additionalProperties": false
},
"output_schema": {
"additionalProperties": true,
"type": "object"
}
}
],
"capabilities": {},
"input_modes": [
"application/json"
],
"output_modes": [
"application/json"
],
"required_secrets": [],
"required_env": [],
"consumer_setup": {
"fields": []
},
"runtime": {
"lifecycle": "ephemeral",
"availability": "on_demand",
"state": "none",
"sandbox": "microsandbox",
"resources": {
"cpu": "100m",
"memory": "256Mi",
"gpu": 0,
"max_runtime_seconds": 600
},
"concurrency": 1,
"egress": {
"allow_hosts": [],
"allow_internal_services": [],
"deny_internet_by_default": true
},
"tools_used": [
"seleniumbase",
"deepagents"
],
"llm_provisioning": "platform",
"pricing": {
"price_per_call_usd": 0.0,
"caller_pays_llm": true,
"notes": ""
},
"wants_cp_jwt": false,
"platform_resources": {
"memory": null,
"databases": []
},
"apt_packages": []
},
"template_lineage": null,
"meta_agent_manifest": null,
"state_schema": null,
"workspace_access": {
"enabled": false,
"max_files": 0,
"allowed_modes": [],
"require_reason": true,
"deny_patterns": [],
"require_human_approval": false,
"max_total_size_bytes": 104857600
},
"config_schema": {
"properties": {},
"title": "GoogleTrendsAgentConfig",
"type": "object"
},
"auth": {
"model": "a2a_pack.auth.NoAuth",
"strategy": "public",
"principal_schema": {
"additionalProperties": false,
"description": "Public agent: no caller identity required.",
"properties": {},
"title": "NoAuth",
"type": "object"
},
"resolver": null,
"required": false
},
"metadata": {
"source": "python-a2a-pack",
"project_manifest": {
"name": "google-trends-agent",
"version": "0.1.0",
"entrypoint": "agent:GoogleTrendsAgent"
}
}
}

51
README.md Normal file
View File

@@ -0,0 +1,51 @@
# google-trends-agent
Google Trends fetcher that uses SeleniumBase UC/CDP mode with Playwright.
The A2A entrypoint exposes one skill, `fetch`, with:
- `task`: a Google Trends URL or plain query string
- `date`: optional Trends date window, default `now 1-d`
- `geo`: optional geo code, default `BR`
Default task:
```text
https://trends.google.com/trends/explore?q=copa%20do%20mundo%20da%20fifa%202026&date=now%201-d&geo=BR
```
## Install
```bash
python -m pip install -r requirements.txt
playwright install chromium
```
## Run Directly
```bash
python agent.py
python agent.py "https://trends.google.com/trends/explore?q=wow&hl=en-US"
python agent.py "copa do mundo da fifa 2026" --date "now 1-d" --geo BR
```
The script starts SeleniumBase with `SB(uc=True, test=True, use_chromium=True)`,
uses the SeleniumBase UC reconnect path by default, activates CDP mode, attaches
Playwright over CDP, performs SeleniumBase GUI mouse movement plus
mouse/scroll/keyboard warmup, and prints JSON with the requested URL, final URL,
page title, visible text, and a screenshot path under `outputs/google-trends.png`.
Pass `--chrome` to use system Chrome instead of SeleniumBase Chromium.
Pass `--incognito` or `--guest` only when you explicitly want those browser
modes.
Pass `--no-humanize` to skip the mouse, scroll, and keyboard warmup.
Pass `--open-mode cdp-goto` to use the direct CDP navigation path instead of
the default SeleniumBase UC reconnect path. `--reconnect-time` controls the UC
disconnect window used by `uc_open_with_reconnect()`.
## Run as A2A
```bash
a2a dev
a2a test --invoke --skill fetch --args-json '{"task":"https://trends.google.com/trends/explore?q=wow&hl=en-US"}'
```

56
a2a.yaml Normal file
View File

@@ -0,0 +1,56 @@
# Project identity for `a2a deploy`. Most metadata (resources, scopes,
# secrets, workspace, etc.) lives on the Python class — this file only
# tells the CLI how to find it.
name: google-trends-agent
version: 0.1.0
entrypoint: agent:GoogleTrendsAgent
expose:
public: true
runtime:
resources:
cpu: 1000m
memory: 2Gi
max_runtime_seconds: 600
egress:
allow_hosts:
- "*"
deny_internet_by_default: false
apt_packages:
- ca-certificates
- fonts-liberation
- libasound2
- libatk-bridge2.0-0
- libatk1.0-0
- libcairo2
- libcups2
- libdbus-1-3
- libdrm2
- libgbm1
- libglib2.0-0
- libgtk-3-0
- libnspr4
- libnss3
- libpango-1.0-0
- libx11-6
- libxcb1
- libxcomposite1
- libxdamage1
- libxext6
- libxfixes3
- libxkbcommon0
- libxrandr2
- xdg-utils
install_commands:
- python -m playwright install chromium
# Optional platform resources:
# resources:
# memory:
# tiers: [kv, vector]
# namespace: google-trends-agent
# databases:
# - name: app
# provider: neon
# engine: postgres
# env:
# url: DATABASE_URL

827
agent.py Normal file
View File

@@ -0,0 +1,827 @@
"""Google Trends fetcher using SeleniumBase CDP Mode."""
from __future__ import annotations
import argparse
import asyncio
import inspect
import json
import random
import re
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
from a2a_pack import A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill
from pydantic import BaseModel, Field
DEFAULT_QUERY = "copa do mundo da fifa 2026"
DEFAULT_DATE = "now 1-d"
DEFAULT_GEO = "BR"
DEFAULT_URL = "https://trends.google.com/trends/explore"
DEFAULT_OUTPUT_DIR = Path("outputs")
WARMUP_URL = "https://example.com"
GOOGLE_URL = "https://www.google.com/"
class GoogleTrendsAgentConfig(BaseModel):
pass
@dataclass
class TrendsResult:
requested_url: str
final_url: str
title: str
visible_text: str
raw_html: str
raw_html_path: str | None
screenshot: str | None
pdf: str | None
pdf_text: str
pdf_text_path: str | None
extracted_data: dict[str, Any]
log: list[str]
class ReadCaptureInput(BaseModel):
offset: int = Field(default=0, ge=0, description="Start character offset.")
max_chars: int = Field(
default=40_000,
ge=1,
le=120_000,
description="Maximum characters to return.",
)
class GoogleTrendsAgent(A2AAgent[GoogleTrendsAgentConfig, NoAuth]):
name = "google-trends-agent"
description = "Fetch Google Trends pages with SeleniumBase CDP Mode"
version = "0.1.0"
config_model = GoogleTrendsAgentConfig
auth_model = NoAuth
llm_provisioning = LLMProvisioning.PLATFORM
pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=True)
tools_used = ("seleniumbase", "deepagents")
@skill(description="Open Google Trends with SeleniumBase CDP Mode")
async def fetch(
self,
ctx: RunContext[NoAuth],
task: str = DEFAULT_QUERY,
date: str = DEFAULT_DATE,
geo: str = DEFAULT_GEO,
stealth_warmup: bool = True,
) -> str:
await ctx.emit_progress("launching CDP Mode browser")
result = await asyncio.to_thread(
fetch_google_trends,
task=task,
date=date,
geo=geo,
output_dir=DEFAULT_OUTPUT_DIR,
stealth_warmup=stealth_warmup,
)
await ctx.emit_progress("google trends fetch complete")
return json.dumps(_trends_result_payload(result), ensure_ascii=False, indent=2)
@skill(
description=(
"Open Google Trends, capture raw HTML/PDF, and use a DeepAgent to "
"extract structured trend data."
),
timeout_seconds=240,
)
async def fetch_structured(
self,
ctx: RunContext[NoAuth],
task: str = DEFAULT_QUERY,
date: str = DEFAULT_DATE,
geo: str = DEFAULT_GEO,
extraction_goal: str = (
"Extract Google Trends structured data: query, date window, geo, "
"interest over time points, subregion rankings, related topics, "
"related queries, and any warnings or data gaps."
),
refresh_once: bool = False,
stealth_warmup: bool = True,
) -> dict[str, Any]:
creds = ctx.llm
if not creds.api_key:
return {
"error": "llm_credentials_missing",
"final": (
"LLM key required. Add an LLM credential in Settings > "
"LLM credentials before running fetch_structured."
),
}
try:
from a2a_pack.deepagents import create_a2a_deep_agent
from langchain_core.tools import StructuredTool
except Exception as exc:
return {
"error": "deepagents_runtime_missing",
"final": (
"The browser/PDF capture path is installed, but this "
"runtime does not provide the DeepAgent helper stack used "
"by agents/blog-openapi-agent."
),
"detail": f"{type(exc).__name__}: {exc}",
}
await ctx.emit_progress("capturing Google Trends page, raw HTML, and PDF")
result = await asyncio.to_thread(
fetch_google_trends,
task=task,
date=date,
geo=geo,
output_dir=DEFAULT_OUTPUT_DIR,
refresh_once=refresh_once,
stealth_warmup=stealth_warmup,
)
await ctx.emit_progress("extracting structured data with DeepAgent")
graph = create_a2a_deep_agent(
ctx,
creds=creds,
tools=_capture_tools(result, StructuredTool),
system_prompt=_structured_extraction_prompt(),
)
deep_result = await graph.ainvoke(
{
"messages": [
{
"role": "user",
"content": (
f"{extraction_goal}\n\n"
"Use the capture reader tools. Return only one JSON "
"object matching the requested schema."
),
}
]
},
{"recursion_limit": 40},
)
messages = deep_result.get("messages", []) if isinstance(deep_result, dict) else []
final = _message_text(messages[-1]) if messages else deep_result
structured = _parse_json_object(final)
return {
"requested_url": result.requested_url,
"final_url": result.final_url,
"title": result.title,
"raw_html_path": result.raw_html_path,
"pdf": result.pdf,
"pdf_text_path": result.pdf_text_path,
"structured_data": structured,
"final": final,
"messages": [_message_to_dict(message) for message in messages[-8:]],
"capture_log": result.log,
}
def build_trends_url(task: str, date: str = DEFAULT_DATE, geo: str = DEFAULT_GEO) -> str:
task = (task or DEFAULT_QUERY).strip()
date = (date or DEFAULT_DATE).strip()
geo = (geo or DEFAULT_GEO).strip()
if task.startswith(("http://", "https://")):
parsed = urlparse(task)
query = dict(parse_qsl(parsed.query, keep_blank_values=True))
query.setdefault("q", DEFAULT_QUERY)
query["date"] = date
query["geo"] = geo
return urlunparse(parsed._replace(query=urlencode(query), fragment=""))
return f"{DEFAULT_URL}?{urlencode({'q': task, 'date': date, 'geo': geo})}"
def fetch_google_trends(
*,
task: str = DEFAULT_QUERY,
date: str = DEFAULT_DATE,
geo: str = DEFAULT_GEO,
output_dir: Path | str | None = DEFAULT_OUTPUT_DIR,
refresh_once: bool = True,
use_chromium: bool = True,
stealth_warmup: bool = True,
) -> TrendsResult:
"""Load Trends in SeleniumBase CDP Mode, then inspect without hard failures."""
from seleniumbase import SB
requested_url = build_trends_url(task, date=date, geo=geo)
locale = _locale_from_url(requested_url)
out_dir = Path(output_dir) if output_dir else None
if out_dir:
out_dir.mkdir(parents=True, exist_ok=True)
log: list[str] = []
final_url = requested_url
title = ""
visible_text = ""
raw_html = ""
raw_html_path: str | None = None
screenshot: str | None = None
pdf: str | None = None
pdf_text = ""
pdf_text_path: str | None = None
with SB(uc=True, test=True, locale=locale, use_chromium=use_chromium) as sb:
log.append("browser: SB(uc=True, test=True) started")
if stealth_warmup:
log.extend(_stealth_google_warmup(sb, requested_url))
else:
log.extend(_open_requested_url_cdp(sb, requested_url))
_sleep(sb, 2.5, 4.0)
log.extend(_human_mouse_cdp(sb))
log.extend(_wait_for_initial_body_cdp(sb))
if refresh_once:
log.extend(_native_refresh(sb))
_sleep(sb, 6.0, 8.0)
log.extend(_human_mouse_cdp(sb))
log.extend(_dismiss_google_interstitials_cdp(sb))
has_challenge, reason = _has_challenge_cdp(sb)
if has_challenge:
log.append(f"captcha: marker found ({reason}); attempting helpers")
log.extend(_solve_captcha_cdp(sb))
_sleep(sb, 2.0, 3.5)
else:
log.append("captcha: no marker found; continuing")
log.extend(_wait_for_trends_cdp(sb))
log.extend(_dismiss_google_interstitials_cdp(sb))
final_url = _safe_cdp_call(sb, "get_current_url", requested_url)
title = _safe_cdp_call(sb, "get_title", "")
visible_text = _safe_cdp_call(sb, "get_text", "", "body")[:20_000]
raw_html = _safe_cdp_call(sb, "get_html", "")
extracted_data = extract_trends_data(visible_text)
if out_dir:
try:
html_path = out_dir / "google-trends.html"
html_path.write_text(raw_html, encoding="utf-8")
raw_html_path = str(html_path)
log.append(f"html: {raw_html_path}")
except Exception as exc:
log.append(f"html warning: {type(exc).__name__}: {exc}")
try:
sb.cdp.save_screenshot("google-trends.png", folder=str(out_dir))
screenshot = str(out_dir / "google-trends.png")
log.append(f"screenshot: {screenshot}")
except Exception as exc:
log.append(f"screenshot warning: {type(exc).__name__}: {exc}")
try:
sb.cdp.print_to_pdf("google-trends.pdf", folder=str(out_dir))
pdf = str(out_dir / "google-trends.pdf")
log.append(f"pdf: {pdf}")
pdf_text = _extract_pdf_text(Path(pdf))
pdf_text_file = out_dir / "google-trends-pdf.txt"
pdf_text_file.write_text(pdf_text, encoding="utf-8")
pdf_text_path = str(pdf_text_file)
log.append(f"pdf text: {pdf_text_path}")
except Exception as exc:
log.append(f"pdf warning: {type(exc).__name__}: {exc}")
return TrendsResult(
requested_url=requested_url,
final_url=final_url,
title=title,
visible_text=visible_text,
raw_html=raw_html,
raw_html_path=raw_html_path,
screenshot=screenshot,
pdf=pdf,
pdf_text=pdf_text,
pdf_text_path=pdf_text_path,
extracted_data=extracted_data,
log=log,
)
def _extract_pdf_text(pdf_path: Path) -> str:
try:
from pypdf import PdfReader
reader = PdfReader(str(pdf_path))
return "\n\n".join(page.extract_text() or "" for page in reader.pages)
except Exception:
return ""
def _trends_result_payload(result: TrendsResult) -> dict[str, Any]:
payload = asdict(result)
payload.pop("raw_html", None)
payload.pop("pdf_text", None)
return payload
def _open_requested_url_cdp(sb: Any, requested_url: str) -> list[str]:
log: list[str] = []
try:
sb.activate_cdp_mode(requested_url)
log.append("cdp: activate_cdp_mode(url)")
except Exception as exc:
log.append(f"cdp activate warning: {type(exc).__name__}: {exc}")
try:
sb.activate_cdp_mode()
sb.goto(requested_url)
log.append("cdp: activate_cdp_mode(); goto(url)")
except Exception as goto_exc:
log.append(f"cdp goto warning: {type(goto_exc).__name__}: {goto_exc}")
return log
def _stealth_google_warmup(sb: Any, requested_url: str) -> list[str]:
"""Warm up a UC/CDP browser with ordinary browsing before Trends."""
log: list[str] = []
try:
sb.uc_open_with_reconnect(WARMUP_URL, reconnect_time=random.uniform(2.5, 4.5))
log.append(f"warmup: uc_open_with_reconnect({WARMUP_URL})")
except Exception as exc:
log.append(f"warmup example warning: {type(exc).__name__}: {exc}")
log.extend(_open_requested_url_cdp(sb, WARMUP_URL))
_sleep(sb, 1.0, 2.2)
try:
sb.activate_cdp_mode()
log.append("warmup: activate_cdp_mode()")
except Exception as exc:
log.append(f"warmup cdp warning: {type(exc).__name__}: {exc}")
log.extend(_human_mouse_cdp(sb))
try:
sb.goto(GOOGLE_URL)
log.append(f"warmup: goto({GOOGLE_URL})")
except Exception as exc:
log.append(f"warmup google warning: {type(exc).__name__}: {exc}")
_sleep(sb, 1.4, 2.8)
log.extend(_dismiss_google_interstitials_cdp(sb))
log.extend(_type_google_search_one_key_at_a_time(sb, _search_query_from_url(requested_url)))
_sleep(sb, 1.8, 3.2)
try:
sb.goto(requested_url)
log.append("warmup: goto requested Trends URL")
except Exception as exc:
log.append(f"warmup trends warning: {type(exc).__name__}: {exc}")
log.extend(_open_requested_url_cdp(sb, requested_url))
return log
def _search_query_from_url(url: str) -> str:
query = dict(parse_qsl(urlparse(url).query, keep_blank_values=True))
return query.get("q") or DEFAULT_QUERY
def _type_google_search_one_key_at_a_time(sb: Any, query: str) -> list[str]:
log: list[str] = []
selectors = (
"textarea[name='q']",
"input[name='q']",
"textarea[title='Search']",
"input[title='Search']",
"textarea[aria-label='Pesquisar']",
"input[aria-label='Pesquisar']",
)
selector = ""
for candidate in selectors:
try:
sb.cdp.wait_for_element(candidate, timeout=3)
selector = candidate
break
except Exception:
continue
if not selector:
log.append("warmup type warning: Google search box not found")
return log
try:
sb.cdp.click(selector)
_sleep(sb, 0.3, 0.8)
for char in query:
sb.cdp.press_keys(selector, char)
_sleep(sb, 0.04, 0.18)
sb.cdp.press_keys(selector, "\n")
log.append("warmup: typed Google query one key at a time")
except Exception as exc:
log.append(f"warmup type warning: {type(exc).__name__}: {exc}")
return log
def _capture_tools(result: TrendsResult, structured_tool: Any) -> list[Any]:
async def read_raw_html(offset: int = 0, max_chars: int = 40_000) -> dict[str, Any]:
return _slice_capture(result.raw_html, offset, max_chars)
async def read_visible_text(offset: int = 0, max_chars: int = 40_000) -> dict[str, Any]:
return _slice_capture(result.visible_text, offset, max_chars)
async def read_pdf_text(offset: int = 0, max_chars: int = 40_000) -> dict[str, Any]:
return _slice_capture(result.pdf_text, offset, max_chars)
return [
structured_tool.from_function(
coroutine=read_raw_html,
name="read_raw_html",
description="Read the rendered Google Trends page HTML captured after browser execution.",
args_schema=ReadCaptureInput,
),
structured_tool.from_function(
coroutine=read_visible_text,
name="read_visible_text",
description="Read visible page text captured from the rendered Google Trends page.",
args_schema=ReadCaptureInput,
),
structured_tool.from_function(
coroutine=read_pdf_text,
name="read_pdf_text",
description="Read text extracted from the captured Google Trends PDF printout.",
args_schema=ReadCaptureInput,
),
]
def _slice_capture(text: str, offset: int, max_chars: int) -> dict[str, Any]:
offset = max(0, offset)
max_chars = max(1, max_chars)
chunk = text[offset : offset + max_chars]
next_offset = offset + len(chunk)
return {
"offset": offset,
"next_offset": next_offset if next_offset < len(text) else None,
"total_chars": len(text),
"content": chunk,
}
def _structured_extraction_prompt() -> str:
return "\n".join(
[
"You extract structured data from a captured Google Trends page.",
"Use read_visible_text, read_raw_html, and read_pdf_text as needed.",
"Do not invent values. If data is missing, use an empty list and add a warning.",
"Prefer exact timestamps, labels, ranking names, and numeric values from the capture.",
"Return only valid JSON, with this shape:",
json.dumps(
{
"query": "string",
"geo": "string",
"date_window": "string",
"interest_over_time": [
{"time": "string", "value": 0}
],
"geo_rankings": [
{"rank": 1, "region": "string", "value": 0}
],
"related_topics": [
{"rank": 1, "topic": "string", "value": "string"}
],
"related_queries": [
{"rank": 1, "query": "string", "value": "string"}
],
"warnings": ["string"],
"sources_used": ["visible_text", "raw_html", "pdf_text"],
},
ensure_ascii=False,
),
]
)
def _parse_json_object(value: Any) -> Any:
if not isinstance(value, str):
return value
text = value.strip()
if text.startswith("```"):
text = re.sub(r"^```(?:json)?\s*", "", text)
text = re.sub(r"\s*```$", "", text)
try:
return json.loads(text)
except ValueError:
match = re.search(r"\{.*\}", text, re.DOTALL)
if not match:
return None
try:
return json.loads(match.group(0))
except ValueError:
return None
def _message_text(message: Any) -> Any:
return getattr(message, "content", message)
def _message_to_dict(message: Any) -> dict[str, Any]:
try:
from langchain_core.messages import BaseMessage
if isinstance(message, BaseMessage):
return message.model_dump(mode="json")
except Exception:
pass
if hasattr(message, "model_dump"):
return message.model_dump(mode="json")
if isinstance(message, dict):
return message
return {"content": str(message)}
def extract_trends_data(text: str) -> dict[str, Any]:
"""Extract numeric and geographic Google Trends data from rendered text."""
return {
"interest_over_time": _extract_interest_over_time(text),
"geo_rankings": _extract_geo_rankings(text),
"related_queries": _extract_related_queries(text),
"related_topics": _extract_related_topics(text),
}
def _extract_interest_over_time(text: str) -> list[dict[str, Any]]:
pattern = re.compile(
r"(Jun|Jul|Aug|Sep|Oct|Nov|Dec|Jan|Feb|Mar|Apr|May)"
r"\s+\d{1,2}\s+at\s+\d{1,2}:\d{2}\s*(?:AM|PM)\s+(\d{1,3})"
)
points: list[dict[str, Any]] = []
for match in pattern.finditer(text):
label = match.group(0).rsplit(" ", 1)[0]
value = int(match.group(2))
points.append({"time": _clean_label(label), "value": value})
return points
def _extract_geo_rankings(text: str) -> list[dict[str, Any]]:
section = _section_between(text, "Interest by subregion", "Related topics")
rankings: list[dict[str, Any]] = []
pattern = re.compile(r"\b(\d{1,2})\s+([A-Za-zÀ-ÿ .'-]+?)\s+(\d{1,3})\b")
for rank, region, value in pattern.findall(section):
region = _clean_label(region)
if region.lower() in {"help", "top", "rising", "subregion", "city"}:
continue
rankings.append(
{"rank": int(rank), "region": region, "value": int(value)}
)
return rankings
def _extract_related_queries(text: str) -> list[dict[str, Any]]:
section = _section_between(text, "Related queries", "var _gaq")
queries: list[dict[str, Any]] = []
pattern = re.compile(
r"\b(\d{1,2})\s+(.+?)\s+(Breakout|\+\d+%|\d{1,3})\b",
re.IGNORECASE,
)
for rank, query, value in pattern.findall(section):
query = _clean_label(query)
if query and not query.lower().startswith(("help", "rising", "top")):
queries.append({"rank": int(rank), "query": query, "value": value})
return queries
def _extract_related_topics(text: str) -> list[dict[str, Any]]:
section = _section_between(text, "Related topics", "Related queries")
topics: list[dict[str, Any]] = []
pattern = re.compile(
r"\b(\d{1,2})\s+(.+?)\s+(Breakout|\+\d+%|\d{1,3})\b",
re.IGNORECASE,
)
for rank, topic, value in pattern.findall(section):
topic = _clean_label(topic)
if topic and not topic.lower().startswith(("help", "rising", "top")):
topics.append({"rank": int(rank), "topic": topic, "value": value})
return topics
def _section_between(text: str, start: str, end: str) -> str:
start_idx = text.find(start)
if start_idx == -1:
return ""
end_idx = text.find(end, start_idx + len(start))
if end_idx == -1:
end_idx = len(text)
return text[start_idx:end_idx]
def _clean_label(value: str) -> str:
value = value.replace("\u202a", "").replace("\u202c", "")
value = re.sub(r"\s+", " ", value)
return value.strip(" \n\t-._*^")
def _locale_from_url(url: str) -> str:
query = dict(parse_qsl(urlparse(url).query, keep_blank_values=True))
return query.get("hl") or ("pt-BR" if query.get("geo", "").upper() == "BR" else "en")
def _sleep(sb: Any, low: float, high: float) -> None:
sb.sleep(random.uniform(low, high))
def _human_mouse_cdp(sb: Any) -> list[str]:
log: list[str] = []
cdp = sb.cdp
try:
cdp.bring_active_window_to_front()
log.append("mouse: browser front")
except Exception as exc:
log.append(f"mouse front warning: {type(exc).__name__}: {exc}")
width, height = 1280, 840
try:
rect = cdp.get_window_rect()
width = int(rect.get("width") or width)
height = int(rect.get("height") or height)
except Exception:
pass
x = random.randint(max(120, width // 5), max(220, width - 260))
y = random.randint(max(140, height // 5), max(240, height - 190))
for _ in range(random.randint(5, 8)):
x = min(max(x + random.randint(-210, 240), 90), max(100, width - 90))
y = min(max(y + random.randint(-150, 170), 100), max(110, height - 100))
try:
cdp.gui_hover_x_y(x, y, timeframe=random.uniform(0.18, 0.45))
_sleep(sb, 0.18, 0.65)
except Exception as exc:
log.append(f"mouse hover warning: {type(exc).__name__}: {exc}")
break
for method_name, amount in (
("scroll_down", random.randint(80, 180)),
("scroll_up", random.randint(25, 90)),
):
try:
getattr(cdp, method_name)(amount)
_sleep(sb, 0.35, 0.9)
except Exception as exc:
log.append(f"{method_name} warning: {type(exc).__name__}: {exc}")
log.append("mouse: movement complete")
return log
def _native_refresh(sb: Any) -> list[str]:
log: list[str] = []
try:
sb.cdp.bring_active_window_to_front()
log.append("refresh: browser front")
except Exception:
pass
try:
sb.cdp.gui_hover_x_y(random.randint(220, 520), random.randint(180, 360))
_sleep(sb, 0.3, 0.8)
sb.cdp.gui_click_element("body", timeframe=random.uniform(0.2, 0.5))
_sleep(sb, 0.5, 1.0)
log.append("refresh: focused page with GUI click")
except Exception as exc:
log.append(f"refresh focus warning: {type(exc).__name__}: {exc}")
try:
import pyautogui
modifier = "command" if sys.platform == "darwin" else "ctrl"
pyautogui.hotkey(modifier, "r")
log.append(f"refresh: native {modifier}+r")
except Exception as exc:
log.append(f"refresh warning: {type(exc).__name__}: {exc}")
try:
_sleep(sb, 1.0, 1.8)
sb.cdp.reload(ignore_cache=True)
log.append("refresh: cdp.reload(ignore_cache=True)")
except Exception as exc:
log.append(f"refresh cdp warning: {type(exc).__name__}: {exc}")
return log
def _wait_for_initial_body_cdp(sb: Any) -> list[str]:
try:
sb.cdp.wait_for_element("body", timeout=10)
return ["initial: body loaded before refresh"]
except Exception as exc:
return [f"initial body warning: {type(exc).__name__}: {exc}"]
def _has_challenge_cdp(sb: Any) -> tuple[bool, str]:
text = ""
url = ""
html = ""
try:
text = sb.cdp.get_text("body").lower()
except Exception:
pass
try:
url = sb.cdp.get_current_url().lower()
except Exception:
pass
try:
html = sb.cdp.get_html().lower()
except Exception:
pass
blob = "\n".join((url, text, html[:10_000]))
for marker in (
"captcha",
"recaptcha",
"unusual traffic",
"too many requests",
"429",
"/sorry/",
"não foi possível",
"nao foi possivel",
):
if marker in blob:
return True, marker
return False, "none"
def _solve_captcha_cdp(sb: Any) -> list[str]:
log: list[str] = []
for method_name in ("solve_captcha", "uc_gui_click_captcha", "uc_gui_handle_captcha"):
method = getattr(sb, method_name, None)
if not callable(method):
continue
try:
result = method()
if inspect.isawaitable(result):
result = asyncio.run(result)
log.append(f"{method_name}: {result!r}")
except Exception as exc:
log.append(f"{method_name} warning: {type(exc).__name__}: {exc}")
if not log:
log.append("captcha helpers: unavailable")
return log
def _dismiss_google_interstitials_cdp(sb: Any) -> list[str]:
log: list[str] = []
for text in ("OK, got it", "Entendi", "Accept all", "I agree", "Stay in classic"):
try:
element = sb.cdp.find_element_by_text(text, timeout=1)
if element:
element.click()
log.append(f"clicked: {text}")
_sleep(sb, 0.8, 1.4)
except Exception:
continue
return log
def _wait_for_trends_cdp(sb: Any) -> list[str]:
for selector in ("trends-widget", "div.fe-atoms-generic-container", "body"):
try:
sb.cdp.wait_for_element(selector, timeout=10)
return [f"content: found {selector}"]
except Exception:
continue
return ["content warning: no Trends marker found; continuing"]
def _safe_cdp_call(sb: Any, method_name: str, default: str, *args: Any) -> str:
try:
return str(getattr(sb.cdp, method_name)(*args))
except Exception:
return default
def main() -> None:
parser = argparse.ArgumentParser(description="Fetch Google Trends with CDP Mode.")
parser.add_argument(
"task",
nargs="?",
default=(
"https://trends.google.com/trends/explore?"
"q=copa%20do%20mundo%20da%20fifa%202026&date=now%201-d&geo=BR"
),
help="Google Trends URL or plain query.",
)
parser.add_argument("--date", default=DEFAULT_DATE)
parser.add_argument("--geo", default=DEFAULT_GEO)
parser.add_argument("--output-dir", default=str(DEFAULT_OUTPUT_DIR))
parser.add_argument("--chrome", action="store_true", help="Use system Chrome.")
parser.add_argument("--no-refresh", action="store_true", help="Skip native Cmd/Ctrl+R.")
parser.add_argument(
"--no-stealth-warmup",
action="store_true",
help="Skip example.com -> Google typed-search warmup.",
)
args = parser.parse_args()
result = fetch_google_trends(
task=args.task,
date=args.date,
geo=args.geo,
output_dir=Path(args.output_dir),
refresh_once=not args.no_refresh,
use_chromium=not args.chrome,
stealth_warmup=not args.no_stealth_warmup,
)
print(json.dumps(_trends_result_payload(result), ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()

0
downloaded_files/pipfinding.lock Normal file
View File

4
requirements.txt Normal file
View File

@@ -0,0 +1,4 @@
# a2a-pack is auto-installed by the deploy build.
seleniumbase==4.49.10
playwright
pypdf>=5.0