Update app.py

app.py

@@ -30,8 +30,13 @@ LOCAL_STATE_FOLDER = Path(".state")
 LOCAL_STATE_FOLDER.mkdir(exist_ok=True)
 
 # Processing configuration
-MAX_UPLOADS_BEFORE_PAUSE = 120  # Pause uploading after 120 files
+MAX_UPLOADS_BEFORE_PAUSE = 120  # Max uploads allowed per hour
 UPLOAD_PAUSE_ENABLED = True
+UPLOAD_WINDOW_SECONDS = 3600  # 1 hour in seconds
+
+# Upload rate limiting (per-hour tracking)
+upload_timestamps: List[float] = []  # Timestamps of recent uploads (for rate limiting)
+upload_lock = asyncio.Lock()  # Lock for upload_timestamps access
 
 # Directory within the HF dataset where the audio files are located
 AUDIO_FILE_PREFIX = "audio/"
@@ -305,6 +310,33 @@ async def upload_transcription_to_hf(wav_filename: str, transcription_data: Dict
     # Use the WAV filename, replacing the extension with .json
     json_filename = Path(wav_filename).with_suffix('.json').name
     
+    # --- Rate limiting: check uploads per hour
+    global upload_timestamps
+    if UPLOAD_PAUSE_ENABLED:
+        async with upload_lock:
+            now = time.time()
+            # Remove uploads older than 1 hour
+            upload_timestamps = [ts for ts in upload_timestamps if now - ts < UPLOAD_WINDOW_SECONDS]
+            
+            # Check if we've hit the limit
+            if len(upload_timestamps) >= MAX_UPLOADS_BEFORE_PAUSE:
+                oldest_upload = upload_timestamps[0]
+                time_until_available = UPLOAD_WINDOW_SECONDS - (now - oldest_upload)
+                print(f"[{FLOW_ID}] ⏸️  Upload limit reached ({len(upload_timestamps)}/{MAX_UPLOADS_BEFORE_PAUSE} in last hour). Pausing for {time_until_available:.0f}s...")
+                
+                # Release lock and wait
+        
+        # Wait until the oldest upload falls outside the window
+        while True:
+            async with upload_lock:
+                now = time.time()
+                upload_timestamps = [ts for ts in upload_timestamps if now - ts < UPLOAD_WINDOW_SECONDS]
+                if len(upload_timestamps) < MAX_UPLOADS_BEFORE_PAUSE:
+                    print(f"[{FLOW_ID}] ✅ Upload limit lifted. Resuming uploads...")
+                    break
+            
+            await asyncio.sleep(5)  # Check every 5 seconds
+    
     try:
         print(f"[{FLOW_ID}] Uploading transcription for {wav_filename} as {json_filename} to {HF_OUTPUT_DATASET_ID}...")
         
@@ -320,6 +352,11 @@ async def upload_transcription_to_hf(wav_filename: str, transcription_data: Dict
             commit_message=f"[{FLOW_ID}] Transcription for {wav_filename}"
         )
         
+        # Record this upload timestamp
+        if UPLOAD_PAUSE_ENABLED:
+            async with upload_lock:
+                upload_timestamps.append(time.time())
+        
         print(f"[{FLOW_ID}] Successfully uploaded transcription for {wav_filename}.")
         return True
         
@@ -620,17 +657,24 @@ async def root():
     total_time = sum(s.total_time for s in servers)
     avg_fps = total_processed / total_time if total_time > 0 else 0
     
+    # Get current uploads in the last hour
+    now = time.time()
+    uploads_in_last_hour = sum(1 for ts in upload_timestamps if now - ts < UPLOAD_WINDOW_SECONDS)
+    is_paused = uploads_in_last_hour >= MAX_UPLOADS_BEFORE_PAUSE
+    
     return {
         "flow_id": FLOW_ID,
         "status": "ready",
         "last_processed_index": progress.get('last_processed_index', 0),
         "total_files_in_list": len(progress['file_list']),
         "uploaded_count": progress.get('uploaded_count', 0),
+        "uploads_in_last_hour": uploads_in_last_hour,
+        "upload_limit": MAX_UPLOADS_BEFORE_PAUSE,
+        "upload_paused": is_paused,
         "total_servers": len(servers),
         "processing_servers": sum(1 for s in servers if s.is_processing),
         "total_files_processed_by_servers": total_processed,
-        "avg_files_per_second": avg_fps,
-        "upload_limit_paused": progress.get('uploaded_count', 0) >= MAX_UPLOADS_BEFORE_PAUSE
+        "avg_files_per_second": avg_fps
     }
 
 @app.post("/start_processing")