API Reference¶

class ExportFormat(StrEnum):
    """Supported export targets for tabular trend data."""

    CSV = "csv"
    JSON = "json"

class GoogleTrendsFetcher:
    """Fetches data via the in-tree :class:`GoogleTrendsHttpSession`."""

    def __init__(
        self,
        language: str = "en",
        timeout: int = 10,
        proxies: Sequence[str] | None = None,
        max_proxy_attempts: int | None = None,
        on_proxy_rotate: Callable[[int, Exception], None] | None = None,
        rpc_ids: Mapping[str, str] | None = None,
    ) -> None:
        """
        ``proxies`` is a list of proxy URLs to rotate through, from any mix of providers.

        One proxy is pinned per query and the pool advances only when a query fails, because
        Google binds its cookie and widget token to the exit IP. ``max_proxy_attempts``
        defaults to the pool size, capped at 5; ``on_proxy_rotate(attempt, error)`` is called
        each time the pool advances.

        ``rpc_ids`` repoints a `batchexecute` identifier that Google has renamed --
        ``{"trending": "...", "geo_list": "...", "suggestions": "..."}``, any subset. This is
        the escape hatch :class:`UnknownRpcError` refers to: a rename can be worked around in
        the caller, with no release and no fork.
        """
        to = (timeout, max(timeout * 2, timeout + 5))
        self._pool = ProxyPool(proxies) if proxies else None
        default_attempts = min(self._pool.size, 5) if self._pool else 1
        self._max_proxy_attempts = max_proxy_attempts if max_proxy_attempts is not None else default_attempts
        self._on_proxy_rotate = on_proxy_rotate
        self._req = GoogleTrendsHttpSession(
            hl=_hl_from_language(language),
            tz=360,
            timeout=to,
            proxies=[self._pool.current()] if self._pool else "",
            rpc_ids=rpc_ids,
        )
        self._rpc_trending: TrendingProvider = RpcTrendingProvider(self._req.rpc_client)
        self._rss_trending: TrendingProvider = RssTrendingProvider(self._req.rss_client)

    @property
    def current_proxy(self) -> str | None:
        """The proxy currently pinned for queries, if a pool is configured."""
        return self._pool.current() if self._pool else None

    def _with_rotation(self, operation: Callable[[], T]) -> T:
        """
        Run a query, moving to the next proxy and re-seeding the cookie jar if it fails in a
        way a different exit IP could fix.
        """
        if self._pool is None:
            return operation()

        attempts = max(1, min(self._pool.size, self._max_proxy_attempts))
        for attempt in range(1, attempts + 1):
            try:
                return operation()
            except Exception as error:
                if not _should_rotate(error) or attempt == attempts:
                    raise
                self._pool.advance()
                # The cookie and any cached widget token belong to the previous exit IP.
                self._req.set_proxy(self._pool.current())
                if self._on_proxy_rotate is not None:
                    self._on_proxy_rotate(attempt, error)
        raise AssertionError("unreachable")  # pragma: no cover

    def interest_over_time(
        self,
        keywords: list[str],
        timeframe: Timeframe | str,
        region: Region | str,
        *,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> InterestOverTimeResult:
        """
        Relative interest for up to five terms over ``timeframe``.

        ``timeframe`` takes a :class:`Timeframe` or a custom ``"YYYY-MM-DD YYYY-MM-DD"`` range,
        and ``region`` any Google geo code -- a country, a sub-region like ``"US-CA"``, or a
        metro code. ``category`` restricts to one subject area, which disambiguates without a
        topic id: "jaguar" under Autos is the car. ``search_property`` chooses the surface;
        results from different properties are separate indexes and not comparable.
        """

        def run() -> InterestOverTimeResult:
            self._req.build_payload(
                keywords,
                cat=category,
                timeframe=_param(timeframe),
                geo=_param(region),
                gprop=_gprop(search_property),
            )
            default = self._req.interest_over_time()
            return _parsers.interest_over_time_to_result(default, keywords, self._req.geo)

        return self._with_rotation(run)

    def interest_by_region(
        self,
        keyword: str,
        resolution: Resolution,
        region: Region | str = Region.US,
        *,
        timeframe: Timeframe | str = Timeframe.PAST_YEAR,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> InterestByRegionResult:
        """
        Where ``keyword`` is searched, broken down at ``resolution``.

        ``timeframe`` defaults to the past year; narrow it to ask where something was searched
        during a specific window rather than across the whole year.
        """

        def run() -> InterestByRegionResult:
            self._req.build_payload(
                [keyword],
                cat=category,
                timeframe=_param(timeframe),
                geo=_param(region),
                gprop=_gprop(search_property),
            )
            default = self._req.interest_by_region(
                resolution=resolution.value,
                inc_low_vol=True,
                inc_geo_code=False,
            )
            if not default.get("geoMapData"):
                return InterestByRegionResult(keyword=keyword, resolution=resolution, rows=[])
            return _parsers.interest_by_region_to_result(default, keyword, [keyword], resolution)

        return self._with_rotation(run)

    def trending_now(
        self,
        region: Region | str = Region.WORLDWIDE,
        window: int = TRENDING_WINDOW_RISING,
        backend: TrendingBackend = "auto",
    ) -> TrendingResult:
        """
        Trending searches for ``region``.

        Accepts any country code, not a fixed list, and worldwide works too. ``window``
        selects fastest-growing (:data:`TRENDING_WINDOW_RISING`) versus highest-volume
        (:data:`TRENDING_WINDOW_TOP`) results.

        ``backend`` selects the source:

        * ``"rpc"`` -- 50 items with growth percentages and volume.
        * ``"rss"`` -- 10 items with the **news articles** behind each trend, which the RPC
          does not carry, but no growth figures. ``window`` does not apply: Google ignores
          it on the feed. There is no worldwide feed, so a country code is required.
        * ``"auto"`` (default) -- the RPC, falling back to RSS if it fails. RPC first
          because it returns five times the items with real growth numbers; defaulting to
          RSS would quietly degrade results.

        :attr:`TrendingResult.source` reports which one answered.
        """
        geo = _trending_geo(region)

        def run(provider: TrendingProvider) -> TrendingResult:
            return TrendingResult(results=provider.fetch(geo, window), source=provider.source)

        if backend == "rpc":
            return self._with_rotation(lambda: run(self._rpc_trending))
        if backend == "rss":
            return self._with_rotation(lambda: run(self._rss_trending))

        def auto() -> TrendingResult:
            try:
                return run(self._rpc_trending)
            except Exception:  # noqa: BLE001 - the feed is a separate source; any RPC failure falls back
                return run(self._rss_trending)

        return self._with_rotation(auto)

    def related_queries(
        self,
        keyword: str,
        *,
        timeframe: Timeframe | str = Timeframe.PAST_YEAR,
        region: Region | str = Region.WORLDWIDE,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> RelatedResult:
        """
        Top and rising searches related to ``keyword``.

        ``region`` defaults to worldwide, which is what this returned unconditionally before it
        was a parameter -- pass a country to ask what is searched alongside the term *there*.
        """

        def run() -> RelatedResult:
            self._req.build_payload(
                [keyword],
                cat=category,
                timeframe=_param(timeframe),
                geo=_param(region),
                gprop=_gprop(search_property),
            )
            raw = self._req.related_queries()
            return _parsers.related_queries_to_result(raw, keyword)

        return self._with_rotation(run)

    def suggestions(self, query: str) -> list[TopicSuggestion]:
        """
        Entity suggestions for a partial query -- the picker behind the UI's "Topic" results.

        Pass a returned ``mid`` as a keyword to any query method to measure the **topic**
        rather than the literal phrase; a topic aggregates every spelling and translation of
        the same concept, so it usually scores far higher than the raw string.

        Needs no cookie and no proxy: this RPC answers on IPs the widgetdata endpoints
        reject.
        """
        return self._with_rotation(
            lambda: _parsers.suggestion_rows_to_topics(self._req.suggestions(query)),
        )

    def geo_list(self) -> Any:
        """Every region Google accepts: ``[code, name, slug]`` per country, with subregions."""
        return self._with_rotation(self._req.geo_list)

@dataclass(frozen=True)
class InterestByRegionResult:
    """Regional popularity for a single keyword."""

    keyword: str
    resolution: Resolution
    rows: list[RegionalInterestRow]

@dataclass(frozen=True)
class InterestOverTimeResult:
    """Interest over time for one or more keywords."""

    keywords: list[str]
    granularity: str
    points: list[TrendPoint]

    def to_dataframe(self) -> pd.DataFrame:
        """Build a pandas DataFrame with a `date` column and one column per keyword."""
        if not self.points:
            return pd.DataFrame(columns=["date", *self.keywords])
        rows: list[dict[str, Any]] = []
        for p in self.points:
            rows.append({"date": p.date, **p.scores})
        return pd.DataFrame(rows)

    def export(self, fmt: ExportFormat, path: str | Path) -> None:
        """Write results to CSV or JSON (UTF-8) via :mod:`trendflow._exporters`."""
        from trendflow._exporters import export_interest_over_time

        export_interest_over_time(self, fmt, Path(path))

class Region(StrEnum):
    """ISO-style geo codes for Google Trends (`hl` / `geo`). Empty string is worldwide."""

    WORLDWIDE = ""
    US = "US"
    GB = "GB"
    DE = "DE"
    FR = "FR"
    IT = "IT"
    ES = "ES"
    CA = "CA"
    AU = "AU"
    JP = "JP"
    IN = "IN"
    BR = "BR"
    MX = "MX"
    NL = "NL"
    SE = "SE"
    PL = "PL"
    TR = "TR"

@dataclass(frozen=True)
class RelatedQuery:
    """A top or rising related query."""

    term: str
    value: int | None = None
    breakout: str | None = None

@dataclass(frozen=True)
class RelatedResult:
    """Related queries for a seed keyword."""

    top: list[RelatedQuery]
    rising: list[RelatedQuery]

class Resolution(StrEnum):
    """Granularity for regional interest breakdowns."""

    COUNTRY = "COUNTRY"
    REGION = "REGION"
    CITY = "CITY"

class SearchProperty(StrEnum):
    """
    Which Google surface to measure.

    These are separate indexes, not filters over one dataset, so the same term can look very
    different across them -- a term may be quiet on web search and busy on YouTube. Values are
    only comparable within a single property.
    """

    WEB = ""
    IMAGES = "images"
    NEWS = "news"
    YOUTUBE = "youtube"
    SHOPPING = "froogle"

class Timeframe(StrEnum):
    """
    Time ranges accepted by Google Trends.

    Named values for the presets. A custom range is a plain string of two ISO dates,
    ``"2023-01-01 2023-06-30"``, and every query method accepts one in place of a member.

    The range chosen also decides the granularity Google returns: the hourly ranges come back
    in minutes, the daily ones hourly, and ``ALL_TIME`` monthly. Ask for five years and you
    cannot see a spike that lasted an afternoon.
    """

    PAST_HOUR = "now 1-H"
    PAST_4_HOURS = "now 4-H"
    PAST_DAY = "now 1-d"
    PAST_WEEK = "now 7-d"
    PAST_MONTH = "today 1-m"
    PAST_3_MONTHS = "today 3-m"
    PAST_YEAR = "today 12-m"
    PAST_5_YEARS = "today 5-y"
    ALL_TIME = "all"

@dataclass(frozen=True)
class TopicSuggestion:
    """
    An entity Google recognises, as returned by search suggestions.

    ``mid`` is the identifier to pass as a keyword to query the **topic** rather than the
    literal phrase -- a topic aggregates every spelling and translation of the same concept.
    """

    #: Freebase-style entity id, e.g. ``"/m/0mkz"``. Pass this as a keyword.
    mid: str
    #: Display name, e.g. ``"Artificial intelligence"``.
    title: str
    #: Disambiguating descriptor, e.g. ``"Professional field"``. ``None`` when Google omits it.
    type: str | None = None

@dataclass(frozen=True)
class TrendPoint:
    """One timestamp in an interest-over-time series."""

    date: datetime
    scores: dict[str, int]

@dataclass(frozen=True)
class TrendingArticle:
    """A news article behind a trending search. Only the RSS backend reports these."""

    title: str
    url: str
    source: str
    picture: str | None = None

@dataclass(frozen=True)
class TrendingItem:
    """
    A single trending search entry.

    Both backends fill ``title`` and ``traffic``; the rest depends on which one answered,
    since Google exposes different fields on each. See :attr:`TrendingResult.source`.
    """

    title: str
    #: Human-readable traffic: ``"+3,950%"`` from the RPC, ``"2000+"`` from RSS.
    traffic: str
    #: News articles behind the trend. RSS backend only; empty from the RPC.
    articles: list[TrendingArticle] = field(default_factory=list)
    #: Percentage increase over the window, e.g. ``3950``. RPC backend only.
    growth: int | None = None
    #: Relative search volume, on Google's own 0-100 style scale. RPC backend only.
    volume: int | None = None
    #: When Google started reporting the trend. RSS backend only.
    started_at: datetime | None = None

@dataclass(frozen=True)
class TrendingResult:
    """Current trending searches for a region."""

    results: list[TrendingItem]
    #: Which backend answered -- useful when ``backend="auto"`` picked for you.
    source: str = "rpc"

@runtime_checkable
class TrendsFetcher(Protocol):
    """Strategy for retrieving Trends data (swap in tests or alternate backends)."""

    def interest_over_time(
        self,
        keywords: list[str],
        timeframe: Timeframe | str,
        region: Region | str,
        *,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> InterestOverTimeResult: ...

    def interest_by_region(
        self,
        keyword: str,
        resolution: Resolution,
        region: Region | str = Region.US,
        *,
        timeframe: Timeframe | str = Timeframe.PAST_YEAR,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> InterestByRegionResult: ...

    def trending_now(
        self,
        region: Region | str = Region.WORLDWIDE,
        window: int = ...,
        backend: TrendingBackend = ...,
    ) -> TrendingResult: ...

    def related_queries(
        self,
        keyword: str,
        *,
        timeframe: Timeframe | str = Timeframe.PAST_YEAR,
        region: Region | str = Region.WORLDWIDE,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> RelatedResult: ...

    def suggestions(self, query: str) -> list[TopicSuggestion]: ...

class UnknownRpcError(Exception):
    """
    Google returned no frame for the RPC that was called.

    That is the signature of an identifier renamed on Google's side.
    """

    def __init__(self, rpc_id: str) -> None:
        super().__init__(
            f"Google returned no data for RPC {rpc_id!r}. This usually means the RPC "
            f"identifier changed on Google's side. Override it with the `rpc_ids` argument, "
            f"and please open an issue at https://github.com/dariomory/trendflow/issues",
        )
        self.rpc_id = rpc_id

class GoogleTrendsFetcher:
    """Fetches data via the in-tree :class:`GoogleTrendsHttpSession`."""

    def __init__(
        self,
        language: str = "en",
        timeout: int = 10,
        proxies: Sequence[str] | None = None,
        max_proxy_attempts: int | None = None,
        on_proxy_rotate: Callable[[int, Exception], None] | None = None,
        rpc_ids: Mapping[str, str] | None = None,
    ) -> None:
        """
        ``proxies`` is a list of proxy URLs to rotate through, from any mix of providers.

        One proxy is pinned per query and the pool advances only when a query fails, because
        Google binds its cookie and widget token to the exit IP. ``max_proxy_attempts``
        defaults to the pool size, capped at 5; ``on_proxy_rotate(attempt, error)`` is called
        each time the pool advances.

        ``rpc_ids`` repoints a `batchexecute` identifier that Google has renamed --
        ``{"trending": "...", "geo_list": "...", "suggestions": "..."}``, any subset. This is
        the escape hatch :class:`UnknownRpcError` refers to: a rename can be worked around in
        the caller, with no release and no fork.
        """
        to = (timeout, max(timeout * 2, timeout + 5))
        self._pool = ProxyPool(proxies) if proxies else None
        default_attempts = min(self._pool.size, 5) if self._pool else 1
        self._max_proxy_attempts = max_proxy_attempts if max_proxy_attempts is not None else default_attempts
        self._on_proxy_rotate = on_proxy_rotate
        self._req = GoogleTrendsHttpSession(
            hl=_hl_from_language(language),
            tz=360,
            timeout=to,
            proxies=[self._pool.current()] if self._pool else "",
            rpc_ids=rpc_ids,
        )
        self._rpc_trending: TrendingProvider = RpcTrendingProvider(self._req.rpc_client)
        self._rss_trending: TrendingProvider = RssTrendingProvider(self._req.rss_client)

    @property
    def current_proxy(self) -> str | None:
        """The proxy currently pinned for queries, if a pool is configured."""
        return self._pool.current() if self._pool else None

    def _with_rotation(self, operation: Callable[[], T]) -> T:
        """
        Run a query, moving to the next proxy and re-seeding the cookie jar if it fails in a
        way a different exit IP could fix.
        """
        if self._pool is None:
            return operation()

        attempts = max(1, min(self._pool.size, self._max_proxy_attempts))
        for attempt in range(1, attempts + 1):
            try:
                return operation()
            except Exception as error:
                if not _should_rotate(error) or attempt == attempts:
                    raise
                self._pool.advance()
                # The cookie and any cached widget token belong to the previous exit IP.
                self._req.set_proxy(self._pool.current())
                if self._on_proxy_rotate is not None:
                    self._on_proxy_rotate(attempt, error)
        raise AssertionError("unreachable")  # pragma: no cover

    def interest_over_time(
        self,
        keywords: list[str],
        timeframe: Timeframe | str,
        region: Region | str,
        *,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> InterestOverTimeResult:
        """
        Relative interest for up to five terms over ``timeframe``.

        ``timeframe`` takes a :class:`Timeframe` or a custom ``"YYYY-MM-DD YYYY-MM-DD"`` range,
        and ``region`` any Google geo code -- a country, a sub-region like ``"US-CA"``, or a
        metro code. ``category`` restricts to one subject area, which disambiguates without a
        topic id: "jaguar" under Autos is the car. ``search_property`` chooses the surface;
        results from different properties are separate indexes and not comparable.
        """

        def run() -> InterestOverTimeResult:
            self._req.build_payload(
                keywords,
                cat=category,
                timeframe=_param(timeframe),
                geo=_param(region),
                gprop=_gprop(search_property),
            )
            default = self._req.interest_over_time()
            return _parsers.interest_over_time_to_result(default, keywords, self._req.geo)

        return self._with_rotation(run)

    def interest_by_region(
        self,
        keyword: str,
        resolution: Resolution,
        region: Region | str = Region.US,
        *,
        timeframe: Timeframe | str = Timeframe.PAST_YEAR,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> InterestByRegionResult:
        """
        Where ``keyword`` is searched, broken down at ``resolution``.

        ``timeframe`` defaults to the past year; narrow it to ask where something was searched
        during a specific window rather than across the whole year.
        """

        def run() -> InterestByRegionResult:
            self._req.build_payload(
                [keyword],
                cat=category,
                timeframe=_param(timeframe),
                geo=_param(region),
                gprop=_gprop(search_property),
            )
            default = self._req.interest_by_region(
                resolution=resolution.value,
                inc_low_vol=True,
                inc_geo_code=False,
            )
            if not default.get("geoMapData"):
                return InterestByRegionResult(keyword=keyword, resolution=resolution, rows=[])
            return _parsers.interest_by_region_to_result(default, keyword, [keyword], resolution)

        return self._with_rotation(run)

    def trending_now(
        self,
        region: Region | str = Region.WORLDWIDE,
        window: int = TRENDING_WINDOW_RISING,
        backend: TrendingBackend = "auto",
    ) -> TrendingResult:
        """
        Trending searches for ``region``.

        Accepts any country code, not a fixed list, and worldwide works too. ``window``
        selects fastest-growing (:data:`TRENDING_WINDOW_RISING`) versus highest-volume
        (:data:`TRENDING_WINDOW_TOP`) results.

        ``backend`` selects the source:

        * ``"rpc"`` -- 50 items with growth percentages and volume.
        * ``"rss"`` -- 10 items with the **news articles** behind each trend, which the RPC
          does not carry, but no growth figures. ``window`` does not apply: Google ignores
          it on the feed. There is no worldwide feed, so a country code is required.
        * ``"auto"`` (default) -- the RPC, falling back to RSS if it fails. RPC first
          because it returns five times the items with real growth numbers; defaulting to
          RSS would quietly degrade results.

        :attr:`TrendingResult.source` reports which one answered.
        """
        geo = _trending_geo(region)

        def run(provider: TrendingProvider) -> TrendingResult:
            return TrendingResult(results=provider.fetch(geo, window), source=provider.source)

        if backend == "rpc":
            return self._with_rotation(lambda: run(self._rpc_trending))
        if backend == "rss":
            return self._with_rotation(lambda: run(self._rss_trending))

        def auto() -> TrendingResult:
            try:
                return run(self._rpc_trending)
            except Exception:  # noqa: BLE001 - the feed is a separate source; any RPC failure falls back
                return run(self._rss_trending)

        return self._with_rotation(auto)

    def related_queries(
        self,
        keyword: str,
        *,
        timeframe: Timeframe | str = Timeframe.PAST_YEAR,
        region: Region | str = Region.WORLDWIDE,
        category: int = 0,
        search_property: SearchProperty | str = SearchProperty.WEB,
    ) -> RelatedResult:
        """
        Top and rising searches related to ``keyword``.

        ``region`` defaults to worldwide, which is what this returned unconditionally before it
        was a parameter -- pass a country to ask what is searched alongside the term *there*.
        """

        def run() -> RelatedResult:
            self._req.build_payload(
                [keyword],
                cat=category,
                timeframe=_param(timeframe),
                geo=_param(region),
                gprop=_gprop(search_property),
            )
            raw = self._req.related_queries()
            return _parsers.related_queries_to_result(raw, keyword)

        return self._with_rotation(run)

    def suggestions(self, query: str) -> list[TopicSuggestion]:
        """
        Entity suggestions for a partial query -- the picker behind the UI's "Topic" results.

        Pass a returned ``mid`` as a keyword to any query method to measure the **topic**
        rather than the literal phrase; a topic aggregates every spelling and translation of
        the same concept, so it usually scores far higher than the raw string.

        Needs no cookie and no proxy: this RPC answers on IPs the widgetdata endpoints
        reject.
        """
        return self._with_rotation(
            lambda: _parsers.suggestion_rows_to_topics(self._req.suggestions(query)),
        )

    def geo_list(self) -> Any:
        """Every region Google accepts: ``[code, name, slug]`` per country, with subregions."""
        return self._with_rotation(self._req.geo_list)