count_eligible_users.py

"""
Count users with verified=True and >=5 interactions.
Streams All_Amazon_Review_5.json.gz without full extraction.
"""
import gzip, json, time
from collections import Counter

REVIEW_PATH = "/workspace/amazon/All_Amazon_Review_5.json.gz"
MIN_INTERACTIONS = 5

print("Streaming review file...", flush=True)
t0 = time.time()

user_counts   = Counter()   # user -> total verified interactions
user_items    = {}          # user -> set of unique ASINs (verified only)
total_rows    = 0
skipped_unverified = 0

with gzip.open(REVIEW_PATH, "rt", encoding="utf-8", errors="replace") as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        try:
            rec = json.loads(line)
        except json.JSONDecodeError:
            continue

        total_rows += 1
        if total_rows % 5_000_000 == 0:
            elapsed = time.time() - t0
            eligible = sum(1 for v in user_counts.values() if v >= MIN_INTERACTIONS)
            print(f"  {total_rows/1e6:.0f}M rows  |  {elapsed:.0f}s  |  "
                  f"eligible users so far: {eligible:,}", flush=True)

        if not rec.get("verified", False):
            skipped_unverified += 1
            continue

        uid  = rec.get("reviewerID", "")
        asin = rec.get("asin", "")
        if not uid or not asin:
            continue

        user_counts[uid] += 1
        if uid not in user_items:
            user_items[uid] = set()
        user_items[uid].add(asin)

elapsed = time.time() - t0
print(f"\nDone in {elapsed:.1f}s", flush=True)

# ── Stats ──────────────────────────────────────────────────────────────────
total_users      = len(user_counts)
eligible_users   = {u for u, c in user_counts.items() if c >= MIN_INTERACTIONS}
n_eligible       = len(eligible_users)
eligible_reviews = sum(user_counts[u] for u in eligible_users)

print(f"\n{'='*55}")
print(f"RESULTS")
print(f"{'='*55}")
print(f"Total rows scanned      : {total_rows:>12,}")
print(f"Skipped (unverified)    : {skipped_unverified:>12,}  ({skipped_unverified/total_rows*100:.1f}%)")
print(f"Verified rows           : {total_rows-skipped_unverified:>12,}  ({(total_rows-skipped_unverified)/total_rows*100:.1f}%)")
print(f"\nUnique users (verified) : {total_users:>12,}")
print(f"Users with >=5 verified : {n_eligible:>12,}  ({n_eligible/total_users*100:.1f}% of verified users)")
print(f"Reviews from eligible   : {eligible_reviews:>12,}")

# interaction count distribution among eligible users
counts = sorted(user_counts[u] for u in eligible_users)
n = len(counts)
if n:
    percentiles = [25, 50, 75, 90, 95, 99]
    print(f"\nInteraction distribution (eligible users only):")
    print(f"  min={counts[0]}  max={counts[-1]}")
    for p in percentiles:
        idx = min(int(n * p / 100), n - 1)
        print(f"  p{p:>2} = {counts[idx]}")

# bucket distribution
print(f"\nBucket breakdown (eligible users):")
buckets = [(5,9),(10,19),(20,49),(50,99),(100,499),(500,9999)]
for lo, hi in buckets:
    cnt = sum(1 for v in counts if lo <= v <= hi)
    print(f"  {lo:>4}–{hi:<4} interactions: {cnt:>8,}  ({cnt/n_eligible*100:.1f}%)")

# unique items per eligible user
unique_item_counts = sorted(len(user_items[u]) for u in eligible_users)
print(f"\nUnique items/user distribution (eligible):")
print(f"  min={unique_item_counts[0]}  "
      f"p50={unique_item_counts[n//2]}  "
      f"p90={unique_item_counts[int(n*.9)]}  "
      f"p99={unique_item_counts[int(n*.99)]}  "
      f"max={unique_item_counts[-1]}")