Spaces:
Running
Running
| from document_to_gloss import DocumentToASLConverter | |
| from vectorizer import Vectorizer | |
| from video_gen import create_multi_stitched_video | |
| import gradio as gr | |
| import asyncio | |
| import re | |
| import boto3 | |
| import os | |
| from botocore.config import Config | |
| from dotenv import load_dotenv | |
| import requests | |
| import tempfile | |
| import uuid | |
| import base64 | |
| # Load environment variables from .env file | |
| load_dotenv() | |
| # Load R2/S3 environment secrets | |
| R2_ENDPOINT = os.environ.get("R2_ENDPOINT") | |
| R2_ACCESS_KEY_ID = os.environ.get("R2_ACCESS_KEY_ID") | |
| R2_SECRET_ACCESS_KEY = os.environ.get("R2_SECRET_ACCESS_KEY") | |
| # Validate that required environment variables are set | |
| if not all([R2_ENDPOINT, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY]): | |
| raise ValueError("Missing required R2 environment variables. Please check your .env file.") | |
| title = "AI-SL" | |
| description = "Convert text to ASL!" | |
| article = ("<p style='text-align: center'><a href='https://github.com/deenasun' " | |
| "target='_blank'>Deena Sun on Github</a></p>") | |
| inputs = gr.File(label="Upload Document (pdf, txt, docx, or epub)") | |
| outputs = [ | |
| gr.JSON(label="Processing Results"), | |
| gr.Video(label="ASL Video Output"), | |
| gr.HTML(label="Download Link") | |
| ] | |
| asl_converter = DocumentToASLConverter() | |
| vectorizer = Vectorizer() | |
| session = boto3.session.Session() | |
| s3 = session.client( | |
| service_name='s3', | |
| region_name='auto', | |
| endpoint_url=R2_ENDPOINT, | |
| aws_access_key_id=R2_ACCESS_KEY_ID, | |
| aws_secret_access_key=R2_SECRET_ACCESS_KEY, | |
| config=Config(signature_version='s3v4') | |
| ) | |
| def clean_gloss_token(token): | |
| """ | |
| Clean a gloss token by removing brackets, newlines, and extra whitespace | |
| """ | |
| # Remove brackets and newlines | |
| cleaned = re.sub(r'[\[\]\n\r]', '', token) | |
| # Remove extra whitespace | |
| cleaned = re.sub(r'\s+', ' ', cleaned).strip() | |
| cleaned = cleaned.lower() | |
| return cleaned | |
| def upload_video_to_r2(video_path, bucket_name="ai-sl-videos"): | |
| """ | |
| Upload a video file to R2 and return a public URL | |
| """ | |
| try: | |
| # Generate a unique filename | |
| file_extension = os.path.splitext(video_path)[1] | |
| unique_filename = f"{uuid.uuid4()}{file_extension}" | |
| # Upload to R2 | |
| with open(video_path, 'rb') as video_file: | |
| s3.upload_fileobj( | |
| video_file, | |
| bucket_name, | |
| unique_filename, | |
| ExtraArgs={'ACL': 'public-read'} | |
| ) | |
| # Generate the public URL | |
| video_url = f"{R2_ENDPOINT}/{bucket_name}/{unique_filename}" | |
| print(f"Video uploaded to R2: {video_url}") | |
| return video_url | |
| except Exception as e: | |
| print(f"Error uploading video to R2: {e}") | |
| return None | |
| def video_to_base64(video_path): | |
| """ | |
| Convert a video file to base64 string for direct download | |
| """ | |
| try: | |
| with open(video_path, 'rb') as video_file: | |
| video_data = video_file.read() | |
| base64_data = base64.b64encode(video_data).decode('utf-8') | |
| return f"data:video/mp4;base64,{base64_data}" | |
| except Exception as e: | |
| print(f"Error converting video to base64: {e}") | |
| return None | |
| def download_video_from_url(video_url): | |
| """ | |
| Download a video from a public R2 URL | |
| Returns the local file path where the video is saved | |
| """ | |
| try: | |
| # Create a temporary file with .mp4 extension | |
| temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') | |
| temp_path = temp_file.name | |
| temp_file.close() | |
| # Download the video | |
| print(f"Downloading video from: {video_url}") | |
| response = requests.get(video_url, stream=True) | |
| response.raise_for_status() | |
| # Save to temporary file | |
| with open(temp_path, 'wb') as f: | |
| for chunk in response.iter_content(chunk_size=8192): | |
| f.write(chunk) | |
| print(f"Video downloaded to: {temp_path}") | |
| return temp_path | |
| except Exception as e: | |
| print(f"Error downloading video: {e}") | |
| return None | |
| def cleanup_temp_video(file_path): | |
| """ | |
| Clean up temporary video file | |
| """ | |
| try: | |
| if file_path and os.path.exists(file_path): | |
| os.unlink(file_path) | |
| print(f"Cleaned up: {file_path}") | |
| except Exception as e: | |
| print(f"Error cleaning up file: {e}") | |
| async def parse_vectorize_and_search(file): | |
| print(file) | |
| gloss = asl_converter.convert_document(file) | |
| print("ASL", gloss) | |
| # Split by spaces and clean each token | |
| gloss_tokens = gloss.split() | |
| cleaned_tokens = [] | |
| for token in gloss_tokens: | |
| cleaned = clean_gloss_token(token) | |
| if cleaned: # Only add non-empty tokens | |
| cleaned_tokens.append(cleaned) | |
| print("Cleaned tokens:", cleaned_tokens) | |
| videos = [] | |
| video_files = [] # Store local file paths for stitching | |
| for g in cleaned_tokens: | |
| print(f"Processing {g}") | |
| try: | |
| result = await vectorizer.vector_query_from_supabase(query=g) | |
| print("result", result) | |
| if result.get("match", False): | |
| video_url = result["video_url"] | |
| videos.append(video_url) | |
| # Download the video | |
| local_path = download_video_from_url(video_url) | |
| if local_path: | |
| video_files.append(local_path) | |
| except Exception as e: | |
| print(f"Error processing {g}: {e}") | |
| continue | |
| # Create stitched video if we have multiple videos | |
| stitched_video_path = None | |
| if len(video_files) > 1: | |
| try: | |
| print(f"Creating stitched video from {len(video_files)} videos...") | |
| stitched_video_path = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4').name | |
| create_multi_stitched_video(video_files, stitched_video_path) | |
| print(f"Stitched video created: {stitched_video_path}") | |
| except Exception as e: | |
| print(f"Error creating stitched video: {e}") | |
| stitched_video_path = None | |
| elif len(video_files) == 1: | |
| # If only one video, just use it directly | |
| stitched_video_path = video_files[0] | |
| # Upload final video to R2 and get public URL | |
| final_video_url = None | |
| if stitched_video_path: | |
| final_video_url = upload_video_to_r2(stitched_video_path) | |
| # Clean up the local file after upload | |
| cleanup_temp_video(stitched_video_path) | |
| # Clean up individual video files after stitching | |
| for video_file in video_files: | |
| if video_file != stitched_video_path: # Don't delete the final output | |
| cleanup_temp_video(video_file) | |
| # Create download link HTML | |
| download_html = "" | |
| if final_video_url: | |
| download_html = f""" | |
| <div style="text-align: center; padding: 20px;"> | |
| <h3>Download Your ASL Video</h3> | |
| <a href="{final_video_url}" download="asl_video.mp4" | |
| style="background-color: #4CAF50; color: white; | |
| padding: 12px 24px; text-decoration: none; | |
| border-radius: 4px; display: inline-block;"> | |
| Download Video | |
| </a> | |
| <p style="margin-top: 10px; color: #666;"> | |
| <small>Right-click and "Save As" if the download doesn't | |
| start automatically</small> | |
| </p> | |
| </div> | |
| """ | |
| return { | |
| "status": "success", | |
| "videos": videos, | |
| "video_count": len(videos), | |
| "gloss": gloss, | |
| "cleaned_tokens": cleaned_tokens, | |
| "final_video_url": final_video_url | |
| }, final_video_url, download_html | |
| # Create a synchronous wrapper for Gradio | |
| def parse_vectorize_and_search_sync(file): | |
| return asyncio.run(parse_vectorize_and_search(file)) | |
| async def parse_vectorize_and_search_base64(file): | |
| """ | |
| Alternative version that returns video as base64 data instead of uploading to R2 | |
| """ | |
| print(file) | |
| gloss = asl_converter.convert_document(file) | |
| print("ASL", gloss) | |
| # Split by spaces and clean each token | |
| gloss_tokens = gloss.split() | |
| cleaned_tokens = [] | |
| for token in gloss_tokens: | |
| cleaned = clean_gloss_token(token) | |
| if cleaned: # Only add non-empty tokens | |
| cleaned_tokens.append(cleaned) | |
| print("Cleaned tokens:", cleaned_tokens) | |
| videos = [] | |
| video_files = [] # Store local file paths for stitching | |
| for g in cleaned_tokens: | |
| print(f"Processing {g}") | |
| try: | |
| result = await vectorizer.vector_query_from_supabase(query=g) | |
| print("result", result) | |
| if result.get("match", False): | |
| video_url = result["video_url"] | |
| videos.append(video_url) | |
| # Download the video | |
| local_path = download_video_from_url(video_url) | |
| if local_path: | |
| video_files.append(local_path) | |
| except Exception as e: | |
| print(f"Error processing {g}: {e}") | |
| continue | |
| # Create stitched video if we have multiple videos | |
| stitched_video_path = None | |
| if len(video_files) > 1: | |
| try: | |
| print(f"Creating stitched video from {len(video_files)} videos...") | |
| stitched_video_path = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4').name | |
| create_multi_stitched_video(video_files, stitched_video_path) | |
| print(f"Stitched video created: {stitched_video_path}") | |
| except Exception as e: | |
| print(f"Error creating stitched video: {e}") | |
| stitched_video_path = None | |
| elif len(video_files) == 1: | |
| # If only one video, just use it directly | |
| stitched_video_path = video_files[0] | |
| # Convert final video to base64 | |
| final_video_base64 = None | |
| if stitched_video_path: | |
| final_video_base64 = video_to_base64(stitched_video_path) | |
| # Clean up the local file after conversion | |
| cleanup_temp_video(stitched_video_path) | |
| # Clean up individual video files after stitching | |
| for video_file in video_files: | |
| if video_file != stitched_video_path: # Don't delete the final output | |
| cleanup_temp_video(video_file) | |
| # Create download link HTML for base64 | |
| download_html = "" | |
| if final_video_base64: | |
| download_html = f""" | |
| <div style="text-align: center; padding: 20px;"> | |
| <h3>Download Your ASL Video</h3> | |
| <a href="{final_video_base64}" download="asl_video.mp4" | |
| style="background-color: #4CAF50; color: white; | |
| padding: 12px 24px; text-decoration: none; | |
| border-radius: 4px; display: inline-block;"> | |
| Download Video | |
| </a> | |
| <p style="margin-top: 10px; color: #666;"> | |
| <small>Video is embedded directly - click to download</small> | |
| </p> | |
| </div> | |
| """ | |
| return { | |
| "status": "success", | |
| "videos": videos, | |
| "video_count": len(videos), | |
| "gloss": gloss, | |
| "cleaned_tokens": cleaned_tokens, | |
| "video_format": "base64" | |
| }, final_video_base64, download_html | |
| def parse_vectorize_and_search_base64_sync(file): | |
| return asyncio.run(parse_vectorize_and_search_base64(file)) | |
| intf = gr.Interface( | |
| fn=parse_vectorize_and_search_sync, | |
| inputs=inputs, | |
| outputs=outputs, | |
| title=title, | |
| description=description, | |
| article=article | |
| ) | |
| intf.launch(share=True) | |