Update app.py

app.py

@@ -29,15 +29,11 @@ HF_STATE_FILE = "processing_state_transcriptions.json"
 LOCAL_STATE_FOLDER = Path(".state")
 LOCAL_STATE_FOLDER.mkdir(exist_ok=True)
 
-# Processing configuration
-MAX_UPLOADS_BEFORE_PAUSE = 120  # Pause uploading after 120 files
-UPLOAD_PAUSE_ENABLED = True
-
 # Directory within the HF dataset where the audio files are located
 AUDIO_FILE_PREFIX = "audio/"
 
 WHISPER_SERVERS = [
-    f"https://eliasishere-makeitfr-mineo-{i}.hf.space/transcribe" for i in range(1, 21)
+    f"https://makeitfr-mineo-{i}.hf.space/transcribe" for i in range(1, 21)
 ]
 
 # Temporary storage for audio files
@@ -61,11 +57,6 @@ class WhisperServer:
         """Files per second"""
         return self.total_processed / self.total_time if self.total_time > 0 else 0
     
-    def assign_file(self, file_index: int):
-        """Assign a file index to this server"""
-        self.is_processing = True
-        self.current_file_index = file_index
-    
     def release(self):
         """Release the server for a new file"""
         self.is_processing = False
@@ -73,42 +64,37 @@ class WhisperServer:
 
 # Global state for whisper servers
 servers = [WhisperServer(url) for url in WHISPER_SERVERS]
-server_lock = asyncio.Lock()  # Lock for thread-safe server state access
+server_lock = asyncio.Lock()
 
 # --- Progress and State Management Functions ---
 
 def load_progress() -> Dict:
-    """Loads the local processing progress from the JSON file."""
     default_structure = {
         "last_processed_index": 0,
-        "processed_files": {}, # {index: repo_path}
-        "file_list": [], # Full list of all zip files found in the dataset
+        "processed_files": {},
+        "file_list": [],
         "uploaded_count": 0
     }
     if PROGRESS_FILE.exists():
         try:
             with PROGRESS_FILE.open('r') as f:
                 data = json.load(f)
-                # Ensure all keys exist
                 for key, value in default_structure.items():
                     if key not in data:
                         data[key] = value
                 return data
         except json.JSONDecodeError:
-            print(f"[{FLOW_ID}] WARNING: Progress file is corrupted. Starting fresh.")
-    
+            print(f"[{FLOW_ID}] WARNING: Progress file is corrupted.")
     return default_structure
 
 def save_progress(progress_data: Dict):
-    """Saves the local processing progress to the JSON file."""
     try:
         with PROGRESS_FILE.open('w') as f:
             json.dump(progress_data, f, indent=4)
     except Exception as e:
-        print(f"[{FLOW_ID}] CRITICAL ERROR: Could not save progress to {PROGRESS_FILE}: {e}")
+        print(f"[{FLOW_ID}] CRITICAL ERROR: Could not save progress: {e}")
 
 def load_json_state(file_path: str, default_value: Dict[str, Any]) -> Dict[str, Any]:
-    """Load state from JSON file with migration logic for new structure."""
     if os.path.exists(file_path):
         try:
             with open(file_path, "r") as f:
@@ -118,17 +104,15 @@ def load_json_state(file_path: str, default_value: Dict[str, Any]) -> Dict[str,
                 if "next_download_index" not in data:
                     data["next_download_index"] = 0
                 return data
-        except json.JSONDecodeError:
-            print(f"[{FLOW_ID}] WARNING: Corrupted state file: {file_path}")
+        except Exception:
+            pass
     return default_value
 
 def save_json_state(file_path: str, data: Dict[str, Any]):
-    """Save state to JSON file"""
     with open(file_path, "w") as f:
         json.dump(data, f, indent=2)
 
 async def download_hf_state() -> Dict[str, Any]:
-    """Downloads the state file from Hugging Face or returns a default state."""
     local_path = LOCAL_STATE_FOLDER / HF_STATE_FILE
     default_state = {"next_download_index": 0, "file_states": {}}
     try:
@@ -142,11 +126,10 @@ async def download_hf_state() -> Dict[str, Any]:
         )
         return load_json_state(str(local_path), default_state)
     except Exception as e:
-        print(f"[{FLOW_ID}] Failed to download state file: {str(e)}. Using local/default.")
+        print(f"[{FLOW_ID}] Failed to download state file: {str(e)}")
         return load_json_state(str(local_path), default_state)
 
 async def upload_hf_state(state: Dict[str, Any]) -> bool:
-    """Uploads the state file to Hugging Face."""
     local_path = LOCAL_STATE_FOLDER / HF_STATE_FILE
     try:
         save_json_state(str(local_path), state)
@@ -162,8 +145,6 @@ async def upload_hf_state(state: Dict[str, Any]) -> bool:
         print(f"[{FLOW_ID}] Failed to upload state file: {str(e)}")
         return False
 
-# --- Hugging Face Utility Functions ---
-
 async def get_audio_file_list(progress_data: Dict) -> List[str]:
     if progress_data.get('file_list'):
         return progress_data['file_list']
@@ -201,7 +182,6 @@ async def transcribe_with_server(server: WhisperServer, wav_path: Path) -> Optio
     return None
 
 async def process_file_task(wav_file: str, state: Dict, progress: Dict):
-    # Find an available server
     server = None
     while server is None:
         async with server_lock:
@@ -214,11 +194,10 @@ async def process_file_task(wav_file: str, state: Dict, progress: Dict):
             await asyncio.sleep(1)
 
     try:
