"""
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 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", "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 ------------------------------------------------------------
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"),
        "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": d.get("lastUpdatedOn") or epoch_to_date(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"),
        "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):
        self.fields = fields
        self.lock = threading.Lock()
        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()

    def close(self) -> None:
        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) -> 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

    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)

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


if __name__ == "__main__":
    main()