1"""Actor entrypoint: Google Trends -> dataset, with a persistent cache.
2
3Contract with the caller, stated plainly because that honesty is the product:
4 * Every item carries ``fromCache`` and ``cachedAt`` so a caller can see exactly what it got.
5 * A keyword that fails does NOT fail the run. It is pushed as an item with ``error`` set, so a
6 100-keyword job returns 99 good rows instead of nothing. Only an empty input or a total
7 upstream outage aborts.
8 * Requests to Google are paced under 1 req/s by design (measured 429 ceiling ~6.6 req/s), so a
9 large keyword list is slow on purpose.
10"""
11
12from __future__ import annotations
13
14import asyncio
15import os
16import time
17import json
18
19from apify import Actor
20
21from .cache import cache_key, open_cache
22from .trends import TrendsBlocked, TrendsClient, TrendsError, TrendsUnavailable
23
24
25
26MAX_BLOCK_ROTATIONS = 4
27
28
29
30
31
32ROTATION_BUDGET = 6
33
34
35
36
37
38
39RUN_BUDGET_SECONDS = float(os.environ.get("TRENDS_RUN_BUDGET_SECONDS") or 210)
40
41
42
43DATA_TYPES = {
44 "interestOverTime": "TIMESERIES",
45 "interestByRegion": "GEO_MAP",
46 "relatedTopics": "RELATED_TOPICS",
47 "relatedQueries": "RELATED_QUERIES",
48}
49
50
51def _shape(data_type: str, keyword: str, payload: dict, terms=None) -> dict:
52 """Flatten Google's widget payload into something a spreadsheet can hold.
53
54 `terms` is set for a COMPARISON: each timeline point then carries one value per term, in the
55 order the terms were submitted, and `values` is emitted alongside `value` so a single-term
56 consumer keeps working unchanged.
57 """
58 default = (payload or {}).get("default") or {}
59 if data_type == "interestOverTime":
60 points = []
61 for p in default.get("timelineData") or []:
62 vals = p.get("value") or []
63 point = {
64 "date": p.get("formattedTime") or p.get("formattedAxisTime"),
65 "timestamp": p.get("time"),
66 "value": vals[0] if vals else None,
67 "isPartial": bool(p.get("isPartial")),
68 }
69 if terms:
70
71
72 point["values"] = {
73 t: (vals[i] if i < len(vals) else None) for i, t in enumerate(terms)
74 }
75 points.append(point)
76 return {"points": points, "pointCount": len(points)}
77 if data_type == "interestByRegion":
78 regions = []
79 for bucket in default.get("geoMapData") or []:
80 vals = bucket.get("value") or []
81 regions.append({
82 "geoCode": bucket.get("geoCode"),
83 "geoName": bucket.get("geoName"),
84 "value": vals[0] if vals else None,
85 "hasData": bool((bucket.get("hasData") or [False])[0]),
86 })
87 return {"regions": regions, "regionCount": len(regions)}
88
89
90
91
92 out = {"top": [], "rising": []}
93 ranked_lists = default.get("rankedList") or []
94 if not ranked_lists:
95
96
97
98
99 raise TrendsError(
100 "Google returned an empty %s widget (throttled, or no data for this query)" % data_type
101 )
102 for idx, bucket in enumerate(ranked_lists[:2]):
103 rows = []
104 for item in bucket.get("rankedKeyword") or []:
105 rows.append({
106 "query": item.get("query") or (item.get("topic") or {}).get("title"),
107 "type": (item.get("topic") or {}).get("type"),
108 "value": item.get("value"),
109 "formattedValue": item.get("formattedValue"),
110 "link": item.get("link"),
111 })
112 out["rising" if idx else "top"] = rows
113 out["topCount"] = len(out["top"])
114 out["risingCount"] = len(out["rising"])
115 return out
116
117
118async def _charge(event: str, count: int = 1) -> None:
119 """Charge a pay-per-event unit. No-ops when the Actor is not monetized (local runs, tests),
120 because a charging failure must never lose data the caller already paid compute for."""
121 if count <= 0:
122 return
123 try:
124 await Actor.charge(event_name=event, count=count)
125 except Exception as exc:
126 Actor.log.debug("charge(%s x%d) skipped: %s", event, count, exc)
127
128
129async def main() -> None:
130 async with Actor:
131 cfg = await Actor.get_input() or {}
132
133 keywords = [str(k).strip() for k in (cfg.get("keywords") or []) if str(k).strip()]
134 compare = [str(k).strip() for k in (cfg.get("compareKeywords") or []) if str(k).strip()]
135 geo = (cfg.get("geo") or "").strip()
136 timeframe = (cfg.get("timeframe") or "today 12-m").strip()
137 category = int(cfg.get("category") or 0)
138 wanted = [d for d in (cfg.get("dataTypes") or ["interestOverTime"]) if d in DATA_TYPES]
139 include_trending = bool(cfg.get("includeTrendingNow"))
140 trending_geo = (cfg.get("trendingGeo") or geo or "US").strip()
141 max_age = float(cfg.get("cacheMaxAgeMinutes", 360)) * 60.0
142 min_gap = float(cfg.get("minRequestGapSeconds") or 1.2)
143
144 if not keywords and not compare and not include_trending:
145 raise ValueError(
146 "Nothing to do: provide `keywords` or `compareKeywords`, or set "
147 "`includeTrendingNow` to true."
148 )
149 if len(compare) > 5:
150 raise ValueError(
151 "Google Trends compares at most 5 terms at once; `compareKeywords` has %d. "
152 "Split them, or use `keywords` for independent (non-comparable) lookups."
153 % len(compare)
154 )
155 if not wanted:
156 wanted = ["interestOverTime"]
157
158
159
160
161 await _charge("actor-start")
162
163 cache = await open_cache(Actor, max_age)
164
165
166
167
168
169
170 proxy_cfg = None
171 if cfg.get("useApifyProxy", True):
172 try:
173 proxy_cfg = await Actor.create_proxy_configuration(
174 groups=cfg.get("apifyProxyGroups") or None,
175 country_code=cfg.get("apifyProxyCountry") or None,
176 )
177 except Exception as exc:
178 Actor.log.warning(
179 "Apify Proxy unavailable (%s) — continuing on the run's own IP, which will "
180 "rate-limit sooner", exc)
181 rotate_every = max(1, int(cfg.get("rotateAfterKeywords") or 5))
182
183 session_seq = 0
184
185 async def new_client() -> TrendsClient:
186 """A fresh client (fresh cookie jar) on a fresh proxy session, i.e. a fresh IP."""
187 nonlocal session_seq
188 session_seq += 1
189 url = None
190 if proxy_cfg is not None:
191 try:
192 url = await proxy_cfg.new_url("trends%d" % session_seq)
193 except Exception as exc:
194 Actor.log.warning("could not get a proxy URL: %s", exc)
195 return TrendsClient(min_gap=min_gap, proxy_url=url)
196
197 client = await new_client()
198 rotations_on_block = 0
199 explored: dict = {}
200
201 async def with_rotation(make_call, tries: int = MAX_BLOCK_ROTATIONS):
202 """Run `make_call(client)`; on a 429 move to a fresh IP and run it again.
203
204 Rotating costs one proxy session. Sleeping costs billed compute — measured at more than
205 half a run's cost — so rotation is the cheaper answer whenever a proxy is configured.
206 Without a proxy there is no new IP to move to, so the error stands.
207
208 `make_call` takes the client rather than being bound to one, because a rotation must be
209 able to redo the WHOLE unit of work on the new IP. That is not a style choice: a widget
210 token minted by `/explore` belongs to the session that minted it, so retrying only the
211 failed `widget_data` on a fresh IP is refused again — observed as a run that rotated
212 once and still returned nothing. More than one rotation is attempted for the same
213 reason a single one was not enough: the shared datacenter pool is small, so the next IP
214 can be just as hot as the last.
215 """
216 nonlocal client, rotations_on_block
217 last = None
218 for attempt in range(max(1, tries)):
219 try:
220 return await asyncio.to_thread(make_call, client)
221 except TrendsBlocked as exc:
222 last = exc
223 if proxy_cfg is None:
224 raise
225 if rotations_on_block >= ROTATION_BUDGET:
226 Actor.log.warning(
227 "rotation budget spent (%d) — reporting the block instead of paying "
228 "for more attempts", rotations_on_block)
229 raise
230 rotations_on_block += 1
231 Actor.log.info(
232 "blocked — rotating to a fresh IP (%d of %d) instead of waiting",
233 attempt + 1, max(1, tries))
234 client = await new_client()
235 raise last
236
237 def widget_call(target, widget_name: str):
238 """One unit of work: `/explore` then the widget fetch, on ONE client.
239
240 The explore result is memoised per client, so a keyword still costs one explore per IP
241 no matter how many dataTypes are asked for — but a rotation misses that cache and
242 re-explores, which is exactly what makes the fresh token valid.
243 """
244 def call(c):
245 ck = (id(c), json.dumps(target, sort_keys=True), geo, timeframe, category)
246 widgets = explored.get(ck)
247 if widgets is None:
248 widgets = c.explore(target, geo, timeframe, category)
249 explored[ck] = widgets
250 w = widgets.get(widget_name)
251 if w is None:
252 raise TrendsError(
253 "Google returned no %s widget for this query" % widget_name)
254 return c.widget_data(w)
255 return call
256
257 ok = failed = 0
258 deadline = time.monotonic() + RUN_BUDGET_SECONDS
259
260 def out_of_time() -> bool:
261 return time.monotonic() >= deadline
262
263
264 if include_trending:
265 key = cache_key("trending", geo=trending_geo, hours=48)
266 rows = await cache.get(key)
267 cached = rows is not None
268 if not cached:
269 try:
270 rows = await with_rotation(lambda c: c.trending_now(trending_geo))
271 await cache.put(key, rows)
272 except (
273 Exception
274 ) as exc:
275 Actor.log.warning(
276 "trending now failed for %s: %s", trending_geo, exc
277 )
278 rows = None
279 failed += 1
280 await Actor.push_data(
281 {
282 "type": "trendingNow",
283 "geo": trending_geo,
284 "error": str(exc),
285 "fromCache": False,
286 }
287 )
288 if rows is not None:
289 await Actor.push_data({
290 "type": "trendingNow", "geo": trending_geo,
291 "trendCount": len(rows), "trends": rows, "fromCache": cached,
292 })
293 ok += 1
294 await _charge("trending-feed")
295
296
297 if compare:
298 key = cache_key("compare", kws=sorted(compare), geo=geo, tf=timeframe, cat=category)
299 payload = await cache.get(key)
300 cached = payload is not None
301 if not cached:
302 try:
303 raw = await with_rotation(widget_call(compare, "TIMESERIES"))
304 payload = _shape("interestOverTime", ", ".join(compare), raw, terms=compare)
305 await cache.put(key, payload)
306 except (
307 Exception
308 ) as exc:
309 Actor.log.warning("comparison %s failed: %s", compare, exc)
310 failed += 1
311 await Actor.push_data(
312 {
313 "type": "comparison",
314 "keywords": compare,
315 "geo": geo,
316 "timeframe": timeframe,
317 "error": str(exc),
318 "fromCache": False,
319 }
320 )
321 payload = None
322 if payload is not None:
323 item = {"type": "comparison", "keywords": compare, "geo": geo,
324 "timeframe": timeframe, "category": category, "fromCache": cached}
325 item.update(payload)
326 await Actor.push_data(item)
327 ok += 1
328
329
330
331 await _charge("result", len(compare))
332
333
334 async def skip_rest(rest, reason: str) -> None:
335 """Report keywords the run is giving up on, one row each. Reporting beats raising: the
336 caller can see exactly which terms are missing and why, and the run still ends
337 SUCCEEDED — an exit code is not a per-keyword error channel."""
338 for kw_left in rest:
339 await Actor.push_data(
340 {
341 "keyword": kw_left,
342 "geo": geo,
343 "timeframe": timeframe,
344 "error": reason,
345 "fromCache": False,
346 }
347 )
348
349 unavailable = False
350 for kw_index, kw in enumerate(keywords):
351 if out_of_time():
352 Actor.log.warning(
353 "hết ngân sách thời gian (%.0fs) — bỏ qua %d keyword còn lại",
354 RUN_BUDGET_SECONDS, len(keywords) - kw_index)
355 await skip_rest(
356 keywords[kw_index:],
357 "run time budget exhausted before this keyword was fetched "
358 "(Google was rate-limiting); re-run to pick it up",
359 )
360 break
361 if kw_index and kw_index % rotate_every == 0:
362 client = await new_client()
363 Actor.log.info("rotated to proxy session #%d after %d keywords",
364 session_seq, kw_index)
365 for dtype in wanted:
366 key = cache_key(dtype, kw=kw, geo=geo, tf=timeframe, cat=category)
367 payload = await cache.get(key)
368 cached = payload is not None
369
370 if not cached:
371 try:
372 raw = await with_rotation(widget_call(kw, DATA_TYPES[dtype]))
373 payload = _shape(dtype, kw, raw)
374 await cache.put(key, payload)
375 except TrendsUnavailable as exc:
376
377
378
379
380
381 Actor.log.error("Google endpoint unavailable: %s", exc)
382 failed += 1
383 await Actor.push_data(
384 {
385 "keyword": kw,
386 "dataType": dtype,
387 "geo": geo,
388 "timeframe": timeframe,
389 "error": str(exc),
390 "fromCache": False,
391 }
392 )
393 await skip_rest(
394 keywords[kw_index + 1 :],
395 "stopped early: %s — every later keyword would hit the same wall"
396 % exc,
397 )
398 unavailable = True
399 break
400 except Exception as exc:
401
402
403
404
405 Actor.log.warning("%s / %s failed: %s", kw, dtype, exc)
406 failed += 1
407 await Actor.push_data(
408 {
409 "keyword": kw,
410 "dataType": dtype,
411 "geo": geo,
412 "timeframe": timeframe,
413 "error": str(exc),
414 "fromCache": False,
415 }
416 )
417 continue
418
419 item = {
420 "keyword": kw, "dataType": dtype, "geo": geo,
421 "timeframe": timeframe, "category": category,
422 "fromCache": cached,
423 }
424 item.update(payload)
425 await Actor.push_data(item)
426 ok += 1
427 await _charge("result")
428
429 if unavailable:
430 break
431
432 Actor.log.info(
433 "done: %d results, %d failed | cache %d/%d hits (%.0f%%) | %d proxy session(s), "
434 "%d rotation(s) to dodge a block",
435 ok, failed, cache.hits, cache.total, 100 * cache.hit_rate, session_seq,
436 rotations_on_block,
437 )
438 if failed and not proxy_cfg:
439 Actor.log.warning(
440 "%d result(s) failed while running WITHOUT a proxy. Google budgets requests per "
441 "IP over hours, not seconds, so a shared IP runs out. Enable Apify Proxy.", failed)
