Skip to content

SERPAPI - Trends

SerpAPIDownload ¤

Download trend using config

Parameters:

Name Type Description Default
parent_folder AnyPath

parent folder for the data

 required
snapshot_date datetime.date

snapshot date for the path

 required
trends_service

trend service

 required
Source code in sm_trendy/use_serpapi/get_trends.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class SerpAPIDownload:
    """Download trend using config

    :params parent_folder: parent folder for the data
    :param snapshot_date: snapshot date for the path
    :param trends_service: trend service
    """

    def __init__(
        self,
        parent_folder: AnyPath,
        snapshot_date: datetime.date,
    ):
        self.parent_folder = parent_folder
        self.snapshot_date = snapshot_date

    def __call__(self, config: SerpAPIConfig):
        """
        :param config: config for the keyword
        """
        api_params = config.serpapi_params.model_dump(exclude_none=True)

        path_params = config.path_params
        target_folder = path_params.path(parent_folder=self.parent_folder)

        sdf = StoreDataFrame(
            target_folder=target_folder, snapshot_date=self.snapshot_date
        )

        logger.info(
            f"keyword: {config.serpapi_params.q}\n"
            f"target_path: {target_folder}\n"
            "..."
        )

        sst = SerpAPISingleTrend(
            serpapi_params=api_params, extra_metadata=config.extra_metadata
        )

        logger.debug("Saving raw json format ...")
        sj = StoreJSON(target_folder=target_folder, snapshot_date=self.snapshot_date)
        sj.save(records=sst.search_results, formats=["json"])

        logger.debug("Saving dataframe ...")
        sdf.save(sst, formats=["csv", "parquet"])
        logger.info(f"Saved to {target_folder}")

__call__(config) ¤

Parameters:

Name Type Description Default
config SerpAPIConfig

config for the keyword

 required
Source code in sm_trendy/use_serpapi/get_trends.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def __call__(self, config: SerpAPIConfig):
    """
    :param config: config for the keyword
    """
    api_params = config.serpapi_params.model_dump(exclude_none=True)

    path_params = config.path_params
    target_folder = path_params.path(parent_folder=self.parent_folder)

    sdf = StoreDataFrame(
        target_folder=target_folder, snapshot_date=self.snapshot_date
    )

    logger.info(
        f"keyword: {config.serpapi_params.q}\n"
        f"target_path: {target_folder}\n"
        "..."
    )

    sst = SerpAPISingleTrend(
        serpapi_params=api_params, extra_metadata=config.extra_metadata
    )

    logger.debug("Saving raw json format ...")
    sj = StoreJSON(target_folder=target_folder, snapshot_date=self.snapshot_date)
    sj.save(records=sst.search_results, formats=["json"])

    logger.debug("Saving dataframe ...")
    sdf.save(sst, formats=["csv", "parquet"])
    logger.info(f"Saved to {target_folder}")

SerpAPISingleTrend ¤

Source code in sm_trendy/use_serpapi/get_trends.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class SerpAPISingleTrend:
    def __init__(
        self, serpapi_params: SerpAPIParams, extra_metadata: Optional[Dict] = {}
    ):
        self.serpapi_params = serpapi_params
        self.extra_metadata = extra_metadata

    @cached_property
    def search_results(self) -> Dict:
        api_params = self.serpapi_params

        search = GoogleSearch(api_params)
        results = search.get_dict()

        self._check_status(results["search_metadata"]["status"])

        return results

    def _check_status(self, status: str):
        if not status == "Success":
            raise Exception(f"Request failed with error {status}")

    @cached_property
    def dataframe(self) -> pd.DataFrame:
        """
        Build the dataframe

        :param keyword: keyword to be searched
        """
        if "interest_over_time" not in self.search_results:
            raise Exception(
                f"interest_over_time is not found: {self.search_results.keys()}\n"
                f"{self.search_results}"
            )

        timeline_data = self.search_results["interest_over_time"]["timeline_data"]

        df = pd.DataFrame(
            sum([self._flatten_record(record) for record in timeline_data], [])
        )

        df["date"] = pd.to_datetime(df.timestamp.astype(int), unit="s", origin="unix")

        return df

    @staticmethod
    def _flatten_record(record: Dict):
        """Convert raw record to flattened dicts"""
        return [
            {**i, **{"date_range": record["date"], "timestamp": record["timestamp"]}}
            for i in record["values"]
        ]

    @cached_property
    def metadata(self) -> Dict[str, Dict]:
        return {
            "search_metadata": self.search_results["search_metadata"],
            "search_parameters": self.search_results["search_parameters"],
            "extra_metadata": self.extra_metadata,
        }

dataframe: pd.DataFrame cached property ¤

Build the dataframe

Parameters:

Name Type Description Default
keyword

keyword to be searched

 required