yt_script_extract.py

import ctypes
import html
import logging
import re
import tempfile
import time
from datetime import datetime
from pathlib import Path
from urllib.error import HTTPError
from urllib.error import URLError
from urllib.parse import parse_qsl
from urllib.parse import urlencode
from urllib.parse import urlsplit
from urllib.parse import urlunsplit
from urllib.request import Request
from urllib.request import urlopen

from yt_dlp import DownloadError, YoutubeDL


APP_TITLE = "YouTube \uc2a4\ud06c\ub9bd\ud2b8 \ucd94\ucd9c"
MSG_NO_URL = (
    "\ud074\ub9bd\ubcf4\ub4dc\uc5d0 \uc720\ud29c\ube0c \ub9c1\ud06c\uac00 "
    "\uc5c6\uc2b5\ub2c8\ub2e4. \ub9c1\ud06c\ub97c \uba3c\uc800 "
    "\ubcf5\uc0ac\ud574 \uc8fc\uc138\uc694."
)
MSG_CLIPBOARD_READ_FAILED = (
    "\ud074\ub9bd\ubcf4\ub4dc\ub97c \uc77d\uc9c0 \ubabb\ud588\uc2b5\ub2c8\ub2e4. "
    "\uc720\ud29c\ube0c \ub9c1\ud06c\ub97c \ub2e4\uc2dc \ubcf5\uc0ac\ud55c \ub4a4 "
    "\uc2e4\ud589\ud574 \uc8fc\uc138\uc694."
)
MSG_NO_SUBTITLE = (
    "\uc774 \uc601\uc0c1\uc5d0\uc11c \uac00\uc838\uc62c \uc218 \uc788\ub294 "
    "\uc790\ub9c9\uc744 \ucc3e\uc9c0 \ubabb\ud588\uc2b5\ub2c8\ub2e4."
)
MSG_DOWNLOAD_FAILED = (
    "\uc790\ub9c9 \ucd94\ucd9c\uc5d0 \uc2e4\ud328\ud588\uc2b5\ub2c8\ub2e4. "
    "\uc601\uc0c1\uc5d0 \uc790\ub9c9\uc774 \uc5c6\uac70\ub098, "
    "\uc720\ud29c\ube0c \uc811\uc18d\uc774 \ucc28\ub2e8\ub418\uc5c8\uc744 "
    "\uc218 \uc788\uc2b5\ub2c8\ub2e4."
)
MSG_NETWORK_FAILED = (
    "\uc778\ud130\ub137 \uc5f0\uacb0\uc744 \ud655\uc778\ud55c \ub4a4 "
    "\ub2e4\uc2dc \uc2dc\ub3c4\ud574 \uc8fc\uc138\uc694."
)
MSG_COPY_FAILED = (
    "\uc790\ub9c9\uc740 \ucd94\ucd9c\ud588\uc9c0\ub9cc \ud074\ub9bd\ubcf4\ub4dc\uc5d0 "
    "\ubcf5\uc0ac\ud558\uc9c0 \ubabb\ud588\uc2b5\ub2c8\ub2e4."
)
MSG_FILE_FAILED = (
    "\ud30c\uc77c \ucc98\ub9ac \uc911 \ubb38\uc81c\uac00 \ubc1c\uc0dd\ud588\uc2b5\ub2c8\ub2e4. "
    "\uc7a0\uc2dc \ud6c4 \ub2e4\uc2dc \uc2dc\ub3c4\ud574 \uc8fc\uc138\uc694."
)
MSG_DONE = "\uc2a4\ud06c\ub9bd\ud2b8\ub97c \ud074\ub9bd\ubcf4\ub4dc\uc5d0 \ubcf5\uc0ac\ud588\uc2b5\ub2c8\ub2e4."
LABEL_NO_TITLE = "\uc81c\ubaa9 \uc5c6\uc74c"
LABEL_TITLE = "\uc81c\ubaa9"
LABEL_EXTRACTED_AT = "\ucd94\ucd9c\uc77c\uc2dc"
LABEL_LANGUAGE = "\uc5b8\uc5b4"
LOG_PATH = Path(tempfile.gettempdir()) / "YouTubeScriptExtractor.log"

