Frank Schwenk
12cf6cab32
Rewrite https proxy URLs to http, allow HOST/PORT/USER/PASSWORD env vars, and surface clearer errors when proxies fail on the wrong port. Co-authored-by: Cursor <cursoragent@cursor.com>
300 lines
9.6 KiB
Python
300 lines
9.6 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
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 normalize_proxy_url(url: str) -> str:
|
|
"""Normalize common proxy URL mistakes (https scheme, missing port hint)."""
|
|
cleaned = url.strip()
|
|
if cleaned.startswith("https://"):
|
|
cleaned = "http://" + cleaned[len("https://") :]
|
|
logger.info("YOUTUBE_PROXY: https:// → http:// (Proxy-Protokoll ist HTTP)")
|
|
return cleaned
|
|
|
|
|
|
def resolve_proxy_url() -> str | None:
|
|
if settings.youtube_proxy.strip():
|
|
return normalize_proxy_url(settings.youtube_proxy)
|
|
|
|
host = settings.youtube_proxy_host.strip()
|
|
port = settings.youtube_proxy_port
|
|
user = settings.youtube_proxy_user.strip()
|
|
password = settings.youtube_proxy_password
|
|
if host and port and user and password:
|
|
return f"http://{quote(user, safe='')}:{quote(password, safe='')}@{host}:{port}"
|
|
|
|
return None
|
|
|
|
|
|
def _proxy_error_message(exc: Exception) -> str:
|
|
msg = str(exc)
|
|
if "Connection refused" in msg and ":443" in msg:
|
|
return (
|
|
"Proxy-Verbindung fehlgeschlagen (Port 443). "
|
|
"Webshare-Dashboard-Proxies nutzen meist einen anderen Port (z.B. 6641). "
|
|
"Setze YOUTUBE_PROXY=http://USER:PASS@HOST:PORT oder "
|
|
"YOUTUBE_PROXY_HOST/PORT/USER/PASSWORD — Schema http://, nicht https://."
|
|
)
|
|
return f"Proxy-Verbindung fehlgeschlagen: {exc}"
|
|
|
|
|
|
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 = resolve_proxy_url()
|
|
if proxy_url:
|
|
return GenericProxyConfig(http_url=proxy_url, https_url=proxy_url)
|
|
|
|
return None
|
|
|
|
|
|
def proxy_url_for_ytdlp() -> str | None:
|
|
direct = resolve_proxy_url()
|
|
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)
|
|
if "ProxyError" in str(exc.reason) or "Unable to connect to proxy" in str(exc.reason):
|
|
raise ValueError(_proxy_error_message(exc)) from 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 ValueError:
|
|
raise
|
|
except Exception as exc:
|
|
if "ProxyError" in type(exc).__name__ or "Unable to connect to proxy" in str(exc):
|
|
raise ValueError(_proxy_error_message(exc)) from 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
|