#!/usr/bin/env python3 """ Hugging Face Spaces app.py - Video Transcription Service Combines Gradio interface with FastAPI for full functionality """ import gradio as gr import asyncio import threading import time import os import logging import socket from datetime import datetime from typing import Optional, Tuple import uvicorn from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.responses import JSONResponse import tempfile # Import our existing modules from config import settings from models import TranscriptionStatus, TranscriptionResponse, TranscriptionResult from storage import storage from transcription_service import transcription_service from logging_config import setup_logging, log_step, log_success, log_error def find_available_port(start_port=7860, max_attempts=100): """Find an available port starting from start_port""" for port in range(start_port, start_port + max_attempts): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('0.0.0.0', port)) log_success(f"Found available port: {port}") return port except OSError: continue # If no port found in range, try system-assigned port with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('0.0.0.0', 0)) port = s.getsockname()[1] log_success(f"Using system-assigned port: {port}") return port # Setup logging for Hugging Face Spaces setup_logging(level=logging.INFO, log_to_file=False) logger = logging.getLogger(__name__) # Configure for Hugging Face Spaces os.environ.setdefault("WHISPER_MODEL", "base") # HF Spaces can handle base model os.environ.setdefault("MODEL_PRELOAD", "true") os.environ.setdefault("DEBUG", "false") # FastAPI app for API functionality api_app = FastAPI( title="Video Transcription API", description="API endpoints for video transcription", version="1.0.0" ) class TranscriptionManager: def __init__(self): self.model_loaded = False self.model_loading = False async def ensure_model_loaded(self): """Ensure Whisper model is loaded""" if self.model_loaded: return True if self.model_loading: while self.model_loading: await asyncio.sleep(0.1) return self.model_loaded self.model_loading = True try: logger.info("🤖 Loading Whisper model for Hugging Face Spaces...") success = await transcription_service.preload_model() self.model_loaded = success return success finally: self.model_loading = False # Global transcription manager transcription_manager = TranscriptionManager() # FastAPI endpoints (preserve existing API functionality) @api_app.post("/transcribe") async def api_transcribe(file: UploadFile = File(...), language: str = None): """API endpoint for video transcription""" try: # Ensure model is loaded if not await transcription_manager.ensure_model_loaded(): raise HTTPException(status_code=503, detail="Model not available") # Validate file if not file.filename: raise HTTPException(status_code=400, detail="No file provided") # Read file content content = await file.read() if len(content) > settings.MAX_FILE_SIZE: raise HTTPException(status_code=413, detail="File too large") # Create transcription transcription_id = storage.create_transcription(language=language) # Start transcription in background asyncio.create_task( transcription_service.transcribe_video(content, transcription_id, language) ) return TranscriptionResponse( id=transcription_id, status=TranscriptionStatus.PENDING, message="Transcription started", created_at=storage.get_transcription(transcription_id).created_at ) except HTTPException: raise except Exception as e: logger.error(f"API transcription error: {e}") raise HTTPException(status_code=500, detail=str(e)) @api_app.get("/transcribe/{transcription_id}") async def api_get_transcription(transcription_id: int): """API endpoint to get transcription status/results""" result = storage.get_transcription(transcription_id) if not result: raise HTTPException(status_code=404, detail="Transcription not found") return result @api_app.get("/health") async def api_health(): """API health check""" return { "status": "healthy", "model_loaded": transcription_manager.model_loaded, "active_transcriptions": len([ t for t in storage._storage.values() if t.status in [TranscriptionStatus.PENDING, TranscriptionStatus.PROCESSING] ]) if hasattr(storage, '_storage') else 0 } # Gradio interface functions (sync versions for Gradio compatibility) def gradio_transcribe(video_file, language): """Gradio transcription function""" if video_file is None: return "❌ Please upload a video file", "", "" try: # Check if model is loaded (sync check) if not transcription_manager.model_loaded: return "❌ Model not loaded yet. Please wait and try again.", "", "" # Read file with open(video_file, 'rb') as f: content = f.read() if len(content) > settings.MAX_FILE_SIZE: return f"❌ File too large. Maximum size: {settings.MAX_FILE_SIZE // (1024*1024)}MB", "", "" # Create transcription transcription_id = storage.create_transcription(language=language if language != "auto" else None) # Start transcription in background loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_in_executor( None, lambda: asyncio.run(transcription_service.transcribe_video( content, transcription_id, language if language != "auto" else None )) ) return f"✅ Transcription started with ID: {transcription_id}", str(transcription_id), "⏳ Processing..." except Exception as e: logger.error(f"Gradio transcription error: {e}") return f"❌ Error: {str(e)}", "", "" def gradio_check_status(transcription_id_str): """Check transcription status for Gradio""" if not transcription_id_str: return "❌ Please provide a transcription ID" try: transcription_id = int(transcription_id_str) result = storage.get_transcription(transcription_id) if not result: return "❌ Transcription not found or expired" if result.status == TranscriptionStatus.COMPLETED: return f"✅ Completed!\n\nLanguage: {result.language}\nDuration: {result.duration}s\n\nText:\n{result.text}" elif result.status == TranscriptionStatus.FAILED: return f"❌ Failed: {result.error_message}" elif result.status == TranscriptionStatus.PROCESSING: return "⏳ Still processing... Please wait and check again." else: return "⏳ Pending... Please wait and check again." except ValueError: return "❌ Invalid transcription ID (must be a number)" except Exception as e: return f"❌ Error: {str(e)}" # Create Gradio interface def create_gradio_interface(): """Create the Gradio interface with error handling""" try: with gr.Blocks( title="tubeMate - Video Transcription Service", theme=gr.themes.Soft(), css=""" .gradio-container { max-width: 1000px !important; } """, analytics_enabled=False # Disable analytics to prevent schema issues ) as interface: gr.Markdown(""" # 🎬 Video Transcription Service Upload your video files and get accurate transcriptions using OpenAI Whisper. **Features:** - 🎥 Multiple video formats (MP4, AVI, MOV, etc.) - 🌐 Automatic language detection or manual selection - 🚀 Fast processing with OpenAI Whisper - 📱 Both web interface and API access """) with gr.Tab("📤 Upload & Transcribe"): with gr.Row(): with gr.Column(): video_input = gr.File( label="Upload Video File", file_types=[".mp4", ".avi", ".mov", ".mkv", ".wmv", ".flv", ".webm", ".m4v"], type="filepath", file_count="single" ) language_input = gr.Dropdown( choices=["auto", "en", "es", "fr", "de", "it", "pt", "ru", "ja", "ko", "zh", "ar", "hi"], value="auto", label="Language (auto-detect or specify)", interactive=True ) transcribe_btn = gr.Button("🚀 Start Transcription", variant="primary") with gr.Column(): status_output = gr.Textbox(label="Status", lines=3) transcription_id_output = gr.Textbox(label="Transcription ID", visible=True) result_output = gr.Textbox(label="Progress", lines=2) with gr.Tab("🔍 Check Status"): with gr.Row(): with gr.Column(): id_input = gr.Textbox(label="Transcription ID", placeholder="Enter transcription ID...") check_btn = gr.Button("📊 Check Status", variant="secondary") with gr.Column(): status_result = gr.Textbox(label="Result", lines=10) with gr.Tab("🔧 API Documentation"): gr.Markdown(""" ## 🌐 API Endpoints You can also use this service programmatically via API calls: ### Upload Video for Transcription ```bash curl -X POST "https://your-space-name.hf.space/api/transcribe" \\ -F "file=@video.mp4" \\ -F "language=en" ``` ### Check Transcription Status ```bash curl "https://your-space-name.hf.space/api/transcribe/123" ``` ### Health Check ```bash curl "https://your-space-name.hf.space/api/health" ``` ### Python Example ```python import requests # Upload video with open('video.mp4', 'rb') as f: response = requests.post( 'https://your-space-name.hf.space/api/transcribe', files={'file': f}, data={'language': 'en'} ) transcription_id = response.json()['id'] # Check status result = requests.get(f'https://your-space-name.hf.space/api/transcribe/{transcription_id}') print(result.json()) ``` """) # Event handlers transcribe_btn.click( fn=gradio_transcribe, inputs=[video_input, language_input], outputs=[status_output, transcription_id_output, result_output], api_name="transcribe" ) check_btn.click( fn=gradio_check_status, inputs=[id_input], outputs=[status_result], api_name="check_status" ) return interface except Exception as e: log_error(f"Error creating Gradio interface: {e}") # Return a simple fallback interface with gr.Blocks(title="tubeMate - Error") as fallback_interface: gr.Markdown("## ❌ Interface Error\nThere was an error creating the interface. Please check the logs.") return fallback_interface # Startup function async def startup(): """Initialize services""" logger.info("🚀 Starting Video Transcription Service on Hugging Face Spaces") # Start storage cleanup await storage.start_cleanup_task() # Preload model log_step("Preloading Whisper model") success = await transcription_manager.ensure_model_loaded() if success: log_success("Model preloaded successfully") else: log_error("Model preload failed") def run_fastapi(port=None): """Run FastAPI in a separate thread""" if port is None: port = find_available_port(7860) log_step(f"Starting FastAPI server on port {port}") uvicorn.run(api_app, host="0.0.0.0", port=port, log_level="info") # Main execution if __name__ == "__main__": # Run startup asyncio.run(startup()) # Find available port available_port = find_available_port(7860) log_step(f"Using port {available_port} for both FastAPI and Gradio") # Start FastAPI in background thread for API access api_thread = threading.Thread(target=run_fastapi, args=(available_port,), daemon=True) api_thread.start() # Create and launch Gradio interface interface = create_gradio_interface() # Launch with error handling try: interface.launch( server_name="0.0.0.0", server_port=available_port, share=False, # HF Spaces handles sharing show_api=False, # Disable API docs to prevent schema errors show_error=True, quiet=False ) except Exception as e: log_error(f"Error launching Gradio interface: {e}") # Try launching without API features interface.launch( server_name="0.0.0.0", server_port=available_port, share=False, show_api=False, show_error=False, quiet=True )