youtubeTranscription.py

import os
import re
import time
from googleapiclient.discovery import build
from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api._errors import TranscriptsDisabled, NoTranscriptAvailable, VideoUnavailable
from langdetect import detect
from tqdm import tqdm
from dotenv import load_dotenv
# Use deep-translator instead of googletrans
from deep_translator import GoogleTranslator

load_dotenv()

# Set your YouTube Data API key and Channel ID
YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
CHANNEL_URLS = os.getenv("CHANNEL_URLS").split(",")

# Create YouTube API service
youtube = build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)

def get_channel_id(channel_url):
    channel_url = channel_url.strip()
    if "youtube.com/channel/" in channel_url:
        return channel_url.split("youtube.com/channel/")[1]
    elif "youtube.com/user/" in channel_url or "youtube.com/c/" in channel_url:
        request = youtube.search().list(
            part="id",
            q=channel_url,
            type="channel",
            maxResults=1
        )
        response = request.execute()
        if response["items"]:
            return response["items"][0]["id"]["channelId"]
    else:
        request = youtube.search().list(
            part="id",
            q=channel_url,
            type="channel",
            maxResults=1
        )
        response = request.execute()
        if response["items"]:
            return response["items"][0]["id"]["channelId"]
    return None

CHANNEL_IDS = []
for url in CHANNEL_URLS:
    channel_id = get_channel_id(url)
    if channel_id:
        CHANNEL_IDS.append(channel_id)
    else:
        print(f"Warning: Unable to determine Channel ID for {url}")

if not CHANNEL_IDS:
    raise ValueError("Unable to determine any valid Channel IDs from the provided URLs or names.")

# Initialize Translator
translator = GoogleTranslator(source='auto', target='en')

def get_channel_name(channel_id):
    try:
        request = youtube.channels().list(
            part="snippet",
            id=channel_id
        )
        response = request.execute()
        return response['items'][0]['snippet']['title']
    except Exception as e:
        print(f"Error fetching channel name for {channel_id}: {e}")
        return channel_id  # Fallback to using channel_id as folder name

def get_all_video_ids(channel_id):
    video_ids = []
    next_page_token = None

    while True:
        request = youtube.search().list(
            part="id",
            channelId=channel_id,
            maxResults=50,
            pageToken=next_page_token,
            type="video"
        )
        response = request.execute()

        for item in response["items"]:
            if "videoId" in item["id"]:
                video_ids.append(item["id"]["videoId"])

        next_page_token = response.get("nextPageToken")
        if not next_page_token:
            break

    return video_ids

def clean_description(description):
    # Remove URLs
    description = re.sub(r'http[s]?://\S+', '', description)
    # Remove social media handles (e.g., @username)
    description = re.sub(r'@\w+', '', description)
    # Remove any remaining unwanted characters (optional)
    description = re.sub(r'\s+', ' ', description).strip()
    return description

# Fetch transcripts for the given video IDs
def fetch_video_data(video_ids, channel_id):
    video_data = {}
    failed_videos = []
    channel_name = get_channel_name(channel_id)
    
    # Create a subfolder for the channel
    channel_folder = os.path.join('transcriptions', channel_name)
    os.makedirs(channel_folder, exist_ok=True)

    # Create a progress bar
    pbar = tqdm(total=len(video_ids), desc=f"Processing videos for {channel_name}", unit="video")
    
    for video_id in video_ids:
        # Check if transcription file already exists
        transcription_file = os.path.join(channel_folder, f"{video_id}_transcription.txt")
        if os.path.exists(transcription_file):
            pbar.write(f"Skipping Video ID: {video_id} (transcription already exists)")
            pbar.update(1)
            continue  # Skip to the next video

        try:
            # Fetch video details
            video_response = youtube.videos().list(
                part="snippet",
                id=video_id
            ).execute()

            if not video_response['items']:
                pbar.write(f"No video details found for Video ID: {video_id}")
                failed_videos.append(video_id)
                continue  # Skip to the next video

            video_item = video_response['items'][0]
            snippet = video_item.get('snippet', {})

            video_title = snippet.get('title', '')
            video_description = snippet.get('description', '')

            # Clean the description to remove social handles and links
            cleaned_description = clean_description(video_description)

            # Attempt to fetch transcript using different methods
            transcript_entries = None
            for method in range(1, 4):
                try:
                    if method == 1:
                        # Method 1: Try to get all transcripts
                        transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
                        transcript = transcript_list.find_transcript(['en'])
                        transcript_entries = transcript.fetch()
                    elif method == 2:
                        # Method 2: Try to get transcript directly
                        transcript_entries = YouTubeTranscriptApi.get_transcript(video_id)
                    else:
                        # Method 3: Try to get any available transcript and translate if necessary
                        transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
                        transcript = transcript_list.find_generated_transcript(['hi', 'en'])
                        transcript_entries = transcript.fetch()
                    
                    if transcript_entries:
                        break
                except Exception:
                    continue

            if transcript_entries is not None and len(transcript_entries) > 0:
                # Detect language of the transcript
                transcript_text = " ".join([entry['text'] for entry in transcript_entries if entry.get('text')])
                detected_language = detect(transcript_text)

                if detected_language != 'en':
                    # Translate each entry individually
                    for entry in transcript_entries:
                        text_to_translate = entry.get('text', '')
                        if text_to_translate:
                            try:
                                # Use deep-translator
                                translated_text = translator.translate(text_to_translate)
                                entry['text'] = translated_text
                            except Exception as e:
                                print(f"Translation failed for video {video_id} at entry starting at {entry.get('start')}: {e}")
                                import traceback
                                traceback.print_exc()
                                entry['text'] = text_to_translate  # Fallback to original text
                            # Include a delay to prevent rate limiting
                            time.sleep(0.5)
                        else:
                            entry['text'] = ''

                # Write the file in the channel's subfolder with timestamps
                with open(transcription_file, "w", encoding='utf-8') as f:
                    f.write(f"Title: {video_title}\n\nDescription: {cleaned_description}\n\nTranscript:\n")
                    for entry in transcript_entries:
                        start_time = entry.get('start')
                        duration = entry.get('duration')
                        text = entry.get('text')
                        if start_time is not None and duration is not None and text is not None:
                            f.write(f"[{start_time:.2f} - {start_time + duration:.2f}] {text}\n")

                # Update video data
                video_data[video_id] = {
                    'title': video_title,
                    'description': cleaned_description,
                    'transcript_entries': transcript_entries
                }

            else:
                pbar.write(f"Could not retrieve transcript for Video ID: {video_id}")
                failed_videos.append(video_id)  # Log the failed video ID

        except Exception as e:
            pbar.write(f"An error occurred while processing video {video_id}: {e}")
            import traceback
            pbar.write(traceback.format_exc())  # Output the traceback
            failed_videos.append(video_id)  # Log the failed video ID

        finally:
            # Update the progress bar
            pbar.update(1)

    # Close the progress bar
    pbar.close()

    # Write failed video IDs to a separate file
    if failed_videos:
        with open('failed_videos.txt', 'w') as fv:
            for vid in failed_videos:
                fv.write(f"{vid}\n")
        print(f"\nLogged failed video IDs to 'failed_videos.txt'.")

    return video_data

# Main Function
if __name__ == "__main__":
    for channel_id in CHANNEL_IDS:
        # Step 1: Get all video IDs for the channel
        video_ids = get_all_video_ids(channel_id)
        
        # Step 2: Fetch transcripts for the videos
        video_data = fetch_video_data(video_ids, channel_id)

    print("Transcription process completed.")