app.py

import googleapiclient.discovery
import re
import yt_dlp
import whisper
from pydub import AudioSegment
import tempfile
from transformers import pipeline
from youtube_transcript_api import YouTubeTranscriptApi
import torch
import openai
import json
from urllib.parse import urlparse, parse_qs
import os
import gradio as gr

# Set up API keys (ensure these are provided as environment variables)
youtube_api_key = os.getenv("YOUTUBE_API_KEY")
openai_api_key = os.getenv("OPENAI_API_KEY")
openai.api_key = openai_api_key

# Validate API keys
if not youtube_api_key:
    raise ValueError("YOUTUBE_API_KEY is not set. Please set it as an environment variable.")

if not openai_api_key:
    raise ValueError("OPENAI_API_KEY is not set. Please set it as an environment variable.")


def extract_video_id(url):
    """Extracts the video ID from a YouTube URL."""
    try:
        parsed_url = urlparse(url)
        if "youtube.com" in parsed_url.netloc:
            query_params = parse_qs(parsed_url.query)
            return query_params.get('v', [None])[0]
        elif "youtu.be" in parsed_url.netloc:
            return parsed_url.path.strip("/")
        else:
            return None
    except Exception as e:
        print(f"Error parsing URL: {e}")
        return None


def get_video_duration(video_id, api_key):
    """Fetches the video duration in minutes."""
    try:
        youtube = googleapiclient.discovery.build("youtube", "v3", developerKey=api_key)
        request = youtube.videos().list(part="contentDetails", id=video_id)
        response = request.execute()
        if response["items"]:
            duration = response["items"][0]["contentDetails"]["duration"]
            match = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?', duration)
            hours = int(match.group(1)) if match.group(1) else 0
            minutes = int(match.group(2)) if match.group(2) else 0
            seconds = int(match.group(3)) if match.group(3) else 0
            return hours * 60 + minutes + seconds / 60
        else:
            return None
    except Exception as e:
        print(f"Error fetching video duration: {e}")
        return None


def download_and_transcribe_with_whisper(youtube_url):
    """Downloads audio from YouTube and transcribes it using Whisper."""
    try:
        with tempfile.TemporaryDirectory() as temp_dir:
            temp_audio_file = os.path.join(temp_dir, "audio.mp3")
            
            ydl_opts = {
                'format': 'bestaudio/best',
                'outtmpl': temp_audio_file,
                'postprocessors': [{
                    'key': 'FFmpegExtractAudio',
                    'preferredcodec': 'mp3',
                    'preferredquality': '192',
                }],
            }
            # Download audio
            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                ydl.download([youtube_url])
            # Convert to WAV
            audio = AudioSegment.from_file(temp_audio_file)
            wav_file = os.path.join(temp_dir, "audio.wav")
            audio.export(wav_file, format="wav")
            # Transcribe using Whisper
            model = whisper.load_model("large")
            result = model.transcribe(wav_file)
            return result['text']
    except Exception as e:
        print(f"Error during transcription: {e}")
        return None


def get_transcript_from_youtube_api(video_id, video_length):
    """Fetches transcript using YouTube API if available."""
    try:
        transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
        for transcript in transcript_list:
            if not transcript.is_generated:
                segments = transcript.fetch()
                return " ".join(segment['text'] for segment in segments)
        if video_length > 15:  # Use generated transcript for longer videos
            auto_transcript = transcript_list.find_generated_transcript(['en'])
            if auto_transcript:
                segments = auto_transcript.fetch()
                return " ".join(segment['text'] for segment in segments)
        return None
    except Exception as e:
        print(f"Error fetching transcript: {e}")
        return None


def get_transcript(youtube_url):
    """Gets transcript using YouTube API or Whisper."""
    video_id = extract_video_id(youtube_url)
    if not video_id:
        return "Invalid YouTube URL."
    video_length = get_video_duration(video_id, youtube_api_key)
    if video_length is not None:
        transcript = get_transcript_from_youtube_api(video_id, video_length)
        if transcript:
            return transcript
        return download_and_transcribe_with_whisper(youtube_url)
    return "Error fetching video duration."


def summarize_text(text):
    """Summarizes text using Hugging Face pipeline."""
    summarizer = pipeline("summarization", model="facebook/bart-large-cnn", device=0 if torch.cuda.is_available() else -1)
    max_input_length = 1024
    chunk_overlap = 100
    text_chunks = [
        text[i:i + max_input_length]
        for i in range(0, len(text), max_input_length - chunk_overlap)
    ]
    summaries = [
        summarizer(chunk, max_length=100, min_length=50, do_sample=False)[0]['summary_text']
        for chunk in text_chunks
    ]
    return " ".join(summaries)


def generate_optimized_content(summary):
    """Generates optimized content using OpenAI GPT."""
    prompt = f"""
    Analyze the following summarized YouTube video transcript and:
    1. Extract the top 10 keywords.
    2. Generate an optimized title (less than 65 characters).
    3. Create an engaging description.
    4. Generate related tags for the video.

    Summarized Transcript:
    {summary}

    Provide the results in JSON format:
    {{
        "keywords": ["keyword1", "keyword2", ..., "keyword10"],
        "title": "Generated Title",
        "description": "Generated Description",
        "tags": ["tag1", "tag2", ..., "tag10"]
    }}
    """
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "You are an SEO expert."},
                {"role": "user", "content": prompt}
            ]
        )
        return json.loads(response['choices'][0]['message']['content'])
    except Exception as e:
        return {"error": str(e)}


def process_video(youtube_url):
    """Processes video and returns optimized metadata."""
    transcript = get_transcript(youtube_url)
    if not transcript:
        return {"error": "Could not fetch the transcript."}
    summary = summarize_text(transcript)
    return generate_optimized_content(summary)


# Gradio Interface
iface = gr.Interface(
    fn=process_video,
    inputs=gr.Textbox(label="Enter a YouTube video URL"),
    outputs=gr.JSON(label="Optimized Content"),
    title="YouTube Video Optimization Tool",
    description="Enter a YouTube URL to generate SEO-optimized titles, descriptions, and tags."
)

if __name__ == "__main__":
    iface.launch()