YOUTUBE_URL_RE = re.compile(
    r"^https?://("
    r"(www\.)?youtube\.com/watch\?[^ \n\r\t]*v=[^ \n\r\t&]+"
    r"|m\.youtube\.com/watch\?[^ \n\r\t]*v=[^ \n\r\t&]+"
    r"|youtu\.be/[^ \n\r\t/?#]+"
    r"|youtube\.com/shorts/[^ \n\r\t/?#]+"
    r"|www\.youtube\.com/shorts/[^ \n\r\t/?#]+"
    r")",
    re.IGNORECASE,
)
LANGUAGE_PRIORITY = ("ko", "en")
TIMESTAMP_RE = re.compile(
    r"^\d{2}:\d{2}:\d{2}\.\d{3}\s+-->\s+\d{2}:\d{2}:\d{2}\.\d{3}"
)
TAG_RE = re.compile(r"<[^>]+>")
HTTP_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/125.0 Safari/537.36"
    ),
}
CF_UNICODETEXT = 13
GMEM_MOVEABLE = 0x0002

user32 = ctypes.windll.user32
kernel32 = ctypes.windll.kernel32

user32.OpenClipboard.argtypes = [ctypes.c_void_p]
user32.OpenClipboard.restype = ctypes.c_bool
user32.CloseClipboard.argtypes = []
user32.CloseClipboard.restype = ctypes.c_bool
user32.EmptyClipboard.argtypes = []
user32.EmptyClipboard.restype = ctypes.c_bool
user32.GetClipboardData.argtypes = [ctypes.c_uint]
user32.GetClipboardData.restype = ctypes.c_void_p
user32.SetClipboardData.argtypes = [ctypes.c_uint, ctypes.c_void_p]
user32.SetClipboardData.restype = ctypes.c_void_p
user32.MessageBoxW.argtypes = [ctypes.c_void_p, ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint]
user32.MessageBoxW.restype = ctypes.c_int
kernel32.GlobalAlloc.argtypes = [ctypes.c_uint, ctypes.c_size_t]
kernel32.GlobalAlloc.restype = ctypes.c_void_p
kernel32.GlobalLock.argtypes = [ctypes.c_void_p]
kernel32.GlobalLock.restype = ctypes.c_void_p
kernel32.GlobalUnlock.argtypes = [ctypes.c_void_p]
kernel32.GlobalUnlock.restype = ctypes.c_bool
kernel32.GlobalFree.argtypes = [ctypes.c_void_p]
kernel32.GlobalFree.restype = ctypes.c_void_p


class UserFacingError(Exception):
    pass


logging.basicConfig(
    filename=LOG_PATH,
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    encoding="utf-8",
)


def show_error(message: str) -> None:
    user32.MessageBoxW(None, message, APP_TITLE, 0x10)


def open_clipboard() -> None:
    for _ in range(10):
        if user32.OpenClipboard(None):
            return
        time.sleep(0.05)
    raise UserFacingError(MSG_CLIPBOARD_READ_FAILED)


def read_clipboard_text() -> str:
    open_clipboard()
    try:
        handle = user32.GetClipboardData(CF_UNICODETEXT)
        if not handle:
            raise UserFacingError(MSG_CLIPBOARD_READ_FAILED)

        pointer = kernel32.GlobalLock(handle)
        if not pointer:
            raise UserFacingError(MSG_CLIPBOARD_READ_FAILED)

        try:
            value = ctypes.wstring_at(pointer)
        finally:
            kernel32.GlobalUnlock(handle)

        logging.info("Read clipboard text length=%s", len(value))
        return value
    finally:
        user32.CloseClipboard()


def write_clipboard_text(text: str) -> None:
    data = text.encode("utf-16-le") + b"\x00\x00"
    handle = kernel32.GlobalAlloc(GMEM_MOVEABLE, len(data))
    if not handle:
        raise UserFacingError(MSG_COPY_FAILED)

    pointer = kernel32.GlobalLock(handle)
    if not pointer:
        kernel32.GlobalFree(handle)
        raise UserFacingError(MSG_COPY_FAILED)

    try:
        ctypes.memmove(pointer, data, len(data))
    finally:
        kernel32.GlobalUnlock(handle)

    try:
        open_clipboard()
    except UserFacingError:
        kernel32.GlobalFree(handle)
        raise UserFacingError(MSG_COPY_FAILED)

    try:
        user32.EmptyClipboard()
        if not user32.SetClipboardData(CF_UNICODETEXT, handle):
            kernel32.GlobalFree(handle)
            raise UserFacingError(MSG_COPY_FAILED)
        handle = None
        logging.info("Wrote clipboard text length=%s", len(text))
    finally:
        user32.CloseClipboard()
        if handle:
            kernel32.GlobalFree(handle)


