eda.py

"""
EDA script for Amazon Review + Meta datasets.
Streams both files without full extraction to disk.
"""
import gzip, io, json, sys
from collections import Counter, defaultdict
from pathlib import Path

REVIEW_PATH = Path("/workspace/amazon/All_Amazon_Review_5.json.gz")
META_PATH   = Path("/workspace/amazon/All_Amazon_Meta.json.gz")

SAMPLE_REVIEWS = 500_000   # rows to sample for review EDA
SAMPLE_META    = 200_000   # rows to sample for meta EDA


# ─── helpers ────────────────────────────────────────────────────────────────

def open_review(path):
    """Single gzip — standard JSONL."""
    return gzip.open(path, "rt", encoding="utf-8", errors="replace")


def open_meta(path):
    """Double gzip — outer gz → inner gz → JSONL, fully streaming."""
    f_outer = gzip.open(path, "rb")
    f_inner = gzip.open(f_outer, "rt", encoding="utf-8", errors="replace")
    return f_inner


def stream_jsonl(fh, max_rows, label=""):
    for i, line in enumerate(fh):
        if i >= max_rows:
            break
        if i % 100_000 == 0 and i > 0:
            print(f"  [{label}] {i:,} rows...", flush=True)
        line = line.strip()
        if not line:
            continue
        try:
            yield json.loads(line)
        except json.JSONDecodeError:
            continue


# ─── Review EDA ─────────────────────────────────────────────────────────────

def eda_reviews():
    print("\n" + "="*60)
    print("REVIEW FILE EDA")
    print("="*60)

    fields_seen    = Counter()
    overall_dist   = Counter()
    verified_dist  = Counter()
    user_counts    = Counter()
    item_counts    = Counter()
    year_counts    = Counter()
    has_text       = 0
    has_vote       = 0
    total          = 0

    with open_review(REVIEW_PATH) as fh:
        for rec in stream_jsonl(fh, SAMPLE_REVIEWS, "review"):
            total += 1
            for k in rec:
                fields_seen[k] += 1

            overall_dist[int(rec.get("overall", 0))] += 1
            verified_dist[rec.get("verified", None)] += 1
            user_counts[rec.get("reviewerID", "")] += 1
            item_counts[rec.get("asin", "")] += 1

            ts = rec.get("unixReviewTime")
            if ts:
                import datetime
                year_counts[datetime.datetime.fromtimestamp(ts).year] += 1

            if rec.get("reviewText", "").strip():
                has_text += 1
            if rec.get("vote"):
                has_vote += 1

    print(f"\nRows sampled      : {total:,}")
    print(f"Unique users      : {len(user_counts):,}")
    print(f"Unique items      : {len(item_counts):,}")

    print("\nField presence (%):")
    for f, cnt in sorted(fields_seen.items(), key=lambda x: -x[1]):
        print(f"  {f:<20} {cnt/total*100:6.1f}%")

    print("\nRating distribution:")
    for star in sorted(overall_dist):
        bar = "#" * int(overall_dist[star] / total * 50)
        print(f"  {star}★  {overall_dist[star]:>7,}  {bar}")

    print(f"\nVerified purchases: {verified_dist.get(True,0)/total*100:.1f}%")
    print(f"Has review text  : {has_text/total*100:.1f}%")
    print(f"Has vote field   : {has_vote/total*100:.1f}%")

    print("\nYear distribution (sampled):")
    for yr in sorted(year_counts):
        bar = "#" * int(year_counts[yr] / total * 40)
        print(f"  {yr}  {year_counts[yr]:>7,}  {bar}")

    print("\nUser review count distribution:")
    cnt_vals = list(user_counts.values())
    cnt_vals.sort()
    n = len(cnt_vals)
    print(f"  min={cnt_vals[0]}  p25={cnt_vals[n//4]}  median={cnt_vals[n//2]}  "
          f"p75={cnt_vals[3*n//4]}  p90={cnt_vals[int(n*.9)]}  "
          f"p99={cnt_vals[int(n*.99)]}  max={cnt_vals[-1]}")
    print(f"  Users with >=5 reviews: {sum(1 for v in cnt_vals if v>=5):,} "
          f"({sum(1 for v in cnt_vals if v>=5)/n*100:.1f}%)")

    print("\nTop-10 most reviewed items:")
    for asin, cnt in item_counts.most_common(10):
        print(f"  {asin}  {cnt:,}")


