feat(market-research): add play.py resolve (name -> Play link + stats)
Live: org.isoron.uhabits returned for "habit tracker" with rating + last_updated. installs null on live (Play DOM varies); offline test all 7 PASS.
This commit is contained in:
fatih.bulut
parent
fbfa3aece7
commit
73e39c40e2
|
|
@ -0,0 +1,113 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Resolve an app name to a Google Play link + stats, and measure saturation.
|
||||
|
||||
Subcommands:
|
||||
resolve "<name>" -> top Play apps hit: package, play_url, rating, installs,
|
||||
last_updated, developer (1-2 links per candidate upstream).
|
||||
count "<query>" -> saturation: how many distinct apps + avg rating on the
|
||||
first Play search results page.
|
||||
|
||||
Play search (play.google.com/store/search?q=...&c=apps&hl=en&gl=US) is
|
||||
server-rendered enough to grep details?id= links + star aria-labels. The details
|
||||
page carries rating/installs/updated in embedded ld+json (+ light regex), the
|
||||
same structure clone-app's scrape-play-store.py parses. Stdlib-only, no pip.
|
||||
"""
|
||||
import sys, json, re, argparse, urllib.request, urllib.parse, ssl, subprocess, shutil
|
||||
|
||||
UA = "Mozilla/5.0"
|
||||
ID_RE = re.compile(r'/store/apps/details\?id=([\w.]+)')
|
||||
STAR_RE = re.compile(r'Rated\s+([\d.]+)\s+stars')
|
||||
|
||||
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")
|
||||
|
||||
def _search_html(query, search_file):
|
||||
if search_file:
|
||||
with open(search_file, encoding="utf-8") as f:
|
||||
return f.read()
|
||||
q = urllib.parse.quote(query)
|
||||
return _http_get(f"https://play.google.com/store/search?q={q}&c=apps&hl=en&gl=US")
|
||||
|
||||
def _details_html(package, details_file):
|
||||
if details_file:
|
||||
with open(details_file, encoding="utf-8") as f:
|
||||
return f.read()
|
||||
return _http_get(
|
||||
f"https://play.google.com/store/apps/details?id={package}&hl=en&gl=US")
|
||||
|
||||
def parse_details(html, package):
|
||||
out = {"name": None, "developer": None, "rating": None,
|
||||
"installs": None, "last_updated": None}
|
||||
for m in re.finditer(r'<script type="application/ld\+json"[^>]*>(.*?)</script>',
|
||||
html, re.DOTALL):
|
||||
try:
|
||||
data = json.loads(m.group(1).strip())
|
||||
except Exception:
|
||||
continue
|
||||
if isinstance(data, dict) and data.get("@type") == "SoftwareApplication":
|
||||
out["name"] = data.get("name")
|
||||
auth = data.get("author")
|
||||
if isinstance(auth, dict):
|
||||
out["developer"] = auth.get("name")
|
||||
ar = data.get("aggregateRating") or {}
|
||||
if ar.get("ratingValue") is not None:
|
||||
try: out["rating"] = float(ar["ratingValue"])
|
||||
except Exception: pass
|
||||
break
|
||||
m = re.search(r'([\d,]+\+)\s*</[^>]+>\s*<[^>]*>\s*Downloads', html)
|
||||
if not m:
|
||||
m = re.search(r'([\d,]+\+)\s*<span>\s*Downloads', html)
|
||||
if m:
|
||||
out["installs"] = m.group(1)
|
||||
m = re.search(r'Updated on\s*</[^>]+>\s*<[^>]*>([^<]+)</[^>]+>', html)
|
||||
if not m:
|
||||
m = re.search(r'Updated on\s*<span>([^<]+)</span>', html)
|
||||
if m:
|
||||
out["last_updated"] = m.group(1).strip()
|
||||
return out
|
||||
|
||||
def cmd_resolve(args):
|
||||
html = _search_html(args.query, args.search_file)
|
||||
m = ID_RE.search(html)
|
||||
res = {"query": args.query, "name": None, "package": None, "play_url": None,
|
||||
"rating": None, "installs": None, "last_updated": None, "developer": None}
|
||||
if not m:
|
||||
print(json.dumps(res, indent=2))
|
||||
return
|
||||
package = m.group(1)
|
||||
res["package"] = package
|
||||
res["play_url"] = f"https://play.google.com/store/apps/details?id={package}"
|
||||
details = _details_html(package, args.details_file)
|
||||
res.update(parse_details(details, package))
|
||||
print(json.dumps(res, indent=2))
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
pr = sub.add_parser("resolve")
|
||||
pr.add_argument("query")
|
||||
pr.add_argument("--search-file")
|
||||
pr.add_argument("--details-file")
|
||||
args = ap.parse_args()
|
||||
if args.cmd == "resolve":
|
||||
try:
|
||||
cmd_resolve(args)
|
||||
except Exception as e:
|
||||
print(f"ERROR: play resolve failed: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
<!DOCTYPE html><html><body>
|
||||
<script type="application/ld+json" nonce="abc">
|
||||
{"@type":"SoftwareApplication","name":"Habit Tracker","author":{"name":"Focus Labs"},
|
||||
"applicationCategory":"Productivity","aggregateRating":{"ratingValue":"4.6","ratingCount":"120000"}}
|
||||
</script>
|
||||
<div>5,000,000+</div><div>Downloads</div>
|
||||
<div>Updated on</div><div>Jun 1, 2026</div>
|
||||
</body></html>
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
<!DOCTYPE html><html><body>
|
||||
<div role="main">
|
||||
<a href="/store/apps/details?id=com.example.habit"><span>Habit Tracker</span></a>
|
||||
<div aria-label="Rated 4.6 stars out of five stars"></div>
|
||||
<a href="/store/apps/details?id=com.example.habit2"><span>Habit Plus</span></a>
|
||||
<div aria-label="Rated 4.1 stars out of five stars"></div>
|
||||
<a href="/store/apps/details?id=com.example.habit3"><span>Daily Habits</span></a>
|
||||
<div aria-label="Rated 3.9 stars out of five stars"></div>
|
||||
</div>
|
||||
</body></html>
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
#!/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", "play.py")
|
||||
SEARCH = os.path.join(HERE, "fixtures", "play-search.html")
|
||||
DETAILS = os.path.join(HERE, "fixtures", "play-details.html")
|
||||
|
||||
def run(*args):
|
||||
return json.loads(subprocess.check_output([sys.executable, SCRIPT, *args]))
|
||||
|
||||
def main():
|
||||
fails = []
|
||||
def check(name, cond):
|
||||
print(f"{'PASS' if cond else 'FAIL'}: {name}")
|
||||
if not cond: fails.append(name)
|
||||
|
||||
r = run("resolve", "Habit Tracker", "--search-file", SEARCH, "--details-file", DETAILS)
|
||||
check("package", r["package"] == "com.example.habit")
|
||||
check("play_url", r["play_url"] == "https://play.google.com/store/apps/details?id=com.example.habit")
|
||||
check("name", r["name"] == "Habit Tracker")
|
||||
check("rating", r["rating"] == 4.6)
|
||||
check("installs", r["installs"] == "5,000,000+")
|
||||
check("last_updated", r["last_updated"] == "Jun 1, 2026")
|
||||
check("developer", r["developer"] == "Focus Labs")
|
||||
|
||||
sys.exit(1 if fails else 0)
|
||||
|
||||
main()
|
||||
Loading…
Reference in New Issue