Create app.py

app.py

@@ -0,0 +1,205 @@
+import yt_dlp
+import requests
+import os
+import time
+from google.oauth2.credentials import Credentials
+from google_auth_oauthlib.flow import InstalledAppFlow
+from google.auth.transport.requests import Request
+from googleapiclient.discovery import build
+from googleapiclient.http import MediaFileUpload
+import streamlit as st
+
+API_KEY = "d0ec0d1455bc43a48b6596efb16abcd2"  # ← Replace with your actual API key
+BASE_URL = "https://api.aimlapi.com/v1"
+HEADERS = {"Authorization": f"Bearer {API_KEY}"}
+
+# Streamlit UI
+st.title("YouTube Video to Audio and Transcription")
+
+# ✅ Download YouTube Audio
+def download_audio(youtube_url, output_file="audio.mp3"):
+    ydl_opts = {
+        'format': 'bestaudio/best',
+        'outtmpl': output_file,
+        'postprocessors': [{
+            'key': 'FFmpegExtractAudio',
+            'preferredcodec': 'mp3',
+            'preferredquality': '192',
+        }],
+        'quiet': False,  # Set to False for more verbose output to help debug
+    }
+    
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            st.write("Downloading audio...")
+            ydl.download([youtube_url])
+        # Fixing the extension issue by renaming the file to 'audio.mp3'
+        if os.path.exists("audio.mp3.mp3"):
+            os.rename("audio.mp3.mp3", "audio.mp3")
+            st.write(f"Renamed file to audio.mp3")
+        st.success(f"Audio downloaded successfully: {output_file}")
+        return output_file
+    except Exception as e:
+        st.error(f"Error downloading audio: {str(e)}")
+        return None
+
+
+# Function to upload file to Google Drive
+def upload_to_google_drive(file_path, credentials_file="credentials.json"):
+    SCOPES = ['https://www.googleapis.com/auth/drive.file']
+    creds = None
+    if os.path.exists('token.json'):
+        creds = Credentials.from_authorized_user_file('token.json', SCOPES)
+    
+    if not creds or not creds.valid:
+      if creds and creds.expired and creds.refresh_token:
+          creds.refresh(Request())
+      else:
+          flow = InstalledAppFlow.from_client_secrets_file(credentials_file, SCOPES)
+          creds = flow.run_local_server(port=0)
+
+      with open('token.json', 'w') as token:
+            token.write(creds.to_json())
+
+    try:
+        drive_service = build('drive', 'v3', credentials=creds)
+        file_metadata = {'name': os.path.basename(file_path)}
+        media = MediaFileUpload(file_path, mimetype='audio/mp3')
+        
+        file = drive_service.files().create(body=file_metadata, media_body=media, fields='id').execute()
+        
+        st.success(f"File uploaded successfully: https://drive.google.com/file/d/{file['id']}/view")
+        return f"https://drive.google.com/file/d/{file['id']}/view"
+    
+    except Exception as e:
+        st.error(f"Error uploading to Google Drive: {e}")
+        return None
+
+# Updated function to handle upload to STT (using Google Drive URL)
+def upload_audio_to_stt(file_path):
+    file_url = upload_to_google_drive(file_path)
+    
+    if file_url:
+        st.success(f"File uploaded successfully to Google Drive: {file_url}")
+        return file_url
+    else:
+        st.error("Failed to upload audio to Google Drive.")
+        return None
+
+def get_stt_result(file_url):
+    if file_url is None:
+        st.error("No valid file URL received. Skipping transcription.")
+        return None
+
+    data = {
+        "model": "#g1_whisper-small",
+        "url": file_url
+    }
+    try:
+        response = requests.post(f"{BASE_URL}/stt/create", headers=HEADERS, json=data)
+        if response.status_code == 200:
+            return response.json().get("generation_id")
+        else:
+            st.error(f"Error during transcription request: {response.status_code}")
+            return None
+    except Exception as e:
+        st.error(f"Error during transcription request: {str(e)}")
+        return None
+
+def get_transcription_result(gen_id):
+    time.sleep(10)  # Adjust the wait time based on the API's response time
+    try:
+        response = requests.get(f"{BASE_URL}/stt/{gen_id}", headers=HEADERS)
+        if response.status_code == 200:
+            return response.json().get("result")
+        else:
+            st.error(f"Error retrieving transcription result: {response.status_code}")
+            return None
+    except Exception as e:
+        st.error(f"Error getting transcription result: {str(e)}")
+        return None
+
+# ✅ Summarize Transcript
+def summarize_text(transcript_text):
+    data = {
+        "model": "gpt-4o",
+        "messages": [
+            {"role": "system", "content": "You are a helpful assistant who summarizes YouTube videos."},
+            {"role": "user", "content": f"Summarize the following video transcript:\n\n{transcript_text}"}
+        ],
+        "temperature": 0.7
+    }
+    try:
+        response = requests.post(f"{BASE_URL}/chat/completions", headers=HEADERS, json=data)
+        if response.status_code == 200:
+            return response.json()["choices"][0]["message"]["content"]
+        else:
+            st.error(f"Error summarizing transcript: {response.status_code}")
+            return None
+    except Exception as e:
+        st.error(f"Error summarizing transcript: {str(e)}")
+        return None
+
+# ✅ Text-to-Speech
+def generate_tts_audio(text, output_file="summary.mp3"):
+    data = {
+        "model": "#g1_aura-angus-en",
+        "input": text
+    }
+    try:
+        response = requests.post(f"{BASE_URL}/tts", headers=HEADERS, json=data)
+        if response.status_code == 200:
+            with open(output_file, "wb") as f:
+                f.write(response.content)
+            st.success(f"Audio summary saved as: {output_file}")
+            return output_file
+        else:
+            st.error(f"Error generating TTS audio: {response.status_code}")
+            return None
+    except Exception as e:
+        st.error(f"Error generating TTS audio: {str(e)}")
+        return None
+
+# ✅ Full Pipeline Execution
+def main():
+    video_url = st.text_input("Please enter the YouTube video URL:")
+
+    if video_url:
+        # Step 1: Download Audio from YouTube
+        audio_file = download_audio(video_url)
+        if audio_file:
+            # Step 2: Upload the Audio for Transcription
+            file_url = upload_audio_to_stt(audio_file)
+            if file_url:
+                st.write("Transcribing...")
+                # Step 3: Get Transcription ID
+                gen_id = get_stt_result(file_url)
+                if gen_id:
+                    st.write("Transcription ID generated.")
+                    # Step 4: Get the Transcription
+                    transcript = get_transcription_result(gen_id)
+                    if transcript:
+                        st.write("Transcript:\n", transcript)
+
+                        # Step 5: Summarize the Transcript
+                        summary = summarize_text(transcript)
+                        if summary:
+                            st.write("\nSummary:\n", summary)
+
+                            # Step 6: Generate TTS Audio for the Summary
+                            summary_audio = generate_tts_audio(summary)
+                        else:
+                            st.error("Failed to generate summary.")
+                    else:
+                        st.error("Failed to retrieve transcription result.")
+                else:
+                    st.error("Failed to generate transcription ID.")
+            else:
+                st.error("Failed to upload audio to STT service.")
+        else:
+            st.error("Failed to download audio from the video.")
+
+
+# Run the full pipeline when the button is clicked
+if st.button('Start Processing'):
+    main()