Update app.py

app.py

@@ -1,17 +1,183 @@
-import httpx
-import pandas as pd
-import io
-
-url = "https://data.commoncrawl.org/cc-index/table/cc-main/warc/crawl=CC-MAIN-2013-20/subset=warc/part-00000-6ac52f25-05a1-4998-adf1-b8c830c08eec.c000.gz.parquet"
-
-print("Downloading...")
-with httpx.stream("GET", url) as r:
-    data = b""
-    for chunk in r.iter_bytes():
-        data += chunk
-
-print("Reading parquet...")
-df = pd.read_parquet(io.BytesIO(data))
-print("Columns:", df.columns.tolist())
-print("Shape:", df.shape)
-print(df.head(3))
+"""
+Common Crawl URL Extractor
+--------------------------
+Streams every URL from the latest Common Crawl crawl into chunked .gz files.
+Resumable: skips shards whose output file already exists.
+
+Output: urls_chunk_000.gz, urls_chunk_001.gz, ... (one per CDX shard)
+        Each file contains one URL per line, gzip-compressed.
+
+Estimated totals (latest single crawl):
+  - ~300 CDX shards
+  - ~800 MB per shard (compressed download)
+  - ~3–5 billion URLs total
+  - ~240 GB total download bandwidth
+"""
+
+import gzip
+import json
+import os
+import time
+import requests
+
+# ── Config ────────────────────────────────────────────────────────────────────
+
+OUTPUT_DIR   = "/data/cc_urls"          # where chunked .gz files are written
+COLLINFO_URL = "https://index.commoncrawl.org/collinfo.json"
+BASE_URL     = "https://data.commoncrawl.org/"
+STREAM_CHUNK = 1 * 1024 * 1024     # 1 MB read chunks while streaming
+MAX_RETRIES  = 5
+RETRY_DELAY  = 10                   # seconds between retries
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+def get_latest_crawl_paths():
+    """Fetch the list of CDX shard paths for the most recent crawl."""
+    print("Fetching crawl list from collinfo.json ...")
+    info = requests.get(COLLINFO_URL, timeout=30).json()
+    # collinfo.json is newest-first
+    latest_id = info[0]["id"]
+    paths_url = f"{BASE_URL}crawl-data/{latest_id}/cc-index.paths.gz"
+    print(f"Latest crawl: {latest_id}")
+    print(f"Fetching shard list from: {paths_url}")
+
+    r = requests.get(paths_url, timeout=60)
+    r.raise_for_status()
+
+    # The paths file is a gzip'd plain-text list, one path per line
+    raw = gzip.decompress(r.content).decode("utf-8")
+    paths = [
+        line.strip()
+        for line in raw.splitlines()
+        if line.strip().endswith(".gz") and "/indexes/cdx-" in line
+    ]
+    print(f"Found {len(paths)} CDX shards.")
+    return latest_id, paths
+
+
+def shard_output_path(output_dir, shard_index):
+    return os.path.join(output_dir, f"urls_chunk_{shard_index:03d}.gz")
+
+
+def extract_urls_from_shard(shard_url, out_path):
+    """
+    Stream a CDX shard, extract the URL from each JSON line,
+    write compressed output. Returns count of URLs written.
+    """
+    count = 0
+    leftover = b""
+
+    for attempt in range(1, MAX_RETRIES + 1):
+        try:
+            r = requests.get(shard_url, stream=True, timeout=120)
+            r.raise_for_status()
+
+            with gzip.open(out_path + ".tmp", "wb") as out_gz:
+                # The CDX file is itself gzip-compressed, so decompress on the fly
+                decompressor = gzip.GzipFile(fileobj=r.raw)
+                while True:
+                    chunk = decompressor.read(STREAM_CHUNK)
+                    if not chunk:
+                        break
+                    # Process complete lines; carry over incomplete tail
+                    block = leftover + chunk
+                    lines = block.split(b"\n")
+                    leftover = lines[-1]  # last may be incomplete
+                    for line in lines[:-1]:
+                        line = line.strip()
+                        if not line:
+                            continue
+                        try:
+                            # CDX line format:
+                            #   <surt-url> <timestamp> <json-blob>
+                            # We only need the JSON blob (3rd field onward)
+                            parts = line.split(b" ", 2)
+                            if len(parts) < 3:
+                                continue
+                            obj = json.loads(parts[2])
+                            url = obj.get("url")
+                            if url:
+                                out_gz.write((url + "\n").encode("utf-8"))
+                                count += 1
+                        except (json.JSONDecodeError, UnicodeDecodeError):
+                            continue
+
+                # Flush leftover
+                if leftover.strip():
+                    try:
+                        parts = leftover.split(b" ", 2)
+                        if len(parts) >= 3:
+                            obj = json.loads(parts[2])
+                            url = obj.get("url")
+                            if url:
+                                out_gz.write((url + "\n").encode("utf-8"))
+                                count += 1
+                    except Exception:
+                        pass
+
+            # Rename temp file to final on success
+            os.replace(out_path + ".tmp", out_path)
+            return count
+
+        except Exception as e:
+            print(f"  Attempt {attempt}/{MAX_RETRIES} failed: {e}")
+            if os.path.exists(out_path + ".tmp"):
+                os.remove(out_path + ".tmp")
+            if attempt < MAX_RETRIES:
+                time.sleep(RETRY_DELAY * attempt)
+            else:
+                print(f"  Giving up on this shard after {MAX_RETRIES} attempts.")
+                return 0
+
+
+# ── Main ──────────────────────────────────────────────────────────────────────
+
+def main():
+    os.makedirs(OUTPUT_DIR, exist_ok=True)
+
+    crawl_id, shard_paths = get_latest_crawl_paths()
+
+    total_urls   = 0
+    skipped      = 0
+    failed       = 0
+    start_time   = time.time()
+
+    print(f"\nStarting extraction → {OUTPUT_DIR}/\n{'─'*60}")
+
+    for i, shard_path in enumerate(shard_paths):
+        out_path   = shard_output_path(OUTPUT_DIR, i)
+        shard_url  = BASE_URL + shard_path
+
+        # ── Resume: skip already-done shards ──────────────────────────────
+        if os.path.exists(out_path):
+            skipped += 1
+            print(f"[{i+1:03d}/{len(shard_paths)}] SKIP  {os.path.basename(shard_path)}")
+            continue
+
+        print(f"[{i+1:03d}/{len(shard_paths)}] Fetching {os.path.basename(shard_path)} ...")
+        t0 = time.time()
+
+        count = extract_urls_from_shard(shard_url, out_path)
+        elapsed = time.time() - t0
+
+        if count > 0:
+            total_urls += count
+            size_mb = os.path.getsize(out_path) / 1024**2
+            print(f"           ✓ {count:,} URLs  |  {size_mb:.1f} MB  |  {elapsed:.0f}s")
+        else:
+            failed += 1
+            print(f"           ✗ Failed — shard skipped")
+
+    # ── Summary ───────────────────────────────────────────────────────────
+    total_elapsed = time.time() - start_time
+    print(f"\n{'─'*60}")
+    print(f"Crawl:      {crawl_id}")
+    print(f"Shards:     {len(shard_paths)} total  |  {skipped} skipped  |  {failed} failed")
+    print(f"URLs:       {total_urls:,}")
+    print(f"Output dir: {OUTPUT_DIR}/")
+    print(f"Time:       {total_elapsed/3600:.2f} hours")
+    print(f"{'─'*60}")
+
+
+if __name__ == "__main__":
+    main()