Compare commits

..

2 Commits

Author SHA1 Message Date
dhruv.godvani 22582bdcd7 Add developer email and website fields to CSV output
- 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.
2026-06-18 12:36:17 +05:30
dhruv.godvani 159c00762d feat: enhance CSV handling with backup snapshots and last updated date parsing
- Added support for parsing multiple date formats for the "lastUpdatedOn" field from Google Play.
- Implemented a backup mechanism in CsvSink to create snapshots of the CSV file every N rows, with a configurable number of backups to keep.
- Updated the CsvSink class to handle file flushing and backup creation safely.
- Modified the run function to allow configuration of backup settings via command-line arguments.
2026-06-17 19:25:06 +05:30
3 changed files with 212 additions and 400025 deletions
+122
View File
@@ -0,0 +1,122 @@
# Run Plan — Collect ~2 Lakh (200k) Apps + Scrape Overnight
A step-by-step runbook for one full-time laptop. Goal: discover **~200,000 apps**
(Google Play + Apple App Store) across multiple countries, then scrape full data
for **US + AU** (add more as needed) overnight.
> Full architecture & how everything works: see **ARCHITECTURE.html**.
---
## The strategy in one line
**App Store** is the volume engine (~180 apps per search term) → it carries you to 200k.
**Google Play** is slow (~30 per query, hard cap) → it adds ~4060k using the fast `--deep` set.
Running Play with the full wordlist would waste ~11 hours on duplicates, so we **don't**.
---
## PHASE 1 — Discover the apps (run during the day, ~36 hrs)
### Step 1. App Store — the big pull (full wordlist)
```powershell
python discover.py --skip-play --target 160000 --deep --terms-file english_words.txt --countries us,au,gb,in,ca,de
```
### Step 2. Google Play — bounded & fast (adds Play apps)
```powershell
python discover.py --skip-appstore --deep --countries us,au,gb,in,ca,de
```
Both steps **merge into the same `apps.json`** (Step 2 keeps Step 1's results).
More countries = more unique apps → edit the `--countries` list to match where you operate.
### Step 3. Back up the list immediately (it's a multi-hour artifact!)
```powershell
Copy-Item apps.json apps_backup.json
```
### Check the count anytime
```powershell
python -c "import json;c=json.load(open('apps.json',encoding='utf-8'));print('Play',len(c['google_play']),'+ App Store',len(c['app_store']),'=',len(c['google_play'])+len(c['app_store']))"
```
If under 200k, add more countries (e.g. `,fr,br,jp,mx`) and re-run **Step 1**.
---
## PHASE 2 — Scrape the data (overnight)
### Before you start: set a faster pace (one-time)
Open `apps.json` and set the delay in `settings` to `0.5` (currently `1.0`):
```jsonc
"delay_seconds": 0.5
```
Effective request rate ≈ `workers ÷ delay`. At `workers 8` and `delay 0.5` → ~16 req/s.
### Run the scrape (US + AU)
```powershell
python scraper.py --countries us,au --workers 8
```
- Produces **one row per app per country** → ~200k apps × 2 countries ≈ 400k rows.
- Output: **`output/app_data.csv`**
- Auto-backups every 1,000 rows to **`output/backups/`**.
### If it doesn't finish in one night
Just **run the exact same command again** the next night — it **resumes** and skips
everything already done. It's totally fine if this takes 12 nights.
**Rough timing:** ~400k rows at ~16 req/s ≈ **7 hours** (one night). 3 countries ≈ 1011 hrs.
---
## Safety (unattended overnight)
- The **circuit breaker** auto-pauses all workers if a store rate-limits, then resumes.
Seeing occasional `THROTTLING: pausing all workers` in the log = protection working, not an error.
- If you wake up to **constant** throttling messages → lower to `--workers 6` next run.
- Don't push workers too high while you sleep — an IP block at 2am costs more time than running slower.
---
## Recovery (if something gets deleted)
- **`output/app_data.csv` lost?** Copy the newest snapshot back, then re-run (it resumes):
```powershell
Copy-Item output\backups\app_data.bak3.csv output\app_data.csv
python scraper.py --countries us,au --workers 8
```
- **`apps.json` lost?** Restore your backup:
```powershell
Copy-Item apps_backup.json apps.json
```
---
## Output columns (`output/app_data.csv`)
`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`
- `developer_email` → filled for almost all **Google Play** apps; **blank for App Store** (Apple's API has none).
- `developer_website` → filled for both stores.
---
## Command cheat sheet
```powershell
# DISCOVER (App Store volume + Play)
python discover.py --skip-play --target 160000 --deep --terms-file english_words.txt --countries us,au,gb,in,ca,de
python discover.py --skip-appstore --deep --countries us,au,gb,in,ca,de
Copy-Item apps.json apps_backup.json
# SCRAPE (overnight, resumable)
python scraper.py --countries us,au --workers 8
# CHECK COUNT
python -c "import json;c=json.load(open('apps.json',encoding='utf-8'));print(len(c['google_play'])+len(c['app_store']))"
```
---
## Notes / limits (honest)
- No store has an "all apps" list — you only get what discovery finds. 200k is realistic; *every* app is not.
- Google Play search is capped at ~30 results/query — Play volume comes from many terms, not from one big query.
- Developer **phone numbers** are not available from either source.
- Data collected is **public, non-personal** app metadata. Using `developer_email` for bulk
outreach is governed by anti-spam law (CAN-SPAM / GDPR / CASL) — get sign-off before any marketing use.
-400018
View File
File diff suppressed because it is too large Load Diff
+90 -7
View File
@@ -36,6 +36,8 @@ import argparse
import csv import csv
import json import json
import logging import logging
import os
import shutil
import threading import threading
import time import time
from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -60,7 +62,8 @@ MAX_RATE_LIMIT_WAITS = 12
MAX_TRANSIENT_RETRIES = 4 MAX_TRANSIENT_RETRIES = 4
CSV_FIELDS = [ CSV_FIELDS = [
"store", "country", "app_id", "title", "developer", "category", "store", "country", "app_id", "title", "developer",
"developer_email", "developer_website", "category",
"price", "currency", "free", "price", "currency", "free",
"avg_rating", "total_ratings", "text_review_count", "avg_rating", "total_ratings", "text_review_count",
"last_updated", "version", "url", "last_updated", "version", "url",
@@ -74,6 +77,33 @@ def epoch_to_date(ts):
return None 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 ------------------------------------------------------------ # --- Google Play ------------------------------------------------------------
def play_row(app_id: str, lang: str, country: str) -> dict: def play_row(app_id: str, lang: str, country: str) -> dict:
d = gp_app(app_id, lang=lang, country=country) d = gp_app(app_id, lang=lang, country=country)
@@ -83,6 +113,8 @@ def play_row(app_id: str, lang: str, country: str) -> dict:
"app_id": app_id, "app_id": app_id,
"title": d.get("title"), "title": d.get("title"),
"developer": d.get("developer"), "developer": d.get("developer"),
"developer_email": d.get("developerEmail"),
"developer_website": d.get("developerWebsite"),
"category": d.get("genre"), "category": d.get("genre"),
"price": d.get("price"), "price": d.get("price"),
"currency": d.get("currency"), "currency": d.get("currency"),
@@ -90,7 +122,7 @@ def play_row(app_id: str, lang: str, country: str) -> dict:
"avg_rating": d.get("score"), "avg_rating": d.get("score"),
"total_ratings": d.get("ratings"), "total_ratings": d.get("ratings"),
"text_review_count": d.get("reviews"), "text_review_count": d.get("reviews"),
"last_updated": d.get("lastUpdatedOn") or epoch_to_date(d.get("updated")), "last_updated": play_last_updated(d.get("lastUpdatedOn"), d.get("updated")),
"version": d.get("version"), "version": d.get("version"),
"url": d.get("url"), "url": d.get("url"),
} }
@@ -117,6 +149,8 @@ def appstore_row(app_id: int, country: str) -> dict | None:
"app_id": meta.get("trackId"), "app_id": meta.get("trackId"),
"title": meta.get("trackName"), "title": meta.get("trackName"),
"developer": meta.get("sellerName"), "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"), "category": meta.get("primaryGenreName"),
"price": price, "price": price,
"currency": meta.get("currency"), "currency": meta.get("currency"),
@@ -134,9 +168,16 @@ def appstore_row(app_id: int, country: str) -> dict | None:
class CsvSink: class CsvSink:
"""Append rows one at a time, flushing each so progress survives a crash.""" """Append rows one at a time, flushing each so progress survives a crash."""
def __init__(self, path: Path, fields: list): def __init__(self, path: Path, fields: list,
backup_every: int = 1000, keep: int = 5):
self.path = path
self.fields = fields self.fields = fields
self.lock = threading.Lock() 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 write_header = not path.exists() or path.stat().st_size == 0
self.fh = path.open("a", newline="", encoding="utf-8-sig") self.fh = path.open("a", newline="", encoding="utf-8-sig")
self.writer = csv.DictWriter(self.fh, fieldnames=fields, extrasaction="ignore") self.writer = csv.DictWriter(self.fh, fieldnames=fields, extrasaction="ignore")
@@ -147,10 +188,35 @@ class CsvSink:
def write(self, row: dict) -> None: def write(self, row: dict) -> None:
with self.lock: with self.lock:
self.writer.writerow({k: row.get(k) for k in self.fields}) 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() 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: def close(self) -> None:
self.fh.close() 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 ---------------------------------------------------------- # --- orchestration ----------------------------------------------------------
@@ -220,7 +286,8 @@ def fetch_one(job, lang: str, delay: float, throttle: Throttle):
def run(config_path: Path, out_dir: Path, countries_override: list | None, def run(config_path: Path, out_dir: Path, countries_override: list | None,
workers: int, shard: int, shards: int, resume: bool) -> None: workers: int, shard: int, shards: int, resume: bool,
backup_every: int, backups_keep: int) -> None:
if not config_path.exists(): if not config_path.exists():
log.error("Config %s not found — run discover.py first to create it.", log.error("Config %s not found — run discover.py first to create it.",
config_path) config_path)
@@ -239,6 +306,17 @@ def run(config_path: Path, out_dir: Path, countries_override: list | None,
name = "app_data.csv" if shards == 1 else f"app_data_shard{shard}of{shards}.csv" name = "app_data.csv" if shards == 1 else f"app_data_shard{shard}of{shards}.csv"
out_path = out_dir / name 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 skipped = 0
if resume: if resume:
done = load_done(out_path) done = load_done(out_path)
@@ -248,7 +326,7 @@ def run(config_path: Path, out_dir: Path, countries_override: list | None,
total = len(jobs) total = len(jobs)
throttle = Throttle() throttle = Throttle()
sink = CsvSink(out_path, CSV_FIELDS) sink = CsvSink(out_path, CSV_FIELDS, backup_every=backup_every, keep=backups_keep)
log.info( log.info(
"shard %d/%d | %d jobs (%d already done, skipped) | %d workers | %s", "shard %d/%d | %d jobs (%d already done, skipped) | %d workers | %s",
@@ -308,6 +386,10 @@ def main() -> None:
help="total number of machines splitting the work") help="total number of machines splitting the work")
parser.add_argument("--no-resume", action="store_true", parser.add_argument("--no-resume", action="store_true",
help="ignore existing CSV and scrape everything again") 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() args = parser.parse_args()
if not 0 <= args.shard < args.shards: if not 0 <= args.shard < args.shards:
@@ -318,7 +400,8 @@ def main() -> None:
if args.countries else None if args.countries else None
) )
run(Path(args.config), Path(args.out), countries, run(Path(args.config), Path(args.out), countries,
args.workers, args.shard, args.shards, resume=not args.no_resume) args.workers, args.shard, args.shards, resume=not args.no_resume,
backup_every=args.backup_every, backups_keep=args.backups_keep)
if __name__ == "__main__": if __name__ == "__main__":