Update app.py

app.py

@@ -1,108 +1,100 @@
 import os
 import json
-import requests
-import subprocess
-import shutil
 import time
-import sys
-import threading
-from typing import Dict, List, Optional, Any
-from huggingface_hub import HfApi, hf_hub_url
-from fastapi import FastAPI, HTTPException
-from fastapi.responses import JSONResponse
-import uvicorn
-
-# Fix Unicode encoding for Windows
-if sys.platform == 'win32':
-    import io
-    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
-
-# Initialize FastAPI app
-app = FastAPI(title="Audio Transcriber", description="Audio transcription and upload service")
-
-# ==== CONFIGURATION ====
-HF_TOKEN = os.environ.get("HF_TOKEN", "")
-SOURCE_REPO_ID = "Samfredoly/BG_Vid"  # Fetch audio files from here
-TARGET_REPO_ID = "samfred2/A_Text"  # Upload transcriptions here
-REFERENCE_REPO_ID = "Fred808/BG3"  # Reference repo to match audio filenames
-
-# Path Configuration
-DOWNLOAD_FOLDER = "downloads_audio"
-TRANSCRIPTIONS_FOLDER = "transcriptions"
-LOCAL_STATE_FOLDER = ".state_audio"
-
-os.makedirs(DOWNLOAD_FOLDER, exist_ok=True)
-os.makedirs(TRANSCRIPTIONS_FOLDER, exist_ok=True)
-os.makedirs(LOCAL_STATE_FOLDER, exist_ok=True)
-
-# State Files
-FAILED_FILES_LOG = "failed_audio_files.log"
-HF_STATE_FILE = "processing_audio_state.json"
-
-# Processing Parameters
-PROCESSING_DELAY = 2
-MAX_RETRIES = 3
-MIN_FREE_SPACE_GB = 1
-WHISPER_MODEL = "small"  # Whisper model size
-
-# Initialize HF API
-hf_api = HfApi(token=HF_TOKEN)
-
-# Global State
-processing_status = {
-    "is_running": False,
-    "current_file": None,
-    "total_files": 0,
-    "processed_files": 0,
-    "failed_files": 0,
-    "transcribed_files": 0,
-    "last_update": None,
-    "logs": []
-}
-
-def log_message(message: str, level: str = "INFO"):
-    """Log messages with timestamp"""
-    timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
-    log_entry = f"[{timestamp}] {level}: {message}"
-    print(log_entry)
-    processing_status["logs"].append(log_entry)
-    processing_status["last_update"] = timestamp
-    if len(processing_status["logs"]) > 100:
-        processing_status["logs"] = processing_status["logs"][-100:]
-
-def log_failed_file(filename: str, error: str):
-    """Log failed files to persistent file"""
-    with open(FAILED_FILES_LOG, "a") as f:
-        f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - {filename}: {error}\n")
-
-def get_disk_usage(path: str) -> Dict[str, float]:
-    """Get disk usage statistics in GB"""
-    statvfs = os.statvfs(path)
-    total = statvfs.f_frsize * statvfs.f_blocks / (1024**3)
-    free = statvfs.f_frsize * statvfs.f_bavail / (1024**3)
-    used = total - free
-    return {"total": total, "free": free, "used": used}
-
-def check_disk_space(path: str = ".") -> bool:
-    """Check if there's enough disk space"""
-    disk_info = get_disk_usage(path)
-    if disk_info["free"] < MIN_FREE_SPACE_GB:
-        log_message(f'⚠️ Low disk space: {disk_info["free"]:.2f}GB free, {disk_info["used"]:.2f}GB used')
-        return False
-    return True
-
-def cleanup_temp_files():
-    """Clean up temporary files to free space"""
-    log_message("🧹 Cleaning up temporary files...", "INFO")
+import asyncio
+import aiohttp
+import zipfile
+import shutil
+from typing import Dict, List, Set, Optional, Tuple, Any
+from urllib.parse import quote
+from datetime import datetime
+from pathlib import Path
+import io
+
+from fastapi import FastAPI, BackgroundTasks, HTTPException, status
+from pydantic import BaseModel, Field
+from huggingface_hub import HfApi, hf_hub_download
+
+# --- Configuration ---
+AUTO_START_INDEX = 0  # Hardcoded default start index if no progress is found
+FLOW_ID = os.getenv("FLOW_ID", "flow_default")
+FLOW_PORT = int(os.getenv("FLOW_PORT", 8001))
+HF_TOKEN = os.getenv("HF_TOKEN", "")
+HF_AUDIO_DATASET_ID = os.getenv("HF_AUDIO_DATASET_ID", "Samfredoly/BG_Vid")  # Source dataset for audio files
+HF_OUTPUT_DATASET_ID = os.getenv("HF_OUTPUT_DATASET_ID", "samfred2/AT2")  # Target dataset for transcriptions
+
+# Progress and State Tracking
+PROGRESS_FILE = Path("processing_progress.json")
+HF_STATE_FILE = "processing_state_transcriptions.json"  # State file in output dataset
+LOCAL_STATE_FOLDER = Path(".state")  # Local folder for state file
+LOCAL_STATE_FOLDER.mkdir(exist_ok=True)
+
+# Directory within the HF dataset where audio files are located
+AUDIO_FILE_PREFIX = "audio/"
+
+# Reference dataset for filename mapping
+REFERENCE_REPO_ID = os.getenv("REFERENCE_REPO_ID", "Fred808/BG3")  # For matching audio to reference files
+
+# Whisper server endpoints
+WHISPER_SERVERS = [
+    "https://fred1012-switch3.hf.space/transcribe",
+]
+
+MODEL_TYPE = "whisper-small"
+ZIP_UPLOAD_THRESHOLD = 100  # Upload and zip after this many transcriptions
+
+# Temporary storage for audio files
+TEMP_DIR = Path(f"temp_audio_{FLOW_ID}")
+TEMP_DIR.mkdir(exist_ok=True)
+
+# Temporary storage for transcription results
+RESULTS_DIR = Path(f"transcription_results_{FLOW_ID}")
+RESULTS_DIR.mkdir(exist_ok=True)
+
+# --- Models ---
+class WhisperServer:
+    def __init__(self, url):
+        self.url = url
+        self.busy = False
+        self.total_processed = 0
+        self.total_time = 0
+        self.model = MODEL_TYPE
+
+    @property
+    def fps(self):
+        return self.total_processed / self.total_time if self.total_time > 0 else 0
+
+# Global state for whisper servers
+servers = [WhisperServer(url) for url in WHISPER_SERVERS]
+server_index = 0
+
+# --- Progress and State Management Functions ---
+
+def load_progress() -> Dict:
+    """Loads the local processing progress from the JSON file."""
+    if PROGRESS_FILE.exists():
+        try:
+            with PROGRESS_FILE.open('r') as f:
+                return json.load(f)
+        except json.JSONDecodeError:
+            print(f"[{FLOW_ID}] WARNING: Progress file is corrupted. Starting fresh.")
     