-        wav_filename = Path(wav_file).name
-        wav_path = TEMP_DIR / wav_filename
-        
-        # Download
-        hf_hub_download(
+        # FIX: Ensure we use the correct local path for the downloaded file
+        # hf_hub_download returns the absolute path to the downloaded file
+        print(f"[{FLOW_ID}] Downloading {wav_file}...")
+        downloaded_path_str = hf_hub_download(
             repo_id=HF_AUDIO_DATASET_ID,
             filename=wav_file,
             repo_type="dataset",
@@ -226,12 +205,15 @@ async def process_file_task(wav_file: str, state: Dict, progress: Dict):
             local_dir_use_symlinks=False,
             token=HF_TOKEN
         )
+        wav_path = Path(downloaded_path_str)
         
+        if not wav_path.exists():
+            raise FileNotFoundError(f"Downloaded file not found at {wav_path}")
+
         # Transcribe
         result = await transcribe_with_server(server, wav_path)
         
         if result:
-            # Upload transcription result to HF
             json_filename = Path(wav_file).with_suffix('.json').name
             json_content = json.dumps(result, indent=2, ensure_ascii=False).encode('utf-8')
             
@@ -251,6 +233,7 @@ async def process_file_task(wav_file: str, state: Dict, progress: Dict):
             state["file_states"][wav_file] = "failed_transcription"
             print(f"[{FLOW_ID}] ❌ Failed: {wav_file}")
             
+        # Cleanup
         if wav_path.exists():
             wav_path.unlink()
             
@@ -270,7 +253,6 @@ async def main_processing_loop():
             file_list = await get_audio_file_list(progress)
             
             if not file_list:
-                print(f"[{FLOW_ID}] File list empty, retrying in 60s...")
                 await asyncio.sleep(60)
                 continue
 
@@ -288,16 +270,13 @@ async def main_processing_loop():
             # 1. Handpick failed_transcription files
             failed_files = [f for f, s in state.get("file_states", {}).items() if s == "failed_transcription"]
             
-            # 2. Also check for new files based on next_download_index
+            # 2. Check for new files
             next_idx = state.get("next_download_index", 0)
-            # We take a larger chunk to allow for more skipping without re-fetching the list
-            new_files_chunk = file_list[next_idx:next_idx + 500] 
+            new_files_chunk = file_list[next_idx:next_idx + 1000] 
             
-            # Combine: Prioritize failed files, then add new ones
             files_to_check = failed_files + [f for f in new_files_chunk if f not in state["file_states"]]
             
             if not files_to_check:
-                print(f"[{FLOW_ID}] No files to process. Sleeping...")
                 await asyncio.sleep(60)
                 continue
 
@@ -310,71 +289,57 @@ async def main_processing_loop():
                 expected_json_name = Path(f).with_suffix('.json').name
                 
                 if expected_json_name in existing_json_files:
-                    # Mark locally but DO NOT upload yet
                     if state["file_states"].get(f) != "processed":
                         state["file_states"][f] = "processed"
                         state_changed_locally = True
-                        
-                        # Update next_download_index if it's a new file
                         if f in new_files_chunk:
                             current_idx = file_list.index(f)
                             if current_idx >= state.get("next_download_index", 0):
                                 state["next_download_index"] = current_idx + 1
                     continue
                 
-                # If we reach here, we found an UNPROCESSED file
+                # Found an UNPROCESSED file
                 print(f"[{FLOW_ID}] Found unprocessed file: {f}")
                 
-                # Before processing, if we have local changes (skips), upload the state once
+                # Upload skipped state before processing
                 if state_changed_locally:
-                    print(f"[{FLOW_ID}] Synchronizing skipped files to HF state before processing...")
+                    print(f"[{FLOW_ID}] Synchronizing skipped files to HF state...")
                     await upload_hf_state(state)
                     state_changed_locally = False
                 
                 files_to_process.append(f)
-                
-                # Once we find an unprocessed file, we stop the skip-scan and start processing
-                # This ensures we process files as soon as we find them
                 break 
 
-            # If we scanned everything and only found skips, upload the state once at the end
             if state_changed_locally and not files_to_process:
-                print(f"[{FLOW_ID}] Uploading final batch of skips to HF state...")
+                print(f"[{FLOW_ID}] Uploading final batch of skips...")
                 await upload_hf_state(state)
             
             if not files_to_process:
-                # If we only found skips, the loop will restart and check the next chunk
                 continue
 
-            print(f"[{FLOW_ID}] Processing batch of {len(files_to_process)} unprocessed files...")
-            
-            # Process the found unprocessed file(s)
-            # (In this logic, it's usually just 1 file at a time to ensure frequent state updates)
+            # Process the unprocessed file
             batch_size = len(servers)
             for i in range(0, len(files_to_process), batch_size):
                 batch = files_to_process[i:i + batch_size]
                 tasks = [process_file_task(f, state, progress) for f in batch]
                 await asyncio.gather(*tasks)
                 
-                # Update next_download_index
                 for f in batch:
                     if f in file_list:
                         current_idx = file_list.index(f)
                         if current_idx >= state.get("next_download_index", 0):
                             state["next_download_index"] = current_idx + 1
                 
-                # Save and upload state after processing the unprocessed file
                 await upload_hf_state(state)
                 save_progress(progress)
                 
-            await asyncio.sleep(2) # Short sleep before looking for the next unprocessed file
+            await asyncio.sleep(2)
             
         except Exception as e:
             print(f"[{FLOW_ID}] Error in main loop: {e}")
             await asyncio.sleep(60)
 
 # --- FastAPI App ---
-
 app = FastAPI(title=f"Flow Server {FLOW_ID} API")
 
 @app.on_event("startup")