ytrecap/server/transcripts.py

from __future__ import annotations

import hashlib
import logging
import re
from dataclasses import dataclass
from typing import Any

from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api._errors import (
    IpBlocked,
    NoTranscriptFound,
    RequestBlocked,
    TranscriptsDisabled,
    VideoUnavailable,
    YouTubeRequestFailed,
)
from youtube_transcript_api.proxies import GenericProxyConfig, ProxyConfig, WebshareProxyConfig

from server.settings import settings

logger = logging.getLogger(__name__)

PREFERRED_LANGUAGES = ("de", "en", "de-DE", "en-US", "en-GB")


@dataclass
class TranscriptSegment:
    start: float
    text: str


@dataclass
class Transcript:
    language: str
    segments: list[TranscriptSegment]
    source: str


def _hash_transcript(segments: list[TranscriptSegment]) -> str:
    payload = "\n".join(f"{s.start:.2f}:{s.text}" for s in segments)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def transcript_hash(segments: list[TranscriptSegment]) -> str:
    return _hash_transcript(segments)


def format_transcript_for_prompt(segments: list[TranscriptSegment], *, max_chars: int = 120_000) -> str:
    lines: list[str] = []
    total = 0
    for seg in segments:
        minutes = int(seg.start // 60)
        seconds = int(seg.start % 60)
        line = f"[{minutes:02d}:{seconds:02d}] {seg.text.strip()}"
        if total + len(line) + 1 > max_chars:
            lines.append("[... Transcript gekürzt wegen Länge ...]")
            break
        lines.append(line)
        total += len(line) + 1
    return "\n".join(lines)


def _normalize_ytt_segments(raw: list[dict]) -> list[TranscriptSegment]:
    return [TranscriptSegment(start=float(item["start"]), text=str(item["text"])) for item in raw]


def _proxy_dict() -> dict[str, str] | None:
    config = build_proxy_config()
    if config is None:
        return None
    return config.to_requests_dict()


def build_proxy_config() -> ProxyConfig | None:
    if settings.webshare_proxy_username and settings.webshare_proxy_password:
        locations = [
            code.strip().lower()
            for code in settings.webshare_proxy_locations.split(",")
            if code.strip()
        ]
        return WebshareProxyConfig(
            proxy_username=settings.webshare_proxy_username,
            proxy_password=settings.webshare_proxy_password,
            filter_ip_locations=locations or None,
        )

    proxy_url = settings.youtube_proxy.strip()
    if proxy_url:
        return GenericProxyConfig(http_url=proxy_url, https_url=proxy_url)

    return None


def proxy_url_for_ytdlp() -> str | None:
    direct = settings.youtube_proxy.strip()
    if direct:
        return direct
    config = build_proxy_config()
    if config is None:
        return None
    proxies = config.to_requests_dict()
    return proxies.get("https") or proxies.get("http")


def build_youtube_transcript_api() -> YouTubeTranscriptApi:
    config = build_proxy_config()
    if config is None:
        return YouTubeTranscriptApi()
    return YouTubeTranscriptApi(proxy_config=config)


def _blocked_message() -> str:
    if build_proxy_config() is not None:
        return (
            "YouTube blockiert auch über den Proxy. "
            "Statische/datacenter Proxies halten oft nicht lange — "
            "für zuverlässigere Ergebnisse rotating residential nutzen."
        )
    return (
        "YouTube blockiert deine Server-IP. "
        "Setze YOUTUBE_PROXY in der .env (siehe README)."
    )


def _fetch_with_youtube_transcript_api(video_id: str) -> Transcript:
    api = build_youtube_transcript_api()
    try:
        fetched = api.fetch(video_id, languages=list(PREFERRED_LANGUAGES))
        segments = _normalize_ytt_segments(
            [{"start": s.start, "text": s.text} for s in fetched]
        )
        return Transcript(language=fetched.language_code, segments=segments, source="youtube-transcript-api")
    except (RequestBlocked, IpBlocked) as exc:
        logger.warning("YouTube IP block for %s: %s", video_id, exc)
        raise ValueError(_blocked_message()) from exc
    except (TranscriptsDisabled, NoTranscriptFound, VideoUnavailable):
        raise
    except YouTubeRequestFailed as exc:
        logger.warning("youtube-transcript-api failed for %s: %s", video_id, exc)
        raise


def _parse_vtt(content: str) -> list[TranscriptSegment]:
    segments: list[TranscriptSegment] = []
    blocks = re.split(r"\n\n+", content.strip())
    for block in blocks:
        lines = [ln.strip() for ln in block.splitlines() if ln.strip()]
        if len(lines) < 2:
            continue
        time_line = lines[0] if "-->" in lines[0] else (lines[1] if len(lines) > 1 and "-->" in lines[1] else "")
        if "-->" not in time_line:
            continue
        start_raw = time_line.split("-->")[0].strip()
        match = re.match(r"(?:(\d+):)?(\d+):(\d+(?:\.\d+)?)", start_raw)
        if not match:
            continue
        hours = int(match.group(1) or 0)
        minutes = int(match.group(2))
        seconds = float(match.group(3))
        start = hours * 3600 + minutes * 60 + seconds
        text_lines = [ln for ln in lines if "-->" not in ln and not ln.isdigit()]
        text = " ".join(text_lines).strip()
        text = re.sub(r"<[^>]+>", "", text)
        if text:
            segments.append(TranscriptSegment(start=start, text=text))
    return segments


def _fetch_with_ytdlp(video_id: str) -> Transcript:
    import yt_dlp

    url = f"https://www.youtube.com/watch?v={video_id}"
    opts: dict[str, Any] = {
        "skip_download": True,
        "quiet": True,
        "no_warnings": True,
        "writesubtitles": True,
        "writeautomaticsub": True,
        "subtitleslangs": list(PREFERRED_LANGUAGES),
        "subtitlesformat": "vtt",
    }
    proxy = proxy_url_for_ytdlp()
    if proxy:
        opts["proxy"] = proxy

    with yt_dlp.YoutubeDL(opts) as ydl:
        info = ydl.extract_info(url, download=False)

    subtitles = info.get("subtitles") or {}
    automatic = info.get("automatic_captions") or {}
    tracks = {**automatic, **subtitles}

    chosen_lang = None
    for lang in PREFERRED_LANGUAGES:
        if lang in tracks:
            chosen_lang = lang
            break
    if not chosen_lang:
        for lang in tracks:
            chosen_lang = lang
            break
    if not chosen_lang:
        raise ValueError("Keine Untertitel gefunden.")

    formats = tracks[chosen_lang]
    vtt_url = None
    for fmt in formats:
        if fmt.get("ext") == "vtt" or "vtt" in (fmt.get("url") or ""):
            vtt_url = fmt.get("url")
            break
    if not vtt_url and formats:
        vtt_url = formats[0].get("url")
    if not vtt_url:
        raise ValueError("Untertitel-URL nicht verfügbar.")

    import httpx

    client_kwargs: dict[str, Any] = {"timeout": 30.0, "follow_redirects": True}
    proxies = _proxy_dict()
    if proxies:
        client_kwargs["proxy"] = proxies.get("https") or proxies.get("http")

    with httpx.Client(**client_kwargs) as client:
        resp = client.get(vtt_url)
        resp.raise_for_status()
        vtt_text = resp.text

    segments = _parse_vtt(vtt_text)
    if not segments:
        raise ValueError("Untertitel konnten nicht geparst werden.")

    return Transcript(language=chosen_lang, segments=segments, source="yt-dlp")


def fetch_transcript(video_id: str) -> Transcript:
    try:
        return _fetch_with_youtube_transcript_api(video_id)
    except (TranscriptsDisabled, NoTranscriptFound):
        pass
    except VideoUnavailable as exc:
        raise ValueError("Video nicht verfügbar.") from exc

    try:
        return _fetch_with_ytdlp(video_id)
    except Exception as exc:
        logger.exception("yt-dlp transcript fallback failed for %s", video_id)
        raise ValueError(
            "Dieses Video hat keine Captions. Dein Goldfisch kann leider nicht ins Leere starren."
        ) from exc


def estimate_duration_sec(segments: list[TranscriptSegment]) -> int | None:
    if not segments:
        return None
    last = segments[-1]
    return int(last.start) + 30