backend/tools/youtube_tools.py

# backend/tools/youtube_tools.py
# Returns structured status keys so the React layer can use i18n.t() for display text.
import os
import asyncio
import logging
from backend.tools.browser_tools import get_ephemeral_context, stealth_async
from backend.tools.captcha_solver import solve_captcha_if_present

async def scrape_youtube_transcript(url: str) -> dict:
    """
    Navigates to a YT video, clicks show transcript, and scrapes it.
    If DOM scraping fails, falls back to downloading audio via yt-dlp and transcribing via Whisper.
    Returns status_key for i18n rendering on the frontend (youtube.scrape_success or youtube.no_transcript).
    """
    try:
        async with get_ephemeral_context() as context:
            page = await context.new_page()
            if stealth_async:
                await stealth_async(page)

            await page.goto(url)
            await page.wait_for_timeout(2000)
            await solve_captcha_if_present(page)
            
            try:
                await page.wait_for_selector('#expand', timeout=5000)
                await page.click('#expand')
                await page.wait_for_timeout(1000)
                await page.click('button[aria-label="Show transcript"]')
                await page.wait_for_selector('ytd-transcript-segment-renderer', timeout=5000)

                segments = await page.locator('ytd-transcript-segment-renderer').all_inner_texts()
                transcript = "\n".join(segments)
                await page.close()
                return {
                    "status": "ok",
                    "status_key": "youtube.scrape_success",
                    "transcript": transcript[:2000],
                    "truncated": len(transcript) > 2000
                }
            except Exception as e:
                logging.warning(f"YouTube DOM transcript failed: {e}. Falling back to yt-dlp + Whisper...")
                await page.close()
                
                # Fallback: yt-dlp + Whisper
                return await _fallback_whisper_transcript(url)

    except Exception as e:
        return {"status": "error", "status_key": "youtube.no_transcript", "detail": str(e)}

async def _fallback_whisper_transcript(url: str) -> dict:
    import tempfile
    import subprocess

    with tempfile.TemporaryDirectory() as tmpdirname:
        audio_path = os.path.join(tmpdirname, "audio.wav")
        # Use yt-dlp to download the best audio format and convert to wav
        cmd = [
            "yt-dlp",
            "-x",
            "--audio-format", "wav",
            "-o", audio_path,
            url
        ]
        try:
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            stdout, stderr = await process.communicate()
            
            if process.returncode != 0:
                logging.error(f"yt-dlp failed: {stderr.decode()}")
                return {"status": "error", "status_key": "youtube.no_transcript", "detail": "yt-dlp download failed"}
                
            # If successful, transcription
            # yt-dlp might append .wav to the path depending on how it handles -o, let's ensure we find the file
            actual_audio_path = audio_path
            if not os.path.exists(audio_path):
                # Search for the downloaded file
                files = os.listdir(tmpdirname)
                if files:
                    actual_audio_path = os.path.join(tmpdirname, files[0])
                else:
                    return {"status": "error", "status_key": "youtube.no_transcript", "detail": "Audio file not found after yt-dlp"}
            
            from backend.voice.stt import STTPipeline
                
            stt_engine = STTPipeline()
            transcript = await stt_engine.transcribe(actual_audio_path)
            
            if transcript:
                return {
                    "status": "ok",
                    "status_key": "youtube.scrape_success",
                    "transcript": transcript[:2000],
                    "truncated": len(transcript) > 2000
                }
            else:
                return {"status": "error", "status_key": "youtube.no_transcript", "detail": "Whisper transcription returned empty"}
        except Exception as e:
            logging.error(f"Whisper fallback failed: {e}")
            return {"status": "error", "status_key": "youtube.no_transcript", "detail": str(e)}

async def search_youtube(query: str, sort_by: str = "relevance", limit: int = 5) -> dict:
    """
    Searches YouTube using yt-dlp's ytsearch capability.
    sort_by can be: 'relevance' (default), 'views', 'latest'
    Returns a dict with 'status', 'status_key', and 'results' (list of dicts).
    """
    import asyncio
    import json
    import subprocess
    import logging

    try:
        # Fetch up to 15 results to have enough to sort if needed
        fetch_count = limit if sort_by == "relevance" else limit * 3
        search_query = f"ytsearch{fetch_count}:{query}"
        
        if sort_by == "latest":
            search_query = f"ytsearchdate{fetch_count}:{query}"

        cmd = [
            "yt-dlp",
            search_query,
            "--dump-json",
            "--flat-playlist",
            "--no-warnings"
        ]

        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )
        stdout, stderr = await process.communicate()
        
        if process.returncode != 0 and not stdout:
            logging.error(f"yt-dlp search failed: {stderr.decode()}")
            return {"status": "error", "status_key": "youtube.search_failed", "detail": "Search execution failed"}

        results = []
        for line in stdout.decode('utf-8').splitlines():
            if not line.strip():
                continue
            try:
                data = json.loads(line)
                results.append({
                    "title": data.get("title", ""),
                    "url": data.get("url", f"https://www.youtube.com/watch?v={data.get('id')}"),
                    "views": data.get("view_count", 0),
                    "uploader": data.get("uploader", ""),
                    "duration": data.get("duration", 0),
                    "upload_date": data.get("upload_date", "")
                })
            except Exception as e:
                import logging; logging.getLogger(__name__).error(f"Swallowed exception: {e}")

        if sort_by == "views":
            results.sort(key=lambda x: x.get("views") or 0, reverse=True)
            
        results = results[:limit]

        return {
            "status": "ok",
            "status_key": "youtube.search_success",
            "results": results
        }
    except Exception as e:
        logging.error(f"YouTube search error: {e}")
        return {"status": "error", "status_key": "youtube.search_failed", "detail": str(e)}