def show_auto_close_notice(message: str, milliseconds: int = 650) -> None:
    message_box_timeout = getattr(user32, "MessageBoxTimeoutW", None)
    if not message_box_timeout:
        return

    message_box_timeout.argtypes = [
        ctypes.c_void_p,
        ctypes.c_wchar_p,
        ctypes.c_wchar_p,
        ctypes.c_uint,
        ctypes.c_ushort,
        ctypes.c_uint,
    ]
    message_box_timeout.restype = ctypes.c_int
    message_box_timeout(None, message, APP_TITLE, 0x40, 0, milliseconds)


def get_clipboard_youtube_url() -> str:
    value = read_clipboard_text().strip()

    if not YOUTUBE_URL_RE.match(value):
        logging.info("Clipboard did not match YouTube URL: %r", value[:200])
        raise UserFacingError(MSG_NO_URL)

    logging.info("Accepted YouTube URL: %s", value)
    return value


def pick_subtitle(info: dict) -> tuple[str, bool]:
    subtitles = info.get("subtitles") or {}
    automatic = info.get("automatic_captions") or {}
    logging.info(
        "Available subtitles=%s automatic=%s",
        sorted(subtitles.keys()),
        sorted(automatic.keys()),
    )

    for lang in LANGUAGE_PRIORITY:
        if lang in subtitles:
            logging.info("Picked manual subtitle lang=%s", lang)
            return lang, False
        if lang in automatic:
            logging.info("Picked automatic subtitle lang=%s", lang)
            return lang, True

    for lang in subtitles:
        logging.info("Picked fallback manual subtitle lang=%s", lang)
        return lang, False
    for lang in automatic:
        logging.info("Picked fallback automatic subtitle lang=%s", lang)
        return lang, True

    raise UserFacingError(MSG_NO_SUBTITLE)


def iter_subtitle_candidates(info: dict) -> list[tuple[str, bool, list[dict]]]:
    subtitles = info.get("subtitles") or {}
    automatic = info.get("automatic_captions") or {}
    candidates: list[tuple[str, bool, list[dict]]] = []
    seen: set[tuple[str, bool]] = set()

    def add(lang: str, automatic_subtitle: bool) -> None:
        key = (lang, automatic_subtitle)
        if key in seen:
            return
        source = automatic if automatic_subtitle else subtitles
        tracks = source.get(lang) or []
        if tracks:
            candidates.append((lang, automatic_subtitle, tracks))
            seen.add(key)

    for lang in LANGUAGE_PRIORITY:
        add(lang, False)
        add(lang, True)

    for lang in subtitles:
        add(lang, False)
    for lang in automatic:
        add(lang, True)

    return candidates


def prefer_vtt_tracks(tracks: list[dict]) -> list[dict]:
    return sorted(tracks, key=lambda track: 0 if track.get("ext") == "vtt" else 1)


def force_vtt_url(url: str) -> str:
    parts = urlsplit(url)
    query = dict(parse_qsl(parts.query, keep_blank_values=True))
    query["fmt"] = "vtt"
    return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment))


def fetch_subtitle_text(url: str) -> str:
    request = Request(url, headers=HTTP_HEADERS)
    with urlopen(request, timeout=30) as response:
        charset = response.headers.get_content_charset() or "utf-8"
        return response.read().decode(charset, errors="replace")


def download_first_available_subtitle(
    info: dict,
    tmp_dir: Path,
) -> tuple[Path, str, bool]:
    candidates = iter_subtitle_candidates(info)
    if not candidates:
        raise UserFacingError(MSG_NO_SUBTITLE)

    for lang, automatic, tracks in candidates:
        skip_candidate = False
        tried_urls: set[str] = set()
        for track in prefer_vtt_tracks(tracks):
            if skip_candidate:
                break

            url = track.get("url")
            if not url or url in tried_urls:
                continue

            urls_to_try = [url]
            if track.get("ext") != "vtt":
                urls_to_try.insert(0, force_vtt_url(url))

            for subtitle_url in urls_to_try:
                if subtitle_url in tried_urls:
                    continue
                tried_urls.add(subtitle_url)

                try:
                    logging.info(
                        "Trying subtitle lang=%s automatic=%s ext=%s",
                        lang,
                        automatic,
                        track.get("ext"),
                    )
                    subtitle_text = fetch_subtitle_text(subtitle_url)
                except HTTPError as exc:
                    logging.warning(
                        "Subtitle fetch HTTP error lang=%s automatic=%s status=%s",
                        lang,
                        automatic,
                        exc.code,
                    )
                    if exc.code == 429:
                        skip_candidate = True
                        break
                    continue
                except (OSError, URLError) as exc:
                    logging.warning(
                        "Subtitle fetch failed lang=%s automatic=%s error=%r",
                        lang,
                        automatic,
                        exc,
                    )
                    continue

                if "WEBVTT" not in subtitle_text[:200]:
                    logging.info("Fetched subtitle was not VTT; skipping")
                    continue

                subtitle_path = tmp_dir / f"subtitle.{lang}.vtt"
                subtitle_path.write_text(subtitle_text, encoding="utf-8")
                logging.info("Fetched subtitle lang=%s automatic=%s", lang, automatic)
                return subtitle_path, lang, automatic

    raise UserFacingError(MSG_DOWNLOAD_FAILED)


