#!/usr/bin/env python3 """ youtube-transcribe.py 범용 유튜브 음성 추출 CLI 스크립트 yt-dlp로 유튜브 영상에서 오디오를 다운로드하고, 로컬 Whisper GPU 서비스(http://localhost:8200/v1/transcribe)에 multipart/form-data로 전송하여 전사 결과를 반환합니다. 사용법: python3 youtube-transcribe.py --url "https://youtube.com/watch?v=..." python3 youtube-transcribe.py --url "..." --format json python3 youtube-transcribe.py --url "..." --format srt --output /path/to/out.srt python3 youtube-transcribe.py --url "..." --format text --language en """ import argparse import shutil import sys import tempfile from pathlib import Path from typing import TYPE_CHECKING, Any, cast import requests import yt_dlp if TYPE_CHECKING: from yt_dlp import _Params # 로컬 Whisper GPU 서비스 엔드포인트 WHISPER_SERVICE_URL = "http://localhost:8200/v1/transcribe" # HTTP 요청 타임아웃 (초) – 긴 영상 대응 REQUEST_TIMEOUT = 600 # --------------------------------------------------------------------------- # CLI 인자 파싱 # --------------------------------------------------------------------------- def parse_args(argv: list[str] | None = None) -> argparse.Namespace: """CLI 인자를 파싱하여 Namespace 반환.""" parser = argparse.ArgumentParser( description="유튜브 영상 음성 추출 및 Whisper 전사 CLI", formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--url", required=True, help="YouTube URL (필수)", ) parser.add_argument( "--format", choices=["text", "json", "srt"], default="text", help="출력 형식 (text|json|srt, 기본: text)", ) parser.add_argument( "--output", default=None, help="출력 파일 경로 (없으면 stdout)", ) parser.add_argument( "--language", default="ko", help="언어 코드 (기본: ko)", ) return parser.parse_args(argv) # --------------------------------------------------------------------------- # 오디오 다운로드 # --------------------------------------------------------------------------- def download_audio(url: str, output_dir: str) -> str: """ yt-dlp로 YouTube URL에서 오디오를 WAV 포맷으로 다운로드. Args: url: YouTube 영상 URL output_dir: 임시 디렉토리 경로 Returns: 다운로드된 WAV 파일의 절대 경로 Raises: Exception: yt-dlp 다운로드 실패 시 """ ydl_opts = cast( "yt_dlp._Params", { "format": "bestaudio/best", "outtmpl": str(Path(output_dir) / "%(title)s.%(ext)s"), "postprocessors": [ { "key": "FFmpegExtractAudio", "preferredcodec": "wav", } ], "quiet": True, "no_warnings": True, }, ) with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=False) filename = ydl.prepare_filename(info) ydl.download([url]) # .wav 확장자로 변환된 파일 탐색 stem = Path(filename).stem wav_path = Path(output_dir) / f"{stem}.wav" if wav_path.exists(): return str(wav_path) # fallback: 디렉토리에서 .wav 파일 검색 wav_files = list(Path(output_dir).glob("*.wav")) if wav_files: return str(wav_files[0]) raise FileNotFoundError(f"WAV 파일을 찾을 수 없습니다: {output_dir}") # --------------------------------------------------------------------------- # Whisper 전사 # --------------------------------------------------------------------------- def transcribe_audio(audio_path: str, language: str = "ko") -> dict[str, Any]: """ 로컬 Whisper GPU 서비스에 오디오 파일을 전송하여 전사 결과 반환. 연결 실패 또는 타임아웃 시 경고 로그를 출력하고 fallback 메시지를 포함한 dict를 반환합니다. Args: audio_path: 오디오 파일 경로 language: 언어 코드 (기본: ko) Returns: 전사 결과 dict {text, segments} 또는 fallback dict """ try: with open(audio_path, "rb") as audio_file: files = {"file": (Path(audio_path).name, audio_file, "audio/wav")} data = {"language": language} response = requests.post( WHISPER_SERVICE_URL, files=files, data=data, timeout=REQUEST_TIMEOUT, ) response.raise_for_status() return response.json() except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc: print( f"[경고] 로컬 Whisper 서비스 연결 실패: {exc}", file=sys.stderr, ) return { "text": "로컬 Whisper 서비스가 응답하지 않습니다", "segments": [], } # --------------------------------------------------------------------------- # 출력 형식 변환 # --------------------------------------------------------------------------- def _seconds_to_srt_time(seconds: float) -> str: """초를 SRT 타임코드 형식(HH:MM:SS,mmm)으로 변환.""" hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) secs = int(seconds % 60) millis = int(round((seconds - int(seconds)) * 1000)) return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}" def format_output(result: dict[str, Any], fmt: str) -> str: """ 전사 결과를 지정된 형식으로 문자열 변환. Args: result: 전사 결과 dict fmt: 출력 형식 (text|json|srt) Returns: 형식화된 문자열 """ import json as _json if fmt == "json": return _json.dumps(result, ensure_ascii=False, indent=2) if fmt == "srt": segments = result.get("segments", []) lines: list[str] = [] for idx, seg in enumerate(segments, start=1): start = _seconds_to_srt_time(seg.get("start", 0.0)) end = _seconds_to_srt_time(seg.get("end", 0.0)) text = seg.get("text", "").strip() lines.append(f"{idx}\n{start} --> {end}\n{text}\n") return "\n".join(lines) # default: text return result.get("text", "") # --------------------------------------------------------------------------- # 메인 진입점 # --------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> None: """CLI 메인 함수.""" args = parse_args(argv) tmp_dir = tempfile.mkdtemp(prefix="yt_transcribe_") try: # 1. 오디오 다운로드 audio_path = download_audio(args.url, tmp_dir) # 2. Whisper 전사 result = transcribe_audio(audio_path, language=args.language) # 3. 형식 변환 output_text = format_output(result, args.format) # 4. 출력 (파일 또는 stdout) if args.output: Path(args.output).write_text(output_text, encoding="utf-8") else: print(output_text) except Exception as exc: print(f"[ERROR] {exc}", file=sys.stderr) sys.exit(1) finally: shutil.rmtree(tmp_dir, ignore_errors=True) if __name__ == "__main__": main()