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.")