#! /usr/bin/env python3
import copy
import dataclasses
import json
import logging
import os.path
import pathlib
import re
import time
import traceback
from abc import ABC
from dataclasses import dataclass, field
from datetime import timedelta
from typing import Tuple, List, Dict, Optional
from urllib.parse import urlparse, parse_qs
import cachetools
import yt_dlp.utils
from cachetools.keys import hashkey
from sqlalchemy.orm import Session
from yt_dlp import YoutubeDL, DownloadError
from yt_dlp.extractor import YoutubeTabIE # noqa
from wrolpi.captions import strip_youtube_caption_positioning
from wrolpi.cmd import YT_DLP_BIN
from wrolpi.common import logger, get_media_directory, escape_file_name, resolve_generators, background_task, \
trim_file_name, cached_multiprocessing_result, get_absolute_media_path, aiohttp_post, normalize_domain
from wrolpi.dates import now
from wrolpi.db import get_db_session
from wrolpi.downloader import Downloader, Download, DownloadContext, DownloadResult, \
clear_download_progress, _parse_size, make_progress_callback, get_download_cache_config
from wrolpi.errors import UnrecoverableDownloadError, BotBlockedDownloadError, ValidationError
from wrolpi.files.lib import glob_shared_stem, split_path_stem_and_suffix
from wrolpi.files.models import FileGroup
from wrolpi.files.worker import file_worker
from wrolpi.vars import PYTEST, YTDLP_CACHE_DIR
from .channel.lib import create_channel, get_channel
from .common import get_no_channel_directory, update_view_counts_and_censored, \
ffmpeg_video_complete, get_or_create_ffprobe_json
from .errors import UnknownChannel, ChannelDirectoryConflict, ChannelNameConflict, ChannelURLConflict, \
ChannelSourceIdConflict
from .cookies import cookies_unlocked, cookies_for_download
from .lib import get_videos_downloader_config, get_effective_video_settings, CHANNEL_INHERITABLE_SETTINGS, YDL, \
ydl_logger, format_videos_destination, get_yt_dlp_http_headers, get_yt_dlp_sleep_opts
from .models import Video, Channel
from .normalize_video_url import normalize_video_url
from .schema import ChannelPostRequest
from .transcode import codecs_match, get_transcode_target, transcode_video_file, \
transcode_can_satisfy_codecs, TRANSCODE_VIDEO_TARGETS, TRANSCODE_AUDIO_TARGETS
from .video.lib import download_video_info_json
logger = logger.getChild(__name__)
YTDLP_PROGRESS_RE = re.compile(
r'\[download\]\s+'
r'([\d.]+)%\s+' # percent
r'of\s+~?\s*([\d.]+\w+)\s+' # total size (with optional ~)
r'at\s+([\d.]+\w+/s)\s+' # speed
r'ETA\s+(\S+)' # ETA
)
def parse_ytdlp_progress(line: str) -> dict | None:
"""Parse a yt-dlp progress line (with --newline) and return a dict, or None if not a progress line."""
m = YTDLP_PROGRESS_RE.search(line)
if not m:
return None
percent = float(m.group(1))
total_bytes = _parse_size(m.group(2))
speed_str = m.group(3).rstrip('/s')
speed = _parse_size(speed_str)
eta = m.group(4)
if eta == 'Unknown':
eta = None
return dict(
bytes_downloaded=int(percent / 100.0 * total_bytes),
total_bytes=total_bytes,
percent=int(percent),
speed=speed,
eta=eta,
)
def _log_ytdlp_versions():
"""Log yt-dlp and yt-dlp-ejs versions for debugging download issues."""
try:
import yt_dlp.version
ytdlp_version = yt_dlp.version.__version__
except Exception:
ytdlp_version = 'unknown'
try:
from importlib.metadata import version
ejs_version = version('yt-dlp-ejs')
except Exception:
ejs_version = 'not installed'
logger.info(f'yt-dlp={ytdlp_version} yt-dlp-ejs={ejs_version}')
# yt-dlp codec names that differ from their output file extension.
AUDIO_CODEC_EXTENSION_MAP = {
'vorbis': 'ogg',
}
def audio_output_extension(audio_format: str) -> str:
"""Map a yt-dlp audio codec name to the file extension it actually produces."""
return AUDIO_CODEC_EXTENSION_MAP.get(audio_format, audio_format)
VIDEO_RESOLUTION_MAP = {
'360p': [
'134+bestaudio', '230+bestaudio', 'mp4-360p', 'res:360',
'hls-175+bestaudio', # 360p avc1 + best audio
'396+bestaudio', '243+bestaudio', # Vertical
'hls-655+bestaudio', 'hls-655',
'bestvideo[height=360]+bestaudio/bestvideo[width=360]+bestaudio/best[height=360]',
],
'480p': [
'hls-96', # 480p (sort of) Vertical
'135+bestaudio', '231+bestaudio', '135+139', 'mp4-480p', 'res:480',
'hls-312+bestaudio', # 480p avc1 + best audio
'788+bestaudio', '780+bestaudio', '397+bestaudio', # Vertical
'bestvideo[height=480]+bestaudio/bestvideo[width=480]+bestaudio/best[height=480]',
],
'720p': [
'hls-258+bestaudio', # 720p (sort of) Vertical
'136+bestaudio', '22', '311+bestaudio', 'res:720',
'298+bestaudio', '232+bestaudio', # 720@60 avc1 + best audio
'hls-517+bestaudio', 'hls-637+bestaudio', # 720p avc1 + best audio
'398+bestaudio', '247+bestaudio', # Vertical
'hls-2890+bestaudio', 'hls-2890',
'bestvideo[height=720]+bestaudio/bestvideo[width=720]+bestaudio/best[height=720]',
],
'1080p': [
'hls-509+bestaudio', # 1080p Vertical
'137+bestaudio', '614+bestaudio', 'res:1080',
'299+bestaudio', '312+bestaudio', '270+bestaudio', # 1080@60 avc1 + best audio
'hls-899+bestaudio', 'hls-1271+bestaudio', # 1080p avc1 + best audio
'399+bestaudio', '616+bestaudio', # Vertical
'hls-4026+bestaudio', 'hls-4026',
'bestvideo[height=1080]+bestaudio/bestvideo[width=1080]+bestaudio/best[height=1080]',
],
'1440p': [
'639+bestaudio', # vp9, best audio, hdr
'623+bestaudio', # vp9, best audio, no hdr
'620+bestaudio', 'res:1440',
'bestvideo[height=1440]+bestaudio/bestvideo[width=1440]+bestaudio/best[height=1440]',
],
'2160p': [
'642+bestaudio', # vp9, best audio, hdr
'628+bestaudio', # vp9, best audio, no hdr
'625+bestaudio', 'res:2160',
'bestvideo[height=2160]+bestaudio/bestvideo[width=2160]+bestaudio/best[height=2160]',
],
'maximum': ['bestvideo*+bestaudio/best'],
}
# yt-dlp `vcodec`/`acodec` values vary by extractor (avc1.64001f, vp09.00.10.08, mp4a.40.2, ...);
# match them by prefix regex. These are used inside `-f` bracket filters, so they must never
# contain a comma or a forward slash.
VIDEO_CODEC_YT_DLP_PATTERNS = {
'h264': r'^(avc|h264)',
'hevc': r'^(hev|hvc|h265|hevc)',
'vp9': r'^vp0?9',
'av1': r'^av0?1',
'vp8': r'^vp0?8',
}
AUDIO_CODEC_YT_DLP_PATTERNS = {
'aac': r'^(aac|mp4a\.40)',
'opus': r'^opus',
'mp3': r'^(mp3|mp4a\.69|mp4a\.6[bB])', # extractors report both mp4a.6b and mp4a.6B
'vorbis': r'^vorbis',
}
RESOLUTION_HEIGHTS = {'360p': 360, '480p': 480, '720p': 720, '1080p': 1080,
'1440p': 1440, '2160p': 2160, 'maximum': None}
# VTT is the only format that a browser can use to display captions on a .
DEFAULT_CAPTION_FORMAT = 'vtt'
DEFAULT_POSTER_FORMAT = 'jpg'
DEFAULT_CHANNEL_DOWNLOAD_ORDER = 'newest'
def format_selector(video_resolutions: List[str], video_codecs: List[str] = None,
audio_codecs: List[str] = None, strict_codecs: bool = False) -> str:
"""Flatten the requested resolutions (and codec preferences) into a single yt-dlp `-f`
format selector.
The selectors are joined with "/" so yt-dlp downloads only the first one that matches.
Joining with "," would instead ask yt-dlp to download *every* matching selector, which
downloads (and post-processes) the same video once per match.
Codec preference is codec-major: a preferred codec at a lower resolution beats a
non-preferred codec at a higher resolution. Unless `strict_codecs`, the legacy
resolution selectors are appended as a fallback so a video is still downloaded when no
preferred codec is available.
"""
legacy = '/'.join(j for i in video_resolutions for j in VIDEO_RESOLUTION_MAP[i])
if not video_codecs and not audio_codecs:
return legacy
audio_selectors = [f"bestaudio[acodec~='{AUDIO_CODEC_YT_DLP_PATTERNS[i]}']" for i in audio_codecs or []]
if not strict_codecs or not audio_codecs:
audio_selectors.append('bestaudio')
video_filters = [f"[vcodec~='{VIDEO_CODEC_YT_DLP_PATTERNS[i]}']" for i in video_codecs] \
if video_codecs else ['']
selectors = []
for video_filter in video_filters:
for resolution in video_resolutions:
height = RESOLUTION_HEIGHTS[resolution]
dimensions = [f'[height={height}]', f'[width={height}]'] if height else ['']
for dimension in dimensions:
for audio_selector in audio_selectors:
selectors.append(f'bestvideo{dimension}{video_filter}+{audio_selector}')
if not strict_codecs:
selectors.append(legacy)
return '/'.join(selectors)
async def enforce_codecs(video_path: pathlib.Path, video_paths: List[pathlib.Path], effective: dict) \
-> Tuple[pathlib.Path, List[pathlib.Path]]:
"""Verify the downloaded file's codecs against the effective codec preferences.
The format selector already prefers matching sources, but the legacy fallback selectors (and
lying extractor metadata) can still deliver a non-preferred codec. On mismatch: transcode when
enabled, fail permanently when strict, otherwise keep the file (best-effort preference).
Runs before the Video record is created and holds no DB session, so ffmpeg may run for a long
time without blocking writers.
"""
video_codecs = effective.get('video_codecs') or []
audio_codecs = effective.get('audio_codecs') or []
transcode = effective.get('transcode', False)
strict_codecs = effective.get('strict_codecs', False)
data, _ = await get_or_create_ffprobe_json(video_path)
video_match, audio_match = codecs_match(data, video_codecs, audio_codecs)
if video_match and audio_match:
return video_path, video_paths
if transcode:
target_vcodec = None if video_match else get_transcode_target(video_codecs, TRANSCODE_VIDEO_TARGETS)
target_acodec = None if audio_match else get_transcode_target(audio_codecs, TRANSCODE_AUDIO_TARGETS)
# Transcode only when it can fix every mismatched stream: converting just the audio of a
# video whose codec preference has no target would report success while leaving the video
# codec wrong. An unfixable mismatch falls through to strict/keep below.
if (video_match or target_vcodec) and (audio_match or target_acodec):
new_path = await transcode_video_file(video_path, target_vcodec, target_acodec,
container=effective.get('video_format') or 'mp4')
await get_or_create_ffprobe_json(new_path) # Refresh the sidecar for the new streams.
video_paths = [new_path if i == video_path else i for i in video_paths]
return new_path, video_paths
logger.warning(f'Cannot transcode {video_path}: a mismatched preference'
f' (video={video_codecs}, audio={audio_codecs}) has no supported transcode target')
if strict_codecs:
actual_video = [i.get('codec_name') for i in data.get('streams', []) if i.get('codec_type') == 'video']
actual_audio = [i.get('codec_name') for i in data.get('streams', []) if i.get('codec_type') == 'audio']
raise UnrecoverableDownloadError(
f'Downloaded video has codecs (video={actual_video}, audio={actual_audio}) but strict codecs'
f' (video={video_codecs}, audio={audio_codecs}) was required and transcoding cannot produce them')
logger.warning(f'{video_path} does not match preferred codecs'
f' (video={video_codecs}, audio={audio_codecs}); keeping it as downloaded')
return video_path, video_paths
def get_youtube_video_id(url: str) -> Optional[str]:
"""Extract YouTube video ID from various URL formats."""
parsed = urlparse(url)
# Handle youtu.be/VIDEO_ID
if parsed.netloc in ('youtu.be', 'www.youtu.be'):
return parsed.path.lstrip('/')
# Handle youtube.com/watch?v=VIDEO_ID
if parsed.netloc in ('youtube.com', 'www.youtube.com', 'm.youtube.com'):
if parsed.path == '/watch':
query = parse_qs(parsed.query)
if 'v' in query:
return query['v'][0]
# Handle youtube.com/shorts/VIDEO_ID
if parsed.path.startswith('/shorts/'):
return parsed.path.split('/shorts/')[1].split('/')[0]
# Handle youtube.com/v/VIDEO_ID
if parsed.path.startswith('/v/'):
return parsed.path.split('/v/')[1].split('/')[0]
# Handle youtube.com/embed/VIDEO_ID
if parsed.path.startswith('/embed/'):
return parsed.path.split('/embed/')[1].split('/')[0]
return None
YOUTUBE_HOSTS = ('youtube.com', 'youtu.be')
def is_youtube_url(url: str) -> bool:
"""Return True if the URL's host is a YouTube domain (youtube.com / youtu.be) or a subdomain of one.
Parses the hostname so unrelated hosts like `youtube.com.example.org` or `notyoutube.com` are rejected."""
if not url:
return False
host = (urlparse(url).hostname or '').lower()
return any(host == h or host.endswith(f'.{h}') for h in YOUTUBE_HOSTS)
async def get_youtube_duration(video_url: str) -> Optional[int]:
"""Get video duration using YouTube's internal API. Returns None if URL is not YouTube or API fails."""
video_id = get_youtube_video_id(video_url)
if not video_id:
logger.debug(f'Could not extract YouTube video ID from {video_url}')
return None
payload = {
"context": {
"client": {
"clientName": "WEB",
"clientVersion": "2.20250116.10.00"
}
},
"videoId": video_id
}
try:
async with aiohttp_post("https://www.youtube.com/youtubei/v1/player", payload, timeout=10) as response:
response.raise_for_status()
data = await response.json()
length_seconds = data.get("videoDetails", {}).get("lengthSeconds")
if length_seconds is not None:
return int(length_seconds)
else:
# Log why we couldn't get duration
status = data.get("playabilityStatus", {}).get("status", "UNKNOWN")
reason = data.get("playabilityStatus", {}).get("reason", "")
logger.warning(f'YouTube API returned no duration for {video_url}: status={status} reason={reason}')
except Exception as e:
logger.debug(f'YouTube API request failed for {video_url}: {e}')
return None
def extract_info_key(url: str, ydl: YoutubeDL = YDL, process: bool = False):
return hashkey(url, process=process)
extract_info_cache = cachetools.TTLCache(maxsize=1_000, ttl=timedelta(minutes=5).total_seconds())
# Cache results for each URL for a few minutes.
@cachetools.cached(cache=extract_info_cache, key=extract_info_key)
def extract_info(url: str, ydl: YoutubeDL = YDL, process: bool = False) -> dict:
"""Get info about a video, channel, or playlist. Separated for testing."""
if PYTEST:
raise RuntimeError(f'Refusing to download {url} during testing! {ydl}')
if cookies_unlocked():
logger.info(f'Using encrypted cookies for extract_info: {url}')
with cookies_for_download() as cookies_path:
ydl_opts = dict(
cachedir=YTDLP_CACHE_DIR,
cookiefile=str(cookies_path),
**get_yt_dlp_sleep_opts(),
)
http_headers = get_yt_dlp_http_headers()
if http_headers:
ydl_opts['http_headers'] = http_headers
# If a custom ydl was passed, copy its relevant options for format selection and output
if ydl != YDL:
for key in ['outtmpl', 'format', 'merge_output_format']:
if key in ydl.params:
ydl_opts[key] = ydl.params[key]
ydl_with_cookies = YoutubeDL(ydl_opts)
ydl_with_cookies.params['logger'] = logger.getChild('youtube-dl')
ydl_with_cookies.add_default_info_extractors()
return ydl_with_cookies.extract_info(url, download=False, process=process)
logger.info(f'No cookies available for extract_info: {url}')
# Create a new YDL instance with sleep options
ydl_opts = dict(
cachedir=YTDLP_CACHE_DIR,
**get_yt_dlp_sleep_opts(),
)
if ydl != YDL:
for key in ['outtmpl', 'format', 'merge_output_format']:
if key in ydl.params:
ydl_opts[key] = ydl.params[key]
ydl = YoutubeDL(ydl_opts)
ydl.params['logger'] = logger.getChild('youtube-dl')
ydl.add_default_info_extractors()
return ydl.extract_info(url, download=False, process=process)
def prepare_video_filename(url: str, out_dir: pathlib.Path, video_resolutions: str, video_format: str,
audio_only: bool = False, audio_format: str = 'mp3') \
-> Tuple[pathlib.Path, dict]:
"""Predict the on-disk path of a video/audio file from its URL using yt-dlp.
Module-level (rather than a method on VideoDownloader) so tests can patch it once
and avoid the descriptor-binding pitfalls of patching `Class.method`. Returns the
absolute path the file will land at plus the yt-dlp entry dict for the URL.
"""
if not out_dir.is_dir():
raise ValueError(f'Output directory does not exist! {out_dir=}')
# YoutubeDL expects specific options, add onto the default options.
# Deep copy to avoid mutating the shared singleton config dict.
config = get_videos_downloader_config()
options = copy.deepcopy(config.yt_dlp_options)
if audio_only:
options['format'] = 'bestaudio'
options['postprocessors'] = [{
'key': 'FFmpegExtractAudio',
'preferredcodec': audio_format,
}]
# merge_output_format is not used for audio-only.
options.pop('merge_output_format', None)
else:
options['merge_output_format'] = video_format
options['format'] = video_resolutions
# Convert WROLPi-specific variables to yt-dlp syntax
converted_format = convert_wrolpi_filename_format(config.file_name_format)
# yt-dlp expects the absolute path.
options['outtmpl'] = f'{out_dir}/{converted_format}'
options['cachdir'] = YTDLP_CACHE_DIR
output_ext = audio_output_extension(audio_format) if audio_only else video_format
# Create a new YoutubeDL for the output directory.
ydl = YoutubeDL(copy.deepcopy(options))
ydl.params['logger'] = ydl_logger
ydl.add_default_info_extractors()
# Get the path where the file will be saved.
try:
entry = extract_info(url, ydl=ydl, process=True)
# If the entry is a playlist (e.g. Twitter/X post with multiple videos),
# use the first entry for filename preparation.
if entry.get('entries'):
entries = list(entry['entries']) if not isinstance(entry['entries'], list) else entry['entries']
if entries:
entry = entries[0]
final_filename = pathlib.Path(prepare_filename(entry, ydl=ydl)).absolute()
except DownloadError as e:
if ' Cannot write ' in str(e):
# yt-dlp does not handle long file names well, get the name from the error (lol)
last_line = str(e).splitlines()[-1]
full_path = last_line.split(' file ')[-1].strip()
if not full_path.startswith('/'):
logger.error(f'Failed to extract filename from {last_line}')
raise
else:
raise
# Split filename from parent.
full_path = pathlib.Path(full_path)
parent = full_path.parent
filename, _ = split_path_stem_and_suffix(full_path.name)
# Trim long filename, add output suffix, add back into parent directory.
filename = escape_file_name(filename)
final_filename = parent / f'{filename}.{output_ext}'
final_filename = trim_file_name(final_filename)
logger.debug(f'File name was too long. Trimmed to: {final_filename.name}')
# Get entry info json.
options['outtmpl'] = str(final_filename)
ydl = YoutubeDL(copy.deepcopy(options))
ydl.params['logger'] = ydl_logger
ydl.add_default_info_extractors()
entry = extract_info(url, ydl=ydl, process=True)
logger.debug(f'Downloading {url} to {repr(str(final_filename))}')
return final_filename, entry
def prepare_filename(entry: dict, ydl: YoutubeDL = YDL) -> str:
"""Get filename from YoutubeDL. Separated for testing."""
dir_name, file_name = os.path.split(ydl.prepare_filename(entry))
file_name = escape_file_name(file_name)
file_name = trim_file_name(file_name)
return f'{dir_name}/{file_name}'
def convert_wrolpi_filename_format(filename_format: str) -> str:
"""Convert WROLPi-specific filename variables to yt-dlp syntax.
WROLPi provides user-friendly variables that map to yt-dlp's strftime syntax:
%(upload_year)s -> %(upload_date>%Y)s
%(upload_month)s -> %(upload_date>%m)s
This allows users to organize videos into year/month subdirectories.
Example:
'%(upload_year)s/%(upload_month)s/%(title)s.%(ext)s'
-> '%(upload_date>%Y)s/%(upload_date>%m)s/%(title)s.%(ext)s'
"""
conversions = {
'%(upload_year)s': '%(upload_date>%Y)s',
'%(upload_month)s': '%(upload_date>%m)s',
}
result = filename_format
for wrolpi_var, ytdlp_var in conversions.items():
result = result.replace(wrolpi_var, ytdlp_var)
return result
def preview_filename(filename_format: str) -> str:
"""Return an example video file name using simple string formatting.
This is a lightweight implementation that doesn't require YoutubeDL instantiation,
which is expensive (~1500 extractors loaded, regexes compiled).
@raise RuntimeError: If the format is invalid.
"""
if not filename_format.endswith('.%(ext)s'):
raise RuntimeError('Filename must end with .%(ext)s')
# Convert WROLPi-specific variables to yt-dlp syntax
converted_format = convert_wrolpi_filename_format(filename_format)
sample_date = now()
variables = dict(
uploader='WROLPi',
channel='WROLPi',
upload_date=sample_date.strftime('%Y%m%d'),
id='Qz-FuenRylQ',
title='The title of the video',
ext='mp4',
description='A description of the video',
timestamp=int(sample_date.timestamp()),
)
# Handle yt-dlp's date extraction syntax %(upload_date>%Y)s and %(upload_date>%m)s
# Replace these with their values before standard Python % formatting
def replace_date_format(match):
date_format = match.group(1)
return sample_date.strftime(date_format)
converted_format = re.sub(r'%\(upload_date>([^)]+)\)s', replace_date_format, converted_format)
try:
filename = converted_format % variables
return filename.lstrip('/')
except KeyError as e:
raise RuntimeError(f'Invalid variable: {e}')
VIDEO_DURATION_CACHE_MAX_SIZE = 1_000
def _cache_video_duration(url: str, duration: int):
"""Persist a video duration to the download cache config. Evicts oldest entries when over the max size."""
cache_config = get_download_cache_config()
entries = [e for e in cache_config.video_durations if e[0] != url]
entries.append([url, duration])
if len(entries) > VIDEO_DURATION_CACHE_MAX_SIZE:
entries = entries[-VIDEO_DURATION_CACHE_MAX_SIZE:]
cache_config.video_durations = entries
@cached_multiprocessing_result
async def fetch_video_duration(url: str) -> int:
"""Get video duration in seconds. Checks persistent cache, then DB, then YouTube API, then yt-dlp."""
# Check persistent cache first (survives restarts).
cached_duration = dict(get_download_cache_config().video_durations).get(url)
if cached_duration is not None:
logger.debug(f'Using cached duration {cached_duration} for {url}')
return cached_duration
with get_db_session() as session:
# Join with Video to get duration; a URL may have other files (such as an Archive).
video = Video.get_by_url(session, url)
if video and (duration := video.file_group.length):
logger.debug(f'Using already-known duration {duration} for {url} from FileGroup')
_cache_video_duration(url, duration)
return duration
# Try YouTube API first (faster than yt-dlp)
duration = await get_youtube_duration(url)
if duration is not None:
logger.info(f'Fetched video duration of {duration} from {url} using YouTube API')
_cache_video_duration(url, duration)
return duration
# Fall back to yt-dlp for non-YouTube URLs or API failures
logger.debug(f'Fetching video duration from {url} using yt-dlp')
info = extract_info(url, process=False)
duration = info['duration']
logger.info(f'Fetched video duration of {duration} from {url}')
_cache_video_duration(url, duration)
return duration
YOUTUBE_RSS_FEED_URL_RE = re.compile(
r'^https?://([\w.-]+\.)?youtube\.com/feeds/videos\.xml(\?|/|$)',
re.IGNORECASE,
)
def is_youtube_rss_feed_url(url: str) -> bool:
"""Return True if `url` is a YouTube RSS feed URL (e.g. /feeds/videos.xml?channel_id=...)."""
return bool(YOUTUBE_RSS_FEED_URL_RE.match(url or ''))
@dataclass
class PreparedChannel:
"""Plan produced by ChannelDownloader.prepare_download.
Cheap pre-flight: validates the URL is not an RSS feed, snapshots the settings
fields execute and finalize need so they don't have to read back through the
Download row. All real work — yt-dlp metadata fetch, channel resolution, file
indexing — happens in execute_download.
"""
url: str
settings: dict
destination: Optional[str]
tag_names: Optional[List[str]]
@dataclass
class ExecutedChannel:
"""Output of ChannelDownloader.execute_download.
`info_json` is the resolved yt-dlp metadata; `channel_id` and `collection_id` are
the IDs assigned by the channel-resolution helper. `child_settings` is the dict
that will be passed to every child VideoDownloader. `error` is set when the
catalog update raised — finalize converts that into a failure DownloadResult
while still mutating the parent Download with sub_downloader/info_json so the
next attempt has them.
"""
info_json: Optional[dict]
channel_id: Optional[int]
collection_id: Optional[int]
location: Optional[str]
child_settings: dict
error: Optional[str] = None
class ChannelDownloader(Downloader, ABC):
"""Handles downloading of videos in a Channel or Playlist."""
name = 'video_channel'
pretty_name = 'Video Channel'
listable = False
def __repr__(self):
return f''
def normalize_domain(self, url: str) -> str:
# Collapse youtu.be / youtube.com/shorts / tracking params to the canonical youtube.com domain.
return normalize_domain(normalize_video_url(url))
@staticmethod
def is_a_playlist(info: dict):
# A playlist may have an id different from its channel.
return info['id'] != info.get('channel_id')
def prepare_download(self, session: Session, download: Download) -> PreparedChannel:
if is_youtube_rss_feed_url(download.url):
# yt-dlp returns no `uploader` or `channel_id` for an RSS feed URL, so falling through
# would create a junk channel named after `webpage_url_basename` ("videos.xml").
raise UnrecoverableDownloadError(
f'Refusing to handle RSS feed URL with ChannelDownloader: {download.url}.'
f' Use the RSS Downloader (sub_downloader=video) instead.'
)
return PreparedChannel(
url=download.url,
settings=download.settings or dict(),
destination=str(download.destination) if download.destination else None,
tag_names=download.tag_names,
)
async def execute_download(self, prepared: PreparedChannel, ctx: DownloadContext,
download: Download = None) -> ExecutedChannel:
"""Fetch the channel/playlist metadata via yt-dlp, get/create the local Channel
record, and update its catalog. Channel creation + catalog update use a
short-lived self-managed session so the manager's outer session isn't held
during the file_worker await.
"""
_log_ytdlp_versions()
# Prefer ctx.extract_info when provided so tests can inject a fake without
# patching the module-level function. Falls back to the global for back-compat.
_extract = ctx.extract_info if ctx.extract_info else extract_info
try:
info = _extract(prepared.url, process=False)
except Exception as e:
if _bot_blocked(e):
raise BotBlockedDownloadError(str(e)) from e
raise
# Resolve the "entries" generator.
info: dict = resolve_generators(info)
entries = info.get('entries')
if entries and isinstance(entries[0], Dict) and entries[0].get('entries'):
# Entries contains more channels/playlist, tell the Maintainer to specify the URL.
raise UnrecoverableDownloadError(f'Unable to download {prepared.url} because the URL is ambiguous.'
f' Try requesting the videos only:'
f' https://example.com -> https://example.com/videos')
channel_tag_name = i[0] if (i := prepared.settings.get('channel_tag_name')) else None
name = info.get('uploader') or info.get('webpage_url_basename')
if not name:
raise ValueError(f'Could not find name')
channel_source_id = info.get('channel_id') or info.get('id')
channel_url = info.get('channel_url') or info.get('uploader_url') or prepared.url
channel_id = None
collection_id = None
location = None
with get_db_session(commit=True) as session:
channel = get_or_create_channel(session, channel_source_id, channel_url, name, channel_tag_name,
destination=prepared.destination)
if channel and channel.id:
channel_id = channel.id
collection_id = channel.collection_id
location = f'/videos/channel/{channel.id}/video'
# The settings to send to the VideoDownloader children.
child_settings = dict()
if channel_id:
child_settings.update(dict(collection_id=collection_id, channel_url=prepared.url))
if prepared.destination:
destination_abs = get_absolute_media_path(prepared.destination)
child_settings['destination'] = str(destination_abs) # Need str for JSON conversion
if prepared.tag_names:
child_settings['tag_names'] = prepared.tag_names
# Pass all channel-level inheritable settings to child video downloads.
for key in CHANNEL_INHERITABLE_SETTINGS:
if key in prepared.settings:
child_settings[key] = prepared.settings[key]
is_a_playlist = self.is_a_playlist(info)
try:
if not is_a_playlist and channel_id is not None:
await self.prepare_channel_for_downloads(info, channel_id)
elif is_a_playlist:
logger.debug('Not updating channel because this is a playlist')
except Exception as e:
if PYTEST:
raise
kind = 'playlist' if is_a_playlist else 'channel'
logger.error(f'Failed to update catalog of {kind} {prepared.url}', exc_info=e)
return ExecutedChannel(
info_json=info, channel_id=channel_id, collection_id=collection_id,
location=location, child_settings=child_settings,
error=str(traceback.format_exc()),
)
return ExecutedChannel(
info_json=info, channel_id=channel_id, collection_id=collection_id,
location=location, child_settings=child_settings,
)
def finalize_download(self, session: Session, download: Download,
executed: ExecutedChannel) -> DownloadResult:
"""Mutate the parent Download (sub_downloader / info_json / collection_id) and
return the list of missing video URLs as child downloads."""
download.sub_downloader = video_downloader.name
download.info_json = executed.info_json
if executed.collection_id and not download.collection_id:
download.collection_id = executed.collection_id
if executed.error:
return DownloadResult(
success=False,
location=executed.location,
error=executed.error,
settings=executed.child_settings,
)
downloads = self.get_missing_videos(session, download)
return DownloadResult(
success=True,
location=executed.location,
downloads=downloads,
settings=executed.child_settings,
)
@classmethod
async def prepare_channel_for_downloads(cls, info_json: dict, channel_id: int):
"""Update the Channel's video catalog. Refresh the Channel's files if necessary.
Takes channel_id (not Channel) and info_json (not Download) so the caller doesn't
have to hold the manager's session open while we await the file_worker job.
"""
with get_db_session() as session:
channel = get_channel(session, channel_id=channel_id, return_dict=False)
logger.debug(f'Preparing {channel} for downloads')
update_channel_catalog(channel, info_json)
if not channel.videos or not channel.refreshed:
logger.warning(f'Refreshing videos in {channel.directory} for channel {channel}')
job_id = channel.refresh_files(channel.id, send_events=False)
await file_worker.wait_for_job(job_id) # Wait for files to be indexed
session.refresh(channel)
@staticmethod
def get_missing_videos(session: Session, download: Download) -> List[str]:
"""
Return all URLs of Videos in the `info_json` which need to be downloaded.
"""
downloads = download.info_json['entries']
total_downloads = len(downloads)
settings = download.settings or dict()
title_exclude = settings.get('title_exclude', '')
title_exclude = [i.lower() for i in title_exclude.split(',') if i]
title_include = settings.get('title_include', '')
title_include = [i.lower() for i in title_include.split(',') if i]
if title_exclude or title_include:
new_downloads = []
for i in downloads:
title = i.get('title', '').lower()
if title and title_exclude and any(i in title for i in title_exclude):
logger.debug(f'Video with title {str(repr(title))} matches {title_exclude=}')
continue
if title and title_include and not any(i in title for i in title_include):
logger.debug(f'Video with title {str(repr(title))} matches {title_include=}')
continue
new_downloads.append(i)
downloads = new_downloads
# Filter videos by their length.
try:
if minimum_duration := settings.get('minimum_duration'):
downloads = [i for i in downloads if (d := i.get('duration')) and int(d) >= minimum_duration]
if maximum_duration := settings.get('maximum_duration'):
downloads = [i for i in downloads if (d := i.get('duration')) and int(d) <= maximum_duration]
except KeyError as e:
raise RuntimeError('Unable to filter videos because there is no duration') from e
# Sort videos now that we have reduced the number necessary to sort, but before we limit count.
sort_key = settings.get('download_order', DEFAULT_CHANNEL_DOWNLOAD_ORDER)
# Default is videos are sorted by newest first.
if sort_key == 'oldest':
logger.debug(f'Downloading oldest videos from {download.url}')
downloads = downloads.reverse()
elif sort_key == 'views':
# Listing view counts are approximate ("3K" -> 3000); good enough for ordering.
logger.debug(f'Downloading most viewed videos from {download.url}')
if not any(i.get('view_count') is not None for i in downloads):
# Old yt-dlp (< 2026.07) does not extract view_count from YouTube's lockupViewModel
# listings. Fail cleanly rather than downloading in arbitrary order.
raise UnrecoverableDownloadError(
'yt-dlp did not report view counts for this channel/playlist. Update yt-dlp '
'(or change this download to "Newest"/"Oldest").')
downloads = sorted(downloads, key=lambda i: i.get('view_count') or 0, reverse=True)
# Limit the videos that will be downloads (this allows the user to download the "top 100" videos of a Channel)
if video_count_limit := settings.get('video_count_limit'):
logger.debug(f'Limiting video count to {video_count_limit}: {download.url}')
video_count_limit = int(video_count_limit)
downloads = downloads[:video_count_limit]
filtered_entries = len(downloads)
if filtered_entries != total_downloads:
logger.info(f'Downloaded channel. Total downloaded reduced to {filtered_entries} from {total_downloads}')
# Prefer `webpage_url` before `url` for all entries.
downloads = [i.get('webpage_url') or i.get('url') for i in downloads]
# YouTube Shorts are handled specially.
downloads = [normalize_video_url(i) for i in downloads]
# Only download those that have not yet been downloaded.
already_downloaded = [i.url for i in video_downloader.already_downloaded(session, *downloads)]
downloads = [i for i in downloads if i not in already_downloaded]
return downloads
@dataclass
class PreparedVideo:
"""Plan produced by VideoDownloader.prepare_download.
`info_json` is None unless the upstream download row already has it (e.g. set by
ChannelDownloader before queuing this child). `location` is the existing
Video.location if any, used for error attribution if the new download fails.
"""
url: str # normalized
settings: dict
tag_names: List[str] = field(default_factory=list)
destination: Optional[pathlib.Path] = None
info_json: Optional[dict] = None
location: Optional[str] = None # existing video, for error reports
wrolpi: dict = field(default_factory=dict) # existing video's wrolpi section, to carry over
@dataclass
class ExecutedVideo:
"""Output of VideoDownloader.execute_download once the file is on disk and validated.
Carries everything finalize_download needs to persist the Video record without
re-running any I/O. When the metadata-only flow runs, execute returns a
DownloadResult directly and the dispatch skips finalize_download entirely.
"""
url: str # normalized
video_path: pathlib.Path
video_paths: List[pathlib.Path] # all files in the FileGroup, post-glob+normalize
info_json: dict
entry: dict
channel_id: Optional[int]
audio_only: bool
tag_names: List[str]
settings: dict
location: Optional[str] # for error reporting
wrolpi: dict = field(default_factory=dict) # existing video's wrolpi section, to carry over
class VideoDownloader(Downloader, ABC):
"""Downloads a single video.
Store the video in its channel's directory, otherwise store it in `videos/NO CHANNEL`.
Videos are validated in a few ways. A video file is valid when it has a complete video stream and an audio stream.
A complete video stream is validated by taking a screenshot at the end of the video.
"""
name = 'video'
pretty_name = 'Videos'
def __repr__(self):
return f''
def normalize_domain(self, url: str) -> str:
# Collapse youtu.be / youtube.com/shorts / tracking params to the canonical youtube.com domain.
return normalize_domain(normalize_video_url(url))
def already_downloaded(self, session: Session, *urls: str) -> List:
# We only consider a video record with a video file as "downloaded".
file_groups = list(session.query(FileGroup).filter(FileGroup.url.in_(urls), FileGroup.model == 'video'))
return file_groups
def prepare_download(self, session: Session, download: Download) -> PreparedVideo:
"""Sync prep: validate attempts cap, normalize URL, copy upstream-provided settings
onto the Download row, and look up any pre-existing Video for error attribution.
Holding the session here is brief: no network or subprocess work runs in this
phase. extract_info and channel resolution move to execute_download.
"""
if (download.attempts or 0) >= 10:
raise UnrecoverableDownloadError('Max download attempts reached')
url = normalize_video_url(download.url)
# ChannelDownloader passes destination/tag_names via settings; copy onto the row
# so the rest of the flow can read them off the Download object as before.
settings = download.settings or dict()
download.destination = download.destination or settings.get('destination')
tag_names = download.tag_names = download.tag_names or settings.get('tag_names') or []
# If a Video already exists at this URL, capture its location so a failure later
# can still tell the user where the previous attempt landed.
existing = Video.get_by_url(session, url)
location = existing.location if existing else None
# yt-dlp rewrites the whole info json; the user's edits (its wrolpi section) must survive.
wrolpi = existing.file_group.get_wrolpi_json() if existing else dict()
return PreparedVideo(
url=url,
settings=settings,
tag_names=list(tag_names),
destination=download.destination,
info_json=download.info_json,
location=location,
wrolpi=wrolpi,
)
async def execute_download(self, prepared: PreparedVideo, ctx: DownloadContext,
download: Download = None):
"""Async I/O phase: extract_info, channel resolution (brief session),
yt-dlp subprocess, file resolution, ffmpeg validation, sibling file globbing.
Returns either an ExecutedVideo (file is on disk; finalize persists) or a
DownloadResult (an early-exit case the dispatch will surface verbatim — used
for the metadata-only refresh flow and recoverable failures like a missing
file or incomplete stream).
"""
_log_ytdlp_versions()
url = prepared.url
settings = prepared.settings
location = prepared.location
# extract_info is sync but expensive; legacy held a session across it. We don't.
# Prefer ctx.extract_info for testability; fall back to the module global.
_extract = ctx.extract_info if ctx.extract_info else extract_info
info_json = prepared.info_json
if not info_json:
try:
info_json = _extract(url)
except yt_dlp.utils.DownloadError as e:
if _bot_blocked(e):
raise BotBlockedDownloadError(str(e)) from e
# Video may be private — surface as a recoverable failure.
try:
raise RuntimeError('Failed to extract_info') from e
except Exception as e2:
return DownloadResult(
success=False,
location=location,
error='\n'.join(traceback.format_exception(e2)),
)
if not info_json:
raise ValueError(f'Cannot download video with no info_json.')
# Brief session for channel resolution. _get_channel reads info_json (already
# populated above) and may create a Channel record.
download_for_channel = download if download is not None else Download(
url=url, settings=settings, destination=prepared.destination,
)
download_for_channel.info_json = info_json
with get_db_session(commit=True) as session:
channel = await self._get_channel(session, download_for_channel)
channel_id = channel.id if channel else None
channel_directory = channel.directory if channel else None
if prepared.destination:
out_dir = prepared.destination
logger.debug(f'Downloading {url} to destination {prepared.destination} from settings')
elif channel_directory:
out_dir = channel_directory
logger.debug(f'Downloading {url} to channel directory: {channel_directory}')
else:
out_dir = get_no_channel_directory()
logger.debug(f'Downloading {url} to default directory')
out_dir = out_dir if isinstance(out_dir, pathlib.Path) else pathlib.Path(out_dir)
out_dir = get_absolute_media_path(out_dir)
out_dir.mkdir(exist_ok=True, parents=True)
if prepared.tag_names and logger.isEnabledFor(logging.DEBUG):
logger.debug(f'Downloading {url} with tag_names={prepared.tag_names}')
effective = get_effective_video_settings(settings)
audio_only = effective.get('audio_only', False)
audio_format = effective.get('audio_format', 'mp3')
video_codecs = effective.get('video_codecs') or []
audio_codecs = effective.get('audio_codecs') or []
transcode = effective.get('transcode', False)
# Strict is suppressed only when transcoding could actually produce every preferred
# codec; a preference like av1-only has no transcode target, so strict must hold.
strict_codecs = effective.get('strict_codecs', False) \
and not (transcode and transcode_can_satisfy_codecs(video_codecs, audio_codecs))
# download_for_log carries .id (for progress) and .url (for log messages).
# Falling back to a stub keeps unit tests viable.
download_for_log = download if download is not None else Download(url=url)
try:
video_resolutions = effective['video_resolutions']
video_resolutions_str = format_selector(video_resolutions, video_codecs=video_codecs,
audio_codecs=audio_codecs, strict_codecs=strict_codecs)
video_format = effective['video_format']
video_path, entry = prepare_video_filename(url, out_dir, video_resolutions_str, video_format,
audio_only=audio_only, audio_format=audio_format)
if settings.get('download_metadata_only'):
# User has requested refresh of video metadata, skip the rest of the video downloader.
# download_info_json manages its own DB sessions; we surface its DownloadResult
# directly and the dispatch skips finalize_download.
download_for_metadata = download if download is not None else Download(
url=url, settings=settings,
)
download_for_metadata.info_json = info_json
return await self.download_info_json(download_for_metadata, video_path)
cmd = (
str(YT_DLP_BIN),
'--match-filter', '!is_live',
'--sub-format', DEFAULT_CAPTION_FORMAT,
'--convert-subs', DEFAULT_CAPTION_FORMAT,
'--convert-thumbnails', DEFAULT_POSTER_FORMAT,
'--no-cache-dir',
'--newline',
'--compat-options', 'no-live-chat',
'--write-comments',
'--extractor-args', 'youtube:max_comments=all,20,all,10;comment_sort=top',
)
if audio_only:
audio_selector = 'bestaudio'
if audio_codecs:
audio_selectors = [f"bestaudio[acodec~='{AUDIO_CODEC_YT_DLP_PATTERNS[i]}']"
for i in audio_codecs]
if not strict_codecs:
audio_selectors.append('bestaudio')
audio_selector = '/'.join(audio_selectors)
cmd = (*cmd, '-f', audio_selector, '-x', '--audio-format', audio_format)
else:
cmd = (*cmd,
'-f', video_resolutions_str,
'--merge-output-format', video_format,
'--ppa', 'Merger+ffmpeg_o1:-strict -2',
)
if effective['continue_dl']:
cmd = (*cmd, '--continue')
if effective['writesubtitles']:
cmd = (*cmd, '--write-subs', '--write-auto-subs')
if effective['writethumbnail']:
cmd = (*cmd, '--write-thumbnail')
if effective['writeinfojson']:
cmd = (*cmd, '--write-info-json')
if effective['yt_dlp_extra_args']:
cmd = (*cmd, *effective['yt_dlp_extra_args'].split(' '))
if effective.get('sleep_requests'):
try:
if float(effective['sleep_requests']) > 0:
cmd = (*cmd, '--sleep-requests', str(effective['sleep_requests']))
except (TypeError, ValueError):
pass
on_stdout = make_progress_callback(ctx.report_progress, parse_ytdlp_progress)
try:
if cookies_unlocked():
logger.info(f'Using encrypted cookies for video download: {url}')
with cookies_for_download() as cookies_path:
cmd = (*cmd, '--cookies', str(cookies_path))
if effective['user_agent']:
cmd = (*cmd, '--add-headers', f'User-Agent:{effective["user_agent"]}')
cmd = (*cmd, '-o', video_path, url)
result = await self.process_runner(download_for_log, cmd, out_dir, debug=True,
stdout_callback=on_stdout, ctx=ctx)
else:
logger.info(f'No cookies available for video download: {url}')
cmd = (*cmd, '-o', video_path, url)
result = await self.process_runner(download_for_log, cmd, out_dir, debug=True,
stdout_callback=on_stdout, ctx=ctx)
finally:
ctx.clear_progress()
stdout = result.stdout.decode()
stderr = result.stderr.decode()
if result.return_code != 0:
error = f'{stdout}\n\n\n{stderr}\n\nvideo downloader process exited with {result.return_code}'
if _bot_blocked(error):
raise BotBlockedDownloadError(f'Bot detection or invalid cookies: {error}')
if strict_codecs and 'Requested format is not available' in error:
# The source will not grow new formats; do not retry.
raise UnrecoverableDownloadError(
f'None of the requested codecs (video={video_codecs}, audio={audio_codecs}) are'
f' available for this video, and transcoding cannot produce them: {url}')
return DownloadResult(success=False, error=error, location=location)
output_format = audio_output_extension(audio_format) if audio_only else video_format
preferred_path = video_path.with_suffix(f'.{output_format}')
if not video_path.is_file() and preferred_path.is_file():
video_path = preferred_path
logger.info(f'Using preferred file which exists: {preferred_path}')
if not video_path.is_file():
# yt-dlp may have appended the real extension to a bogus one.
appended_path = video_path.parent / f'{video_path.name}.{output_format}'
if appended_path.is_file():
video_path = appended_path
logger.info(f'Using file with appended extension: {appended_path}')
if video_path.suffix == '.part':
return DownloadResult(success=False,
error=f'Video file that completed was a .part file: {video_path}')
if not video_path.is_file():
error = f'{stdout}\n\n\n{stderr}\n\nVideo file could not be found! {video_path}'
retry_seconds = None
if '!is_live' in stdout:
error = f'{stdout}\n\nVideo was live and did not finish downloading.'
retry_seconds = 3600
if ' live event ' in stdout:
error = f'{stdout}\n\nVideo will be live and did not finish downloading.'
retry_seconds = 3600
return DownloadResult(success=False, error=error, location=location,
retry_seconds=retry_seconds)
if not audio_only and not ffmpeg_video_complete(video_path):
return DownloadResult(success=False, error='Video was incomplete', location=location)
# File is on disk and complete. Glob siblings, drop part files, normalize names.
video_paths = glob_shared_stem(video_path)
video_paths = self._delete_part_files(video_path, video_paths)
video_paths = self.normalize_video_file_names(video_path, video_paths)
if not audio_only and (video_codecs or audio_codecs):
video_path, video_paths = await enforce_codecs(video_path, video_paths, effective)
if is_youtube_url(url):
# YouTube auto-captions pin the text to the bottom-left; center them at the source. Runs
# before the FileGroup is created, so the first index reads the corrected file.
for path in video_paths:
if path.name.endswith('.vtt'):
try:
strip_youtube_caption_positioning(path)
except Exception:
logger.warning(f'Failed to center YouTube captions in {path}', exc_info=True)
return ExecutedVideo(
url=url,
video_path=video_path,
video_paths=video_paths,
info_json=info_json,
entry=entry,
channel_id=channel_id,
audio_only=audio_only,
tag_names=prepared.tag_names,
settings=settings,
location=location,
wrolpi=prepared.wrolpi,
)
except BotBlockedDownloadError:
raise
except UnrecoverableDownloadError:
raise
except yt_dlp.utils.UnsupportedError as e:
raise UnrecoverableDownloadError('URL is not supported by yt-dlp') from e
except Exception as e:
logger.warning(f'VideoDownloader failed to download: {url}', exc_info=e)
if _bot_blocked(e):
raise BotBlockedDownloadError(str(e)) from e
if _skip_download(e):
# The video failed to download, and the error will never be fixed. Skip it forever.
try:
source_id = (info_json or {}).get('id')
logger.warning(f'Adding video "{source_id}" to skip list for this channel. WROLPi will not '
f'attempt to download it again.')
if download is not None:
download.add_to_skip_list()
except Exception:
logger.warning(f'Could not skip video {url}')
raise UnrecoverableDownloadError() from e
return DownloadResult(success=False, error=str(traceback.format_exc()), location=location)
async def finalize_download(self, session: Session, download: Download,
executed: ExecutedVideo) -> DownloadResult:
"""Async DB-bound persistence: create the Video record, normalize info_json,
update the channel's collection format, apply tags, run ffprobe, validate
streams, and dedupe. Caller commits.
"""
# Create the Video record.
video = Video.from_paths(session, *executed.video_paths)
video.source_id = executed.entry['id']
video.channel_id = executed.channel_id
if video.info_json_path and (new_info_json := video.clean_info_json()):
video.replace_info_json(new_info_json, clean=False)
if executed.wrolpi and video.info_json_path:
# The re-downloaded info json lost the user's edits; put them back, then derive the
# title/description again so the custom ones win (from_paths validated without them).
video.file_group.update_wrolpi_json(executed.wrolpi)
video.validate(session)
if parent_download_url := executed.settings.get('parent_download_url'):
try:
video.file_group.update_wrolpi_json({'parent_download_url': parent_download_url})
except ValueError as e:
logger.warning(f'Failed to update wrolpi json for {video}: {e}')
# Update channel's collection's file_format to track the format used.
if executed.channel_id:
from modules.videos.lib import save_channels_config
channel_obj = session.get(Channel, executed.channel_id)
if channel_obj and channel_obj.collection and not channel_obj.collection.file_format:
video_config = get_videos_downloader_config()
channel_obj.collection.file_format = video_config.file_name_format
save_channels_config.activate_switch()
# Apply tags now that the file_group is bound to the session.
if executed.tag_names:
existing_names = video.file_group.tag_names
for name in executed.tag_names:
if name not in existing_names:
video.add_tag(session, name)
location = video.location
# ffprobe runs the ffprobe subprocess (may persist .ffprobe.json on disk).
await video.get_ffprobe_json()
if not executed.audio_only and not video.get_streams_by_codec_type('video'):
return DownloadResult(success=False,
error='Video was downloaded but did not contain video stream',
location=location)
if not video.get_streams_by_codec_type('audio'):
return DownloadResult(success=False,
error='Download did not contain audio stream',
location=location)
logger.info(f'Successfully downloaded video {executed.url} {video}')
# Delete duplicate Videos that share the URL or source_id. Runs against the same
# session as the new Video record so deletion is atomic with creation.
await Video.delete_duplicate_videos(session, download.url, executed.entry['id'],
executed.video_path)
return DownloadResult(success=True, location=location)
@staticmethod
async def _get_channel(session: Session, download: Download) -> Optional[Channel]:
"""Find the Channel for a video download.
Priority order:
1. Destination directory — if it belongs to a channel, that channel wins (files must live in their channel's dir)
2. yt-dlp info_json metadata (channel/uploader fields)
3. Download settings (collection_id or channel_url from ChannelDownloader)
"""
settings = download.settings or dict()
destination = download.destination
# Destination channel always takes priority — files must live in their channel's directory.
if destination:
try:
channel = get_channel(session, directory=destination, return_dict=False)
logger.debug(f'Found {channel} using destination directory')
return channel
except UnknownChannel:
pass
# Try yt-dlp info_json metadata.
channel_name = download.info_json.get('channel') or download.info_json.get('uploader')
channel_source_id = download.info_json.get('channel_id') or download.info_json.get('uploader_id')
channel_url = download.info_json.get('channel_url') or download.info_json.get('uploader_url')
if channel_name or channel_source_id or channel_url:
tag_name = i[0] if (i := settings.get('channel_tag_name')) else None
try:
channel = get_or_create_channel(session, source_id=channel_source_id, url=channel_url,
name=channel_name, tag_name=tag_name, destination=destination)
logger.debug(f'Found {channel} using yt-dlp info_json')
return channel
except UnknownChannel:
pass
# Try collection_id or channel_url from ChannelDownloader settings.
collection_id = download.collection_id
settings_channel_url = settings.get('channel_url')
if collection_id:
channel = session.query(Channel).filter_by(collection_id=collection_id).one_or_none()
if channel:
logger.debug(f'Found {channel} using collection_id')
return channel
if settings_channel_url:
try:
channel = get_channel(session, url=settings_channel_url, return_dict=False)
logger.debug(f'Found {channel} using settings channel_url')
return channel
except UnknownChannel:
if not destination:
raise
logger.warning('Could not find channel')
return None
@staticmethod
async def download_info_json(download: Download, video_path: pathlib.Path) -> DownloadResult:
"""User has requested to refresh Video metadata. Download only metadata files to temporary directory.
Replace existing metadata files only after successful download."""
url = download.url
source_id = download.info_json['id']
with get_db_session() as session:
# Delete any duplicate videos before attempting to get info json.
deleted = await Video.delete_duplicate_videos(session, download.url, source_id, video_path)
if deleted:
session.commit()
with get_db_session() as session:
# Find the existing video, replace its info json.
video = Video.get_by_url(session, url)
location = video.location
# Download video info json directly using yt-dlp.
try:
info_json = download_video_info_json(url)
except Exception as e:
logger.error('Failed to download video info json', exc_info=e)
return DownloadResult(success=False, error='Failed to download video info json')
settings = download.settings or dict()
with get_db_session(commit=True) as session:
# Find the existing video, replace its info json; keep the user's edits (wrolpi section).
video = Video.get_by_url(session, url)
wrolpi = video.file_group.get_wrolpi_json()
video.replace_info_json(info_json)
if wrolpi:
video.file_group.update_wrolpi_json(wrolpi)
if parent_download_url := settings.get('parent_download_url'):
try:
video.file_group.update_wrolpi_json({'parent_download_url': parent_download_url})
except ValueError as e:
logger.warning(f'Failed to update wrolpi json for {video}: {e}')
if video.get_comments():
video.have_comments = True
await video.get_ffprobe_json()
logger.debug(f'Downloaded video info json {location=}')
logger.info(f'Successfully downloaded video info json {url} {video}')
return DownloadResult(success=True, location=location)
@staticmethod
def _delete_part_files(video_file: pathlib.Path, files: list[pathlib.Path]) -> list[pathlib.Path]:
"""Delete temporary "part" files that are used by yt-dlp when downloading video (and related) files."""
files = list(files)
# Delete any files in this group that end with .part.
for part_file in [i for i in files if i.name.endswith('.part')]:
logger.warning(f'Deleting leftover .part file: {part_file}')
part_file.unlink()
files.remove(part_file)
# Delete files of an unmerged yt-dlp "format" download, whatever their extension, and anything made from them
# (an earlier refresh may have indexed one and made a poster). Example: The Video Title.f313.webm
video_stem, _ = split_path_stem_and_suffix(video_file)
format_file = re.compile(rf'^{re.escape(video_stem)}\.f\d{{2,4}}\.')
for part_file in [i for i in files if i != video_file and format_file.match(i.name)]:
logger.warning(f'Deleting leftover yt-dlp format file: {part_file}')
part_file.unlink()
files.remove(part_file)
return files
@staticmethod
def normalize_video_file_names(video_path: pathlib.Path, files: List[pathlib.Path]) -> List[pathlib.Path]:
"""Video files can have unpredictable names, rename any video files as necessary to make them share the same
stem."""
video_stem, _ = split_path_stem_and_suffix(video_path.name)
new_files = []
for file in files:
suffix = file.name[len(video_stem):]
if suffix.endswith('.vtt') or suffix.endswith('.srt'):
_, normal_suffix = split_path_stem_and_suffix(file.name)
if suffix != normal_suffix:
# Video captions file has extra characters in the suffix from yt-dlp.
# Example: .en-uYU-mmqFLq8.vtt -> .en.vtt
match = VTT_SUFFIX_PARSER.match(suffix.strip())
if match:
lang, _, suffix = match.groups()
suffix = f'.{lang}.{suffix}'
file = file.rename(video_path.with_suffix(suffix))
new_files.append(file)
return new_files
VTT_SUFFIX_PARSER = re.compile(r'^\.([a-z]{2,3})-(.+?)\.(vtt|srt)$', re.IGNORECASE)
channel_downloader = ChannelDownloader()
# Videos may match the ChannelDownloader, give it a higher priority.
video_downloader = VideoDownloader()
def get_or_create_channel(session: Session, source_id: str = None, url: str = None, name: str = None,
tag_name: str = None, destination: pathlib.Path = None) -> Channel:
"""
Attempt to find a Channel using the provided params. The params are in order of reliability.
Creates a new Channel if one cannot be found. If a destination is provided, the Channel will be created in that
directory instead of the default videos destination.
"""
try:
channel = get_channel(session, source_id=source_id, url=url, name=name, return_dict=False)
return channel
except UnknownChannel:
pass
# If destination matches an existing channel, use it rather than creating a new one.
# This handles the case where yt-dlp returns bad metadata (e.g. channel name 'watch')
# but the destination directory already belongs to a known channel.
if destination:
try:
channel = get_channel(session, directory=destination, return_dict=False)
return channel
except UnknownChannel:
pass
if not name:
raise UnknownChannel(f'Cannot create channel without a name')
# Channel does not exist. Create one using the destination if provided, otherwise the default video directory.
if destination:
channel_directory = get_absolute_media_path(destination)
else:
channel_directory = format_videos_destination(name, tag_name, url)
# Reuse the Channel that already owns this directory instead of creating a conflicting one.
try:
channel = get_channel(session, directory=channel_directory, return_dict=False)
return channel
except UnknownChannel:
pass
if not channel_directory.is_dir():
channel_directory.mkdir(parents=True)
data = ChannelPostRequest(
source_id=source_id,
name=name,
url=url,
directory=str(channel_directory),
tag_name=tag_name,
)
try:
channel = create_channel(session, data=data, return_dict=False)
except (ValidationError, ChannelDirectoryConflict, ChannelNameConflict, ChannelURLConflict, ChannelSourceIdConflict) as e:
raise UnrecoverableDownloadError(
f'Cannot create channel {name!r}: conflicts with an existing channel') from e
# Create the directory now that the channel is approved.
channel_directory.mkdir(exist_ok=True)
return channel
def update_channel_catalog(channel: Channel, info: dict):
"""
Connect to the Channel's host website and pull a catalog of all videos. Insert any new videos into the DB.
It is expected that any missing videos will be downloaded later.
"""
logger.info(f'Downloading video list for {channel} at {channel.url}')
# Resolve all entries to dictionaries.
entries = info['entries'] = list(info['entries'])
# yt-dlp may hand back a list of URLs, lets use the "Uploads" URL, if available.
try:
entries[0]['id']
except Exception:
logger.warning('yt-dlp did not return a list of URLs')
for entry in entries:
if entry['title'] == 'Uploads':
logger.info('Youtube-DL gave back a list of URLs, found the "Uploads" URL and using it.')
info = extract_info(entry['url'])
break
with get_db_session(commit=True) as session:
# Get the channel in this new context.
channel: Channel = session.query(Channel).filter_by(id=channel.id).one()
channel.info_json = info
channel.info_date = now()
channel.source_id = info.get('id')
channel_id = channel.id
# Write the Channel's info to a JSON file.
if channel.directory:
info_json_path = channel.info_json_path
with info_json_path.open('wt') as fh:
json.dump(info, fh, indent=2, sort_keys=True)
logger.debug(f'Wrote channel info json to {info_json_path}')
else:
logger.debug(f'Skipping channel info json because it does not have a directory: {channel}')
logger.info(f'Finished downloading video list for {channel} found {len(entries)} videos')
# Update all view counts using the latest from the Channel's info_json.
background_task(update_view_counts_and_censored(channel_id))
UNRECOVERABLE_ERRORS = {
'404: Not Found',
'requires payment',
'Content Warning',
'Did not get any data blocks',
'Sign in',
'This live stream recording is not available.',
'members-only content',
"You've asked yt-dlp to download the URL",
}
# Errors that indicate bot detection or invalid cookies. All pending downloads should be stopped.
BOT_BLOCKED_ERRORS = {
"Sign in to confirm you're not a bot",
'Sign in to confirm you\u2019re not a bot',
'cookies are no longer valid',
}
def _bot_blocked(error):
"""Return True if the error indicates bot detection or invalid cookies."""
error_str = str(error)
for msg in BOT_BLOCKED_ERRORS:
if msg in error_str:
return True
return False
def _skip_download(error):
"""Return True if the error is unrecoverable and the video should be skipped in the future."""
if _bot_blocked(error):
# Bot detection errors should stop all downloads, not skip individual videos.
return False
error_str = str(error)
for msg in UNRECOVERABLE_ERRORS:
if msg in error_str:
return True
return False