dhruv.godvani
22582bdcd7
- Updated CSV_FIELDS to include "developer_email" and "developer_website". - Modified play_row function to extract developer email and website from the response. - Updated appstore_row function to include developer website, with a note that developer email is not available from Apple's API.
409 lines
16 KiB
Python
409 lines
16 KiB
Python
"""
|
|
App store scraper — Google Play + Apple App Store.
|
|
|
|
For a list of apps (apps.json) it collects, into ONE combined CSV:
|
|
- core metadata (title, developer, category, price, version, LAST UPDATED date)
|
|
- the final rating (average) + total number of ratings
|
|
|
|
Input : apps.json (Play app IDs + App Store numeric IDs)
|
|
Output: output/app_data.csv (one row per app x country, both stores together)
|
|
|
|
SAFETY (never get IP-blocked):
|
|
- A shared circuit breaker (netutil.Throttle) pauses EVERY worker the moment a
|
|
rate-limit (HTTP 429/403) or a run of errors appears, then relaxes on success.
|
|
- 404/not-found is skipped (no retry); rate limits are waited out; timeouts/5xx
|
|
get a short backoff.
|
|
- Every row is flushed to disk immediately, so a crash/block loses nothing.
|
|
|
|
RESUME:
|
|
- On restart it reads the existing CSV and SKIPS apps already scraped, so you
|
|
just re-run after an interruption.
|
|
|
|
SPEED / SCALE:
|
|
--workers N concurrent requests (network-wait tasks -> ~Nx throughput)
|
|
--shard i with --shards N, this machine only does jobs i, i+N, i+2N, ...
|
|
--shards N run shard 0 on PC #1 and shard 1 on PC #2 (each its own IP),
|
|
then combine with merge_csv.py.
|
|
|
|
Run: python scraper.py
|
|
python scraper.py --workers 12
|
|
python scraper.py --shard 0 --shards 2 --workers 10 # PC #1
|
|
python scraper.py --shard 1 --shards 2 --workers 10 # PC #2
|
|
python merge_csv.py
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from google_play_scraper import app as gp_app
|
|
|
|
from netutil import Throttle, is_not_found, is_rate_limited
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%H:%M:%S"
|
|
)
|
|
log = logging.getLogger("scraper")
|
|
|
|
ITUNES_LOOKUP = "https://itunes.apple.com/lookup"
|
|
|
|
# Per-app retry budgets. Rate limits are waited out generously; transient
|
|
# network errors get a few quick retries before we give up on that one app.
|
|
MAX_RATE_LIMIT_WAITS = 12
|
|
MAX_TRANSIENT_RETRIES = 4
|
|
|
|
CSV_FIELDS = [
|
|
"store", "country", "app_id", "title", "developer",
|
|
"developer_email", "developer_website", "category",
|
|
"price", "currency", "free",
|
|
"avg_rating", "total_ratings", "text_review_count",
|
|
"last_updated", "version", "url",
|
|
]
|
|
|
|
|
|
def epoch_to_date(ts):
|
|
try:
|
|
return datetime.fromtimestamp(int(ts), tz=timezone.utc).strftime("%Y-%m-%d")
|
|
except (TypeError, ValueError, OSError):
|
|
return None
|
|
|
|
|
|
# Google Play's displayed "lastUpdatedOn" string can appear in several locale
|
|
# formats. We parse it to ISO (YYYY-MM-DD) because it matches what the store
|
|
# shows; the raw epoch is only a fallback (it can be off by a day in UTC).
|
|
_PLAY_DATE_FORMATS = (
|
|
"%b %d, %Y", # Jun 16, 2026
|
|
"%B %d, %Y", # June 16, 2026
|
|
"%d %b %Y", # 16 Jun 2026
|
|
"%d %B %Y", # 16 June 2026
|
|
"%Y-%m-%d", # 2026-06-16
|
|
"%d-%m-%Y", # 16-06-2026
|
|
"%m/%d/%Y", # 06/16/2026
|
|
"%d/%m/%Y", # 16/06/2026
|
|
)
|
|
|
|
|
|
def play_last_updated(text, epoch):
|
|
"""Return the last-updated date as ISO YYYY-MM-DD."""
|
|
if text:
|
|
t = str(text).strip()
|
|
for fmt in _PLAY_DATE_FORMATS:
|
|
try:
|
|
return datetime.strptime(t, fmt).strftime("%Y-%m-%d")
|
|
except ValueError:
|
|
continue
|
|
return epoch_to_date(epoch) # fallback when the string can't be parsed
|
|
|
|
|
|
# --- Google Play ------------------------------------------------------------
|
|
def play_row(app_id: str, lang: str, country: str) -> dict:
|
|
d = gp_app(app_id, lang=lang, country=country)
|
|
return {
|
|
"store": "google_play",
|
|
"country": country,
|
|
"app_id": app_id,
|
|
"title": d.get("title"),
|
|
"developer": d.get("developer"),
|
|
"developer_email": d.get("developerEmail"),
|
|
"developer_website": d.get("developerWebsite"),
|
|
"category": d.get("genre"),
|
|
"price": d.get("price"),
|
|
"currency": d.get("currency"),
|
|
"free": d.get("free"),
|
|
"avg_rating": d.get("score"),
|
|
"total_ratings": d.get("ratings"),
|
|
"text_review_count": d.get("reviews"),
|
|
"last_updated": play_last_updated(d.get("lastUpdatedOn"), d.get("updated")),
|
|
"version": d.get("version"),
|
|
"url": d.get("url"),
|
|
}
|
|
|
|
|
|
# --- Apple App Store --------------------------------------------------------
|
|
def appstore_metadata(app_id: int, country: str) -> dict | None:
|
|
resp = requests.get(
|
|
ITUNES_LOOKUP, params={"id": app_id, "country": country}, timeout=20
|
|
)
|
|
resp.raise_for_status() # turns 429/403/5xx into exceptions we classify
|
|
results = resp.json().get("results", [])
|
|
return results[0] if results else None
|
|
|
|
|
|
def appstore_row(app_id: int, country: str) -> dict | None:
|
|
meta = appstore_metadata(app_id, country)
|
|
if not meta:
|
|
return None
|
|
price = meta.get("price")
|
|
return {
|
|
"store": "app_store",
|
|
"country": country,
|
|
"app_id": meta.get("trackId"),
|
|
"title": meta.get("trackName"),
|
|
"developer": meta.get("sellerName"),
|
|
"developer_email": None, # Apple's lookup API does not expose a dev email
|
|
"developer_website": meta.get("sellerUrl"),
|
|
"category": meta.get("primaryGenreName"),
|
|
"price": price,
|
|
"currency": meta.get("currency"),
|
|
"free": (price == 0.0) if price is not None else None,
|
|
"avg_rating": meta.get("averageUserRating"),
|
|
"total_ratings": meta.get("userRatingCount"),
|
|
"text_review_count": None,
|
|
"last_updated": (meta.get("currentVersionReleaseDate") or "")[:10] or None,
|
|
"version": meta.get("version"),
|
|
"url": meta.get("trackViewUrl"),
|
|
}
|
|
|
|
|
|
# --- thread-safe, crash-safe CSV sink --------------------------------------
|
|
class CsvSink:
|
|
"""Append rows one at a time, flushing each so progress survives a crash."""
|
|
|
|
def __init__(self, path: Path, fields: list,
|
|
backup_every: int = 1000, keep: int = 5):
|
|
self.path = path
|
|
self.fields = fields
|
|
self.lock = threading.Lock()
|
|
self.backup_every = max(0, backup_every)
|
|
self.keep = max(1, keep)
|
|
self.backup_dir = path.parent / "backups"
|
|
self.n = 0 # rows written this session
|
|
self._bak_idx = -1 # ring-buffer index for rotating snapshots
|
|
write_header = not path.exists() or path.stat().st_size == 0
|
|
self.fh = path.open("a", newline="", encoding="utf-8-sig")
|
|
self.writer = csv.DictWriter(self.fh, fieldnames=fields, extrasaction="ignore")
|
|
if write_header:
|
|
self.writer.writeheader()
|
|
self.fh.flush()
|
|
|
|
def write(self, row: dict) -> None:
|
|
with self.lock:
|
|
self.writer.writerow({k: row.get(k) for k in self.fields})
|
|
self.fh.flush() # row is on disk immediately
|
|
self.n += 1
|
|
if self.backup_every and self.n % self.backup_every == 0:
|
|
self._snapshot()
|
|
|
|
def _snapshot(self) -> None:
|
|
"""Force to disk, then copy the live CSV into backups/ (ring of `keep`)."""
|
|
try:
|
|
self.fh.flush()
|
|
os.fsync(self.fh.fileno()) # guarantee physical write
|
|
self.backup_dir.mkdir(parents=True, exist_ok=True)
|
|
self._bak_idx = (self._bak_idx + 1) % self.keep
|
|
dest = self.backup_dir / f"{self.path.stem}.bak{self._bak_idx + 1}.csv"
|
|
shutil.copyfile(self.path, dest)
|
|
log.info("backup snapshot -> %s", dest)
|
|
except Exception as exc: # never let a backup break the run
|
|
log.warning("backup snapshot failed (continuing): %s", exc)
|
|
|
|
def close(self) -> None:
|
|
with self.lock:
|
|
if self.backup_every:
|
|
self._snapshot() # final snapshot before closing
|
|
else:
|
|
try:
|
|
self.fh.flush()
|
|
os.fsync(self.fh.fileno())
|
|
except Exception:
|
|
pass
|
|
self.fh.close()
|
|
|
|
|
|
# --- orchestration ----------------------------------------------------------
|
|
def job_key(store: str, app_id, country: str) -> tuple:
|
|
return (store, country, str(app_id))
|
|
|
|
|
|
def build_jobs(cfg: dict, countries: list) -> list:
|
|
"""Flat (store, app_id, country) list so sharding balances stores+countries."""
|
|
jobs = []
|
|
for country in countries:
|
|
for entry in cfg.get("google_play", []):
|
|
jobs.append(("google_play", entry["app_id"], country))
|
|
for entry in cfg.get("app_store", []):
|
|
jobs.append(("app_store", entry["app_id"], country))
|
|
return jobs
|
|
|
|
|
|
def load_done(path: Path) -> set:
|
|
"""Keys already present in the output CSV (for resume)."""
|
|
done = set()
|
|
if path.exists() and path.stat().st_size > 0:
|
|
with path.open(encoding="utf-8-sig", newline="") as fh:
|
|
for row in csv.DictReader(fh):
|
|
done.add((row.get("store"), row.get("country"), str(row.get("app_id"))))
|
|
return done
|
|
|
|
|
|
def fetch_one(job, lang: str, delay: float, throttle: Throttle):
|
|
"""
|
|
Scrape one app. Returns a row dict, or None if the app has no data (404).
|
|
Raises only after exhausting retries. Rate limits trip the shared breaker
|
|
so ALL workers slow down together.
|
|
"""
|
|
store, app_id, country = job
|
|
rate_waits = 0
|
|
transient = 0
|
|
while True:
|
|
throttle.wait() # honor any global cooldown before requesting
|
|
if delay:
|
|
time.sleep(delay) # steady per-thread pacing
|
|
try:
|
|
if store == "google_play":
|
|
row = play_row(app_id, lang, country)
|
|
else:
|
|
row = appstore_row(app_id, country)
|
|
throttle.record_success()
|
|
return row
|
|
except Exception as exc:
|
|
if is_not_found(exc):
|
|
throttle.record_success() # a 404 is a healthy answer
|
|
return None
|
|
rate = is_rate_limited(exc)
|
|
cd = throttle.record_fail(rate)
|
|
if cd:
|
|
log.warning("THROTTLING: pausing all workers ~%.0fs (%s)",
|
|
cd, "rate-limit" if rate else "error burst")
|
|
if rate:
|
|
rate_waits += 1
|
|
if rate_waits > MAX_RATE_LIMIT_WAITS:
|
|
raise RuntimeError(f"persistent rate-limit on {app_id}") from exc
|
|
continue # loop; throttle.wait() enforces the cooldown
|
|
transient += 1
|
|
if transient > MAX_TRANSIENT_RETRIES:
|
|
raise
|
|
time.sleep(min(2 ** transient, 20))
|
|
|
|
|
|
def run(config_path: Path, out_dir: Path, countries_override: list | None,
|
|
workers: int, shard: int, shards: int, resume: bool,
|
|
backup_every: int, backups_keep: int) -> None:
|
|
if not config_path.exists():
|
|
log.error("Config %s not found — run discover.py first to create it.",
|
|
config_path)
|
|
return
|
|
cfg = json.loads(config_path.read_text(encoding="utf-8"))
|
|
s = cfg.get("settings", {})
|
|
lang = s.get("lang", "en")
|
|
delay = float(s.get("delay_seconds", 1.0))
|
|
countries = countries_override or s.get("countries") or [s.get("country", "us")]
|
|
|
|
jobs = build_jobs(cfg, countries)
|
|
if shards > 1:
|
|
jobs = jobs[shard::shards]
|
|
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
name = "app_data.csv" if shards == 1 else f"app_data_shard{shard}of{shards}.csv"
|
|
out_path = out_dir / name
|
|
|
|
# If the live CSV is gone but snapshots exist, the user can recover instead
|
|
# of re-scraping from zero.
|
|
backup_dir = out_dir / "backups"
|
|
if not out_path.exists() and backup_dir.exists():
|
|
baks = sorted(backup_dir.glob(f"{out_path.stem}.bak*.csv"),
|
|
key=lambda p: p.stat().st_mtime, reverse=True)
|
|
if baks:
|
|
log.warning("%s is missing but a backup exists: %s", out_path, baks[0])
|
|
log.warning("To resume instead of restarting, copy that backup back to "
|
|
"%s, then re-run.", out_path)
|
|
|
|
skipped = 0
|
|
if resume:
|
|
done = load_done(out_path)
|
|
before = len(jobs)
|
|
jobs = [j for j in jobs if job_key(*j) not in done]
|
|
skipped = before - len(jobs)
|
|
|
|
total = len(jobs)
|
|
throttle = Throttle()
|
|
sink = CsvSink(out_path, CSV_FIELDS, backup_every=backup_every, keep=backups_keep)
|
|
|
|
log.info(
|
|
"shard %d/%d | %d jobs (%d already done, skipped) | %d workers | %s",
|
|
shard, shards, total, skipped, workers, ", ".join(countries),
|
|
)
|
|
if total == 0:
|
|
log.info("Nothing to do. -> %s", out_path)
|
|
sink.close()
|
|
return
|
|
|
|
stats = {"ok": 0, "empty": 0, "failed": 0}
|
|
done_n = 0
|
|
executor = ThreadPoolExecutor(max_workers=workers)
|
|
try:
|
|
futures = {
|
|
executor.submit(fetch_one, job, lang, delay, throttle): job
|
|
for job in jobs
|
|
}
|
|
for fut in as_completed(futures):
|
|
job = futures[fut]
|
|
try:
|
|
row = fut.result()
|
|
if row:
|
|
sink.write(row)
|
|
stats["ok"] += 1
|
|
else:
|
|
stats["empty"] += 1
|
|
except Exception as exc:
|
|
stats["failed"] += 1
|
|
log.error("FAILED %s -> %s", job, exc)
|
|
done_n += 1
|
|
if done_n % 50 == 0 or done_n == total:
|
|
log.info("progress %d/%d (ok=%d empty=%d failed=%d)",
|
|
done_n, total, stats["ok"], stats["empty"], stats["failed"])
|
|
except KeyboardInterrupt:
|
|
log.warning("Interrupted — saving progress and stopping. "
|
|
"Re-run to resume from where you left off.")
|
|
executor.shutdown(wait=False, cancel_futures=True)
|
|
finally:
|
|
sink.close()
|
|
|
|
log.info("DONE ok=%d empty=%d failed=%d -> %s",
|
|
stats["ok"], stats["empty"], stats["failed"], out_path)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Play Store + App Store scraper")
|
|
parser.add_argument("--config", default="apps.json", help="path to apps.json")
|
|
parser.add_argument("--out", default="output", help="output directory")
|
|
parser.add_argument("--countries", default=None,
|
|
help="comma-separated, e.g. us,gb,in (overrides settings)")
|
|
parser.add_argument("--workers", type=int, default=10,
|
|
help="concurrent requests (network-wait, so go high)")
|
|
parser.add_argument("--shard", type=int, default=0,
|
|
help="this machine's slice index (0-based)")
|
|
parser.add_argument("--shards", type=int, default=1,
|
|
help="total number of machines splitting the work")
|
|
parser.add_argument("--no-resume", action="store_true",
|
|
help="ignore existing CSV and scrape everything again")
|
|
parser.add_argument("--backup-every", type=int, default=1000,
|
|
help="snapshot the CSV to backups/ every N rows (0 = off)")
|
|
parser.add_argument("--backups-keep", type=int, default=5,
|
|
help="how many rolling snapshots to keep")
|
|
args = parser.parse_args()
|
|
|
|
if not 0 <= args.shard < args.shards:
|
|
parser.error("--shard must be between 0 and --shards-1")
|
|
|
|
countries = (
|
|
[c.strip() for c in args.countries.split(",") if c.strip()]
|
|
if args.countries else None
|
|
)
|
|
run(Path(args.config), Path(args.out), countries,
|
|
args.workers, args.shard, args.shards, resume=not args.no_resume,
|
|
backup_every=args.backup_every, backups_keep=args.backups_keep)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|