""" 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()