750 lines
31 KiB
Python
750 lines
31 KiB
Python
"""
|
|
extractors.py - Per-artifact extraction logic for YouTube SIPOC Tool
|
|
Uses yt-dlp, youtube-transcript-api, and OpenRouter LLM.
|
|
"""
|
|
import os
|
|
import re
|
|
import json
|
|
import time
|
|
import logging
|
|
import requests
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Tuple, List
|
|
|
|
import yt_dlp
|
|
from youtube_transcript_api import YouTubeTranscriptApi
|
|
try:
|
|
from youtube_transcript_api import TranscriptsDisabled, NoTranscriptFound
|
|
except ImportError:
|
|
# v1.x moved/renamed exceptions
|
|
TranscriptsDisabled = Exception
|
|
NoTranscriptFound = Exception
|
|
from openai import OpenAI
|
|
|
|
import config
|
|
|
|
logger = logging.getLogger("extractors")
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Utilities
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def sanitize_filename(name: str, max_length: int = config.MAX_TITLE_LENGTH) -> str:
|
|
"""Make a string safe to use in a file name."""
|
|
name = re.sub(r'[\\/:*?"<>|]', "_", name)
|
|
name = re.sub(r'\s+', "_", name.strip())
|
|
name = re.sub(r'_+', "_", name)
|
|
return name[:max_length].rstrip("._")
|
|
|
|
|
|
def make_base_name(title: str, video_id: str) -> str:
|
|
"""Return the standard base filename (no extension)."""
|
|
safe_title = sanitize_filename(title) if title else "untitled"
|
|
return f"{safe_title}__{video_id}"
|
|
|
|
|
|
def ensure_dir(path: str):
|
|
Path(path).mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def video_url(video_id: str) -> str:
|
|
return f"https://www.youtube.com/watch?v={video_id}"
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Input Resolution: URL → list of video IDs
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def resolve_input(input_type: str, input_value: str) -> List[Dict]:
|
|
"""
|
|
Resolve any input type to a list of dicts: {video_id, url, title, position}
|
|
Supported input_types: url, playlist, csv, handle
|
|
"""
|
|
if input_type == "url":
|
|
return _resolve_single_url(input_value)
|
|
elif input_type == "playlist":
|
|
return _resolve_playlist(input_value)
|
|
elif input_type == "csv":
|
|
return _resolve_csv(input_value)
|
|
elif input_type == "handle":
|
|
return _resolve_handle(input_value)
|
|
else:
|
|
raise ValueError(f"Unknown input_type: {input_type}")
|
|
|
|
|
|
def _extract_video_id(url: str) -> Optional[str]:
|
|
"""Extract YouTube video ID from various URL formats."""
|
|
patterns = [
|
|
r'(?:v=|youtu\.be/|embed/|shorts/)([A-Za-z0-9_-]{11})',
|
|
]
|
|
for pat in patterns:
|
|
m = re.search(pat, url)
|
|
if m:
|
|
return m.group(1)
|
|
return None
|
|
|
|
|
|
def _resolve_single_url(url: str) -> List[Dict]:
|
|
"""Handle single YouTube URL (video or short)."""
|
|
url = url.strip()
|
|
vid = _extract_video_id(url)
|
|
if vid:
|
|
return [{"video_id": vid, "url": video_url(vid), "title": "", "position": 0}]
|
|
# Fall through to yt-dlp for unusual URLs
|
|
return _resolve_with_ytdlp(url)
|
|
|
|
|
|
def _resolve_playlist(playlist_url: str) -> List[Dict]:
|
|
return _resolve_with_ytdlp(playlist_url.strip())
|
|
|
|
|
|
def _resolve_handle(handle: str) -> List[Dict]:
|
|
"""Resolve @handle or channel URL to video list."""
|
|
handle = handle.strip()
|
|
if not handle.startswith("http"):
|
|
if handle.startswith("@"):
|
|
url = f"https://www.youtube.com/{handle}/videos"
|
|
else:
|
|
url = f"https://www.youtube.com/@{handle}/videos"
|
|
else:
|
|
url = handle
|
|
return _resolve_with_ytdlp(url)
|
|
|
|
|
|
def _resolve_csv(csv_content: str) -> List[Dict]:
|
|
"""Parse CSV content: each line may be a URL or video_id."""
|
|
lines = []
|
|
for line in csv_content.splitlines():
|
|
for part in line.split(","):
|
|
part = part.strip().strip('"').strip("'")
|
|
if part:
|
|
lines.append(part)
|
|
results = []
|
|
for i, entry in enumerate(lines):
|
|
if entry.startswith("http"):
|
|
items = _resolve_single_url(entry)
|
|
for item in items:
|
|
item["position"] = len(results)
|
|
results.append(item)
|
|
else:
|
|
# Treat as raw video_id
|
|
vid = entry[:11]
|
|
results.append({"video_id": vid, "url": video_url(vid), "title": "", "position": len(results)})
|
|
return results
|
|
|
|
|
|
def _resolve_with_ytdlp(url: str) -> List[Dict]:
|
|
"""Use yt-dlp to extract video list from any URL (playlist, channel, etc.)."""
|
|
captured_errors = []
|
|
|
|
class ErrorCollector:
|
|
"""Custom yt-dlp logger to capture warning/error messages."""
|
|
def debug(self, msg): pass
|
|
def info(self, msg): pass
|
|
def warning(self, msg):
|
|
captured_errors.append(f"WARNING: {msg}")
|
|
def error(self, msg):
|
|
captured_errors.append(f"ERROR: {msg}")
|
|
|
|
ydl_opts = {
|
|
"quiet": True,
|
|
"extract_flat": True,
|
|
"skip_download": True,
|
|
"ignoreerrors": False, # Don't swallow errors silently
|
|
"logger": ErrorCollector(),
|
|
}
|
|
results = []
|
|
info = None
|
|
try:
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(url, download=False)
|
|
except Exception as e:
|
|
captured_errors.append(str(e))
|
|
|
|
if not info:
|
|
# Build a meaningful error message
|
|
err_summary = "; ".join(captured_errors) if captured_errors else "No information returned by yt-dlp."
|
|
# Surface private/login-required as a clear user-facing error
|
|
lower = err_summary.lower()
|
|
if "private" in lower:
|
|
raise ValueError(
|
|
f"This playlist or channel is PRIVATE. "
|
|
f"YouTube requires you to be logged in to access it. "
|
|
f"To process private content, add your YouTube cookies file path in Settings. "
|
|
f"Details: {err_summary}"
|
|
)
|
|
elif "sign in" in lower or "login" in lower or "confirm your age" in lower:
|
|
raise ValueError(
|
|
f"YouTube requires authentication (sign-in or age verification). "
|
|
f"Add your YouTube cookies file path in Settings. "
|
|
f"Details: {err_summary}"
|
|
)
|
|
elif "not exist" in lower or "removed" in lower or "unavailable" in lower:
|
|
raise ValueError(f"Content not available or removed. Details: {err_summary}")
|
|
else:
|
|
raise ValueError(f"Could not resolve URL '{url}'. Details: {err_summary}")
|
|
|
|
entries = info.get("entries") or [info]
|
|
for i, entry in enumerate(entries):
|
|
if not entry:
|
|
continue
|
|
vid = entry.get("id") or entry.get("video_id")
|
|
if not vid:
|
|
continue
|
|
title = entry.get("title", "")
|
|
results.append({
|
|
"video_id": vid,
|
|
"url": video_url(vid),
|
|
"title": title,
|
|
"position": i,
|
|
})
|
|
|
|
if not results:
|
|
raise ValueError(
|
|
f"URL resolved successfully but contained 0 videos. "
|
|
f"The playlist/channel may be empty. URL: {url}"
|
|
)
|
|
|
|
return results
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Metadata Fetch
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def fetch_metadata(video_id: str) -> Dict:
|
|
"""Fetch full video metadata using yt-dlp."""
|
|
ydl_opts = {
|
|
"quiet": True,
|
|
"skip_download": True,
|
|
"ignoreerrors": True,
|
|
}
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(video_url(video_id), download=False)
|
|
return info or {}
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Artifact: Title
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def extract_title(video_id: str, meta: Dict, output_folder: str) -> Tuple[str, str]:
|
|
"""
|
|
Save title as a .txt file.
|
|
Returns (file_path, title_string)
|
|
"""
|
|
title = meta.get("title", "")
|
|
base = make_base_name(title, video_id)
|
|
file_path = os.path.join(output_folder, f"{base}__title.txt")
|
|
ensure_dir(output_folder)
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
f.write(f"Title: {title}\n")
|
|
f.write(f"Video ID: {video_id}\n")
|
|
f.write(f"URL: {video_url(video_id)}\n")
|
|
return file_path, title
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Artifact: Stats
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def extract_stats(video_id: str, meta: Dict, output_folder: str) -> str:
|
|
"""Save video stats + full metadata as a human-readable .txt file."""
|
|
title = meta.get("title", "")
|
|
base = make_base_name(title, video_id)
|
|
file_path = os.path.join(output_folder, f"{base}__stats.txt")
|
|
ensure_dir(output_folder)
|
|
|
|
# Format helpers
|
|
def fmt_num(n) -> str:
|
|
if n is None:
|
|
return "N/A"
|
|
try:
|
|
return f"{int(n):,}"
|
|
except Exception:
|
|
return str(n)
|
|
|
|
def fmt_duration(secs) -> str:
|
|
if not secs:
|
|
return "N/A"
|
|
try:
|
|
secs = int(secs)
|
|
h, rem = divmod(secs, 3600)
|
|
m, s = divmod(rem, 60)
|
|
return f"{h}h {m}m {s}s" if h else f"{m}m {s}s"
|
|
except Exception:
|
|
return str(secs)
|
|
|
|
def fmt_date(d) -> str:
|
|
if not d:
|
|
return "N/A"
|
|
try:
|
|
d = str(d)
|
|
return f"{d[:4]}-{d[4:6]}-{d[6:8]}"
|
|
except Exception:
|
|
return str(d)
|
|
|
|
tags = meta.get("tags") or []
|
|
categories = meta.get("categories") or []
|
|
chapters = meta.get("chapters") or []
|
|
description = meta.get("description", "") or ""
|
|
formats = meta.get("formats") or []
|
|
subtitles = meta.get("subtitles") or {}
|
|
automatic_captions = meta.get("automatic_captions") or {}
|
|
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
f.write("=" * 70 + "\n")
|
|
f.write(f" VIDEO STATS & METADATA\n")
|
|
f.write("=" * 70 + "\n\n")
|
|
|
|
f.write(f"Title : {meta.get('title', '')}\n")
|
|
f.write(f"Video ID : {video_id}\n")
|
|
f.write(f"URL : {video_url(video_id)}\n")
|
|
f.write(f"Webpage URL : {meta.get('webpage_url', '')}\n\n")
|
|
|
|
f.write("── Channel ──────────────────────────────────────────────────\n")
|
|
f.write(f"Channel : {meta.get('channel', meta.get('uploader', ''))}\n")
|
|
f.write(f"Channel ID : {meta.get('channel_id', '')}\n")
|
|
f.write(f"Channel URL : {meta.get('channel_url', meta.get('uploader_url', ''))}\n")
|
|
f.write(f"Subscribers : {fmt_num(meta.get('channel_follower_count'))}\n\n")
|
|
|
|
f.write("── Content ──────────────────────────────────────────────────\n")
|
|
f.write(f"Upload Date : {fmt_date(meta.get('upload_date'))}\n")
|
|
f.write(f"Duration : {fmt_duration(meta.get('duration'))} ({meta.get('duration_string', '')})\n")
|
|
f.write(f"Age Limit : {meta.get('age_limit', 0)}\n")
|
|
f.write(f"Is Live : {meta.get('is_live', False)}\n")
|
|
f.write(f"Was Live : {meta.get('was_live', False)}\n")
|
|
f.write(f"Live Status : {meta.get('live_status', '')}\n\n")
|
|
|
|
f.write("── Engagement ───────────────────────────────────────────────\n")
|
|
f.write(f"Views : {fmt_num(meta.get('view_count'))}\n")
|
|
f.write(f"Likes : {fmt_num(meta.get('like_count'))}\n")
|
|
f.write(f"Comments : {fmt_num(meta.get('comment_count'))}\n")
|
|
f.write(f"Average Rating : {meta.get('average_rating', 'N/A')}\n\n")
|
|
|
|
f.write("── Media Info ───────────────────────────────────────────────\n")
|
|
f.write(f"Thumbnail URL : {meta.get('thumbnail', '')}\n")
|
|
# Best format info
|
|
best_fmt = next((fmt for fmt in reversed(formats) if fmt.get("vcodec") not in (None, "none")), None)
|
|
if best_fmt:
|
|
f.write(f"Best Resolution : {best_fmt.get('width', '?')}x{best_fmt.get('height', '?')} "
|
|
f"@ {best_fmt.get('fps', '?')}fps\n")
|
|
f.write(f"Best Video Codec : {best_fmt.get('vcodec', 'N/A')}\n")
|
|
f.write(f"Best Audio Codec : {best_fmt.get('acodec', 'N/A')}\n")
|
|
f.write(f"Available Formats: {len(formats)}\n\n")
|
|
|
|
f.write("── Classification ───────────────────────────────────────────\n")
|
|
f.write(f"Categories : {', '.join(categories) if categories else 'N/A'}\n")
|
|
if tags:
|
|
tag_str = ", ".join(tags[:30])
|
|
if len(tags) > 30:
|
|
tag_str += f" ... (+{len(tags)-30} more)"
|
|
f.write(f"Tags : {tag_str}\n")
|
|
else:
|
|
f.write(f"Tags : N/A\n")
|
|
f.write("\n")
|
|
|
|
# Subtitles / transcript languages
|
|
all_sub_langs = sorted(set(list(subtitles.keys()) + list(automatic_captions.keys())))
|
|
if all_sub_langs:
|
|
f.write("── Subtitles / Captions ─────────────────────────────────────\n")
|
|
f.write(f"Languages ({len(all_sub_langs)}): {', '.join(all_sub_langs[:20])}")
|
|
if len(all_sub_langs) > 20:
|
|
f.write(f" ... (+{len(all_sub_langs)-20} more)")
|
|
f.write("\n\n")
|
|
|
|
# Chapters
|
|
if chapters:
|
|
f.write("── Chapters ─────────────────────────────────────────────────\n")
|
|
for ch in chapters:
|
|
start = ch.get("start_time", 0)
|
|
m, s = divmod(int(start), 60)
|
|
h, m = divmod(m, 60)
|
|
ts = f"{h:02d}:{m:02d}:{s:02d}" if h else f"{m:02d}:{s:02d}"
|
|
f.write(f" [{ts}] {ch.get('title', '')}\n")
|
|
f.write("\n")
|
|
|
|
# Description
|
|
if description:
|
|
f.write("── Description ──────────────────────────────────────────────\n")
|
|
# Limit description to 3000 chars to keep the file manageable
|
|
if len(description) > 3000:
|
|
f.write(description[:3000] + "\n... [truncated]\n")
|
|
else:
|
|
f.write(description + "\n")
|
|
f.write("\n")
|
|
|
|
f.write("── Extraction Info ──────────────────────────────────────────\n")
|
|
f.write(f"Extracted At : {time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime())}\n")
|
|
f.write(f"Extractor : {meta.get('extractor', 'youtube')}\n")
|
|
f.write(f"Extractor Key : {meta.get('extractor_key', 'Youtube')}\n")
|
|
f.write("=" * 70 + "\n")
|
|
|
|
return file_path
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Artifact: Transcript
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def _get_snippet_text(snippet) -> str:
|
|
"""Safely extract text from a transcript snippet (dict or object)."""
|
|
if isinstance(snippet, dict):
|
|
return snippet.get("text", "")
|
|
return getattr(snippet, "text", "") or ""
|
|
|
|
|
|
def _get_snippet_start(snippet) -> float:
|
|
"""Safely extract start time from a transcript snippet (dict or object)."""
|
|
if isinstance(snippet, dict):
|
|
return snippet.get("start", 0.0)
|
|
return getattr(snippet, "start", 0.0) or 0.0
|
|
|
|
|
|
def _fetch_transcript(video_id: str):
|
|
"""
|
|
Fetch transcript snippets using youtube-transcript-api v1.x.
|
|
v1.x API: YouTubeTranscriptApi().list(video_id) and .fetch(video_id, languages=[...])
|
|
Returns (FetchedTranscript iterable, errors_list) — iterable may be None on failure.
|
|
"""
|
|
errors = []
|
|
api = YouTubeTranscriptApi()
|
|
|
|
# Fast path: try direct fetch with preferred languages
|
|
for lang_group in [["en", "en-US", "en-GB"], ["en-US"], ["en-GB"]]:
|
|
try:
|
|
fetched = api.fetch(video_id, languages=lang_group)
|
|
return fetched, errors
|
|
except Exception:
|
|
pass
|
|
|
|
# Slower path: list all available transcripts and pick the best
|
|
try:
|
|
transcript_list = api.list(video_id)
|
|
except Exception as e:
|
|
errors.append(f"list error: {e}")
|
|
# Last-ditch: fetch with no language preference
|
|
try:
|
|
fetched = api.fetch(video_id)
|
|
return fetched, errors
|
|
except Exception as e2:
|
|
errors.append(f"fetch error: {e2}")
|
|
return None, errors
|
|
|
|
# Prefer any manual transcript
|
|
all_transcripts = list(transcript_list)
|
|
manual = [t for t in all_transcripts if not getattr(t, "is_generated", True)]
|
|
auto = [t for t in all_transcripts if getattr(t, "is_generated", False)]
|
|
|
|
for t in (manual + auto):
|
|
try:
|
|
# Try fetching via the transcript object's own fetch method
|
|
fetched = t.fetch()
|
|
return fetched, errors
|
|
except Exception as e:
|
|
errors.append(f"t.fetch() error for {getattr(t, 'language_code','?')}: {e}")
|
|
continue
|
|
|
|
errors.append("No fetchable transcripts found")
|
|
return None, errors
|
|
|
|
|
|
def extract_transcript(video_id: str, meta: Dict, output_folder: str) -> str:
|
|
"""
|
|
Download transcript using youtube-transcript-api.
|
|
Falls back to auto-generated captions. Saves as .txt with timestamps.
|
|
"""
|
|
title = meta.get("title", "")
|
|
base = make_base_name(title, video_id)
|
|
file_path = os.path.join(output_folder, f"{base}__transcript.txt")
|
|
ensure_dir(output_folder)
|
|
|
|
transcript, errors = _fetch_transcript(video_id)
|
|
|
|
if not transcript:
|
|
raise RuntimeError(f"Could not retrieve transcript for {video_id}. Errors: {'; '.join(errors)}")
|
|
|
|
# Write formatted transcript
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
f.write(f"Transcript: {meta.get('title', video_id)}\n")
|
|
f.write(f"Video ID: {video_id}\n")
|
|
f.write(f"URL: {video_url(video_id)}\n")
|
|
f.write("=" * 60 + "\n\n")
|
|
for entry in transcript:
|
|
start = _get_snippet_start(entry)
|
|
text = _get_snippet_text(entry).replace("\n", " ")
|
|
minutes, seconds = divmod(int(start), 60)
|
|
hours, minutes = divmod(minutes, 60)
|
|
ts = f"[{hours:02d}:{minutes:02d}:{seconds:02d}]" if hours else f"[{minutes:02d}:{seconds:02d}]"
|
|
f.write(f"{ts} {text}\n")
|
|
|
|
return file_path
|
|
|
|
|
|
def get_transcript_text(video_id: str) -> str:
|
|
"""Get plain transcript text for summarization."""
|
|
try:
|
|
transcript, _ = _fetch_transcript(video_id)
|
|
if transcript:
|
|
return " ".join(_get_snippet_text(e) for e in transcript)
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Artifact: Summary
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def extract_summary(
|
|
video_id: str,
|
|
meta: Dict,
|
|
output_folder: str,
|
|
api_key: str,
|
|
model: str,
|
|
system_prompt: str,
|
|
) -> str:
|
|
"""Generate a summary using OpenRouter LLM and save as .txt."""
|
|
if not api_key:
|
|
raise RuntimeError("OpenRouter API key is not configured. Set it in the Settings panel.")
|
|
|
|
title = meta.get("title", "")
|
|
base = make_base_name(title, video_id)
|
|
file_path = os.path.join(output_folder, f"{base}__summary.txt")
|
|
ensure_dir(output_folder)
|
|
|
|
# Get transcript text for summarization
|
|
transcript_text = get_transcript_text(video_id)
|
|
if not transcript_text:
|
|
transcript_text = meta.get("description", "")
|
|
if not transcript_text:
|
|
raise RuntimeError(f"No transcript or description available for summarization of {video_id}")
|
|
|
|
# Truncate very long transcripts to ~12000 chars (token budget)
|
|
max_chars = 12000
|
|
if len(transcript_text) > max_chars:
|
|
transcript_text = transcript_text[:max_chars] + "\n\n[Transcript truncated for length]"
|
|
|
|
client = OpenAI(
|
|
api_key=api_key,
|
|
base_url=config.OPENROUTER_BASE_URL,
|
|
)
|
|
|
|
prompt = (
|
|
f"Title: {title}\n"
|
|
f"Video ID: {video_id}\n"
|
|
f"URL: {video_url(video_id)}\n\n"
|
|
f"Transcript:\n{transcript_text}"
|
|
)
|
|
|
|
response = client.chat.completions.create(
|
|
model=model or config.OPENROUTER_MODEL,
|
|
messages=[
|
|
{"role": "system", "content": system_prompt or config.SUMMARY_SYSTEM_PROMPT},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
max_tokens=config.SUMMARY_MAX_TOKENS,
|
|
)
|
|
|
|
summary_text = response.choices[0].message.content or ""
|
|
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
f.write(f"Summary: {title}\n")
|
|
f.write(f"Video ID: {video_id}\n")
|
|
f.write(f"URL: {video_url(video_id)}\n")
|
|
f.write("=" * 60 + "\n\n")
|
|
f.write(summary_text)
|
|
f.write("\n")
|
|
|
|
return file_path
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Artifact: Thumbnail
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def extract_thumbnail(video_id: str, meta: Dict, output_folder: str) -> str:
|
|
"""Download the highest-resolution thumbnail available."""
|
|
title = meta.get("title", "")
|
|
base = make_base_name(title, video_id)
|
|
ensure_dir(output_folder)
|
|
|
|
# Build priority list of thumbnail URLs
|
|
thumbnails = meta.get("thumbnails", [])
|
|
# Sort by resolution (width * height), highest first
|
|
thumbnails_sorted = sorted(
|
|
[t for t in thumbnails if t.get("url")],
|
|
key=lambda t: (t.get("width", 0) or 0) * (t.get("height", 0) or 0),
|
|
reverse=True,
|
|
)
|
|
|
|
# Fallback URLs
|
|
fallback_urls = [
|
|
f"https://img.youtube.com/vi/{video_id}/maxresdefault.jpg",
|
|
f"https://img.youtube.com/vi/{video_id}/sddefault.jpg",
|
|
f"https://img.youtube.com/vi/{video_id}/hqdefault.jpg",
|
|
f"https://img.youtube.com/vi/{video_id}/mqdefault.jpg",
|
|
f"https://img.youtube.com/vi/{video_id}/default.jpg",
|
|
]
|
|
|
|
candidate_urls = [t["url"] for t in thumbnails_sorted] + fallback_urls
|
|
|
|
for url in candidate_urls:
|
|
try:
|
|
response = requests.get(url, timeout=15)
|
|
if response.status_code == 200 and len(response.content) > 1000:
|
|
# Determine extension from content-type or URL
|
|
ct = response.headers.get("content-type", "")
|
|
if "webp" in ct or url.endswith(".webp"):
|
|
ext = "webp"
|
|
elif "png" in ct or url.endswith(".png"):
|
|
ext = "png"
|
|
else:
|
|
ext = "jpg"
|
|
file_path = os.path.join(output_folder, f"{base}__thumbnail.{ext}")
|
|
with open(file_path, "wb") as f:
|
|
f.write(response.content)
|
|
return file_path
|
|
except Exception:
|
|
continue
|
|
|
|
raise RuntimeError(f"Could not download thumbnail for {video_id}")
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Artifact: MP3
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def extract_mp3(video_id: str, meta: Dict, output_folder: str) -> str:
|
|
"""Download audio and convert to MP3 using yt-dlp + ffmpeg."""
|
|
title = meta.get("title", "")
|
|
base = make_base_name(title, video_id)
|
|
ensure_dir(output_folder)
|
|
|
|
ffmpeg_dir = _find_ffmpeg()
|
|
|
|
# yt-dlp will append extension itself; we use outtmpl without extension
|
|
out_template = os.path.join(output_folder, f"{base}__audio")
|
|
expected_path = os.path.join(output_folder, f"{base}__audio.mp3")
|
|
|
|
ydl_opts = {
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
"format": "bestaudio/best",
|
|
"outtmpl": out_template,
|
|
"postprocessors": [
|
|
{
|
|
"key": "FFmpegExtractAudio",
|
|
"preferredcodec": "mp3",
|
|
"preferredquality": "192",
|
|
}
|
|
],
|
|
"ignoreerrors": False,
|
|
}
|
|
|
|
if ffmpeg_dir:
|
|
ydl_opts["ffmpeg_location"] = ffmpeg_dir
|
|
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
ydl.download([video_url(video_id)])
|
|
|
|
# Rename to standard naming convention
|
|
if os.path.exists(expected_path):
|
|
final_path = os.path.join(output_folder, f"{base}__mp3.mp3")
|
|
if not os.path.exists(final_path):
|
|
os.rename(expected_path, final_path)
|
|
return final_path
|
|
|
|
# Search for any mp3 file created in this run
|
|
for fname in os.listdir(output_folder):
|
|
if fname.startswith(base) and fname.endswith(".mp3"):
|
|
final_path = os.path.join(output_folder, f"{base}__mp3.mp3")
|
|
src = os.path.join(output_folder, fname)
|
|
if src != final_path and not os.path.exists(final_path):
|
|
os.rename(src, final_path)
|
|
return final_path
|
|
|
|
raise RuntimeError(f"MP3 file not found after download for {video_id}")
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# Artifact: MP4
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def _find_ffmpeg() -> Optional[str]:
|
|
"""Find ffmpeg binary location."""
|
|
# Common Windows locations
|
|
candidates = [
|
|
r"C:\ProgramData\chocolatey\bin",
|
|
r"C:\ffmpeg\bin",
|
|
r"C:\Program Files\ffmpeg\bin",
|
|
r"C:\Program Files (x86)\ffmpeg\bin",
|
|
]
|
|
for folder in candidates:
|
|
if os.path.isfile(os.path.join(folder, "ffmpeg.exe")):
|
|
return folder
|
|
# Try PATH
|
|
import shutil
|
|
found = shutil.which("ffmpeg")
|
|
if found:
|
|
return str(Path(found).parent)
|
|
return None
|
|
|
|
|
|
def extract_mp4(video_id: str, meta: Dict, output_folder: str, resolution: str = "best") -> str:
|
|
"""Download video as MP4 using yt-dlp. Explicitly uses ffmpeg for merging."""
|
|
title = meta.get("title", "")
|
|
base = make_base_name(title, video_id)
|
|
ensure_dir(output_folder)
|
|
|
|
ffmpeg_dir = _find_ffmpeg()
|
|
|
|
# Build format selector — always require vcodec to avoid audio-only streams
|
|
if resolution == "best":
|
|
fmt = (
|
|
"bestvideo[vcodec^=avc1][ext=mp4]+bestaudio[ext=m4a]"
|
|
"/bestvideo[ext=mp4]+bestaudio[ext=m4a]"
|
|
"/bestvideo+bestaudio"
|
|
"/best[vcodec!=none]"
|
|
"/best"
|
|
)
|
|
else:
|
|
fmt = (
|
|
f"bestvideo[vcodec^=avc1][height<={resolution}][ext=mp4]+bestaudio[ext=m4a]"
|
|
f"/bestvideo[height<={resolution}][ext=mp4]+bestaudio[ext=m4a]"
|
|
f"/bestvideo[height<={resolution}]+bestaudio"
|
|
f"/best[height<={resolution}][vcodec!=none]"
|
|
f"/best[height<={resolution}]"
|
|
f"/best[vcodec!=none]"
|
|
f"/best"
|
|
)
|
|
|
|
out_template = os.path.join(output_folder, f"{base}__mp4.%(ext)s")
|
|
|
|
ydl_opts = {
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
"format": fmt,
|
|
"outtmpl": out_template,
|
|
"merge_output_format": "mp4",
|
|
"ignoreerrors": False,
|
|
"postprocessors": [
|
|
{"key": "FFmpegVideoConvertor", "preferedformat": "mp4"},
|
|
],
|
|
}
|
|
|
|
# Explicitly tell yt-dlp where ffmpeg is
|
|
if ffmpeg_dir:
|
|
ydl_opts["ffmpeg_location"] = ffmpeg_dir
|
|
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
ydl.download([video_url(video_id)])
|
|
|
|
# Find the downloaded file
|
|
for fname in sorted(os.listdir(output_folder)):
|
|
if fname.startswith(f"{base}__mp4") and fname.endswith(".mp4"):
|
|
return os.path.join(output_folder, fname)
|
|
|
|
raise RuntimeError(f"MP4 file not found after download for {video_id}. ffmpeg_dir={ffmpeg_dir}")
|