Spaces:
Building
Building
File size: 7,231 Bytes
a31f556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | # 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)}
|