From f705033061ed0b354645e770011e8b2f72cd67e6 Mon Sep 17 00:00:00 2001 From: "fatih.bulut" Date: Tue, 23 Jun 2026 22:58:09 +0300 Subject: [PATCH] feat(market-research): add best-effort Google Trends fetcher Co-Authored-By: Claude Sonnet 4.6 --- .../skills/market-research/scripts/trends.py | 79 +++++++++++++++++++ .../tests/fixtures/trends-sample.json | 5 ++ plugins/market-research/tests/test-trends.py | 36 +++++++++ 3 files changed, 120 insertions(+) create mode 100755 plugins/market-research/skills/market-research/scripts/trends.py create mode 100644 plugins/market-research/tests/fixtures/trends-sample.json create mode 100644 plugins/market-research/tests/test-trends.py diff --git a/plugins/market-research/skills/market-research/scripts/trends.py b/plugins/market-research/skills/market-research/scripts/trends.py new file mode 100755 index 0000000..83dc942 --- /dev/null +++ b/plugins/market-research/skills/market-research/scripts/trends.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Best-effort Google Trends interest signal. NEVER hard-fails. + +Google Trends has no API key but the unofficial endpoint needs a token dance +and is fragile under stdlib-only. So: try the live token+timeline flow; on ANY +failure print {"ok": false, "fallback": "websearch"} and exit 0, signalling the +skill to fall back to WebSearch for the same momentum signal. The JSON parser is +unit-tested offline via --json-file. Stdlib-only, no pip. +""" +import sys, json, argparse, urllib.request, urllib.parse, ssl, subprocess, shutil + +UA = "Mozilla/5.0" + +def _http_get(url, headers=None): + req = urllib.request.Request(url, headers={"User-Agent": UA, **(headers or {})}) + try: + with urllib.request.urlopen(req, timeout=30) as r: + return r.read().decode("utf-8", "replace") + except urllib.error.URLError as e: + if not isinstance(e.reason, ssl.SSLError): + raise + if not shutil.which("curl"): + raise + out = subprocess.run(["curl", "-sL", "--fail", "-A", UA, url], + capture_output=True, timeout=60) + if out.returncode != 0: + raise + return out.stdout.decode("utf-8", "replace") + +def parse_timeline(raw): + """raw = the timeline JSON (Google prefixes live responses with )]}', — + callers strip it first). Returns the computed interest summary dict.""" + data = json.loads(raw) + points = data["default"]["timelineData"] + values = [int(p["value"][0]) for p in points if p.get("value")] + if not values: + raise ValueError("no timeline values") + avg = round(sum(values) / len(values), 2) + first, latest = values[0], values[-1] + pct = round((latest - first) / first * 100, 2) if first else None + return {"ok": True, "avg_interest": avg, "latest_interest": latest, + "trend_pct": pct, "points": len(values)} + +def fetch_live(term): + """Token dance: explore -> widget token -> multiline timeline.""" + base = "https://trends.google.com/trends/api" + comp = urllib.parse.quote(json.dumps( + {"comparisonItem": [{"keyword": term, "geo": "", "time": "today 3-m"}], + "category": 0, "property": ""})) + explore = _http_get(f"{base}/explore?hl=en-US&tz=0&req={comp}") + widgets = json.loads(explore.lstrip(")]}',\n")) + w = next(x for x in widgets["widgets"] if x.get("id") == "TIMESERIES") + token = w["token"] + reqobj = urllib.parse.quote(json.dumps(w["request"])) + tl = _http_get(f"{base}/widgetdata/multiline?hl=en-US&tz=0&req={reqobj}&token={token}") + return tl.lstrip(")]}',\n") + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("term") + ap.add_argument("--json-file") + args = ap.parse_args() + + try: + if args.json_file: + with open(args.json_file, encoding="utf-8") as f: + raw = f.read() + else: + raw = fetch_live(args.term) + res = parse_timeline(raw) + res["term"] = args.term + print(json.dumps(res, indent=2)) + except Exception: + print(json.dumps({"ok": False, "term": args.term, + "fallback": "websearch"}, indent=2)) + sys.exit(0) + +if __name__ == "__main__": + main() diff --git a/plugins/market-research/tests/fixtures/trends-sample.json b/plugins/market-research/tests/fixtures/trends-sample.json new file mode 100644 index 0000000..8a729c8 --- /dev/null +++ b/plugins/market-research/tests/fixtures/trends-sample.json @@ -0,0 +1,5 @@ +{"default":{"timelineData":[ + {"time":"1","value":[50]}, + {"time":"2","value":[60]}, + {"time":"3","value":[100]} +]}} diff --git a/plugins/market-research/tests/test-trends.py b/plugins/market-research/tests/test-trends.py new file mode 100644 index 0000000..b650f08 --- /dev/null +++ b/plugins/market-research/tests/test-trends.py @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 +import json, subprocess, sys, os + +HERE = os.path.dirname(os.path.abspath(__file__)) +SCRIPT = os.path.join(HERE, "..", "skills", "market-research", "scripts", "trends.py") +FIXTURE = os.path.join(HERE, "fixtures", "trends-sample.json") + +def run(*args): + # exit 0 always (never hard-fails); capture stdout + out = subprocess.run([sys.executable, SCRIPT, *args], + capture_output=True, text=True) + return out.returncode, json.loads(out.stdout) + +def main(): + fails = [] + def check(name, cond): + print(f"{'PASS' if cond else 'FAIL'}: {name}") + if not cond: fails.append(name) + + rc, d = run("habit tracker", "--json-file", FIXTURE) + check("exit 0", rc == 0) + check("ok", d["ok"] is True) + check("points", d["points"] == 3) + check("latest_interest", d["latest_interest"] == 100) + check("avg_interest", d["avg_interest"] == 70.0) # mean(50,60,100)=70 + check("trend_pct", d["trend_pct"] == 100.0) # (100-50)/50*100 + + # bad fixture -> graceful fallback, still exit 0 + rc2, d2 = run("x", "--json-file", os.path.join(HERE, "fixtures", "play-chart.html")) + check("fallback exit 0", rc2 == 0) + check("fallback ok false", d2["ok"] is False) + check("fallback flag", d2["fallback"] == "websearch") + + sys.exit(1 if fails else 0) + +main()