# ─── Meta EDA ───────────────────────────────────────────────────────────────

def eda_meta():
    print("\n" + "="*60)
    print("META FILE EDA  (double-gzip — loading outer layer first...)")
    print("="*60)

    fields_seen    = Counter()
    has_price      = 0
    has_also_buy   = 0
    has_also_view  = 0
    has_desc       = 0
    has_image      = 0
    cat_top        = Counter()
    price_vals     = []
    also_buy_lens  = []
    also_view_lens = []
    total          = 0

    with open_meta(META_PATH) as fh:
        for rec in stream_jsonl(fh, SAMPLE_META, "meta"):
            total += 1
            for k in rec:
                fields_seen[k] += 1

            p = rec.get("price")
            if p not in (None, "", "None"):
                has_price += 1
                try:
                    price_vals.append(float(str(p).replace("$","").replace(",","")))
                except ValueError:
                    pass

            ab = rec.get("also_buy") or []
            av = rec.get("also_view") or []
            desc = rec.get("description") or []

            if ab:
                has_also_buy  += 1
                also_buy_lens.append(len(ab))
            if av:
                has_also_view += 1
                also_view_lens.append(len(av))
            if desc and any(d.strip() for d in (desc if isinstance(desc, list) else [desc])):
                has_desc += 1
            if rec.get("imageURL") or rec.get("imageURLHighRes"):
                has_image += 1

            cats = rec.get("category") or []
            if cats:
                cat_top[cats[0]] += 1

    print(f"\nRows sampled      : {total:,}")

    print("\nField presence (%):")
    for f, cnt in sorted(fields_seen.items(), key=lambda x: -x[1]):
        print(f"  {f:<20} {cnt/total*100:6.1f}%")

    print(f"\nHas price        : {has_price/total*100:.1f}%")
    print(f"Has also_buy     : {has_also_buy/total*100:.1f}%")
    print(f"Has also_view    : {has_also_view/total*100:.1f}%")
    print(f"Has description  : {has_desc/total*100:.1f}%")
    print(f"Has image        : {has_image/total*100:.1f}%")

    if price_vals:
        price_vals.sort()
        n = len(price_vals)
        print(f"\nPrice stats (USD):")
        print(f"  min={price_vals[0]:.2f}  p25={price_vals[n//4]:.2f}  "
              f"median={price_vals[n//2]:.2f}  p75={price_vals[3*n//4]:.2f}  "
              f"p90={price_vals[int(n*.9)]:.2f}  p99={price_vals[int(n*.99)]:.2f}  "
              f"max={price_vals[-1]:.2f}")

    if also_buy_lens:
        also_buy_lens.sort()
        n = len(also_buy_lens)
        print(f"\nalso_buy length: mean={sum(also_buy_lens)/n:.1f}  "
              f"median={also_buy_lens[n//2]}  max={also_buy_lens[-1]}")
    if also_view_lens:
        also_view_lens.sort()
        n = len(also_view_lens)
        print(f"also_view length: mean={sum(also_view_lens)/n:.1f}  "
              f"median={also_view_lens[n//2]}  max={also_view_lens[-1]}")

    print("\nTop-15 categories (first-level):")
    for cat, cnt in cat_top.most_common(15):
        bar = "#" * int(cnt / total * 40)
        print(f"  {cat[:50]:<52} {cnt:>7,}  {bar}")

    print("\nSample records:")
    with open_meta(META_PATH) as fh:
        for i, rec in enumerate(stream_jsonl(fh, 3, "meta-sample")):
            print(f"\n--- Record {i+1} ---")
            for k, v in rec.items():
                val_str = str(v)[:120]
                print(f"  {k:<20}: {val_str}")


# ─── main ────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    mode = sys.argv[1] if len(sys.argv) > 1 else "all"

    if mode in ("all", "review"):
        eda_reviews()

    if mode in ("all", "meta"):
        eda_meta()

    print("\nDone.")