#!/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()