def download_subtitle(url: str, tmp_dir: Path) -> tuple[Path, dict, str, bool]:
    info_opts = {
        "quiet": True,
        "no_warnings": True,
        "skip_download": True,
        "noplaylist": True,
    }

    try:
        with YoutubeDL(info_opts) as ydl:
            info = ydl.extract_info(url, download=False)
    except DownloadError as exc:
        logging.exception("yt-dlp info extraction failed")
        raise UserFacingError(MSG_DOWNLOAD_FAILED) from exc
    except (OSError, URLError) as exc:
        logging.exception("Network error during info extraction")
        raise UserFacingError(MSG_NETWORK_FAILED) from exc

    logging.info("Extracted info id=%s title=%r", info.get("id"), info.get("title"))
    subtitle_path, lang, automatic = download_first_available_subtitle(info, tmp_dir)
    logging.info("Subtitle file created: %s", subtitle_path)
    return subtitle_path, info, lang, automatic


def clean_vtt(vtt_text: str) -> str:
    lines: list[str] = []
    previous = ""

    for raw_line in vtt_text.splitlines():
        line = raw_line.strip()

        if not line or line == "WEBVTT" or line.startswith(("Kind:", "Language:")):
            continue
        if TIMESTAMP_RE.match(line):
            continue
        if line.isdigit():
            continue

        line = TAG_RE.sub("", line)
        line = html.unescape(line)
        line = re.sub(r"\s+", " ", line).strip()

        if not line or line == previous:
            continue

        lines.append(line)
        previous = line

    paragraphs: list[str] = []
    buffer: list[str] = []

    for line in lines:
        buffer.append(line)
        if len(" ".join(buffer)) >= 420:
            paragraphs.append(" ".join(buffer))
            buffer = []

    if buffer:
        paragraphs.append(" ".join(buffer))

    cleaned = "\n\n".join(paragraphs).strip()
    logging.info("Cleaned VTT chars raw=%s cleaned=%s", len(vtt_text), len(cleaned))
    return cleaned


def build_clipboard_text(info: dict, url: str, lang: str, automatic: bool, body: str) -> str:
    title = info.get("title") or LABEL_NO_TITLE
    extracted_at = datetime.now().strftime("%Y-%m-%d %H:%M")
    lang_label = f"auto-{lang}" if automatic else lang

    return (
        f"{LABEL_TITLE}: {title}\n"
        f"URL: {url}\n"
        f"{LABEL_EXTRACTED_AT}: {extracted_at}\n"
        f"{LABEL_LANGUAGE}: {lang_label}\n\n"
        "----------------------------------------\n\n"
        f"{body}"
    )


def copy_script_to_clipboard() -> None:
    logging.info("Started extractor")
    url = get_clipboard_youtube_url()

    with tempfile.TemporaryDirectory(prefix="yt_script_") as tmp:
        subtitle_path, info, lang, automatic = download_subtitle(url, Path(tmp))
        vtt_text = subtitle_path.read_text(encoding="utf-8", errors="replace")

    body = clean_vtt(vtt_text)
    if not body:
        logging.info("Cleaned subtitle body was empty")
        raise UserFacingError(MSG_NO_SUBTITLE)

    write_clipboard_text(build_clipboard_text(info, url, lang, automatic, body))
    logging.info("Finished extractor successfully")


def main() -> int:
    try:
        copy_script_to_clipboard()
    except UserFacingError as exc:
        show_error(str(exc))
        return 1
    except OSError:
        logging.exception("OS error")
        show_error(MSG_FILE_FAILED)
        return 1
    except Exception:
        logging.exception("Unexpected error")
        show_error(MSG_DOWNLOAD_FAILED)
        return 1

    show_auto_close_notice(MSG_DONE)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())