Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import os | |
| import requests | |
| import tempfile | |
| import subprocess | |
| import re | |
| import shutil # Added for rmtree | |
| import modal | |
| def is_youtube_url(url_string: str) -> bool: | |
| """Checks if the given string is a YouTube URL.""" | |
| # More robust regex to find YouTube video ID, accommodating various URL formats | |
| # and additional query parameters. | |
| youtube_regex = ( | |
| r'(?:youtube(?:-nocookie)?\.com/(?:[^/\n\s]+/|watch(?:/|\?(?:[^&\n\s]+&)*v=)|embed(?:/|\?(?:[^&\n\s]+&)*feature=oembed)|shorts/|live/)|youtu\.be/)' | |
| r'([a-zA-Z0-9_-]{11})' # This captures the 11-character video ID | |
| ) | |
| # We use re.search because the video ID might not be at the start of the query string part of the URL. | |
| # re.match only matches at the beginning of the string (or beginning of line in multiline mode). | |
| # The regex now directly looks for the 'v=VIDEO_ID' or youtu.be/VIDEO_ID structure. | |
| # The first part of the regex matches the domain and common paths, the second part captures the ID. | |
| return bool(re.search(youtube_regex, url_string)) | |
| def download_video(url_string: str, temp_dir: str) -> str | None: | |
| """Downloads video from a URL (YouTube or direct link) to a temporary directory.""" | |
| if is_youtube_url(url_string): | |
| print(f"Attempting to download YouTube video: {url_string}") | |
| # Define a fixed output filename pattern within the temp_dir | |
| output_filename_template = "downloaded_video.%(ext)s" # yt-dlp replaces %(ext)s | |
| output_path_template = os.path.join(temp_dir, output_filename_template) | |
| cmd = [ | |
| "yt-dlp", | |
| "-f", "bestvideo[ext=mp4]+bestaudio[ext=m4a]/mp4/best", # Prefer mp4 format | |
| "--output", output_path_template, | |
| url_string | |
| ] | |
| print(f"Executing yt-dlp command: {' '.join(cmd)}") | |
| try: | |
| result = subprocess.run(cmd, capture_output=True, text=True, timeout=300, check=False) | |
| print(f"yt-dlp STDOUT:\n{result.stdout}") | |
| print(f"yt-dlp STDERR:\n{result.stderr}") | |
| if result.returncode == 0: | |
| # Find the actual downloaded file based on the template | |
| downloaded_file_path = None | |
| for item in os.listdir(temp_dir): | |
| if item.startswith("downloaded_video."): | |
| potential_path = os.path.join(temp_dir, item) | |
| if os.path.isfile(potential_path): | |
| downloaded_file_path = potential_path | |
| print(f"YouTube video successfully downloaded to: {downloaded_file_path}") | |
| break | |
| if downloaded_file_path: | |
| return downloaded_file_path | |
| else: | |
| print(f"yt-dlp seemed to succeed (exit code 0) but the output file 'downloaded_video.*' was not found in {temp_dir}.") | |
| return None | |
| else: | |
| print(f"yt-dlp failed with return code {result.returncode}.") | |
| return None | |
| except subprocess.TimeoutExpired: | |
| print(f"yt-dlp command timed out after 300 seconds for URL: {url_string}") | |
| return None | |
| except Exception as e: | |
| print(f"An unexpected error occurred during yt-dlp execution for {url_string}: {e}") | |
| return None | |
| elif url_string.startswith(('http://', 'https://')) and url_string.lower().endswith(('.mp4', '.mov', '.avi', '.mkv', '.webm')): | |
| print(f"Attempting to download direct video link: {url_string}") | |
| try: | |
| response = requests.get(url_string, stream=True, timeout=300) # 5 min timeout | |
| response.raise_for_status() # Raises HTTPError for bad responses (4XX or 5XX) | |
| filename = os.path.basename(url_string) or "downloaded_video_direct.mp4" | |
| video_file_path = os.path.join(temp_dir, filename) | |
| with open(video_file_path, 'wb') as f: | |
| for chunk in response.iter_content(chunk_size=8192): | |
| f.write(chunk) | |
| print(f"Direct video downloaded successfully to: {video_file_path}") | |
| return video_file_path | |
| except requests.exceptions.RequestException as e: | |
| print(f"Error downloading direct video link {url_string}: {e}") | |
| return None | |
| except Exception as e: | |
| print(f"An unexpected error occurred during direct video download for {url_string}: {e}") | |
| return None | |
| else: | |
| print(f"Input '{url_string}' is not a recognized YouTube URL or direct video link for download.") | |
| return None | |
| def process_video_input(input_string: str) -> str: | |
| """ | |
| Processes the video (from URL or local file path) and returns its transcription status. | |
| """ | |
| if not input_string: | |
| return "Error: No video URL or file path provided." | |
| video_path_to_process = None | |
| created_temp_dir = None # To store path of temp directory if created for download | |
| try: | |
| if input_string.startswith(('http://', 'https://')): | |
| print(f"Input is a URL: {input_string}") | |
| created_temp_dir = tempfile.mkdtemp() | |
| print(f"Created temporary directory for download: {created_temp_dir}") | |
| downloaded_path = download_video(input_string, created_temp_dir) | |
| if downloaded_path and os.path.exists(downloaded_path): | |
| video_path_to_process = downloaded_path | |
| else: | |
| # Error message is already printed by download_video or this block | |
| print(f"Failed to download or locate video from URL: {input_string}") | |
| # Cleanup is handled in finally, so just return error | |
| return "Error: Failed to download video from URL." | |
| elif os.path.exists(input_string): | |
| print(f"Input is a local file path: {input_string}") | |
| video_path_to_process = input_string | |
| else: | |
| return f"Error: Input '{input_string}' is not a valid URL or an existing file path." | |
| if video_path_to_process: | |
| print(f"Processing video: {video_path_to_process}") | |
| print(f"Video path to process: {video_path_to_process}") | |
| try: | |
| print("Reading video file into bytes...") | |
| with open(video_path_to_process, "rb") as video_file: | |
| video_bytes_content = video_file.read() | |
| print(f"Read {len(video_bytes_content)} bytes from video file.") | |
| # Ensure MODAL_TOKEN_ID and MODAL_TOKEN_SECRET are set as environment variables | |
| # in your Hugging Face Space. For local `python app.py` runs, Modal CLI's | |
| # authenticated state is usually used. | |
| # os.environ["MODAL_TOKEN_ID"] = "your_modal_token_id" # Replace or set in HF Space | |
| # os.environ["MODAL_TOKEN_SECRET"] = "your_modal_token_secret" # Replace or set in HF Space | |
| print("Looking up Modal function 'whisper-transcriber/transcribe_video_audio'...") | |
| # The function name should match what was deployed. | |
| # It's typically 'AppName/FunctionName' or just 'FunctionName' if app is default. | |
| # Based on your deployment log, app name is 'whisper-transcriber' | |
| # and function is 'transcribe_video_audio' | |
| try: | |
| f = modal.Function.from_name("whisper-transcriber", "transcribe_video_audio") | |
| print("Modal function looked up successfully.") | |
| except modal.Error as e: | |
| print("Modal function 'whisper-transcriber/transcribe_video_audio' not found. Trying with just function name.") | |
| # Fallback or alternative lookup, though the above should be correct for named apps. | |
| # This might be needed if the app name context is implicit. | |
| # For a named app 'whisper-transcriber' and function 'transcribe_video_audio', | |
| # the lookup `modal.Function.lookup("whisper-transcriber", "transcribe_video_audio")` is standard. | |
| # If it was deployed as part of the default app, then just "transcribe_video_audio" might work. | |
| # Given the deployment log, the first lookup should be correct. | |
| return "Error: Could not find the deployed Modal function. Please check deployment status and name." | |
| print("Calling Modal function for transcription...") | |
| # Using .remote() for asynchronous execution, .call() for synchronous | |
| # For Gradio, synchronous (.call()) might be simpler to handle the response directly. | |
| transcription = f.remote(video_bytes_content) # Use .remote() for Modal function call | |
| print(f"Received transcription from Modal: {transcription[:100]}...") | |
| return transcription | |
| except FileNotFoundError: | |
| print(f"Error: Video file not found at {video_path_to_process} before sending to Modal.") | |
| return f"Error: Video file disappeared before processing." | |
| except modal.Error as e: # Using modal.Error as the base Modal exception | |
| print(f"Modal specific error: {e}") | |
| return f"Error during Modal operation: {str(e)}" | |
| except Exception as e: | |
| print(f"An unexpected error occurred while calling Modal: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return f"Error: Failed to get transcription. {str(e)}" | |
| else: | |
| # This case should ideally be caught by earlier checks | |
| return "Error: No video available to process after input handling." | |
| finally: | |
| if created_temp_dir and os.path.exists(created_temp_dir): | |
| print(f"Cleaning up temporary directory: {created_temp_dir}") | |
| try: | |
| shutil.rmtree(created_temp_dir) | |
| print(f"Successfully removed temporary directory: {created_temp_dir}") | |
| except Exception as e: | |
| print(f"Error removing temporary directory {created_temp_dir}: {e}") | |
| # Gradio Interface for the API endpoint | |
| api_interface = gr.Interface( | |
| fn=process_video_input, | |
| inputs=gr.Textbox(label="Video URL or Local File Path for Transcription", | |
| placeholder="Enter YouTube URL, direct video URL (.mp4, .mov, etc.), or local file path..."), | |
| outputs="text", | |
| title="Video Transcription API", | |
| description="Provide a video URL or local file path to get its audio transcription status.", | |
| allow_flagging="never" | |
| ) | |
| # Gradio Interface for a simple user-facing demo | |
| def demo_process_video(input_string: str) -> str: | |
| """ | |
| A simple demo function for the Gradio UI. | |
| It calls the same backend logic as the API. | |
| """ | |
| print(f"Demo received input: {input_string}") | |
| result = process_video_input(input_string) # Call the core logic | |
| return result | |
| demo_interface = gr.Interface( | |
| fn=demo_process_video, | |
| inputs=gr.Textbox(label="Upload Video URL or Local File Path for Demo", | |
| placeholder="Enter YouTube URL, direct video URL (.mp4, .mov, etc.), or local file path..."), | |
| outputs="text", | |
| title="Video Transcription Demo", | |
| description="Provide a video URL or local file path to see its transcription status.", | |
| allow_flagging="never" | |
| ) | |
| # Combine interfaces into a Blocks app | |
| with gr.Blocks() as app: | |
| gr.Markdown("# Contextual Video Data Server") | |
| gr.Markdown("This Hugging Face Space acts as a backend for processing video context for AI models.") | |
| with gr.Tab("API Endpoint (for AI Models)"): | |
| gr.Markdown("### Use this endpoint from another application (e.g., another Hugging Face Space).") | |
| gr.Markdown("The `process_video_input` function is exposed here.") | |
| api_interface.render() | |
| with gr.Tab("Demo (for Manual Testing)"): | |
| gr.Markdown("### Manually test video URLs or paths and observe the response.") | |
| demo_interface.render() | |
| # Launch the Gradio application | |
| if __name__ == "__main__": | |
| app.launch(server_name="0.0.0.0") | |