From 4326bde33f41a1b38de5f72bc59a1910d81465a9 Mon Sep 17 00:00:00 2001 From: "fatih.bulut" Date: Tue, 23 Jun 2026 22:25:20 +0300 Subject: [PATCH] feat(market-research): add AppBrain Play-charts scraper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NOTE: AppBrain returns a Cloudflare JS challenge (HTTP 403) for automated requests; live fetch degrades gracefully to an error (the skill falls back to Apple RSS + web search). Offline test passes against synthetic fixture — that is the test gate per the brief. Co-Authored-By: Claude Sonnet 4.6 --- .../scripts/fetch-play-charts.py | 90 +++++++++++++++++++ .../tests/fixtures/appbrain-popular.html | 18 ++++ .../tests/test-fetch-play-charts.py | 36 ++++++++ 3 files changed, 144 insertions(+) create mode 100755 plugins/market-research/skills/market-research/scripts/fetch-play-charts.py create mode 100644 plugins/market-research/tests/fixtures/appbrain-popular.html create mode 100644 plugins/market-research/tests/test-fetch-play-charts.py diff --git a/plugins/market-research/skills/market-research/scripts/fetch-play-charts.py b/plugins/market-research/skills/market-research/scripts/fetch-play-charts.py new file mode 100755 index 0000000..5ca1cc6 --- /dev/null +++ b/plugins/market-research/skills/market-research/scripts/fetch-play-charts.py @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +"""Fetch an AppBrain Play-store top-chart into a normalized JSON list. + +Google Play has no public chart feed and renders via obfuscated batchexecute +JS. AppBrain (appbrain.com) publishes server-rendered HTML top-chart pages with +REAL Android package names. This scrapes those rows. Stdlib-only, no pip. + +Unlike Apple's RSS (iOS bundle ids), entries here carry Android packages usable +directly for a clone-app handoff. +""" +import sys, json, re, argparse, urllib.request, ssl, subprocess, shutil + +CHARTS = { + "popular": "https://www.appbrain.com/apps/popular/", + "top-grossing": "https://www.appbrain.com/apps/highest-grossing/", + "top-new": "https://www.appbrain.com/apps/new/", +} +UA = "Mozilla/5.0" + +def _http_get(url): + req = urllib.request.Request(url, headers={"User-Agent": UA}) + 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") + +# One app row: a link to /app// followed (within the same row +# block) by developer / category / rating / installs spans. +ROW = re.compile( + r'href="/app/[^"/]+/(?P[\w.]+)"[^>]*>(?P[^<]+)' + r'(?P.*?)(?=href="/app/|\s*|$)', re.DOTALL) +DEV = re.compile(r'class="developer"[^>]*>([^<]+)<') +CAT = re.compile(r'class="category"[^>]*>([^<]+)<') +RAT = re.compile(r'class="rating"[^>]*>\s*([\d.]+)\s*<') +INST = re.compile(r'class="installs"[^>]*>\s*([\d,]+\+)\s*<') + +def _first(rx, text): + m = rx.search(text) + return m.group(1).strip() if m else None + +def parse(html, chart, region, limit): + entries = [] + for i, m in enumerate(ROW.finditer(html), start=1): + if i > limit: + break + rest = m.group("rest") + rating = _first(RAT, rest) + entries.append({ + "rank": i, + "name": m.group("name").strip(), + "developer": _first(DEV, rest), + "category": _first(CAT, rest), + "package": m.group("package"), + "rating": float(rating) if rating else None, + "installs": _first(INST, rest), + }) + return {"source": "appbrain", "chart": chart, "region": region, + "count": len(entries), "entries": entries} + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("chart", choices=sorted(CHARTS)) + ap.add_argument("--region", default="us") + ap.add_argument("--limit", type=int, default=25) + ap.add_argument("--html-file") + args = ap.parse_args() + + if args.html_file: + with open(args.html_file, encoding="utf-8") as f: + html = f.read() + else: + try: + html = _http_get(CHARTS[args.chart]) + except Exception as e: + print(f"ERROR: failed to fetch AppBrain chart: {e}", file=sys.stderr) + sys.exit(1) + + print(json.dumps(parse(html, args.chart, args.region, args.limit), indent=2)) + +if __name__ == "__main__": + main() diff --git a/plugins/market-research/tests/fixtures/appbrain-popular.html b/plugins/market-research/tests/fixtures/appbrain-popular.html new file mode 100644 index 0000000..861ef3c --- /dev/null +++ b/plugins/market-research/tests/fixtures/appbrain-popular.html @@ -0,0 +1,18 @@ + +
+
+ Habit Tracker + Focus Labs + Productivity + 4.6 + 5,000,000+ +
+
+ Budget Buddy + Money Inc + Finance + 4.2 + 1,000,000+ +
+
+ diff --git a/plugins/market-research/tests/test-fetch-play-charts.py b/plugins/market-research/tests/test-fetch-play-charts.py new file mode 100644 index 0000000..23d708a --- /dev/null +++ b/plugins/market-research/tests/test-fetch-play-charts.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", "fetch-play-charts.py") +FIXTURE = os.path.join(HERE, "fixtures", "appbrain-popular.html") + +def run(): + out = subprocess.check_output( + [sys.executable, SCRIPT, "popular", "--region", "us", "--html-file", FIXTURE]) + return json.loads(out) + +def main(): + d = run() + fails = [] + def check(name, cond): + print(f"{'PASS' if cond else 'FAIL'}: {name}") + if not cond: fails.append(name) + + check("source", d["source"] == "appbrain") + check("chart", d["chart"] == "popular") + check("count", d["count"] == 2) + e0 = d["entries"][0] + check("rank 1", e0["rank"] == 1) + check("name", e0["name"] == "Habit Tracker") + check("package", e0["package"] == "com.example.habit") + check("developer", e0["developer"] == "Focus Labs") + check("category", e0["category"] == "Productivity") + check("rating", e0["rating"] == 4.6) + check("installs", e0["installs"] == "5,000,000+") + for k in ["rank", "name", "package", "developer", "category", "rating", "installs"]: + check(f"key present: {k}", k in e0) + check("second package", d["entries"][1]["package"] == "com.example.budget") + sys.exit(1 if fails else 0) + +main()