youtube_transcript_search.py

#!/usr/bin/env python3
"""
YouTube Transcript Keyword Search Tool

This tool extracts transcripts from YouTube videos and searches for specific keywords,
returning the timestamps where those keywords appear.
"""

import re
import sys
import argparse
from urllib.parse import urlparse, parse_qs
from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api.formatters import TextFormatter


def extract_video_id(url):
    """Extract video ID from various YouTube URL formats"""
    
    # Handle different YouTube URL formats
    patterns = [
        r'(?:youtube\.com\/watch\?v=|youtu\.be\/|youtube\.com\/embed\/|youtube\.com\/live\/)([^&\n?#]+)',
        r'youtube\.com\/watch\?.*v=([^&\n?#]+)'
    ]
    
    for pattern in patterns:
        match = re.search(pattern, url)
        if match:
            return match.group(1)
    
    # If no pattern matches, maybe it's already just the video ID
    if len(url) == 11 and url.isalnum():
        return url
    
    return None


def get_transcript(video_id):
    """Get transcript for a YouTube video"""
    
    try:
        # Create an API instance
        api = YouTubeTranscriptApi()
        
        # Try to get transcript list
        transcript_list = api.list(video_id)
        
        # Prefer manual transcripts over auto-generated
        transcript = None
        try:
            # Try to get a manually created transcript first
            transcript = transcript_list.find_manually_created_transcript(['en'])
        except:
            try:
                # Fall back to auto-generated
                transcript = transcript_list.find_generated_transcript(['en'])
            except:
                # Try any available language
                for t in transcript_list:
                    transcript = t
                    break
        
        if transcript is None:
            return None, "No transcript found"
        
        # Fetch the actual transcript data
        transcript_data = transcript.fetch()
        return transcript_data, None
        
    except Exception as e:
        return None, f"Error retrieving transcript: {str(e)}"


def search_keywords(transcript_data, keywords, video_id=None, case_sensitive=False):
    """Search for keywords in transcript and return timestamps"""
    
    results = []
    
    if not case_sensitive:
        keywords = [kw.lower() for kw in keywords]
    
    for entry in transcript_data:
        # Access attributes instead of dictionary keys
        text = entry.text
        start_time = entry.start
        duration = getattr(entry, 'duration', 0)
        
        # Prepare text for searching
        search_text = text if case_sensitive else text.lower()
        
        # Search for each keyword
        for keyword in keywords:
            if keyword in search_text:
                # Find all occurrences in this segment
                start_pos = 0
                while True:
                    pos = search_text.find(keyword, start_pos)
                    if pos == -1:
                        break
                    
                    # Create timestamp URL if video_id is provided
                    timestamp_url = f"https://youtube.com/watch?v={video_id}&t={int(start_time)}s" if video_id else ""
                    
                    results.append({
                        'keyword': keyword,
                        'text_segment': text.strip(),
                        'start_time': start_time,
                        'end_time': start_time + duration,
                        'timestamp_url': timestamp_url
                    })
                    start_pos = pos + 1
    
    return results