-    current_file = processing_status.get("current_file")
-    for file in os.listdir(DOWNLOAD_FOLDER):
-        if file != current_file and file.endswith((".wav", ".mp3")):
-            try:
-                os.remove(os.path.join(DOWNLOAD_FOLDER, file))
-                log_message(f"🗑️ Removed old download: {file}", "INFO")
-            except:
-                pass
+    # Default structure
+    return {
+        "last_processed_index": 0,
+        "processed_files": {},  # {index: audio_file_path}
+        "file_list": [],  # Full list of all audio files found in the dataset
+        "transcription_count": 0,  # Count of transcriptions saved
+        "reference_map": {}  # Mapping from audio filename to reference filename
+    }
+
+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}")
 
 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."""
@@ -111,16 +103,20 @@ def load_json_state(file_path: str, default_value: Dict[str, Any]) -> Dict[str,
             with open(file_path, "r") as f:
                 data = json.load(f)
                 
+                # Migration Logic
                 if "file_states" not in data or not isinstance(data["file_states"], dict):
-                    log_message("ℹ️ Initializing 'file_states' dictionary.", "INFO")
+                    print(f"[{FLOW_ID}] Initializing 'file_states' dictionary.")
                     data["file_states"] = {}
                     
                 if "next_download_index" not in data:
                     data["next_download_index"] = 0
                     
+                if "transcription_count" not in data:
+                    data["transcription_count"] = 0
+                    
                 return data
         except json.JSONDecodeError:
-            log_message(f"⚠️ Corrupted state file: {file_path}", "WARNING")
+            print(f"[{FLOW_ID}] WARNING: Corrupted state file: {file_path}")
     return default_value
 
 def save_json_state(file_path: str, data: Dict[str, Any]):
@@ -128,576 +124,534 @@ def save_json_state(file_path: str, data: Dict[str, Any]):
     with open(file_path, "w") as f:
         json.dump(data, f, indent=2)
 
-def download_hf_state(repo_id: str, filename: str) -> Dict[str, Any]:
+async def download_hf_state() -> Dict[str, Any]:
     """Downloads the state file from Hugging Face or returns a default state."""
-    local_path = os.path.join(LOCAL_STATE_FOLDER, filename)
-    default_state = {"next_download_index": 130, "file_states": {}}
+    local_path = LOCAL_STATE_FOLDER / HF_STATE_FILE
+    default_state = {"next_download_index": 0, "file_states": {}, "transcription_count": 0}
     
     try:
-        files = hf_api.list_repo_files(repo_id=repo_id, repo_type="dataset")
-        if filename not in files:
-            log_message(f"ℹ️ State file {filename} not found in {repo_id}. Starting from default state.", "INFO")
+        # Check if the file exists in the output repo
+        files = HfApi(token=HF_TOKEN).list_repo_files(
+            repo_id=HF_OUTPUT_DATASET_ID,
+            repo_type="dataset"
+        )
+        
+        if HF_STATE_FILE not in files:
+            print(f"[{FLOW_ID}] State file not found in {HF_OUTPUT_DATASET_ID}. Starting fresh.")
             return default_state
 
-        from huggingface_hub import hf_hub_download
+        # Download the file
         hf_hub_download(
-            repo_id=repo_id,
-            filename=filename,
+            repo_id=HF_OUTPUT_DATASET_ID,
+            filename=HF_STATE_FILE,
             repo_type="dataset",
             local_dir=LOCAL_STATE_FOLDER,
-            local_dir_use_symlinks=False
+            local_dir_use_symlinks=False,
+            token=HF_TOKEN
         )
         
-        log_message(f"✅ Successfully downloaded state file from {repo_id}.", "INFO")
-        return load_json_state(local_path, default_state)
+        print(f"[{FLOW_ID}] Successfully downloaded state file.")
+        return load_json_state(str(local_path), default_state)
         
     except Exception as e:
-        log_message(f"⚠️ Failed to download state file from Hugging Face: {str(e)}. Starting from default state.", "WARNING")
+        print(f"[{FLOW_ID}] Failed to download state file: {str(e)}. Starting fresh.")
         return default_state
 
-def upload_hf_state(repo_id: str, filename: str, state: Dict[str, Any]) -> bool:
+async def upload_hf_state(state: Dict[str, Any]) -> bool:
     """Uploads the state file to Hugging Face."""
-    local_path = os.path.join(LOCAL_STATE_FOLDER, filename)
+    local_path = LOCAL_STATE_FOLDER / HF_STATE_FILE
     
     try:
-        save_json_state(local_path, state)
-        
-        hf_api.upload_file(
-            path_or_fileobj=local_path,
-            path_in_repo=filename,
-            repo_id=repo_id,
+        # Save state locally first
+        save_json_state(str(local_path), state)
+        
+        # Upload to output dataset
+        HfApi(token=HF_TOKEN).upload_file(
+            path_or_fileobj=str(local_path),
+            path_in_repo=HF_STATE_FILE,
+            repo_id=HF_OUTPUT_DATASET_ID,
             repo_type="dataset",
-            commit_message=f"Update audio processing state: next_index={state['next_download_index']}"
+            commit_message=f"Update transcription processing state: next_index={state['next_download_index']}, count={state.get('transcription_count', 0)}"
         )
-        log_message(f"✅ Successfully uploaded updated state file to {repo_id}", "INFO")
+        print(f"[{FLOW_ID}] Successfully uploaded state file.")
         return True
     except Exception as e:
-        log_message(f"❌ Failed to upload state file to Hugging Face: {str(e)}", "ERROR")
+        print(f"[{FLOW_ID}] Failed to upload state file: {str(e)}")
         return False
 
-def lock_file_for_processing(wav_filename: str, state: Dict[str, Any]) -> bool:
+async def lock_file_for_processing(audio_filename: str, state: Dict[str, Any]) -> bool:
     """Marks a file as 'processing' in the state file and uploads the lock."""
-    log_message(f"🔒 Attempting to lock file: {wav_filename} (Marking as 'processing')", "INFO")
+    print(f"[{FLOW_ID}] 🔒 Attempting to lock file: {audio_filename}")
     
-    state["file_states"][wav_filename] = "processing"
+    # Update state locally
+    state["file_states"][audio_filename] = "processing"
     
-    if upload_hf_state(TARGET_REPO_ID, HF_STATE_FILE, state):
-        log_message(f"✅ Successfully locked file: {wav_filename}", "INFO")
+    # Upload the updated state file immediately to establish the lock
+    if await upload_hf_state(state):
+        print(f"[{FLOW_ID}] ✅ Successfully locked file: {audio_filename}")
         return True
     else:
-        log_message(f"❌ Failed to upload lock for file: {wav_filename}. Aborting processing.", "ERROR")
-        if wav_filename in state["file_states"]:
-            del state["file_states"][wav_filename]
+        print(f"[{FLOW_ID}] ❌ Failed to lock file: {audio_filename}")
+        # Revert local state
+        if audio_filename in state["file_states"]:
+            del state["file_states"][audio_filename]
         return False
 
-def unlock_file_as_processed(wav_filename: str, state: Dict[str, Any], next_index: int) -> bool:
+async def unlock_file_as_processed(audio_filename: str, state: Dict[str, Any], next_index: int) -> bool:
     """Marks a file as 'processed', updates the index, and uploads the state."""
-    log_message(f"🔓 Attempting to unlock file: {wav_filename} (Marking as 'processed')", "INFO")
+    print(f"[{FLOW_ID}] 🔓 Marking file as processed: {audio_filename}")
     
-    state["file_states"][wav_filename] = "processed"
+    # Update state locally
+    state["file_states"][audio_filename] = "processed"
     state["next_download_index"] = next_index
     
-    if upload_hf_state(TARGET_REPO_ID, HF_STATE_FILE, state):
-        log_message(f"✅ Successfully unlocked and marked as processed: {wav_filename}", "INFO")
+    # Upload the updated state
+    if await upload_hf_state(state):
+        print(f"[{FLOW_ID}] ✅ Successfully marked as processed: {audio_filename}")
         return True
     else:
-        log_message(f"❌ Failed to upload final state for file: {wav_filename}.", "ERROR")
+        print(f"[{FLOW_ID}] ❌ Failed to update state for: {audio_filename}")
         return False
 
-def download_with_retry(url: str, dest_path: str, max_retries: int = 3) -> bool:
-    """Download file with retry logic and disk space checking"""
-    if not check_disk_space():
-        cleanup_temp_files()
-        if not check_disk_space():
-            log_message("❌ Insufficient disk space even after cleanup", "ERROR")
-            return False
-    
-    try:
-        os.makedirs(os.path.dirname(dest_path), exist_ok=True)
-    except Exception as e:
-        log_message(f"❌ Failed to create directory for download path {os.path.dirname(dest_path)}: {str(e)}", "ERROR")
-        return False
-        
-    headers = {"Authorization": f"Bearer {HF_TOKEN}"}
-    for attempt in range(max_retries):
-        try:
-            with requests.get(url, headers=headers, stream=True) as r:
-                r.raise_for_status()
-                
-                with open(dest_path, "wb") as f:
-                    for chunk in r.iter_content(chunk_size=8192):
-                        if chunk:
-                            f.write(chunk)
-                
-                log_message(f"✅ Download successful: {dest_path}", "INFO")
-                return True
-            
-        except requests.exceptions.RequestException as e:
-            log_message(f"❌ Download attempt {attempt + 1} failed for {url}: {str(e)}", "WARNING")
-            time.sleep(PROCESSING_DELAY)
-        except Exception as e:
-            log_message(f"❌ An unexpected error occurred during download: {str(e)}", "ERROR")
-            return False
-            
-    log_message(f"❌ Failed to download {url} after {max_retries} attempts.", "ERROR")
-    return False
+# --- Hugging Face Utility Functions ---
 
-def fetch_reference_files(repo_id: str) -> Dict[str, str]:
-    """Fetch all files from Fred808/BG3 repo to match with audio filenames."""
-    log_message(f"📋 Fetching file list from {repo_id}...", "INFO")
+async def get_reference_map(reference_repo_id: str) -> Dict[str, str]:
+    """
+    Fetches the reference file list from the Hugging Face repo and creates a map
+    from audio filename (without extension) to reference filename.
+    """
+    print(f"[{FLOW_ID}] Fetching reference file list from {reference_repo_id}...")
     
     try:
-        files_list = hf_api.list_repo_files(repo_id=repo_id, repo_type="dataset")
-        
-        # Include all file types (zip, rar, wav, mp3, etc.)
-        all_files = [f for f in files_list]
+        api = HfApi(token=HF_TOKEN)
+        repo_files = api.list_repo_files(repo_id=reference_repo_id, repo_type="dataset")
         
-        # Create a mapping of base filename (without extension) to full path
-        filename_map = {}
-        for file_path in all_files:
-            base_name = os.path.splitext(os.path.basename(file_path))[0]
-            filename_map[base_name] = file_path
+        reference_map = {}
+        for file in repo_files:
+            base_name, ext = os.path.splitext(file)
+            if ext.lower() in ['.txt', '.json']:  # Consider text/json files as reference
+                reference_map[base_name] = file
         
-        log_message(f"✅ Found {len(filename_map)} files in reference repo", "INFO")
-        return filename_map
+        print(f"[{FLOW_ID}] ✅ Successfully created reference map with {len(reference_map)} entries.")
+        return reference_map
         
     except Exception as e:
-        log_message(f"❌ Failed to fetch reference files: {str(e)}", "ERROR")
+        print(f"[{FLOW_ID}] ⚠️ Failed to fetch reference map from Hugging Face: {e}")
         return {}
 
-def find_matching_filename(transcribed_filename: str, reference_map: Dict[str, str]) -> Optional[str]:
-    """Find matching filename in reference map from Fred808/BG3."""
-    base_name = os.path.splitext(transcribed_filename)[0]
-    
-    # Exact match first
-    if base_name in reference_map:
-        full_path = reference_map[base_name]
-        print(f"\n✅ EXACT MATCH FOUND:")
-        print(f"   Audio: {transcribed_filename}")
-        print(f"   File:  {full_path}")
-        log_message(f"✅ Found exact match: {transcribed_filename} -> {full_path}", "INFO")
-        return full_path
-    
-    # Partial/fuzzy match (check if reference contains transcribed as substring)
-    matches = []
-    for ref_base, ref_full_path in reference_map.items():
-        if base_name.lower() in ref_base.lower() or ref_base.lower() in base_name.lower():
-            matches.append((ref_base, ref_full_path))
-    
-    # Return first partial match if found
-    if matches:
-        ref_base, ref_full_path = matches[0]
-        print(f"\n✅ PARTIAL MATCH FOUND:")
-        print(f"   Audio: {transcribed_filename}")
-        print(f"   File:  {ref_full_path}")
-        log_message(f"✅ Found partial match: {transcribed_filename} -> {ref_full_path}", "INFO")
-        return ref_full_path
-    
-    print(f"\n⚠️ NO EXACT/PARTIAL MATCH FOUND (will still process):")
-    print(f"   Audio: {transcribed_filename}")
-    log_message(f"⚠️ No matching filename found for: {transcribed_filename}. Will use original filename.", "WARNING")
-    return None
+def find_matching_filename(audio_filename: str, reference_map: Dict[str, str]) -> Optional[str]:
+    """
+    Finds the matching reference filename for a given audio filename.
+    Returns the reference filename if found, otherwise None.
+    """
+    base_name, _ = os.path.splitext(audio_filename)
+    return reference_map.get(base_name)
 
-def transcribe_audio(wav_path: str) -> Optional[Dict[str, Any]]:
-    """Transcribe audio file using Whisper from Transformers."""
-    log_message(f"🎤 Transcribing audio file: {wav_path}", "INFO")
-    
+async def get_audio_file_list(progress_data: Dict) -> List[str]:
+    """
+    Fetches the list of all audio files from the dataset, or uses the cached list.
+    Updates the progress_data with the file list if a new list is fetched.
+    """
+    if progress_data['file_list']:
+        print(f"[{FLOW_ID}] Using cached file list with {len(progress_data['file_list'])} files.")
+        return progress_data['file_list']
+
+    print(f"[{FLOW_ID}] Fetching full list of audio files from {HF_AUDIO_DATASET_ID}...")
     try:
-        from transformers import pipeline
-        import librosa
-        
-        # Load audio with librosa
-        log_message(f"Loading audio file: {wav_path}", "INFO")
-        audio, sr = librosa.load(wav_path, sr=16000)
-        
-        # Initialize Whisper pipeline
-        log_message(f"Loading Whisper {WHISPER_MODEL} model from Transformers...", "INFO")
-        pipe = pipeline(
-            "automatic-speech-recognition",
-            model=f"openai/whisper-{WHISPER_MODEL}",
-            device=0 if __import__('torch').cuda.is_available() else -1  # GPU if available, else CPU
+        api = HfApi(token=HF_TOKEN)
+        repo_files = api.list_repo_files(
+            repo_id=HF_AUDIO_DATASET_ID,
+            repo_type="dataset"
         )
         
-        # Transcribe
-        log_message("Transcribing audio...", "INFO")
-        result = pipe(audio)
+        # Filter for audio files in the specified directory and sort them alphabetically for consistent indexing
+        audio_extensions = ['.mp3', '.wav', '.m4a', '.flac', '.ogg', '.aac']
+        audio_files = sorted([
+            f for f in repo_files 
+            if f.startswith(AUDIO_FILE_PREFIX) and any(f.lower().endswith(ext) for ext in audio_extensions)
+        ])
+        
+        if not audio_files:
+            raise FileNotFoundError(f"No audio files found in '{AUDIO_FILE_PREFIX}' directory of dataset '{HF_AUDIO_DATASET_ID}'.")
         
-        # Format result to match openai-whisper format
-        formatted_result = {
-            "text": result["text"],
-            "segments": [{"text": result["text"]}]
-        }
+        print(f"[{FLOW_ID}] Found {len(audio_files)} audio files.")
         
-        log_message(f"✅ Successfully transcribed: {wav_path}", "INFO")
-        return formatted_result
+        # Update and save the progress data
+        progress_data['file_list'] = audio_files
+        save_progress(progress_data)
+        
+        return audio_files
         
-    except ImportError as e:
-        missing_lib = str(e)
-        log_message(f"❌ Missing library. Install with: pip install transformers librosa torch torchaudio", "ERROR")
-        log_message(f"   Error: {missing_lib}", "ERROR")
-        return None
     except Exception as e:
-        log_message(f"❌ Failed to transcribe {wav_path}: {str(e)}", "ERROR")
-        return None
+        print(f"[{FLOW_ID}] Error fetching file list from Hugging Face: {e}")
+        return []
 
-def process_audio_file(wav_path: str, reference_map: Dict[str, str], matched_filename: str) -> bool:
-    """
-    Main processing logic for a single audio file:
-    1. Transcribe using Whisper
-    2. Save transcription as JSON
-    3. Upload to HF dataset
-    4. Clean up local files
-    """
-    wav_filename = os.path.basename(wav_path)
+async def download_audio_file(file_index: int, repo_file_full_path: str) -> Optional[Path]:
+    """Downloads the audio file for the given index."""
     
-    # 1. Transcribe audio
-    transcription = transcribe_audio(wav_path)
-    if transcription is None:
-        log_failed_file(wav_filename, "Transcription failed")
-        return False
+    audio_filename = Path(repo_file_full_path).name 
     
-    # 2. Save transcription as JSON
-    json_filename = os.path.splitext(matched_filename)[0] + "_transcription.json"
-    json_output_path = os.path.join(TRANSCRIPTIONS_FOLDER, json_filename)
+    print(f"[{FLOW_ID}] Processing audio file #{file_index}: {repo_file_full_path}")
     
     try:
-        os.makedirs(os.path.dirname(json_output_path), exist_ok=True)
+        # Use hf_hub_download to get the file path
+        audio_path = hf_hub_download(
+            repo_id=HF_AUDIO_DATASET_ID,
+            filename=repo_file_full_path,
+            repo_type="dataset",
+            token=HF_TOKEN,
+        )
         
-        with open(json_output_path, "w", encoding="utf-8") as f:
-            json.dump(transcription, f, indent=2, ensure_ascii=False)
+        print(f"[{FLOW_ID}] Downloaded audio to {audio_path}.")
         
-        log_message(f"✅ Saved transcription: {json_output_path}", "INFO")
+        # Copy to temp directory
+        temp_path = TEMP_DIR / audio_filename
+        shutil.copy2(audio_path, temp_path)
+        
+        return temp_path
         
     except Exception as e:
-        log_message(f"❌ Failed to save transcription JSON: {str(e)}", "ERROR")
-        log_failed_file(wav_filename, f"Failed to save JSON: {str(e)}")
-        return False
+        print(f"[{FLOW_ID}] Error downloading audio file {repo_file_full_path}: {e}")
+        return None
+
+async def upload_transcription_to_hf(audio_filename: str, transcription_data: Dict, reference_filename: Optional[str] = None) -> bool:
+    """
+    Uploads the transcription JSON file to the output dataset.
+    If reference_filename is provided, uses it as the base for the output filename.
+    Otherwise, uses the audio filename.
+    """
+    # Use reference filename if provided, otherwise use audio filename
+    output_base = Path(reference_filename).stem if reference_filename else Path(audio_filename).stem
+    json_filename = f"{output_base}.json"
     
-    # 3. Upload to HF dataset
     try:
-        path_in_repo = f"transcriptions/{json_filename}"
-        commit_message = f"Add transcription for: {matched_filename}"
+        print(f"[{FLOW_ID}] Uploading transcription for {audio_filename} as {json_filename} to {HF_OUTPUT_DATASET_ID}...")
+        
+        # Create JSON content in memory
+        json_content = json.dumps(transcription_data, indent=2, ensure_ascii=False).encode('utf-8')
         
-        hf_api.upload_file(
-            path_or_fileobj=json_output_path,
-            path_in_repo=path_in_repo,
-            repo_id=TARGET_REPO_ID,
+        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=commit_message
+            commit_message=f"[{FLOW_ID}] Transcription for {audio_filename}"
         )
-        log_message(f"✅ Successfully uploaded transcription: {json_filename}", "INFO")
-        processing_status["transcribed_files"] += 1
+        
+        print(f"[{FLOW_ID}] Successfully uploaded transcription for {audio_filename}.")
+        return True
         
     except Exception as e:
-        log_message(f"❌ Failed to upload transcription to HF: {str(e)}", "ERROR")
-        log_failed_file(wav_filename, f"Failed to upload: {str(e)}")
+        print(f"[{FLOW_ID}] Error uploading transcription for {audio_filename}: {e}")
         return False
-    
-    # 4. Clean up local files
-    try:
-        os.remove(json_output_path)
-        log_message(f"🗑️ Cleaned up local transcription file: {json_output_path}", "INFO")
-    except:
-        pass
-    
-    return True
 
-def get_next_file_to_process(repo_id: str, state: Dict[str, Any]) -> Optional[Dict[str, Any]]:
-    """
-    Finds the next audio file to process from the source repo in reverse order (oldest to newest).
-    Returns: { 'filename': str, 'url': str, 'index': int } or None
-    """
-    log_message(f"🔍 Searching for next audio file to process in {repo_id}", "INFO")
+async def zip_and_upload_transcriptions(transcription_files: List[Path]) -> bool:
+    """Zips transcription JSON files and uploads to dataset."""
+    if not transcription_files:
+        print(f"[{FLOW_ID}] No transcription files to zip.")
+        return False
     
     try:
-        files_list = hf_api.list_repo_files(repo_id=repo_id, repo_type="dataset")
-        
-        # Filter for audio files and sort in reverse order (descending)
-        audio_files = sorted([f for f in files_list if f.endswith(('.wav', '.mp3'))], reverse=True)
-        
-        if not audio_files:
-            log_message("ℹ️ No audio files found in the source repository.", "INFO")
-            return None
-            
-        processing_status["total_files"] = len(audio_files)
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        zip_filename = f"transcriptions_{timestamp}.zip"
+        zip_path = RESULTS_DIR / zip_filename
         
-        start_index = state.get("next_download_index", 0)
+        print(f"[{FLOW_ID}] Creating zip file: {zip_path} with {len(transcription_files)} files...")
         
-        for index in range(start_index, len(audio_files)):
-            filename = audio_files[index]
-            file_state = state["file_states"].get(filename)
-            
-            if file_state is None or file_state == "failed":
-                url = hf_hub_url(repo_id=repo_id, filename=filename, repo_type="dataset", subfolder=None)
-                
-                log_message(f"✅ Found next audio file: {filename} at index {index}", "INFO")
-                return {
-                    'filename': filename,
-                    'url': url,
-                    'index': index
-                }
-                
-            elif file_state == "processing":
-                log_message(f"⚠️ File {filename} is currently marked as 'processing'. Skipping for now.", "WARNING")
-                
-            elif file_state == "processed":
-                log_message(f"ℹ️ File {filename} already processed. Skipping.", "INFO")
-                
-        log_message("ℹ️ All files up to the current index have been processed or skipped.", "INFO")
+        with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
+            for file_path in transcription_files:
+                if file_path.exists():
+                    zipf.write(file_path, arcname=file_path.name)
         
-        if start_index >= len(audio_files):
-            log_message("ℹ️ Reached end of file list. Resetting index to 0 for next loop.", "INFO")
-            state["next_download_index"] = 0
-            upload_hf_state(TARGET_REPO_ID, HF_STATE_FILE, state)
-            
-        return None
+        print(f"[{FLOW_ID}] Uploading zip file to {HF_OUTPUT_DATASET_ID}...")
         
-    except Exception as e:
-        log_message(f"❌ Failed to list files from Hugging Face: {str(e)}", "ERROR")
-        return None
-
-def main_processing_loop():
-    """The main loop that orchestrates the download, transcription, and upload cycle."""
-    
-    if processing_status["is_running"]:
-        log_message("⚠️ Processing loop is already running.", "WARNING")
-        return
+        api = HfApi(token=HF_TOKEN)
+        api.upload_file(
+            path_or_fileobj=str(zip_path),
+            path_in_repo=zip_filename,
+            repo_id=HF_OUTPUT_DATASET_ID,
+            repo_type="dataset",
+            commit_message=f"[{FLOW_ID}] Batch transcriptions: {len(transcription_files)} files"
+        )
         
-    processing_status["is_running"] = True
-    
-    try:
-        log_message("🚀 Starting audio transcription processing loop...", "INFO")
+        print(f"[{FLOW_ID}] Successfully uploaded zip file: {zip_filename}")
         
-        # Fetch reference files from BG_Vid repo once at the start
-        reference_map = fetch_reference_files(REFERENCE_REPO_ID)
+        # Cleanup
+        os.remove(zip_path)
         
-        if not reference_map:
-            log_message("❌ No reference files found. Cannot proceed.", "ERROR")
-            return
+        return True
         
-        while processing_status["is_running"]:
+    except Exception as e:
+        print(f"[{FLOW_ID}] Error zipping and uploading transcriptions: {e}")
+        return False
+
+# --- Core Processing Functions ---
+
+async def get_available_server(timeout: float = 300.0) -> WhisperServer:
+    """Round-robin selection of an available whisper server."""
+    global server_index
+    start_time = time.time()
+    while True:
+        # Round-robin check for an available server
+        for _ in range(len(servers)):
+            server = servers[server_index]
+            server_index = (server_index + 1) % len(servers)
+            if not server.busy:
+                return server
+        
+        # If all servers are busy, wait for a short period and check again
+        await asyncio.sleep(0.5)
+        
+        # Check if timeout has been reached
+        if time.time() - start_time > timeout:
+            raise TimeoutError(f"Timeout ({timeout}s) waiting for an available whisper server.")
+
+async def send_audio_for_transcription(audio_path: Path, progress_tracker: Dict) -> Optional[Dict]:
+    """Sends a single audio file to a whisper server for transcription."""
+    MAX_RETRIES = 3
+    for attempt in range(MAX_RETRIES):
+        server = None
+        try:
+            # 1. Get an available server
+            server = await get_available_server()
+            server.busy = True
+            start_time = time.time()
             
-            current_state = download_hf_state(TARGET_REPO_ID, HF_STATE_FILE)
-            next_file_info = get_next_file_to_process(SOURCE_REPO_ID, current_state)
+            if attempt == 0:
+                print(f"[{FLOW_ID}] Starting transcription attempt on {audio_path.name}...")
             
-            if next_file_info is None:
-                log_message("💤 No new audio files to process. Sleeping for a while...", "INFO")
-                time.sleep(PROCESSING_DELAY * 5)
-                continue
-                
-            target_file = next_file_info['filename']
-            audio_url = next_file_info['url']
-            target_index = next_file_info['index']
+            # 2. Prepare request data
+            form_data = aiohttp.FormData()
+            form_data.add_field('file',
+                                audio_path.open('rb'),
+                                filename=audio_path.name,
+                                content_type='audio/mpeg')
             
-            processing_status["current_file"] = target_file
-            success = False
-            matched_filename = None
-            
-            try:
-                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)
-                    continue
-                    
-                local_wav_path = os.path.join(DOWNLOAD_FOLDER, os.path.basename(target_file))
-                log_message(f"⬇️ Downloading audio file: {target_file}", "INFO")
-                
-                if download_with_retry(audio_url, local_wav_path):
-                    
-                    # Extract base filename for matching
-                    base_filename = os.path.basename(target_file)
-                    matched_filename = find_matching_filename(base_filename, reference_map)
-                    
-                    # Use matched filename if found, otherwise use original filename
-                    output_filename = matched_filename if matched_filename else base_filename
-                    
-                    if process_audio_file(local_wav_path, reference_map, output_filename):
-                        success = True
-                        log_message(f"✅ Finished processing: {target_file}", "INFO")
+            # 3. Send request
+            async with aiohttp.ClientSession() as session:
+                async with session.post(server.url, data=form_data, timeout=aiohttp.ClientTimeout(total=600)) as resp:
+                    if resp.status == 200:
+                        result = await resp.json()
+                        
+                        # Check if response contains transcription data
+                        if result.get('text') or result.get('transcription'):
+                            # Update progress counter
+                            progress_tracker['completed'] += 1
+                            if progress_tracker['completed'] % 10 == 0:
+                                print(f"[{FLOW_ID}] PROGRESS: {progress_tracker['completed']}/{progress_tracker['total']} transcriptions completed.")
+                            
+                            print(f"[{FLOW_ID}] Success: {audio_path.name} transcribed by {server.url}")
+                            
+                            # Store the full transcription result
+                            return {
+                                "audio_file": audio_path.name,
+                                "text": result.get('text', result.get('transcription', '')),
+                                "language": result.get('language', 'unknown'),
+                                "confidence": result.get('confidence'),
+                                "duration": result.get('duration'),
+                            }
+                        else:
+                            print(f"[{FLOW_ID}] Server {server.url} returned invalid response format for {audio_path.name}. Response: {result}")
+                            continue
                     else:
-                        log_message(f"❌ Processing failed for: {target_file}", "ERROR")
-                else:
-                    log_message(f"❌ Download failed for: {target_file}", "ERROR")
-                    
-            except Exception as e:
-                log_message(f"🔥 An unhandled error occurred while processing {target_file}: {str(e)}", "ERROR")
-                log_failed_file(target_file, str(e))
-                
-            finally:
-                next_index_to_save = target_index + 1
-                current_state = download_hf_state(TARGET_REPO_ID, HF_STATE_FILE)
-                
-                if success:
-                    unlock_file_as_processed(target_file, current_state, next_index_to_save)
-                    processing_status["processed_files"] += 1
-                else:
-                    log_message(f"⚠️ Processing failed for {target_file}. Marking as 'failed' and advancing index.", "WARNING")
-                    current_state["file_states"][target_file] = "failed"
-                    current_state["next_download_index"] = next_index_to_save
-                    upload_hf_state(TARGET_REPO_ID, HF_STATE_FILE, current_state)
-                    processing_status["failed_files"] += 1
-                    
-                if os.path.exists(local_wav_path):
-                    os.remove(local_wav_path)
-                    log_message(f"🗑️ Cleaned up local file: {local_wav_path}", "INFO")
-
-            time.sleep(PROCESSING_DELAY)
-
-        log_message("🎉 Processing complete!", "INFO")
-        log_message(f"📊 Final stats: {processing_status['transcribed_files']} audio files transcribed, {processing_status['processed_files']} files processed", "INFO")
-        
-    except KeyboardInterrupt:
-        log_message("⏹️ Processing interrupted by user", "WARNING")
-    except Exception as e:
-        log_message(f"❌ Fatal error: {str(e)}", "ERROR")
-    finally:
-        processing_status["is_running"] = False
-        cleanup_temp_files()
+                        error_text = await resp.text()
+                        print(f"[{FLOW_ID}] Error from server {server.url} for {audio_path.name}: {resp.status} - {error_text}. Retrying...")
+                        continue
+                        
+        except (aiohttp.ClientError, asyncio.TimeoutError, TimeoutError) as e:
+            print(f"[{FLOW_ID}] Connection/Timeout error for {audio_path.name} on {server.url if server else 'unknown server'}: {e}. Retrying...")
+            continue
+        except Exception as e:
+            print(f"[{FLOW_ID}] Unexpected error during transcription for {audio_path.name}: {e}. Retrying...")
+            continue
+        finally:
+            if server:
+                end_time = time.time()
+                server.busy = False
+                server.total_processed += 1
+                server.total_time += (end_time - start_time)
+
+    print(f"[{FLOW_ID}] FAILED after {MAX_RETRIES} attempts for {audio_path.name}.")
+    return None
 
-if __name__ == "__main__":
-    main_processing_loop()
+# --- FastAPI App and Endpoints ---
 
-# ===== FASTAPI ENDPOINTS =====
+app = FastAPI(
+    title=f"Flow Server {FLOW_ID} API",
+    description="Processes audio files from a dataset, sends to whisper servers for transcription, and tracks progress.",
+    version="1.0.0"
+)
 
-@app.get("/")
-async def root():
-    """Root endpoint with service info"""
-    return {
-        "service": "Audio Transcriber",
-        "status": "running",
-        "version": "1.0.0",
-        "endpoints": {
-            "status": "/status",
-            "start": "/start",
-            "stop": "/stop",
-            "process": "/process/{filename}",
-            "logs": "/logs"
-        }
-    }
+@app.on_event("startup")
+async def startup_event():
+    print(f"Flow Server {FLOW_ID} started on port {FLOW_PORT}.")
 
-@app.get("/status")
-async def get_status():
-    """Get current processing status"""
+@app.post("/process")
+async def process_audio_files(background_tasks: BackgroundTasks):
+    """
+    Main processing endpoint that orchestrates transcription of audio files.
+    Fetches audio from HF dataset, sends to Whisper servers, and uploads results.
+    Uses reference file mapping for output filename renaming.
+    """
+    background_tasks.add_task(process_audio_files_background)
     return {
-        "is_running": processing_status["is_running"],
-        "current_file": processing_status["current_file"],
-        "total_files": processing_status["total_files"],
-        "processed_files": processing_status["processed_files"],
-        "transcribed_files": processing_status["transcribed_files"],
-        "failed_files": processing_status["failed_files"],
-        "last_update": processing_status["last_update"],
-        "recent_logs": processing_status["logs"][-10:]
+        "status": "processing_started",
+        "flow_id": FLOW_ID,
+        "message": "Background processing task started. Check /status for progress."
     }
 
-@app.post("/start")
-async def start_processing():
-    """Start the main processing loop"""
-    if processing_status["is_running"]:
-        raise HTTPException(status_code=400, detail="Processing already running")
+async def process_audio_files_background():
+    """Background task that processes audio files with reference mapping."""
+    progress_data = load_progress()
+    reference_map = progress_data.get('reference_map', {})
     
-    # Start processing in a separate thread
-    thread = threading.Thread(target=main_processing_loop, daemon=True)
-    thread.start()
+    # Fetch reference map if empty
+    if not reference_map:
+        print(f"[{FLOW_ID}] Reference map is empty. Fetching from {REFERENCE_REPO_ID}...")
+        reference_map = await get_reference_map(REFERENCE_REPO_ID)
+        progress_data['reference_map'] = reference_map
+        save_progress(progress_data)
     
-    return {
-        "message": "Processing started",
-        "status": "started"
-    }
-
-@app.post("/stop")
-async def stop_processing():
-    """Stop the main processing loop"""
-    if not processing_status["is_running"]:
-        raise HTTPException(status_code=400, detail="Processing not running")
+    audio_files = await get_audio_file_list(progress_data)
+    if not audio_files:
+        print(f"[{FLOW_ID}] No audio files found. Exiting.")
+        return
     
-    processing_status["is_running"] = False
+    start_index = progress_data['last_processed_index']
+    transcription_results = []
+    transcription_files = []
     
-    return {
-        "message": "Processing stopped",
-        "status": "stopped"
-    }
-
-@app.get("/logs")
-async def get_logs(limit: int = 50):
-    """Get recent logs"""
-    logs = processing_status["logs"][-limit:]
-    return {
-        "total_logs": len(processing_status["logs"]),
-        "recent_logs": logs
+    # Progress tracking for console output
+    progress_tracker = {
+        'total': len(audio_files) - start_index,
+        'completed': 0
     }
-
-@app.post("/process/{filename}")
-async def process_single_file(filename: str):
-    """Process a single audio file manually"""
-    try:
-        log_message(f"🎯 Manual processing requested for: {filename}", "INFO")
-        
-        # Download and process the file
-        reference_map = fetch_reference_files(REFERENCE_REPO_ID)
-        if not reference_map:
-            raise HTTPException(status_code=500, detail="Could not fetch reference files")
-        
-        # Get file URL
-        audio_url = hf_hub_url(repo_id=SOURCE_REPO_ID, filename=filename, repo_type="dataset", subfolder=None)
-        local_wav_path = os.path.join(DOWNLOAD_FOLDER, os.path.basename(filename))
-        
-        # Download
-        if not download_with_retry(audio_url, local_wav_path):
-            raise HTTPException(status_code=500, detail="Failed to download file")
+    
+    print(f"[{FLOW_ID}] Starting processing from file #{start_index} (out of {len(audio_files)})...")
+    
+    for file_index in range(start_index, len(audio_files)):
+        repo_file_path = audio_files[file_index]
+        audio_filename = Path(repo_file_path).name
         
-        # Find match
-        base_filename = os.path.basename(filename)
-        matched_filename = find_matching_filename(base_filename, reference_map)
+        # Check if already processed
+        state = await download_hf_state()
+        if audio_filename in state.get('file_states', {}) and state['file_states'][audio_filename] == 'processed':
+            print(f"[{FLOW_ID}] Skipping already processed: {audio_filename}")
+            continue
         
-        if not matched_filename:
-            os.remove(local_wav_path)
-            raise HTTPException(status_code=404, detail="No matching filename found")
+        # Lock the file for processing
+        if not await lock_file_for_processing(audio_filename, state):
+            print(f"[{FLOW_ID}] Could not lock file {audio_filename}, skipping.")
+            continue
         
-        # Process
-        if process_audio_file(local_wav_path, reference_map, matched_filename):
-            processing_status["transcribed_files"] += 1
+        try:
+            # Download audio file
+            audio_path = await download_audio_file(file_index, repo_file_path)
+            if not audio_path:
+                print(f"[{FLOW_ID}] Failed to download {audio_filename}")
+                continue
             
-            if os.path.exists(local_wav_path):
-                os.remove(local_wav_path)
+            # Get matching reference filename
+            reference_filename = find_matching_filename(audio_filename, reference_map)
+            if reference_filename:
+                print(f"[{FLOW_ID}] Found reference match: {audio_filename} → {reference_filename}")
+            else:
+                print(f"[{FLOW_ID}] No reference match for {audio_filename}, will use audio filename")
             
-            return {
-                "status": "success",
-                "file": filename,
-                "matched": matched_filename,
-                "message": "Audio transcribed and uploaded successfully"
-            }
-        else:
-            if os.path.exists(local_wav_path):
-                os.remove(local_wav_path)
-            raise HTTPException(status_code=500, detail="Processing failed")
+            # Send for transcription
+            transcription_result = await send_audio_for_transcription(audio_path, progress_tracker)
             
-    except Exception as e:
-        log_message(f"❌ Manual processing error: {str(e)}", "ERROR")
-        raise HTTPException(status_code=500, detail=str(e))
-
-@app.on_event("startup")
-async def startup_event():
-    """Auto-start processing when server starts"""
-    log_message("🚀 Server startup: Checking dependencies...", "INFO")
-    
-    try:
-        import transformers
-        log_message("✅ Transformers found", "INFO")
-    except ImportError:
-        log_message("⚠️ WARNING: Transformers not installed!", "WARNING")
-        log_message("   Install with: pip install transformers librosa torch torchaudio", "WARNING")
+            if transcription_result:
+                # Upload with reference filename if available
+                success = await upload_transcription_to_hf(
+                    audio_filename,
+                    transcription_result,
+                    reference_filename=reference_filename
+                )
+                
+                if success:
+                    transcription_results.append(transcription_result)
+                    json_filename = Path(reference_filename).stem if reference_filename else Path(audio_filename).stem
+                    transcription_files.append(Path(RESULTS_DIR) / f"{json_filename}.json")
+                    
+                    # Mark as processed with fresh state
+                    fresh_state = await download_hf_state()
+                    await unlock_file_as_processed(audio_filename, fresh_state, file_index + 1)
+                    progress_data['transcription_count'] += 1
+                    save_progress(progress_data)
+            
+            # Cleanup
+            if audio_path.exists():
+                os.remove(audio_path)
+            
+            # Check if we've reached the batch threshold
+            if len(transcription_files) >= ZIP_UPLOAD_THRESHOLD:
+                print(f"[{FLOW_ID}] Reached batch threshold ({ZIP_UPLOAD_THRESHOLD}). Creating zip...")
+                await zip_and_upload_transcriptions(transcription_files)
+                transcription_files = []
+                transcription_results = []
+        
+        except Exception as e:
+            print(f"[{FLOW_ID}] Error processing {audio_filename}: {e}")
+        
+        # Save progress after each file
+        progress_data['last_processed_index'] = file_index + 1
+        save_progress(progress_data)
     
-    log_message("🚀 Server startup: Auto-starting processing loop", "INFO")
+    # Upload remaining transcriptions
+    if transcription_files:
+        print(f"[{FLOW_ID}] Uploading final batch of {len(transcription_files)} transcriptions...")
+        await zip_and_upload_transcriptions(transcription_files)
     
-    # Start processing in a separate thread
-    thread = threading.Thread(target=main_processing_loop, daemon=True)
-    thread.start()
+    print(f"[{FLOW_ID}] ✅ Processing complete! Total transcriptions: {progress_data['transcription_count']}")
+
+@app.get("/")
+async def root():
+    progress = load_progress()
+    return {
+        "flow_id": FLOW_ID,
+        "status": "ready",
+        "last_processed_index": progress['last_processed_index'],
+        "total_files_in_list": len(progress['file_list']),
+        "processed_files_count": len(progress['processed_files']),
+        "transcription_count": progress.get('transcription_count', 0),
+        "total_servers": len(servers),
+        "busy_servers": sum(1 for s in servers if s.busy),
+    }
 
-def run_api(host: str = "0.0.0.0", port: int = 8000):
-    """Run the FastAPI server"""
-    log_message(f"🚀 Starting FastAPI server on {host}:{port}", "INFO")
-    uvicorn.run(app, host=host, port=port)
+@app.get("/status")
+async def get_status():
+    """Returns detailed processing status with reference map info."""
+    progress = load_progress()
+    state = await download_hf_state()
+    
+    return {
+        "flow_id": FLOW_ID,
+        "status": "processing" if state['next_download_index'] < len(progress.get('file_list', [])) else "idle",
+        "progress": {
+            "current_index": state['next_download_index'],
+            "total_files": len(progress.get('file_list', [])),
+            "percentage": (state['next_download_index'] / len(progress.get('file_list', [])) * 100) if progress.get('file_list') else 0
+        },
+        "transcription_count": progress.get('transcription_count', 0),
+        "reference_map_size": len(progress.get('reference_map', {})),
+        "server_stats": {
+            "total_servers": len(servers),
+            "busy_servers": sum(1 for s in servers if s.busy),
+            "details": [
+                {
+                    "url": s.url,
+                    "busy": s.busy,
+                    "total_processed": s.total_processed,
+                    "avg_time_per_file": s.total_time / s.total_processed if s.total_processed > 0 else 0
+                }
+                for s in servers
+            ]
+        },
+        "files_in_processing": list(state.get('file_states', {}).keys())
+    }
 
 if __name__ == "__main__":
-    # Run API server (processing will auto-start via startup event)
-    run_api()
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=FLOW_PORT)