yt_transcribe.py

#!/usr/bin/env python3
"""Save YouTube audio and create local text transcripts."""

from __future__ import annotations

import argparse
import re
import shutil
import subprocess
import sys
import tempfile
import unicodedata
from datetime import datetime
from pathlib import Path
from urllib.parse import parse_qs, urlencode, urlparse

__version__ = "1.0.4"

DEFAULT_OUTPUT_DIR = Path.home() / "Downloads" / "YouTube Transcriber"
YOUTUBE_HOSTS = {
    "youtube.com",
    "www.youtube.com",
    "m.youtube.com",
    "music.youtube.com",
    "youtu.be",
    "www.youtube-nocookie.com",
    "youtube-nocookie.com",
}


class CliError(RuntimeError):
    """An expected error that should be shown without a traceback."""


class CommandError(RuntimeError):
    """An external command failed."""

    def __init__(self, action: str, stderr: str):
        super().__init__(action)
        self.action = action
        self.stderr = stderr.strip()


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="yt-transcribe",
        description=(
            "Save YouTube audio and create local text transcripts with faster-whisper."
        ),
    )
    parser.add_argument("url", help="YouTube video URL")
    parser.add_argument(
        "-o",
        "--output-dir",
        default=DEFAULT_OUTPUT_DIR,
        help="output directory (default: ~/Downloads/YouTube Transcriber)",
    )
    parser.add_argument(
        "--mode",
        choices=("audio-only", "transcript-only", "both"),
        default="both",
        help="files to keep (default: both)",
    )
    parser.add_argument(
        "--audio-format",
        choices=("mp3", "wav"),
        default="mp3",
        help="saved audio format (default: mp3)",
    )
    parser.add_argument(
        "--model",
        default="base",
        help="faster-whisper model name or path (default: base)",
    )
    parser.add_argument(
        "--language",
        "--lang",
        dest="language",
        default=None,
        help="spoken language code; omit for automatic detection",
    )
    parser.add_argument(
        "--device",
        choices=("cpu", "cuda"),
        default="cpu",
        help="transcription device (default: cpu)",
    )
    parser.add_argument(
        "--compute-type",
        "--compute",
        dest="compute_type",
        default="int8",
        help="faster-whisper compute type (default: int8)",
    )
    parser.add_argument(
        "--timestamps",
        action="store_true",
        help="include segment timestamps in transcripts",
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {__version__}",
    )
    return parser


def validate_youtube_url(value: str) -> str:
    try:
        parsed = urlparse(value)
    except ValueError as exc:
        raise CliError("Invalid YouTube URL.") from exc

    host = (parsed.hostname or "").lower()
    if parsed.scheme not in {"http", "https"} or host not in YOUTUBE_HOSTS:
        raise CliError(
            "Invalid YouTube URL. Provide an http or https URL from youtube.com "
            "or youtu.be."
        )

    query = parse_qs(parsed.query)
    if host == "youtu.be":
        has_video = bool(parsed.path.strip("/"))
    elif parsed.path == "/watch":
        video_id = query.get("v", [""])[0]
        has_video = bool(video_id)
    else:
        parts = [part for part in parsed.path.split("/") if part]
        has_video = len(parts) >= 2 and parts[0] in {"embed", "live", "shorts"}

    if not has_video:
        if parsed.path.rstrip("/") == "/playlist" or query.get("list"):
            raise CliError(
                "Playlist URLs are not supported yet. Please paste the URL of "
                "an individual video."
            )
        raise CliError("Invalid YouTube URL. The URL does not identify a video.")

    if host != "youtu.be" and parsed.path == "/watch":
        return f"https://www.youtube.com/watch?{urlencode({'v': video_id})}"
    return value


def sanitize_filename(title: str, max_length: int = 100) -> str:
    normalized = unicodedata.normalize("NFKD", title)
    ascii_title = normalized.encode("ascii", "ignore").decode("ascii")
    safe = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "", ascii_title)
    safe = re.sub(r"[\s._-]+", "-", safe).strip("-. ").lower()
    safe = safe[:max_length].rstrip("-. ")
    return safe or "youtube-transcript"


def create_output_directories(base_dir: Path) -> tuple[Path, Path]:
    audio_dir = base_dir / "audio"
    transcript_dir = base_dir / "transcripts"
    try:
        audio_dir.mkdir(parents=True, exist_ok=True)
        transcript_dir.mkdir(parents=True, exist_ok=True)
    except PermissionError as exc:
        raise CliError(f"Cannot create output directory: {base_dir}") from exc
    except OSError as exc:
        raise CliError(f"Cannot create output directory {base_dir}: {exc}") from exc
    return audio_dir, transcript_dir


def require_command(name: str, install_hint: str) -> str:
    command = shutil.which(name)
    if command is None:
        raise CliError(f"Missing dependency: {name}. {install_hint}")
    return command


