Update app.py

app.py

@@ -1,3 +1,39 @@
 import os
-print(os.listdir("/data/raw"))
-print(list(Path("/data/raw").iterdir()) if os.path.exists("/data/raw") else "DIR MISSING")
+import json
+from pathlib import Path
+import pyarrow.parquet as pq
+
+RAW_DIR    = "/data/raw"
+STATE_FILE = "/data/state.json"
+
+deleted = []
+kept    = []
+
+for f in Path(RAW_DIR).glob("*.parquet"):
+    try:
+        pq.read_schema(f)  # lightweight — only reads footer, not full file
+        kept.append(f.name)
+    except Exception as e:
+        print(f"  ✗ Corrupt: {f.name} — {e}")
+        f.unlink()
+        deleted.append(f.name)
+
+print(f"\n✓ Kept: {len(kept)} | Deleted: {len(deleted)}")
+
+# Reset deleted shards back to pending in state.json
+if deleted and os.path.exists(STATE_FILE):
+    with open(STATE_FILE) as f:
+        state = json.load(f)
+    reset = 0
+    for name in deleted:
+        if name in state["shards"]:
+            state["shards"][name]["status"]     = "pending"
+            state["shards"][name]["worker"]     = None
+            state["shards"][name]["claimed_at"] = None
+            state["shards"][name]["error"]      = None
+            reset += 1
+    tmp = STATE_FILE + ".tmp"
+    with open(tmp, "w") as f:
+        json.dump(state, f, indent=2)
+    os.replace(tmp, STATE_FILE)
+    print(f"✓ Reset {reset} shards to pending in state.json")