Update app.py

app.py

@@ -33,7 +33,6 @@ def serve():
         conn.close()
 
 # ── Friendly name ─────────────────────────────────────────────────────────────
-# data/CC-MAIN-2025-05/000_00042.parquet → cc2025-05_000042.parquet
 def friendly_name(hf_path):
     m = re.search(r"CC-MAIN-(\d{4}-\d+)/\d+_(\d+)\.parquet", hf_path)
     if m:
@@ -62,7 +61,7 @@ def save_state(state):
         json.dump(state, f, indent=2)
     os.replace(tmp, STATE_FILE)
 
-# ── Discover — builds queue only, does NOT touch state["shards"] ──────────────
+# ── Discover ──────────────────────────────────────────────────────────────────
 def discover_queue(state):
     print("Discovering shards from HF...")
     files = api.list_repo_files(DATASET_REPO, repo_type="dataset")
@@ -90,7 +89,7 @@ def reclaim_stale(state):
     if reclaimed:
         save_state(state)
 
-# ── Download loop — pops queue, downloads, THEN adds to state as pending ──────
+# ── Download loop ─────────────────────────────────────────────────────────────
 def download_loop(state):
     base_url = f"https://huggingface.co/datasets/{DATASET_REPO}/resolve/main/"
 
@@ -123,6 +122,7 @@ def download_loop(state):
         hf_path  = state["queue"][0]
         name     = friendly_name(hf_path)
         raw_path = Path(RAW_DIR) / name
+        tmp_path = Path(RAW_DIR) / f"{name}.tmp"
         url      = base_url + hf_path
 
         print(f"  Downloading: {hf_path} → {name}")
@@ -134,15 +134,16 @@ def download_loop(state):
                 stream=True,
             )
             resp.raise_for_status()
-            with open(raw_path, "wb") as f:
+            with open(tmp_path, "wb") as f:
                 for chunk in resp.iter_content(chunk_size=8 * 1024 * 1024):
                     f.write(chunk)
+            tmp_path.rename(raw_path)  # ← atomic, only visible when complete
         except Exception as e:
             print(f"  ✗ Failed: {e} — retrying in 30s")
+            tmp_path.unlink(missing_ok=True)  # ← clean up partial
             time.sleep(30)
             continue
 
-        # Only now: pop from queue, add to state as pending
         state["queue"].pop(0)
         state["shards"][name] = {
             "status":     "pending",