Update app.py

app.py

@@ -492,88 +492,89 @@ async def process_batch_dynamic(wav_files: List[str], start_batch_index: int, ba
                         wav_path.unlink()
     except Exception as e:
         print(f"[{FLOW_ID}] Error in process_batch_dynamic: {e}")
-        
-        return batch_end, uploaded_count
     
-    async def process_dataset_task(start_index: int):
-        """Main task to process the dataset using dynamic server assignment."""
+    return batch_end, uploaded_count
+
+async def process_dataset_task(start_index: int):
+    """Main task to process the dataset using dynamic server assignment."""
     
-        # Load both local progress and HF state
-        progress = load_progress()
-        current_state = await download_hf_state()
-        file_list = await get_audio_file_list(progress)
+    # Load both local progress and HF state
+    progress = load_progress()
+    current_state = await download_hf_state()
+    file_list = await get_audio_file_list(progress)
     
-        if not file_list:
-            print(f"[{FLOW_ID}] ERROR: Cannot proceed. File list is empty.")
-            return False
+    if not file_list:
+        print(f"[{FLOW_ID}] ERROR: Cannot proceed. File list is empty.")
+        return False
     
-        # Ensure start_index is within bounds
-        if start_index > len(file_list):
-            print(f"[{FLOW_ID}] WARNING: Start index {start_index} is greater than the total number of files ({len(file_list)}). Exiting.")
-            return True
+    # Ensure start_index is within bounds
+    if start_index > len(file_list):
+        print(f"[{FLOW_ID}] WARNING: Start index {start_index} is greater than the total number of files ({len(file_list)}). Exiting.")
+        return True
     
-        # Determine the actual starting index in the 0-indexed list
-        start_list_index = start_index - 1
+    # Determine the actual starting index in the 0-indexed list
+    start_list_index = start_index - 1
     
-        print(f"[{FLOW_ID}] Starting audio transcription from file index: {start_index} out of {len(file_list)}.")
-        print(f"[{FLOW_ID}] Using {len(servers)} Whisper servers for dynamic processing.")
-        print(f"[{FLOW_ID}] Upload pause enabled: {UPLOAD_PAUSE_ENABLED}, Max uploads before pause: {MAX_UPLOADS_BEFORE_PAUSE}")
+    print(f"[{FLOW_ID}] Starting audio transcription from file index: {start_index} out of {len(file_list)}.")
+    print(f"[{FLOW_ID}] Using {len(servers)} Whisper servers for dynamic processing.")
+    print(f"[{FLOW_ID}] Upload pause enabled: {UPLOAD_PAUSE_ENABLED}, Max uploads before pause: {MAX_UPLOADS_BEFORE_PAUSE}")
     
-        # Initialize progress tracking
-        if 'uploaded_count' not in progress:
-            progress['uploaded_count'] = 0
+    # Initialize progress tracking
+    if 'uploaded_count' not in progress:
+        progress['uploaded_count'] = 0
     
-        # If there was no HF state in the repo, upload a fresh initial state file
-        try:
-            if not current_state.get("file_states") and current_state.get("next_download_index", 0) == 0:
-                print(f"[{FLOW_ID}] No HF state detected; uploading initial state file to {HF_OUTPUT_DATASET_ID}...")
-                # Ensure structure
-                current_state.setdefault("file_states", {})
-                current_state.setdefault("next_download_index", 0)
-                if await upload_hf_state(current_state):
-                    print(f"[{FLOW_ID}] ✅ Initial HF state uploaded.")
-                else:
-                    print(f"[{FLOW_ID}] ❌ Failed to upload initial HF state.")
-        except Exception as e:
-            print(f"[{FLOW_ID}] Error while uploading initial HF state: {e}")
-        global_success = True
-        current_batch_index = start_list_index
-        batch_size = len(servers)  # Batch size = number of servers (20 files per batch)
+    # If there was no HF state in the repo, upload a fresh initial state file
+    try:
+        if not current_state.get("file_states") and current_state.get("next_download_index", 0) == 0:
+            print(f"[{FLOW_ID}] No HF state detected; uploading initial state file to {HF_OUTPUT_DATASET_ID}...")
+            # Ensure structure
+            current_state.setdefault("file_states", {})
+            current_state.setdefault("next_download_index", 0)
+            if await upload_hf_state(current_state):
+                print(f"[{FLOW_ID}] ✅ Initial HF state uploaded.")
+            else:
+                print(f"[{FLOW_ID}] ❌ Failed to upload initial HF state.")
+    except Exception as e:
+        print(f"[{FLOW_ID}] Error while uploading initial HF state: {e}")
     
-        try:
-            while current_batch_index < len(file_list):
-                # Process a batch dynamically
-                next_index, uploaded_count = await process_batch_dynamic(
-                    file_list, 
-                    current_batch_index, 
-                    batch_size, 
-                    current_state, 
-                    progress
-                )
-                
-                # Update progress
-                progress['last_processed_index'] = next_index
-                progress['uploaded_count'] = uploaded_count
-                save_progress(progress)
-                
-                # Update current batch index
-                current_batch_index = next_index
-                
-                # Log statistics
-                print(f"[{FLOW_ID}] Batch complete. Progress: {current_batch_index}/{len(file_list)}, Uploaded: {uploaded_count}")
-                
-                # Print server statistics
-                print(f"[{FLOW_ID}] Server Statistics:")
-                for i, server in enumerate(servers):
-                    print(f"  Server {i+1}: {server.total_processed} files, {server.total_time:.2f}s total, {server.fps:.2f} files/sec")
+    global_success = True
+    current_batch_index = start_list_index
+    batch_size = len(servers)  # Batch size = number of servers (20 files per batch)
+    
+    try:
+        while current_batch_index < len(file_list):
+            # Process a batch dynamically
+            next_index, uploaded_count = await process_batch_dynamic(
+                file_list, 
+                current_batch_index, 
+                batch_size, 
+                current_state, 
+                progress
+            )
+            
+            # Update progress
+            progress['last_processed_index'] = next_index
+            progress['uploaded_count'] = uploaded_count
+            save_progress(progress)
             
-            print(f"[{FLOW_ID}] All files processed successfully!")
-            return True
+            # Update current batch index
+            current_batch_index = next_index
             
-        except Exception as e:
-            print(f"[{FLOW_ID}] Critical error in process_dataset_task: {e}")
-            global_success = False
-            return global_success
+            # Log statistics
+            print(f"[{FLOW_ID}] Batch complete. Progress: {current_batch_index}/{len(file_list)}, Uploaded: {uploaded_count}")
+            
+            # Print server statistics
+            print(f"[{FLOW_ID}] Server Statistics:")
+            for i, server in enumerate(servers):
+                print(f"  Server {i+1}: {server.total_processed} files, {server.total_time:.2f}s total, {server.fps:.2f} files/sec")
+        
+        print(f"[{FLOW_ID}] All files processed successfully!")
+        return True
+        
+    except Exception as e:
+        print(f"[{FLOW_ID}] Critical error in process_dataset_task: {e}")
+        global_success = False
+        return global_success
 
 # --- FastAPI App and Endpoints ---