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.

RUN_PLAN.md

@@ -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 ~40–60k 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, ~3–6 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 1–2 nights.
+
+**Rough timing:** ~400k rows at ~16 req/s ≈ **7 hours** (one night). 3 countries ≈ 10–11 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.

scraper.py

@@ -36,6 +36,8 @@ import argparse
 import csv
 import json
 import logging
+import os
+import shutil
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -60,7 +62,8 @@ MAX_RATE_LIMIT_WAITS = 12
 MAX_TRANSIENT_RETRIES = 4
 
 CSV_FIELDS = [
-    "store", "country", "app_id", "title", "developer", "category",
+    "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",
@@ -74,6 +77,33 @@ def epoch_to_date(ts):
         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)
@@ -83,6 +113,8 @@ def play_row(app_id: str, lang: str, country: str) -> dict:
         "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"),
@@ -90,7 +122,7 @@ def play_row(app_id: str, lang: str, country: str) -> dict:
         "avg_rating": d.get("score"),
         "total_ratings": d.get("ratings"),
         "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"),
         "url": d.get("url"),
     }
@@ -117,6 +149,8 @@ def appstore_row(app_id: int, country: str) -> dict | None:
         "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"),
@@ -134,9 +168,16 @@ def appstore_row(app_id: int, country: str) -> dict | None:
 class CsvSink:
     """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.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")
@@ -147,9 +188,34 @@ class CsvSink:
     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()
 
 
@@ -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,
-        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():
         log.error("Config %s not found — run discover.py first to create it.",
                   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"
     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)
@@ -248,7 +326,7 @@ def run(config_path: Path, out_dir: Path, countries_override: list | None,
 
     total = len(jobs)
     throttle = Throttle()
-    sink = CsvSink(out_path, CSV_FIELDS)
+    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",
@@ -308,6 +386,10 @@ def main() -> None:
                         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:
@@ -318,7 +400,8 @@ def main() -> None:
         if args.countries else None
     )
     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__":