Update app.py

app.py

@@ -37,7 +37,7 @@ UPLOAD_PAUSE_ENABLED = True
 AUDIO_FILE_PREFIX = "audio/"
 
 WHISPER_SERVERS = [
-    f"https://makeitfr-mineo-{i}.hf.space/transcribe" for i in range(1, 21)
+    f"https://eliasishere-makeitfr-mineo-{i}.hf.space/transcribe" for i in range(1, 21)
 ]
 
 # Temporary storage for audio files
@@ -231,6 +231,19 @@ async def process_file_task(wav_file: str, state: Dict, progress: Dict):
         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')
+            
+            api = HfApi(token=HF_TOKEN)
+            api.upload_file(
+                path_or_fileobj=io.BytesIO(json_content),
+                path_in_repo=json_filename,
+                repo_id=HF_OUTPUT_DATASET_ID,
+                repo_type="dataset",
+                commit_message=f"[{FLOW_ID}] Transcription for {wav_file}"
+            )
+            
             state["file_states"][wav_file] = "processed"
             progress["uploaded_count"] = progress.get("uploaded_count", 0) + 1
             print(f"[{FLOW_ID}] ✅ Success: {wav_file}")
@@ -261,41 +274,101 @@ async def main_processing_loop():
                 await asyncio.sleep(60)
                 continue
 
+            # Check HF_OUTPUT_DATASET_ID for existing JSON outputs
+            print(f"[{FLOW_ID}] Checking {HF_OUTPUT_DATASET_ID} for existing JSON outputs...")
+            try:
+                api = HfApi(token=HF_TOKEN)
+                existing_files = api.list_repo_files(repo_id=HF_OUTPUT_DATASET_ID, repo_type="dataset")
+                existing_json_files = {f for f in existing_files if f.endswith('.json')}
+                print(f"[{FLOW_ID}] Found {len(existing_json_files)} existing JSON files.")
+            except Exception as e:
+                print(f"[{FLOW_ID}] Warning: Could not fetch existing files: {e}")
+                existing_json_files = set()
+
             # 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
             next_idx = state.get("next_download_index", 0)
-            new_files = file_list[next_idx:next_idx + 100] # Take a chunk of new files
+            # 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] 
             
             # Combine: Prioritize failed files, then add new ones
-            files_to_process = failed_files + [f for f in new_files if f not in state["file_states"]]
+            files_to_check = failed_files + [f for f in new_files_chunk if f not in state["file_states"]]
             
-            if not files_to_process:
+            if not files_to_check:
                 print(f"[{FLOW_ID}] No files to process. Sleeping...")
                 await asyncio.sleep(60)
                 continue
 
-            print(f"[{FLOW_ID}] Processing {len(files_to_process)} files ({len(failed_files)} failed, {len(files_to_process)-len(failed_files)} new)...")
+            files_to_process = []
+            state_changed_locally = False
+            
+            print(f"[{FLOW_ID}] Scanning {len(files_to_check)} files for existing results...")
+            
+            for f in files_to_check:
+                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
+                print(f"[{FLOW_ID}] Found unprocessed file: {f}")
+                
+                # Before processing, if we have local changes (skips), upload the state once
+                if state_changed_locally:
+                    print(f"[{FLOW_ID}] Synchronizing skipped files to HF state before processing...")
+                    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...")
+                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 in batches of server count
+            # Process the found unprocessed file(s)
+            # (In this logic, it's usually just 1 file at a time to ensure frequent state updates)
             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 if we processed new files
-                processed_new = [f for f in batch if f in new_files]
-                if processed_new:
-                    last_new_file = processed_new[-1]
-                    state["next_download_index"] = file_list.index(last_new_file) + 1
+                # 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 each batch
+                # Save and upload state after processing the unprocessed file
                 await upload_hf_state(state)
                 save_progress(progress)
                 
-            await asyncio.sleep(10)
+            await asyncio.sleep(2) # Short sleep before looking for the next unprocessed file
+            
         except Exception as e:
             print(f"[{FLOW_ID}] Error in main loop: {e}")
             await asyncio.sleep(60)