def format_timestamp(seconds):
    """Convert seconds to MM:SS or HH:MM:SS format"""
    hours = int(seconds // 3600)
    minutes = int((seconds % 3600) // 60)
    secs = int(seconds % 60)
    
    if hours > 0:
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
    else:
        return f"{minutes:02d}:{secs:02d}"


def process_single_video(video_url, keywords, case_sensitive=False, show_progress=True):
    """Process a single video and return results"""
    
    # Extract video ID
    video_id = extract_video_id(video_url)
    if not video_id:
        if show_progress:
            print(f"❌ Invalid YouTube URL: {video_url}")
        return None, f"Invalid YouTube URL: {video_url}"
    
    if show_progress:
        print(f"📹 Processing Video ID: {video_id}")
    
    # Get transcript
    if show_progress:
        print("🔄 Fetching transcript...")
    transcript_data, error = get_transcript(video_id)
    
    if error:
        if show_progress:
            print(f"❌ {error}")
        return None, error
    
    if show_progress:
        print(f"✅ Transcript found! ({len(transcript_data)} segments)")
    
    # Search for keywords
    if show_progress:
        print(f"🔍 Searching for: {', '.join(keywords)}")
    results = search_keywords(transcript_data, keywords, video_id=video_id, case_sensitive=case_sensitive)
    
    return {
        'video_id': video_id,
        'video_url': f"https://youtube.com/watch?v={video_id}",
        'transcript_segments': len(transcript_data),
        'results': results,
        'total_matches': len(results)
    }, None


def main():
    """Main function to run the transcript search"""
    
    print("YouTube Transcript Keyword Search Tool")
    print("=" * 50)
    
    # Ask if user wants single or batch mode
    mode = input("Search mode - (s)ingle video or (b)atch multiple videos? [s/b]: ").strip().lower()
    
    if mode.startswith('b'):
        # Batch mode
        print("\n📋 Batch Mode Selected")
        input_method = input("Input method - (m)anual entry or (f)ile input? [m/f]: ").strip().lower()
        
        if input_method.startswith('f'):
            # File input
            file_path = input("Enter path to file containing YouTube URLs (one per line): ").strip()
            try:
                with open(file_path, 'r', encoding='utf-8') as f:
                    video_urls = [line.strip() for line in f if line.strip() and not line.strip().startswith('#')]
                print(f"✅ Loaded {len(video_urls)} URLs from file")
            except FileNotFoundError:
                print(f"❌ File not found: {file_path}")
                return
            except Exception as e:
                print(f"❌ Error reading file: {str(e)}")
                return
        else:
            # Manual input
            print("Enter YouTube video URLs (one per line). Press Enter twice when done:")
            
            video_urls = []
            while True:
                url = input().strip()
                if not url:
                    break
                video_urls.append(url)
        
        if not video_urls:
            print("❌ No video URLs provided.")
            return
        
        print(f"\n📝 Processing {len(video_urls)} videos...")
        
    else:
        # Single mode
        video_url = input("Enter YouTube video URL: ").strip()
        video_urls = [video_url]
    
    # Get keywords from user
    keywords_input = input("Enter keywords to search for (comma-separated): ").strip()
    keywords = [kw.strip() for kw in keywords_input.split(',') if kw.strip()]
    
    if not keywords:
        print("❌ No keywords provided.")
        return
    
    # Ask about case sensitivity
    case_sensitive = input("Case sensitive search? (y/n): ").strip().lower() == 'y'
    
    # Process videos
    all_results = []
    successful_videos = 0
    failed_videos = []
    
    for i, video_url in enumerate(video_urls, 1):
        if len(video_urls) > 1:
            print(f"\n{'='*60}")
            print(f"Processing Video {i}/{len(video_urls)}")
            print(f"{'='*60}")
        
        video_data, error = process_single_video(video_url, keywords, case_sensitive, show_progress=True)
        
        if error:
            failed_videos.append({'url': video_url, 'error': error})
            continue
        
        successful_videos += 1
        all_results.append(video_data)
        
        if len(video_urls) > 1:
            print(f"✅ Video {i} completed: {video_data['total_matches']} matches found")
    
    # Display summary for batch mode
    if len(video_urls) > 1:
        print(f"\n{'='*80}")
        print(f"📊 BATCH PROCESSING SUMMARY")
        print(f"{'='*80}")
        print(f"✅ Successfully processed: {successful_videos}/{len(video_urls)} videos")
        if failed_videos:
            print(f"❌ Failed videos: {len(failed_videos)}")
            for failed in failed_videos:
                print(f"   • {failed['url']}: {failed['error']}")
        
        total_matches = sum(video['total_matches'] for video in all_results)
        print(f"\n🔍 Total matches across all videos: {total_matches}")
    
    if not all_results:
        print("❌ No videos were successfully processed.")
        return
    
    # Display results
    if len(video_urls) == 1:
        # Single video mode - use existing display logic
        video_data = all_results[0]
        results = video_data['results']
        video_id = video_data['video_id']
        
        print(f"\n📊 Found {len(results)} matches:")
        print("=" * 80)
        
        if not results:
            print("No matches found.")
            return
        
        # Group results by keyword
        keyword_groups = {}
        for result in results:
            keyword = result['keyword']
            if keyword not in keyword_groups:
                keyword_groups[keyword] = []
            keyword_groups[keyword].append(result)
        
        # Display results grouped by keyword
        for keyword, matches in keyword_groups.items():
            print(f"\n🔑 Keyword: '{keyword}' ({len(matches)} matches)")
            print("-" * 50)
            
            for i, match in enumerate(matches, 1):
                timestamp_formatted = format_timestamp(match['start_time'])
                print(f"{i:2d}. [{timestamp_formatted}] {match['text_segment']}")
                print(f"    🔗 {match['timestamp_url']}")
                print()
        
    else:
        # Batch mode - display summary and detailed results
        print(f"\n{'='*80}")
        print(f"📊 DETAILED RESULTS")
        print(f"{'='*80}")
        
        # Combine all results for keyword grouping
        all_combined_results = []
        for video_data in all_results:
            for result in video_data['results']:
                result['source_video_id'] = video_data['video_id']
                result['source_video_url'] = video_data['video_url']
                all_combined_results.append(result)
        
        if not all_combined_results:
            print("No matches found across all videos.")
            return
        
        # Group by keyword across all videos
        keyword_groups = {}
        for result in all_combined_results:
            keyword = result['keyword']
            if keyword not in keyword_groups:
                keyword_groups[keyword] = []
            keyword_groups[keyword].append(result)
        
        # Display results grouped by keyword
        for keyword, matches in keyword_groups.items():
            print(f"\n🔑 Keyword: '{keyword}' ({len(matches)} matches across {len(set(m['source_video_id'] for m in matches))} videos)")
            print("-" * 80)
            
            # Group by video within keyword
            video_groups = {}
            for match in matches:
                vid_id = match['source_video_id']
                if vid_id not in video_groups:
                    video_groups[vid_id] = []
                video_groups[vid_id].append(match)
            
            for vid_id, vid_matches in video_groups.items():
                print(f"\n📹 Video: {vid_id} ({len(vid_matches)} matches)")
                print(f"   🔗 {vid_matches[0]['source_video_url']}")
                
                for i, match in enumerate(vid_matches[:5], 1):  # Show first 5 matches per video
                    timestamp_formatted = format_timestamp(match['start_time'])
                    print(f"   {i:2d}. [{timestamp_formatted}] {match['text_segment'][:100]}{'...' if len(match['text_segment']) > 100 else ''}")
                    print(f"      🔗 {match['timestamp_url']}")
                
                if len(vid_matches) > 5:
                    print(f"      ... and {len(vid_matches) - 5} more matches")
                print()
    
    # Option to save results
    save_option = input("Save results to file? (y/n): ").strip().lower()
    if save_option == 'y':
        if len(video_urls) == 1:
            # Single video filename
            filename = f"transcript_search_results_{all_results[0]['video_id']}.txt"
        else:
            # Batch filename with timestamp
            from datetime import datetime
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"transcript_search_batch_results_{timestamp}.txt"
        
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(f"YouTube Transcript Search Results\n")
            f.write(f"{'='*50}\n")
            f.write(f"Search Mode: {'Single Video' if len(video_urls) == 1 else 'Batch Processing'}\n")
            f.write(f"Videos Processed: {len(all_results)}\n")
            f.write(f"Keywords: {', '.join(keywords)}\n")
            f.write(f"Case Sensitive: {case_sensitive}\n")
            
            if len(video_urls) > 1:
                total_matches = sum(video['total_matches'] for video in all_results)
                f.write(f"Total Matches: {total_matches}\n")
                f.write(f"\nVideo Summary:\n")
                for i, video_data in enumerate(all_results, 1):
                    f.write(f"  {i}. {video_data['video_id']} - {video_data['total_matches']} matches\n")
            else:
                f.write(f"Total Matches: {len(all_results[0]['results'])}\n")
                f.write(f"Video ID: {all_results[0]['video_id']}\n")
                f.write(f"Video URL: {all_results[0]['video_url']}\n")
            
            f.write(f"\n{'='*80}\n\n")
            
            # Write detailed results
            if len(video_urls) == 1:
                # Single video format
                results = all_results[0]['results']
                keyword_groups = {}
                for result in results:
                    keyword = result['keyword']
                    if keyword not in keyword_groups:
                        keyword_groups[keyword] = []
                    keyword_groups[keyword].append(result)
                
                for keyword, matches in keyword_groups.items():
                    f.write(f"Keyword: '{keyword}' ({len(matches)} matches)\n")
                    f.write("-" * 50 + "\n")
                    
                    for i, match in enumerate(matches, 1):
                        timestamp_formatted = format_timestamp(match['start_time'])
                        f.write(f"{i:2d}. [{timestamp_formatted}] {match['text_segment']}\n")
                        f.write(f"    URL: {match['timestamp_url']}\n\n")
                    
                    f.write("\n")
            else:
                # Batch format
                for keyword, matches in keyword_groups.items():
                    f.write(f"Keyword: '{keyword}' ({len(matches)} matches)\n")
                    f.write("=" * 50 + "\n")
                    
                    video_groups = {}
                    for match in matches:
                        vid_id = match['source_video_id']
                        if vid_id not in video_groups:
                            video_groups[vid_id] = []
                        video_groups[vid_id].append(match)
                    
                    for vid_id, vid_matches in video_groups.items():
                        f.write(f"\nVideo: {vid_id}\n")
                        f.write(f"URL: {vid_matches[0]['source_video_url']}\n")
                        f.write("-" * 30 + "\n")
                        
                        for i, match in enumerate(vid_matches, 1):
                            timestamp_formatted = format_timestamp(match['start_time'])
                            f.write(f"{i:2d}. [{timestamp_formatted}] {match['text_segment']}\n")
                            f.write(f"    URL: {match['timestamp_url']}\n\n")
                    
                    f.write("\n")
        
        print(f"💾 Results saved to: {filename}")


def parse_args():
    """Parse command line arguments"""
    parser = argparse.ArgumentParser(
        description='YouTube Transcript Keyword Search Tool',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog='''
Examples:
  # Interactive mode (default)
  python youtube_transcript_search.py
  
  # Single video search
  python youtube_transcript_search.py -u "https://www.youtube.com/watch?v=VIDEO_ID" -k "keyword1,keyword2"
  
  # Batch search from file
  python youtube_transcript_search.py -f urls.txt -k "keyword1,keyword2" -o results.txt
  
  # Case sensitive search
  python youtube_transcript_search.py -u "https://www.youtube.com/watch?v=VIDEO_ID" -k "Keyword" --case-sensitive
'''
    )
    
    parser.add_argument('-u', '--url', help='Single YouTube video URL')
    parser.add_argument('-f', '--file', help='File containing YouTube URLs (one per line)')
    parser.add_argument('-k', '--keywords', help='Comma-separated keywords to search for')
    parser.add_argument('-c', '--case-sensitive', action='store_true', help='Enable case-sensitive search')
    parser.add_argument('-o', '--output', help='Output file path (auto-generated if not specified)')
    parser.add_argument('--no-save', action='store_true', help='Do not save results to file')
    
    return parser.parse_args()


def cli_mode(args):
    """Command-line interface mode"""
    
    # Get video URLs
    video_urls = []
    if args.url:
        video_urls = [args.url]
    elif args.file:
        try:
            with open(args.file, 'r', encoding='utf-8') as f:
                video_urls = [line.strip() for line in f if line.strip() and not line.strip().startswith('#')]
            print(f"✅ Loaded {len(video_urls)} URLs from {args.file}")
        except FileNotFoundError:
            print(f"❌ File not found: {args.file}")
            return
        except Exception as e:
            print(f"❌ Error reading file: {str(e)}")
            return
    else:
        print("❌ Either --url or --file must be specified in CLI mode")
        return
    
    # Get keywords
    if not args.keywords:
        print("❌ Keywords must be specified with --keywords in CLI mode")
        return
    
    keywords = [kw.strip() for kw in args.keywords.split(',') if kw.strip()]
    
    print(f"YouTube Transcript Search - CLI Mode")
    print(f"Videos to process: {len(video_urls)}")
    print(f"Keywords: {', '.join(keywords)}")
    print(f"Case sensitive: {args.case_sensitive}")
    print("=" * 50)
    
    # Process videos
    all_results = []
    successful_videos = 0
    failed_videos = []
    
    for i, video_url in enumerate(video_urls, 1):
        if len(video_urls) > 1:
            print(f"Processing Video {i}/{len(video_urls)}: {video_url}")
        
        video_data, error = process_single_video(video_url, keywords, args.case_sensitive, show_progress=(len(video_urls) == 1))
        
        if error:
            failed_videos.append({'url': video_url, 'error': error})
            if len(video_urls) > 1:
                print(f"❌ Failed: {error}")
            continue
        
        successful_videos += 1
        all_results.append(video_data)
        
        if len(video_urls) > 1:
            print(f"✅ Completed: {video_data['total_matches']} matches found\n")
    
    # Display summary
    total_matches = sum(video['total_matches'] for video in all_results)
    print(f"\nSummary:")
    print(f"Successfully processed: {successful_videos}/{len(video_urls)} videos")
    print(f"Total matches: {total_matches}")
    
    if failed_videos:
        print(f"Failed videos: {len(failed_videos)}")
        for failed in failed_videos:
            print(f"  • {failed['url']}: {failed['error']}")
    
    # Save results
    if not args.no_save and all_results:
        if args.output:
            filename = args.output
        elif len(video_urls) == 1:
            filename = f"transcript_search_results_{all_results[0]['video_id']}.txt"
        else:
            from datetime import datetime
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"transcript_search_batch_results_{timestamp}.txt"
        
        # Save logic (similar to main function)
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(f"YouTube Transcript Search Results (CLI Mode)\n")
            f.write(f"{'='*50}\n")
            f.write(f"Videos Processed: {len(all_results)}\n")
            f.write(f"Keywords: {', '.join(keywords)}\n")
            f.write(f"Case Sensitive: {args.case_sensitive}\n")
            f.write(f"Total Matches: {total_matches}\n")
            f.write(f"\n{'='*80}\n\n")
            
            # Combine all results
            all_combined_results = []
            for video_data in all_results:
                for result in video_data['results']:
                    result['source_video_id'] = video_data['video_id']
                    result['source_video_url'] = video_data['video_url']
                    all_combined_results.append(result)
            
            # Group by keyword
            keyword_groups = {}
            for result in all_combined_results:
                keyword = result['keyword']
                if keyword not in keyword_groups:
                    keyword_groups[keyword] = []
                keyword_groups[keyword].append(result)
            
            for keyword, matches in keyword_groups.items():
                f.write(f"Keyword: '{keyword}' ({len(matches)} matches)\n")
                f.write("=" * 50 + "\n")
                
                if len(video_urls) > 1:
                    # Group by video for batch results
                    video_groups = {}
                    for match in matches:
                        vid_id = match['source_video_id']
                        if vid_id not in video_groups:
                            video_groups[vid_id] = []
                        video_groups[vid_id].append(match)
                    
                    for vid_id, vid_matches in video_groups.items():
                        f.write(f"\nVideo: {vid_id}\n")
                        f.write(f"URL: {vid_matches[0]['source_video_url']}\n")
                        f.write("-" * 30 + "\n")
                        
                        for i, match in enumerate(vid_matches, 1):
                            timestamp_formatted = format_timestamp(match['start_time'])
                            f.write(f"{i:2d}. [{timestamp_formatted}] {match['text_segment']}\n")
                            f.write(f"    URL: {match['timestamp_url']}\n\n")
                else:
                    # Single video format
                    for i, match in enumerate(matches, 1):
                        timestamp_formatted = format_timestamp(match['start_time'])
                        f.write(f"{i:2d}. [{timestamp_formatted}] {match['text_segment']}\n")
                        f.write(f"    URL: {match['timestamp_url']}\n\n")
                
                f.write("\n")
        
        print(f"💾 Results saved to: {filename}")


if __name__ == "__main__":
    try:
        args = parse_args()
        
        # Check if any CLI arguments were provided
        if args.url or args.file or args.keywords:
            cli_mode(args)
        else:
            main()
    except KeyboardInterrupt:
        print("\n\n👋 Goodbye!")
    except Exception as e:
        print(f"\n❌ An error occurred: {str(e)}")