from __future__ import annotations import urllib.request from pathlib import Path from app.errors import JobCancelled, VaaniError from app.links import is_video_id from app.text import format_upload_date _UA = "Mozilla/5.0 (compatible; Vaani/0.1; +https://localhost)" _THUMBS = ( "https://i.ytimg.com/vi/{video_id}/hqdefault.jpg", "https://i.ytimg.com/vi/{video_id}/mqdefault.jpg", ) def _ytdlp(): try: import yt_dlp except ImportError as exc: raise VaaniError( "The YouTube reader is not installed on this computer.", "tool_missing", 500, ) from exc return yt_dlp def explain_download_failure(raw: str) -> tuple[str, str, str | None]: text = (raw or "").lower() if "ffmpeg" in text or "ffprobe" in text: return ( "audio_tool_missing", "FFmpeg is not installed, so the audio cannot be prepared.", "retry", ) if "private video" in text or "video is private" in text: return ( "video_unavailable", "This video is private. If you have a copy you are allowed to use, upload it.", "upload", ) if "age" in text or "sign in" in text or "confirm your age" in text: return ( "audio_unavailable", "YouTube asked for a signed-in session before it would release the audio. " "If you have a copy you are allowed to use, upload it.", "upload", ) if "country" in text or "not made this video available" in text: return ( "audio_unavailable", "YouTube does not offer this video's audio in this country. " "If you have a copy you are allowed to use, upload it.", "upload", ) if "unavailable" in text or "removed" in text or "deleted" in text: return ("video_unavailable", "This video is unavailable.", None) if "format" in text: return ( "audio_unavailable", "YouTube did not return an audio file for this video. " "If you have a copy you are allowed to use, upload it.", "upload", ) return ( "audio_unavailable", "The audio could not be retrieved from YouTube. " "If you have a copy you are allowed to use, upload it.", "upload", ) def normalize_video(info: dict, fallback_url: str = "") -> dict: video_id = str(info.get("id") or "") duration = info.get("duration") try: duration = int(duration) if duration is not None else None except (TypeError, ValueError): duration = None return { "video_id": video_id, "title": (info.get("title") or "Untitled").strip()[:300], "channel": (info.get("channel") or info.get("uploader") or "").strip()[:200], "duration": duration, "upload_date": format_upload_date(info.get("upload_date")), "webpage_url": info.get("webpage_url") or fallback_url, "live_status": info.get("live_status") or "", "is_live": bool(info.get("is_live")), "availability": info.get("availability") or "", } def fetch_video(url: str) -> dict: yt_dlp = _ytdlp() options = { "quiet": True, "no_warnings": True, "noplaylist": True, "skip_download": True, "socket_timeout": 20, } try: with yt_dlp.YoutubeDL(options) as ydl: info = ydl.extract_info(url, download=False) except Exception as exc: code, message, recovery = explain_download_failure(str(exc)) if code == "audio_unavailable": lowered = str(exc).lower() if "country" in lowered or "age" in lowered or "sign in" in lowered: code = "video_unavailable" recovery = None else: code, message, recovery = ( "video_unavailable", "This video could not be opened.", None, ) raise VaaniError(message, code, 400, recovery) from exc if not info or not is_video_id(str(info.get("id") or "")): raise VaaniError("This video could not be opened.", "video_unavailable", 400) return normalize_video(info, url) def fetch_collection(url: str, limit: int = 24) -> dict: yt_dlp = _ytdlp() options = { "quiet": True, "no_warnings": True, "extract_flat": "in_playlist", "playlistend": limit, "socket_timeout": 25, } try: with yt_dlp.YoutubeDL(options) as ydl: info = ydl.extract_info(url, download=False) except Exception as exc: raise VaaniError( "That channel or playlist could not be opened.", "collection_unavailable", 400, ) from exc title = (info or {}).get("title") or (info or {}).get("channel") or "Videos" videos = [] for entry in _flat_entries((info or {}).get("entries") or []): video_id = str(entry.get("id") or "") if not is_video_id(video_id): continue duration = entry.get("duration") try: duration = int(duration) if duration is not None else None except (TypeError, ValueError): duration = None videos.append( { "video_id": video_id, "title": (entry.get("title") or "Untitled").strip()[:300], "channel": (entry.get("channel") or entry.get("uploader") or "").strip()[:200], "duration": duration, "url": f"https://www.youtube.com/watch?v={video_id}", } ) if len(videos) >= limit: break return {"title": str(title).strip()[:300], "url": url, "videos": videos} def _flat_entries(entries) -> list[dict]: flat: list[dict] = [] for entry in entries: if not isinstance(entry, dict): continue nested = entry.get("entries") if nested: flat.extend(_flat_entries(nested)) continue flat.append(entry) return flat def save_thumbnail(video_id: str, dest: Path) -> bool: if not is_video_id(video_id): return False for template in _THUMBS: url = template.format(video_id=video_id) try: request = urllib.request.Request(url, headers={"User-Agent": _UA}) with urllib.request.urlopen(request, timeout=20) as response: data = response.read() except Exception: continue if len(data) < 800 or not data.startswith(b"\xff\xd8"): continue dest.write_bytes(data) return True return False def download_audio(url: str, directory: Path, cancel, on_progress) -> Path: yt_dlp = _ytdlp() def hook(event: dict) -> None: if cancel(): raise JobCancelled() status = event.get("status") if status == "downloading": on_progress("Downloading audio") elif status == "finished": on_progress("Preparing the download") options = { "format": "bestaudio/best", "outtmpl": str(directory / "source.%(ext)s"), "noplaylist": True, "quiet": True, "no_warnings": True, "noprogress": True, "retries": 3, "socket_timeout": 30, "restrictfilenames": True, "windowsfilenames": True, "progress_hooks": [hook], "overwrites": True, } try: with yt_dlp.YoutubeDL(options) as ydl: ydl.extract_info(url, download=True) except JobCancelled: raise except Exception as exc: code, message, recovery = explain_download_failure(str(exc)) raise VaaniError(message, code, 400, recovery) from exc matches = [ path for path in directory.glob("source.*") if path.is_file() and path.stat().st_size > 0 and not path.name.endswith((".part", ".ytdl")) ] if not matches: raise VaaniError( "YouTube did not return an audio file for this video. " "If you have a copy you are allowed to use, upload it.", "audio_unavailable", 400, "upload", ) return matches[0]