def require_transcription_engine() -> None:
    try:
        import faster_whisper  # noqa: F401
    except ImportError as exc:
        raise CliError(
            "Missing dependency: faster-whisper. Reinstall this package with "
            "'python -m pip install .'."
        ) from exc


def run_command(command: list[str], action: str) -> str:
    try:
        result = subprocess.run(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            check=False,
        )
    except FileNotFoundError as exc:
        raise CliError(f"Missing dependency: {command[0]}.") from exc
    except OSError as exc:
        raise CliError(f"Could not start {command[0]}: {exc}") from exc

    if result.returncode != 0:
        raise CommandError(action, result.stderr)
    return result.stdout.strip()


def explain_video_error(error: CommandError) -> CliError:
    detail = error.stderr.lower()
    if any(
        marker in detail
        for marker in (
            "sign in to confirm you're not a bot",
            "sign in to confirm you’re not a bot",
            "confirm you are not a bot",
        )
    ):
        return CliError(
            "YouTube requested browser verification for this video.\n\n"
            "Possible solutions:\n\n"
            "• Sign into YouTube in Chrome and try again.\n"
            "• Retry later if YouTube is temporarily rate limiting requests.\n"
            "• If necessary, rerun using yt-dlp browser cookies.\n\n"
            "The transcription tool is functioning correctly.\n"
            "This message comes from YouTube's access restrictions."
        )
    if "private video" in detail or "this video is private" in detail:
        return CliError("The video is private and cannot be accessed.")
    if any(
        marker in detail
        for marker in (
            "video unavailable",
            "this video is unavailable",
            "has been removed",
            "not available in your country",
            "members-only",
        )
    ):
        return CliError("The video is unavailable.")
    return CliError(
        f"{error.action} failed. Check the URL and your network connection."
    )


def get_video_title(yt_dlp: str, url: str) -> str:
    try:
        title = run_command(
            [
                yt_dlp,
                "--no-playlist",
                "--skip-download",
                "--print",
                "%(title)s",
                url,
            ],
            "Video lookup",
        )
    except CommandError as exc:
        raise explain_video_error(exc) from exc
    return title.splitlines()[-1].strip() if title else "youtube-transcript"


def download_audio(yt_dlp: str, url: str, temporary_dir: Path) -> Path:
    template = str(temporary_dir / "source.%(ext)s")
    try:
        run_command(
            [
                yt_dlp,
                "--no-playlist",
                "-f",
                "bestaudio/best",
                "-o",
                template,
                url,
            ],
            "Audio download",
        )
    except CommandError as exc:
        raise explain_video_error(exc) from exc

    candidates = [
        path
        for path in temporary_dir.glob("source.*")
        if path.is_file() and not path.name.endswith((".part", ".ytdl"))
    ]
    if not candidates:
        raise CliError("Audio download failed: yt-dlp did not create an audio file.")
    return max(candidates, key=lambda path: path.stat().st_mtime)


def convert_audio(
    ffmpeg: str, source: Path, destination: Path, audio_format: str
) -> None:
    if audio_format == "mp3":
        encoding_options = ["-vn", "-b:a", "192k"]
    else:
        encoding_options = ["-vn", "-ac", "1", "-ar", "16000"]

    try:
        run_command(
            [
                ffmpeg,
                "-nostdin",
                "-hide_banner",
                "-loglevel",
                "error",
                "-n",
                "-i",
                str(source),
                *encoding_options,
                str(destination),
            ],
            "Audio conversion",
        )
    except CommandError as exc:
        if "permission denied" in exc.stderr.lower():
            raise CliError(f"Cannot write output file: {destination}") from exc
        raise CliError(
            "Audio conversion failed. Verify that ffmpeg is working."
        ) from exc


def transcribe_audio(
    wav_path: Path,
    model_size: str,
    language: str | None,
    device: str,
    compute_type: str,
):
    try:
        from faster_whisper import WhisperModel

        model = WhisperModel(
            model_size,
            device=device,
            compute_type=compute_type,
        )
        segments, _ = model.transcribe(
            str(wav_path),
            language=language,
            vad_filter=True,
            beam_size=5,
        )
        return list(segments)
    except Exception as exc:
        raise CliError(
            "Transcription failed. Check the model name, device, available memory, "
            "and network access for the first model download."
        ) from exc


def verify_audio_file(path: Path) -> None:
    try:
        is_valid = path.is_file() and path.stat().st_size > 0
    except OSError as exc:
        raise CliError(f"Could not verify audio file: {path}") from exc
    if not is_valid:
        raise CliError(f"Audio conversion failed: no audio file was created at {path}")


def quote_for_shell(value: str) -> str:
    escaped = (
        value.replace("\\", "\\\\")
        .replace('"', '\\"')
        .replace("$", "\\$")
        .replace("`", "\\`")
    )
    return f'"{escaped}"'


