"""Video recipe extraction: yt-dlp → caption check → Whisper → GPT-4.""" from __future__ import annotations import asyncio import json import logging import os import uuid from pathlib import Path from typing import AsyncGenerator from app.config import get_settings from app.models import Recipe, StatusEvent from app.services.openai_service import ( check_caption_has_recipe, extract_recipe_from_text, transcribe_audio, ) log = logging.getLogger(__name__) MAX_DURATION_WARN = 20 * 60 # 20 minutes in seconds MAX_FILESIZE = "50m" WHISPER_MAX_BYTES = 24 * 1024 * 1024 # 25MB limit, buffer to 24MB async def extract_video(url: str) -> AsyncGenerator[StatusEvent | Recipe, None]: yield StatusEvent(stage="detecting", message="Erkenne Plattform...", progress=5) # Step 1: Get metadata without downloading yield StatusEvent(stage="metadata", message="Lade Video-Metadaten...", progress=10) metadata = await _get_metadata(url) title = metadata.get("title", "") description = metadata.get("description", "") or "" tags = metadata.get("tags") or [] duration = metadata.get("duration") or 0 # Step 2: Check if caption/description contains a complete recipe yield StatusEvent(stage="caption", message="Prüfe Video-Beschreibung...", progress=20) caption_text = f"{title}\n\n{description}" has_recipe_in_caption = await check_caption_has_recipe(caption_text) if has_recipe_in_caption: yield StatusEvent(stage="extracting", message="Extrahiere Rezept aus Beschreibung...", progress=50) enriched_text = f"Titel: {title}\n\nBeschreibung:\n{description}" recipe = await extract_recipe_from_text(enriched_text, url) if recipe.image_url is None: recipe.image_url = _get_thumbnail(metadata) if not recipe.tags and tags: recipe.tags = [str(t) for t in tags[:5]] yield StatusEvent(stage="done", message="Rezept aus Beschreibung extrahiert.", progress=95) yield recipe return # Step 3: Audio download + transcription fallback if duration > MAX_DURATION_WARN: yield StatusEvent( stage="warning", message=f"Video ist {duration // 60} Min. lang – Transkription kann dauern.", progress=25, ) yield StatusEvent(stage="downloading", message="Lade Audio herunter...", progress=30) settings = get_settings() audio_path = Path(settings.tmp_dir) / f"{uuid.uuid4()}.mp3" try: await _download_audio(url, audio_path) # Check file size for Whisper limit if audio_path.exists(): size = audio_path.stat().st_size if size > WHISPER_MAX_BYTES: raise RuntimeError( f"Audio-Datei zu groß ({size // 1024 // 1024} MB). " "Whisper unterstützt max. 25MB. Bitte kürzeres Video verwenden." ) else: raise RuntimeError("Audio-Download fehlgeschlagen.") yield StatusEvent(stage="transcribing", message="Transkribiere Audio (Whisper)...", progress=55) transcript = await transcribe_audio(audio_path) if not transcript.strip(): raise RuntimeError("Transkription leer. Kein gesprochener Text erkannt.") yield StatusEvent(stage="extracting", message="Extrahiere Rezept aus Transkription...", progress=75) enriched_text = ( f"Video-Titel: {title}\n" f"Beschreibung: {description[:500]}\n\n" f"Transkription:\n{transcript}" ) recipe = await extract_recipe_from_text(enriched_text, url) if recipe.image_url is None: recipe.image_url = _get_thumbnail(metadata) if not recipe.tags and tags: recipe.tags = [str(t) for t in tags[:5]] yield StatusEvent(stage="done", message="Rezept extrahiert.", progress=95) yield recipe except FileNotFoundError: raise RuntimeError( "yt-dlp nicht gefunden. Bitte sicherstellen, dass yt-dlp installiert ist." ) except Exception: raise finally: if audio_path.exists(): try: audio_path.unlink() except Exception as e: log.warning("Could not delete tmp file %s: %s", audio_path, e) def _ffmpeg_args() -> list[str]: """Return --ffmpeg-location args if configured, else empty list.""" loc = get_settings().ffmpeg_location return ["--ffmpeg-location", loc] if loc else [] async def _get_metadata(url: str) -> dict: """Run yt-dlp --dump-json to get metadata without downloading.""" proc = await asyncio.create_subprocess_exec( "yt-dlp", "--dump-json", "--no-playlist", *_ffmpeg_args(), url, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60) if proc.returncode != 0: err = stderr.decode(errors="replace") _check_tiktok_error(err, url) raise RuntimeError(f"yt-dlp Metadaten-Fehler: {err[:500]}") try: return json.loads(stdout.decode()) except json.JSONDecodeError: return {} async def _download_audio(url: str, output_path: Path) -> None: """Download audio only via yt-dlp subprocess (non-blocking).""" output_path.parent.mkdir(parents=True, exist_ok=True) proc = await asyncio.create_subprocess_exec( "yt-dlp", "--no-playlist", "--extract-audio", "--audio-format", "mp3", "--audio-quality", "5", "--max-filesize", MAX_FILESIZE, *_ffmpeg_args(), "--output", str(output_path).replace(".mp3", ".%(ext)s"), url, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) try: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300) except asyncio.TimeoutError: proc.kill() raise RuntimeError("Audio-Download Timeout (5 Min.) überschritten.") if proc.returncode != 0: err = stderr.decode(errors="replace") _check_tiktok_error(err, url) raise RuntimeError(f"Audio-Download fehlgeschlagen: {err[:500]}") # yt-dlp may output with different extension before conversion # Try to find the actual file if not output_path.exists(): parent = output_path.parent stem = output_path.stem for f in parent.glob(f"{stem}*"): if f.suffix in (".mp3", ".m4a", ".webm", ".opus"): f.rename(output_path) break def _get_thumbnail(metadata: dict) -> str | None: thumbnail = metadata.get("thumbnail") if thumbnail: return thumbnail thumbnails = metadata.get("thumbnails") or [] if thumbnails: # prefer largest best = max(thumbnails, key=lambda t: (t.get("width", 0) or 0), default=None) if best: return best.get("url") return None def _check_tiktok_error(stderr: str, url: str) -> None: """Raise a clear error for TikTok login/cookie issues.""" if "tiktok" in url.lower(): if any(phrase in stderr.lower() for phrase in ["login", "cookie", "captcha", "403"]): raise RuntimeError( "TikTok erfordert Cookies für dieses Video. " "Bitte exportiere deine Browser-Cookies und konfiguriere yt-dlp entsprechend." )