Update app.py

app.py

@@ -387,28 +387,40 @@ def get_next_file_to_process(source_repo_id: str, state: Dict[str, Any]) -> Opti
         next_index = state.get("next_download_index", 0)
         file_states = state.get("file_states", {})
         
-        # 3. Find the next file that hasn't been processed or is not currently being processed
-        for i in range(next_index, len(audio_files)):
-            filename = audio_files[i]
+        # 3. Skip forward past all processed and processing files starting from next_index
+        # This ensures we don't repeatedly find files that have already been handled
+        current_index = next_index
+        while current_index < len(audio_files):
+            filename = audio_files[current_index]
             status = file_states.get(filename, "unprocessed")
             
-            # Skip files that are already processed or currently being processed
+            # If this file is processed or currently processing, skip it
             if status in ["processed", "processing"]:
+                current_index += 1
                 continue
             
-            # Found an unprocessed file
-            file_url = hf_hub_url(repo_id=source_repo_id, filename=filename, repo_type="dataset")
+            # If this file failed, we can retry it, so return it
+            if status == "failed":
+                file_url = hf_hub_url(repo_id=source_repo_id, filename=filename, repo_type="dataset")
+                log_message(f"Found failed file for retry at index {current_index}: {filename}", "INFO")
+                return {
+                    "filename": filename,
+                    "url": file_url,
+                    "index": current_index
+                }
             
-            log_message(f"Found next file at index {i}: {filename}", "INFO")
+            # If this file is unprocessed, we found our next file
+            file_url = hf_hub_url(repo_id=source_repo_id, filename=filename, repo_type="dataset")
+            log_message(f"Found next file at index {current_index}: {filename}", "INFO")
             return {
                 "filename": filename,
                 "url": file_url,
-                "index": i
+                "index": current_index
             }
-            
-        log_message("All files up to the current index have been processed or are locked.", "INFO")
         
-        # If we reach the end, check from the beginning for any failed files
+        log_message("All files have been processed or are locked. Checking for any failed files from the start.", "INFO")
+        
+        # 4. If we've processed all files from next_index to end, check from beginning for failed files
         for i in range(0, next_index):
             filename = audio_files[i]
             status = file_states.get(filename, "unprocessed")
@@ -421,7 +433,8 @@ def get_next_file_to_process(source_repo_id: str, state: Dict[str, Any]) -> Opti
                     "url": file_url,
                     "index": i
                 }
-                
+        
+        log_message("All files have been processed. Waiting for new files...", "INFO")
         return None
         
     except Exception as e:
@@ -585,6 +598,9 @@ def main_processing_loop():
             
             try:
                 # 2. Lock file by updating state on the API
+                # IMPORTANT: Update next_download_index when locking to prevent other workers from picking same file
+                current_state["next_download_index"] = target_index + 1
+                
                 if not lock_file_for_processing(target_file, current_state):
                     log_message(f"❌ Failed to lock file {target_file}. Skipping.", "ERROR")
                     time.sleep(PROCESSING_DELAY)