""" Pass 1: collect unique ASINs from sampled_1m_reviews.jsonl Pass 2: stream All_Amazon_Meta.json.gz (double-gzip), write matching records Output: sampled_meta.jsonl — meta records cho các ASIN trong 1M users meta_stats.json — thống kê coverage """ import gzip, json, time from pathlib import Path REVIEWS_PATH = Path("/workspace/amazon/sampled_1m_reviews.jsonl") META_GZ = Path("/workspace/amazon/All_Amazon_Meta.json.gz") OUT_META = Path("/workspace/amazon/sampled_meta.jsonl") OUT_STATS = Path("/workspace/amazon/meta_stats.json") # ── Pass 1: collect ASINs from reviews ──────────────────────────────────── print("Pass 1: collecting ASINs from sampled reviews...", flush=True) t0 = time.time() review_asins = set() total_reviews = 0 with open(REVIEWS_PATH, "r", encoding="utf-8", errors="replace") as f: for line in f: line = line.strip() if not line: continue try: rec = json.loads(line) except: continue asin = rec.get("asin", "") if asin: review_asins.add(asin) total_reviews += 1 if total_reviews % 3_000_000 == 0: print(f" {total_reviews/1e6:.0f}M reviews, {len(review_asins):,} ASINs ({time.time()-t0:.0f}s)", flush=True) print(f"Pass 1 done in {time.time()-t0:.1f}s", flush=True) print(f" Reviews scanned : {total_reviews:,}", flush=True) print(f" Unique ASINs : {len(review_asins):,}", flush=True) # ── Pass 2: stream meta (double-gzip), write matching records ───────────── print(f"\nPass 2: streaming meta (double-gzip)...", flush=True) t1 = time.time() meta_total = 0 meta_written = 0 found_asins = set() # field quality counters has_title = 0 has_price = 0 has_also_buy = 0 has_also_view = 0 has_desc = 0 has_brand = 0 has_category = 0 with gzip.open(META_GZ, "rb") as outer, \ gzip.open(outer, "rt", encoding="utf-8", errors="replace") as f, \ open(OUT_META, "w", encoding="utf-8") as fout: for line in f: line = line.strip() if not line: continue meta_total += 1 if meta_total % 2_000_000 == 0: print(f" {meta_total/1e6:.0f}M meta rows, {meta_written:,} matched ({time.time()-t1:.0f}s)", flush=True) try: rec = json.loads(line) except: continue asin = rec.get("asin", "") if asin not in review_asins: continue # clean up noisy fields before saving clean = { "asin" : asin, "title" : rec.get("title", ""), "brand" : rec.get("brand", ""), "price" : rec.get("price", ""), "category" : rec.get("category") or [], "main_cat" : rec.get("main_cat", ""), "description": rec.get("description") or [], "feature" : rec.get("feature") or [], "also_buy" : rec.get("also_buy") or [], "also_view" : rec.get("also_view") or [], "rank" : rec.get("rank", ""), } fout.write(json.dumps(clean, ensure_ascii=False) + "\n") found_asins.add(asin) meta_written += 1 # field quality tracking if clean["title"].strip(): has_title += 1 if clean["price"] not in ("", None, "None"): has_price += 1 if clean["also_buy"]: has_also_buy += 1 if clean["also_view"]: has_also_view += 1 if clean["brand"].strip(): has_brand += 1 if clean["category"]: has_category += 1 desc = clean["description"] if isinstance(desc, list): desc = " ".join(desc) if str(desc).strip(): has_desc += 1 print(f"Pass 2 done in {time.time()-t1:.1f}s", flush=True) # ── Coverage & stats ─────────────────────────────────────────────────────── missing_asins = review_asins - found_asins coverage = len(found_asins) / len(review_asins) * 100 if review_asins else 0 print(f"\n{'='*50}") print(f"RESULTS") print(f"{'='*50}") print(f"Review ASINs : {len(review_asins):,}") print(f"Meta matched : {meta_written:,} ({coverage:.1f}% coverage)") print(f"Missing from meta : {len(missing_asins):,} ({100-coverage:.1f}%)") print(f"\nField quality (matched records):") n = meta_written or 1 print(f" title : {has_title/n*100:.1f}%") print(f" brand : {has_brand/n*100:.1f}%") print(f" price : {has_price/n*100:.1f}%") print(f" description : {has_desc/n*100:.1f}%") print(f" category : {has_category/n*100:.1f}%") print(f" also_buy : {has_also_buy/n*100:.1f}%") print(f" also_view : {has_also_view/n*100:.1f}%") stats = { "review_asins": len(review_asins), "meta_matched": meta_written, "meta_coverage_pct": round(coverage, 2), "missing_asins": len(missing_asins), "field_quality": { "title": round(has_title/n*100, 1), "brand": round(has_brand/n*100, 1), "price": round(has_price/n*100, 1), "description": round(has_desc/n*100, 1), "category": round(has_category/n*100, 1), "also_buy": round(has_also_buy/n*100, 1), "also_view": round(has_also_view/n*100, 1), } } with open(OUT_STATS, "w") as f: json.dump(stats, f, indent=2) print(f"\nOutput: {OUT_META} ({OUT_META.stat().st_size/1e9:.1f} GB)") print(f"Total time: {time.time()-t0:.1f}s")