Update app.py

app.py

@@ -1,56 +1,187 @@
-from fastapi import FastAPI
-import os
+from fastapi import FastAPI, HTTPException, BackgroundTasks
+from fastapi.responses import StreamingResponse
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
 import subprocess
+import os
+import tempfile
+import uuid
+import time
+import asyncio
+from typing import Optional
+import whisper
+from googletrans import Translator
+from gtts import gTTS
+import yt_dlp
+import logging
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(title="YouTube Streaming Translator API")
+
+# Enable CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Load whisper model (small version for speed)
+try:
+    model = whisper.load_model("tiny")
+    logger.info("Whisper model loaded successfully")
+except Exception as e:
+    logger.error(f"Failed to load whisper model: {e}")
+    model = None
 
-app = FastAPI()
+# Initialize translator
+translator = Translator()
 
-# Head node address
-RAY_HEAD_ADDRESS = "ray://172.28.0.12:6379"  # <-- Apna correct head node address yahan daalo
+# Temporary directory for storing audio chunks
+TEMP_DIR = tempfile.gettempdir()
+os.makedirs(os.path.join(TEMP_DIR, "youtube_translator"), exist_ok=True)
 
-# Worker connect status
-connected = False
+class VideoRequest(BaseModel):
+    url: str
+    timestamp: Optional[int] = 0  # Start time in seconds
+    chunk_size: Optional[int] = 15  # Size of each chunk in seconds
+    target_language: str = "en"  # Default target language
 
-def get_ray_status():
-    """Check if ray is running or stopped."""
+@app.post("/process-chunk/")
+async def process_chunk(request: VideoRequest, background_tasks: BackgroundTasks):
+    """Process a single chunk of a YouTube video"""
     try:
-        output = subprocess.check_output(["ray", "status"], stderr=subprocess.STDOUT)
-        if b"cluster" in output or b"Connected" in output:
-            return "running"
+        # Generate a unique ID for this request
+        request_id = str(uuid.uuid4())
+        chunk_path = os.path.join(TEMP_DIR, "youtube_translator", f"{request_id}.mp3")
+        
+        # Extract audio chunk from YouTube video
+        start_time = request.timestamp
+        end_time = start_time + request.chunk_size
+        
+        logger.info(f"Extracting audio chunk from {request.url} from {start_time}s to {end_time}s")
+        
+        # Use yt-dlp to download only the specific chunk
+        ydl_opts = {
+            'format': 'bestaudio/best',
+            'outtmpl': chunk_path,
+            'postprocessors': [{
+                'key': 'FFmpegExtractAudio',
+                'preferredcodec': 'mp3',
+                'preferredquality': '192',
+            }],
+            'download_ranges': {
+                'chapters': [],
+                'ranges': {
+                    'start_time': start_time,
+                    'end_time': end_time
+                }
+            },
+            'quiet': True,
+            'no_warnings': True
+        }
+        
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([request.url])
+        
+        # Process the audio chunk in background
+        background_tasks.add_task(
+            process_audio_chunk, 
+            chunk_path, 
+            request.target_language,
+            request_id
+        )
+        
+        # Return a response with the request ID
+        return {"request_id": request_id, "status": "processing"}
+    
+    except Exception as e:
+        logger.error(f"Error processing chunk: {e}")
+        raise HTTPException(status_code=500, detail=f"Error processing chunk: {str(e)}")
+
+async def process_audio_chunk(chunk_path, target_language, request_id):
+    """Process an audio chunk: transcribe, translate, and convert to speech"""
+    try:
+        # Step 1: Transcribe the audio chunk
+        logger.info(f"Transcribing audio chunk: {chunk_path}")
+        result = model.transcribe(chunk_path)
+        transcription = result["text"]
+        
+        # Step 2: Translate the transcription
+        logger.info(f"Translating text to {target_language}: {transcription[:50]}...")
+        translation = translator.translate(transcription, dest=target_language).text
+        
+        # Step 3: Convert translation to speech
+        logger.info(f"Converting translation to speech: {translation[:50]}...")
+        tts_output_path = os.path.join(TEMP_DIR, "youtube_translator", f"{request_id}_tts.mp3")
+        tts = gTTS(text=translation, lang=target_language)
+        tts.save(tts_output_path)
+        
+        logger.info(f"Audio processing completed for request {request_id}")
+        
+    except Exception as e:
+        logger.error(f"Error processing audio chunk: {e}")
+        # Cleanup
+        if os.path.exists(chunk_path):
+            os.remove(chunk_path)
+
+@app.get("/get-audio/{request_id}")
+async def get_audio(request_id: str):
+    """Get the processed audio for a specific request"""
+    tts_output_path = os.path.join(TEMP_DIR, "youtube_translator", f"{request_id}_tts.mp3")
+    
+    # Check if the file exists
+    if not os.path.exists(tts_output_path):
+        raise HTTPException(status_code=404, detail="Audio processing not completed yet or request ID invalid")
+    
+    # Stream the audio file
+    def iterfile():
+        with open(tts_output_path, "rb") as f:
+            yield from f
+        
+        # Clean up the files after streaming
+        try:
+            os.remove(tts_output_path)
+            chunk_path = os.path.join(TEMP_DIR, "youtube_translator", f"{request_id}.mp3")
+            if os.path.exists(chunk_path):
+                os.remove(chunk_path)
+        except Exception as e:
+            logger.error(f"Error cleaning up files: {e}")
+    
+    return StreamingResponse(
+        iterfile(),
+        media_type="audio/mpeg",
+        headers={"Content-Disposition": f"attachment; filename={request_id}.mp3"}
+    )
+
+@app.get("/status/{request_id}")
+async def check_status(request_id: str):
+    """Check the status of a processing request"""
+    tts_output_path = os.path.join(TEMP_DIR, "youtube_translator", f"{request_id}_tts.mp3")
+    
+    if os.path.exists(tts_output_path):
+        return {"status": "completed", "request_id": request_id}
+    else:
+        # Check if the original chunk exists (meaning processing is in progress)
+        chunk_path = os.path.join(TEMP_DIR, "youtube_translator", f"{request_id}.mp3")
+        if os.path.exists(chunk_path):
+            return {"status": "processing", "request_id": request_id}
         else:
-            return "unknown"
-    except subprocess.CalledProcessError:
-        return "stopped"
-    except FileNotFoundError:
-        return "ray-not-installed"
-
-@app.post("/worker")
-async def connect_worker():
-    global connected
-    if not connected:
-        os.system(f"ray start --address='{RAY_HEAD_ADDRESS}'")
-        connected = True
-    return {
-        "message": "Worker connection attempt finished",
-        "connection": 1 if connected else 0,
-        "ray_status": get_ray_status()
-    }
-
-@app.post("/noworker")
-async def disconnect_worker():
-    global connected
-    if connected:
-        os.system("ray stop")
-        connected = False
-    return {
-        "message": "Worker disconnection attempt finished",
-        "connection": 1 if connected else 0,
-        "ray_status": get_ray_status()
-    }
+            raise HTTPException(status_code=404, detail="Request ID not found")
 
 @app.get("/")
 async def root():
-    return {
-        "message": "Worker Node Ready",
-        "connection": 1 if connected else 0,
-        "ray_status": get_ray_status()
-    }
+    return {"message": "YouTube Streaming Translator API"}
+
+# Simple health check endpoint
+@app.get("/health")
+async def health_check():
+    return {"status": "healthy"}
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)