moldovan-qwen/scripts/extract_youtube_transcripts.py

#!/usr/bin/env python3
"""
extract_youtube_transcripts.py — Moldovan YouTube Transcript Extractor

Extracts subtitles and transcripts from Moldovan podcasts, stand-up shows,
interviews, and vlogs to build the raw training corpus for Moldovan Qwen.
"""

import os
import sys
import re
import json
import time
import argparse
from typing import List, Dict, Optional
from urllib.parse import urlparse, parse_qs

try:
    from youtube_transcript_api import YouTubeTranscriptApi, TranscriptsDisabled, NoTranscriptFound
except ImportError:
    print("Please install youtube-transcript-api: pip install youtube-transcript-api")
    sys.exit(1)

# Curated seed list of top Moldovan YouTube channels & videos
MOLDOVAN_CHANNELS_SEED = [
    {
        "channel": "Dorin Galben",
        "category": "Interviews & Podcasts",
        "sample_video_ids": [
            "dQw4w9WgXcQ",  # Placeholder IDs will be populated or passed via CLI
        ]
    },
    {
        "channel": "Standupovka",
        "category": "Stand-Up Comedy & Slang",
        "sample_video_ids": []
    },
    {
        "channel": "Banca de Bancuri",
        "category": "Popular Humor & Folklore",
        "sample_video_ids": []
    },
    {
        "channel": "Zebra Show",
        "category": "Satire & Sketches",
        "sample_video_ids": []
    },
    {
        "channel": "Titania Podcast",
        "category": "Urban Culture & Youth Talk",
        "sample_video_ids": []
    }
]

def extract_video_id(url_or_id: str) -> Optional[str]:
    """Extract 11-char YouTube video ID from various URL formats."""
    url_or_id = url_or_id.strip()
    if len(url_or_id) == 11 and re.match(r'^[a-zA-Z0-9_-]{11}$', url_or_id):
        return url_or_id
    
    parsed = urlparse(url_or_id)
    if parsed.hostname in ('youtu.be', 'www.youtu.be'):
        return parsed.path.lstrip('/')
    if parsed.hostname in ('youtube.com', 'www.youtube.com'):
        if parsed.path == '/watch':
            return parse_qs(parsed.query).get('v', [None])[0]
        if parsed.path.startswith('/embed/'):
            return parsed.path.split('/')[2]
        if parsed.path.startswith('/v/'):
            return parsed.path.split('/')[2]
        if parsed.path.startswith('/shorts/'):
            return parsed.path.split('/')[2]
    return None

def fetch_transcript(video_id: str, languages: List[str] = None) -> Optional[Dict]:
    """Fetch transcript for a given video ID with language fallback."""
    if languages is None:
        languages = ['ro', 'ro-MD', 'mo', 'ru', 'en']

    try:
        api = YouTubeTranscriptApi()
        try:
            transcript_list = api.list(video_id)
        except Exception:
            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)

        # Priority: Manual Romanian -> Auto Romanian -> Manual Russian -> Auto Russian
        transcript = None
        used_lang = None

        for lang in languages:
            try:
                transcript = transcript_list.find_transcript([lang])
                used_lang = lang
                break
            except Exception:
                continue

        # If no direct match, try finding any generated transcript
        if not transcript:
            try:
                transcript = transcript_list.find_generated_transcript(languages)
                used_lang = getattr(transcript, 'language_code', 'ro')
            except Exception:
                pass

        if not transcript:
            # Fallback to the first available transcript
            for t in transcript_list:
                transcript = t
                used_lang = getattr(t, 'language_code', 'unknown')
                break

        if not transcript:
            return None

        data = transcript.fetch()

        # Merge segments into readable paragraphs
        full_text_chunks = []
        current_chunk = []
        current_word_count = 0

        for item in data:
            if hasattr(item, 'text'):
                text = str(item.text).strip()
            elif isinstance(item, dict):
                text = item.get('text', '').strip()
            else:
                text = str(item).strip()

            current_word_count += len(text.split())
            
            # Group into ~100-word logical paragraphs
            if current_word_count >= 100 or text.endswith(('.', '?', '!')):
                full_text_chunks.append(' '.join(current_chunk))
                current_chunk = []
                current_word_count = 0
                
        if current_chunk:
            full_text_chunks.append(' '.join(current_chunk))
            
        merged_text = '\n\n'.join(full_text_chunks)
        
        return {
            "video_id": video_id,
            "language": used_lang,
            "is_generated": transcript.is_generated,
            "raw_segments_count": len(data),
            "paragraphs_count": len(full_text_chunks),
            "word_count": len(merged_text.split()),
            "full_text": merged_text,
            "extracted_at": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        
    except (TranscriptsDisabled, NoTranscriptFound) as e:
        print(f"[-] Subtitles not available for video {video_id}: {e}")
        return None
    except Exception as e:
        print(f"[!] Error fetching transcript for {video_id}: {e}")
        return None

def main():
    parser = argparse.ArgumentParser(description="Extract YouTube transcripts for Moldovan corpus")
    parser.add_argument("--urls", nargs="+", help="List of YouTube URLs or video IDs to transcribe")
    parser.add_argument("--file", help="File with list of YouTube URLs/IDs (one per line)")
    parser.add_argument("--output-dir", default="moldovan-qwen/data/raw_transcripts", help="Output directory")
    args = parser.parse_args()

    os.makedirs(args.output_dir, exist_ok=True)
    
    video_ids = []
    if args.urls:
        for u in args.urls:
            vid = extract_video_id(u)
            if vid:
                video_ids.append(vid)
            else:
                print(f"[!] Warning: Could not parse video ID from: {u}")
                
    if args.file and os.path.exists(args.file):
        with open(args.file, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    vid = extract_video_id(line)
                    if vid:
                        video_ids.append(vid)
                        
    if not video_ids:
        print("No videos specified. Example usage:")
        print("  python3 scripts/extract_youtube_transcripts.py --urls https://www.youtube.com/watch?v=VIDEO_ID")
        print("  python3 scripts/extract_youtube_transcripts.py --file data/video_sources.txt")
        return

    print(f"[*] Found {len(video_ids)} video IDs to process.")
    results = []
    
    for idx, vid in enumerate(video_ids, 1):
        print(f"[{idx}/{len(video_ids)}] Fetching transcript for video: {vid}...")
        res = fetch_transcript(vid)
        if res:
            out_file = os.path.join(args.output_dir, f"{vid}.json")
            with open(out_file, 'w', encoding='utf-8') as f:
                json.dump(res, f, ensure_ascii=False, indent=2)
            print(f"    [+] Saved transcript ({res['word_count']} words, lang: {res['language']}) -> {out_file}")
            results.append(res)
        time.sleep(1) # Gentle delay between requests

    print(f"\n[✓] Successfully processed {len(results)}/{len(video_ids)} transcripts.")

if __name__ == "__main__":
    main()