def format_timestamp(seconds: float) -> str:
    total_seconds = max(0, int(seconds))
    hours = total_seconds // 3600
    minutes = (total_seconds % 3600) // 60
    remainder = total_seconds % 60
    return f"{hours:02d}:{minutes:02d}:{remainder:02d}"


def choose_output_paths(
    audio_dir: Path,
    transcript_dir: Path,
    stem: str,
    audio_format: str,
    mode: str,
) -> tuple[Path, Path, Path]:
    number = 1
    while True:
        candidate = stem if number == 1 else f"{stem}-{number}"
        audio_path = audio_dir / f"{candidate}.{audio_format}"
        text_path = transcript_dir / f"{candidate}.txt"
        markdown_path = transcript_dir / f"{candidate}.md"
        expected = []
        if mode in {"audio-only", "both"}:
            expected.append(audio_path)
        if mode in {"transcript-only", "both"}:
            expected.extend((text_path, markdown_path))
        if not any(path.exists() for path in expected):
            return audio_path, text_path, markdown_path
        number += 1


def write_transcripts(
    title: str,
    url: str,
    segments,
    timestamps: bool,
    text_path: Path,
    markdown_path: Path,
) -> None:
    lines: list[str] = []
    for segment in segments:
        text = segment.text.strip()
        if not text:
            continue
        if timestamps:
            start = format_timestamp(segment.start)
            end = format_timestamp(segment.end)
            lines.append(f"[{start} -> {end}] {text}")
        else:
            lines.append(text)

    transcript = "\n".join(lines).strip()
    generated = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M %Z")
    markdown = "\n".join(
        [
            f"# {title}",
            "",
            f"Source: {url}",
            f"Generated: {generated}",
            "",
            transcript,
            "",
        ]
    )
    try:
        text_path.write_text(f"{transcript}\n", encoding="utf-8")
        markdown_path.write_text(markdown, encoding="utf-8")
    except PermissionError as exc:
        text_path.unlink(missing_ok=True)
        markdown_path.unlink(missing_ok=True)
        raise CliError(f"Cannot write transcript files in: {text_path.parent}") from exc
    except OSError as exc:
        text_path.unlink(missing_ok=True)
        markdown_path.unlink(missing_ok=True)
        raise CliError(f"Could not write transcript files: {exc}") from exc


def run(args: argparse.Namespace) -> None:
    video_url = validate_youtube_url(args.url)
    yt_dlp = require_command(
        "yt-dlp",
        "Install the project dependencies with 'python -m pip install .'.",
    )
    ffmpeg = require_command(
        "ffmpeg",
        "Install ffmpeg with your operating system package manager.",
    )
    if args.mode != "audio-only":
        require_transcription_engine()

    output_dir = Path(args.output_dir).expanduser().resolve()
    audio_dir, transcript_dir = create_output_directories(output_dir)
    title = get_video_title(yt_dlp, video_url)
    stem = sanitize_filename(title)
    audio_path, text_path, markdown_path = choose_output_paths(
        audio_dir,
        transcript_dir,
        stem,
        args.audio_format,
        args.mode,
    )

    with tempfile.TemporaryDirectory(prefix="yt-transcribe-") as temporary:
        temporary_dir = Path(temporary)
        print("Downloading audio...")
        source_audio = download_audio(yt_dlp, video_url, temporary_dir)

        if args.mode in {"audio-only", "both"}:
            convert_audio(ffmpeg, source_audio, audio_path, args.audio_format)
            verify_audio_file(audio_path)

        if args.mode in {"transcript-only", "both"}:
            wav_path = temporary_dir / "transcription.wav"
            convert_audio(ffmpeg, source_audio, wav_path, "wav")
            verify_audio_file(wav_path)

        print("✓ Audio download complete.")

        if args.mode in {"transcript-only", "both"}:
            print()
            print("Preparing transcription model...")
            print("Transcribing...")
            segments = transcribe_audio(
                wav_path,
                args.model,
                args.language,
                args.device,
                args.compute_type,
            )
            write_transcripts(
                title,
                args.url,
                segments,
                args.timestamps,
                text_path,
                markdown_path,
            )
    print("✓ Transcription complete.")
    print()
    if args.mode in {"audio-only", "both"}:
        print(f"Saved audio: {audio_path.resolve()}")
    if args.mode in {"transcript-only", "both"}:
        print(f"Saved text transcript: {text_path.resolve()}")
        print(f"Saved Markdown transcript: {markdown_path.resolve()}")
    print()
    print("To open this folder in Finder, run:")
    print()
    print(f"open {quote_for_shell(str(output_dir))}")


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        run(args)
    except CliError as exc:
        parser.exit(1, f"yt-transcribe: error: {exc}\n")
    return 0


if __name__ == "__main__":
    sys.exit(main())