import contextlib
import dataclasses
import html
import json
import pathlib
import re
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Generator, Type, List, Set
from typing import Tuple, Optional
import pytz
from sqlalchemy.orm import Session, joinedload
from yt_dlp import YoutubeDL
from modules.videos.models import Video
from wrolpi import dates, flags
from wrolpi.captions import extract_captions
from wrolpi.common import ConfigFile, extract_domain, logger, \
escape_file_name, get_media_directory, background_task, Base, get_wrolpi_config, trim_file_name
from wrolpi.collections import Collection
from wrolpi.collections.models import validate_collection_directory
from wrolpi.config_import import ConfigSection, Unreadable, chunks, media_relative
from wrolpi.config_version import register_ownership
from wrolpi.dates import Seconds, from_timestamp
from wrolpi.db import get_db_curs, get_db_session, session_curs
from wrolpi.downloader import Download, download_manager, save_downloads_config
from wrolpi.errors import UnknownDirectory
from wrolpi.events import Events
from wrolpi.files.lib import split_path_stem_and_suffix
from wrolpi.switches import register_switch_handler, ActivateSwitchMethod
from wrolpi.tags import Tag, unknown_tag_names
from wrolpi.vars import YTDLP_CACHE_DIR, PYTEST
from .common import is_valid_poster, convert_image, \
generate_video_poster, ConfigError, \
extract_video_duration, ffprobe_json_sync, read_ffprobe_json_file, write_ffprobe_json_file
from .cookies import cookies_unlocked, cookies_for_download
from .errors import UnknownChannel
from .models import Channel
logger = logger.getChild(__name__)
DEFAULT_DOWNLOAD_FREQUENCY = Seconds.week
REQUIRED_OPTIONS = ['name', 'directory']
@dataclasses.dataclass
class VideoInfoJSON:
channel_source_id: str | None = None
channel_url: str | None = None
duration: int | None = None
epoch: int | None = None
timestamp: datetime | None = None
title: str | None = None
upload_date: datetime | None = None
url: str | None = None
view_count: int | None = None
def extract_video_info_json(video: Video) -> VideoInfoJSON:
"""
Parse the Video's info json file, return the relevant data.
"""
video_info_json = VideoInfoJSON()
if info_json := video.get_info_json():
# A title the user set (Edit > Edit) is kept apart from yt-dlp's keys, and wins.
wrolpi_section = info_json.get('wrolpi') or dict()
title = wrolpi_section.get('custom_title') or info_json.get('fulltitle') or info_json.get('title')
video_info_json.title = html.unescape(title) if title else None
upload_date = dates.strpdate(i) if (i := info_json.get('upload_date')) else None
if upload_date:
upload_date = upload_date.astimezone(pytz.UTC)
timestamp = None
try:
timestamp = dates.from_timestamp(info_json['timestamp']) if info_json.get('timestamp') else None
except Exception:
pass
video_info_json.channel_source_id = info_json.get('channel_id') or info_json.get('uploader_id') or None
video_info_json.channel_url = info_json.get('channel_url') or info_json.get('uploader_url') or None
video_info_json.duration = int(i) if (i := info_json.get('duration')) else None
video_info_json.epoch = int(i) if (i := info_json.get('epoch')) else None
video_info_json.timestamp = timestamp
video_info_json.upload_date = upload_date
video_info_json.url = info_json.get('webpage_url') or info_json.get('url') or None
video_info_json.view_count = int(i) if (i := info_json.get('view_count')) else None
return video_info_json
EXTRACT_SUBTITLES = False
def validate_video(session: Session, video: Video, channel_generate_poster: bool):
"""
Validate a single video.
A Video is validated when we have attempted to find its: title, duration, view_count, url, caption, size, and
channel.
A Video is also valid when it has a JPEG poster, if any. If no poster can be
found, it will be generated from the video file.
"""
json_published_datetime = None
if video.info_json_path:
# These properties can be found in the info json.
video_info_json = extract_video_info_json(video)
video.file_group.title = video_info_json.title
video.file_group.length = video_info_json.duration
video.file_group.url = video_info_json.url
json_published_datetime = video.file_group.published_datetime = (
video_info_json.timestamp # The exact second the video was published.
or video_info_json.upload_date # The day the video was published.
or video.file_group.published_datetime
)
video.file_group.download_datetime = from_timestamp(video_info_json.epoch) if video_info_json.epoch else None
# View count will probably be overwritten by more recent data when this Video's Channel is
# updated.
video.view_count = video.view_count or video_info_json.view_count
if video_info_json.channel_source_id or video_info_json.channel_url:
from modules.videos.channel.lib import get_channel
try:
if channel := get_channel(
session,
source_id=video_info_json.channel_source_id,
url=video_info_json.channel_url,
directory=video.video_path.parent,
return_dict=False,
):
video.channel_id = channel.id
except UnknownChannel:
pass
video_path = video.video_path
if not video_path:
logger.error(f'{video} does not have video_path!')
if video_path and (not video.file_group.title or not video.file_group.published_datetime or not video.source_id):
# Video is missing things that can be extracted from the video file name.
# These are the least trusted, so anything already on the video should be trusted.
_, published_date, source_id, title = parse_video_file_name(video_path)
published_date = published_date or video.file_group.published_datetime
if published_date and isinstance(published_date, str):
published_date = dates.strpdate(published_date)
if not published_date.tzinfo:
published_date = published_date.astimezone(pytz.UTC)
video.file_group.title = video.file_group.title or html.unescape(title)
# Trust info json upload_date before file name datetime.
video.file_group.published_datetime = json_published_datetime or published_date
video.source_id = video.source_id or source_id
is_audio = video.file_group.mimetype and video.file_group.mimetype.startswith('audio/')
if channel_generate_poster and not is_audio:
# Try to convert/generate, but keep the old poster if those fail.
new_poster_path, duration = convert_or_generate_poster(video) or video.poster_path
if new_poster_path:
# Poster was created/updated.
video.file_group.append_files(new_poster_path)
if duration:
video.file_group.length = duration
if video_path and not video.ffprobe_json:
# Extract ffprobe data if not already present (e.g., from .ffprobe.json cache file)
ffprobe_path = video.ffprobe_json_path
# Try to read from cache file first
if ffprobe_path and ffprobe_path.is_file():
video.ffprobe_json = read_ffprobe_json_file(ffprobe_path)
# If still no data, run ffprobe
if not video.ffprobe_json:
try:
video.ffprobe_json = ffprobe_json_sync(video_path)
# Write cache file if enabled
if ffprobe_path and get_wrolpi_config().save_ffprobe_json:
try:
write_ffprobe_json_file(ffprobe_path, video.ffprobe_json)
video.file_group.append_files(ffprobe_path)
except IOError as e:
logger.warning(f'Failed to write ffprobe json cache file {ffprobe_path}', exc_info=e)
except Exception as e:
logger.warning(f'Failed to extract ffprobe data for {video}', exc_info=e)
# Extract duration from ffprobe data
if video_path and not video.file_group.length:
if video.ffprobe_json:
if duration := video.ffprobe_json['format'].get('duration'):
video.file_group.length = float(duration)
elif (video_streams := video.get_streams_by_codec_type('video')) and 'duration' in video_streams[0]:
video.file_group.length = float(video_streams[0]['duration'])
else:
# Fallback if ffprobe extraction failed
video.file_group.length = extract_video_duration(video_path)
if video_path and not video.caption_paths and video.file_group.d_text and not EXTRACT_SUBTITLES:
# Caption file was deleted, clear out old captions.
video.file_group.d_text = None
elif video_path and not video.file_group.d_text and EXTRACT_SUBTITLES:
# Captions were not found, extract them from the video.
video.file_group.d_text = extract_captions(video_path)
elif video_path and video.caption_paths and not video.file_group.d_text:
video.file_group.d_text = video.get_caption_text()
def convert_or_generate_poster(video: Video) -> Tuple[Optional[pathlib.Path], Optional[int]]:
"""
If a Video has a poster, but the poster is invalid, convert it. If a Video has no poster, generate one from the
video file.
Returns None if the poster was not converted, and not generated.
"""
video_path = video.video_path
# Modification/generation of poster is enabled for this channel.
if video.poster_path:
# Check that the poster is a more universally supported JPEG.
old: pathlib.Path = video.poster_path
new = old.with_suffix('.jpg')
if old != new and new.exists():
# Destination JPEG already exists (it may have the wrong format).
video.file_group.unlink_tag_directory_links(old)
old.unlink()
old = new
if not is_valid_poster(old):
# Poster is not valid, convert it and place it in the new location. Conversion removes `old`, so its
# Tags Directory links are found now and removed only once the conversion succeeds.
old_links = video.file_group.get_tag_directory_links(old) if old != new else []
try:
convert_image(old, new)
if not new.is_file() or not new.stat().st_size:
raise FileNotFoundError(f'Failed to convert poster: {new}')
for link in old_links:
link.unlink(missing_ok=True)
logger.info(f'Converted invalid poster {repr(str(old))} to {repr(str(new))}')
return new, None
except Exception as e:
logger.error(f'Failed to convert invalid poster {old} to {new}', exc_info=e)
return None, None
if not video.poster_path:
# Video poster was not discovered, or converted. Let's generate it.
try:
poster_path, duration = generate_video_poster(video_path)
logger.debug(f'Generated poster for {video}')
return poster_path, duration
except Exception as e:
logger.error(f'Failed to generate poster for {video}', exc_info=e)
return None, None
@dataclass
class ChannelDictValidator:
name: str
directory: str
download_frequency: int = 604800 # Default to weekly
url: str = None
def _channel_key(data: dict) -> str:
"""A Channel's key: its directory, relative to the media directory."""
if not isinstance(data, dict) or not data.get('directory'):
raise ValueError('A Channel requires a directory')
return media_relative(validate_collection_directory(data['directory']))
# The key of a channel Collection without a Channel or a directory.
ORPHAN_KEY = '\0orphan:'
# Missing from an entry: the Channel keeps the database's value.
CHANNEL_KEPT_IF_MISSING = ('source_id', 'generate_posters', 'calculate_duration', 'download_missing_data')
CHANNEL_COMPARED = ('name', 'url', 'tag_name', 'file_format') + CHANNEL_KEPT_IF_MISSING
def _channel_row(name, url, source_id, tag_name, file_format, generate_posters, calculate_duration,
download_missing_data) -> dict:
return dict(name=name, url=url or None, source_id=source_id or None, tag_name=tag_name or None,
file_format=file_format or None, generate_posters=generate_posters,
calculate_duration=calculate_duration, download_missing_data=download_missing_data)
class ChannelsSection(ConfigSection):
"""channels.yaml's Channels (a Collection with kind 'channel' and its Channel), keyed by directory. A channel
Collection without a Channel is in the database's rows too: listed, it gets its Channel; not listed, it is
deleted."""
name = 'channels'
def file_rows(self, config: dict) -> Tuple[Dict[str, dict], Set[str]]:
rows, invalid = dict(), set()
for position, data in enumerate(config.get('channels') or []):
try:
key = _channel_key(data)
except Exception as e:
channel_import_logger.error(f'Skipping invalid Channel in config: {data!r}', exc_info=e)
# Counted as skipped (see `Unreadable`). The entry's name keeps its Channel (see `prepare`).
invalid.add(Unreadable(position))
continue
if key in rows or key in invalid:
raise RuntimeError('Refusing to import channels config because it contains duplicate directories!')
if not data.get('name'):
channel_import_logger.error(f'Channel {key!r} is required to have "name", skipping')
invalid.add(key)
continue
rows[key] = _channel_row(data['name'], data.get('url'), data.get('source_id'), data.get('tag_name'),
data.get('file_format'), data.get('generate_posters'),
data.get('calculate_duration'), data.get('download_missing_data'))
return rows, invalid
def db_rows(self, session: Session) -> Dict[str, dict]:
query = session.query(Collection.directory, Collection.name, Channel.url, Channel.source_id, Tag.name,
Collection.file_format, Channel.generate_posters, Channel.calculate_duration,
Channel.download_missing_data, Collection.id, Channel.id) \
.select_from(Collection) \
.outerjoin(Channel, Channel.collection_id == Collection.id) \
.outerjoin(Tag, Tag.id == Collection.tag_id) \
.filter(Collection.kind == 'channel')
rows = dict()
for directory, *values, collection_id, channel_id in query:
if directory:
key = media_relative(directory)
elif channel_id is None:
# An orphan without a directory: no entry can list it (`ORPHAN_KEY`).
key = f'{ORPHAN_KEY}{collection_id}'
else:
continue
rows[key] = row = _channel_row(*values)
if channel_id is None:
row['orphan'] = True
return rows
def same(self, file_row: dict, db_row: dict) -> bool:
if db_row.get('orphan') or file_row.get('directory'):
return False
return all(file_row[i] == db_row[i] for i in CHANNEL_COMPARED
if not (i in CHANNEL_KEPT_IF_MISSING and file_row[i] is None))
def upsert(self, session: Session, rows: Dict[str, dict]):
media_directory = get_media_directory()
collections = {media_relative(i.directory): i for i in session.query(Collection).filter(
Collection.kind == 'channel', Collection.directory.in_([str(media_directory / i) for i in rows]))}
channels = {i.collection_id: i for i in session.query(Channel).filter(
Channel.collection_id.in_([i.id for i in collections.values()]))}
tags = {i.name: i for i in
session.query(Tag).filter(Tag.name.in_(list({i['tag_name'] for i in rows.values()} - {None})))}
for key, row in rows.items():
if not (collection := collections.get(key)):
collection = Collection(kind='channel')
session.add(collection)
# A moved Channel's new directory (see `ChannelsConfig.prepare`).
collection.directory = media_directory / row.get('directory', key)
collection.name = row['name']
collection.tag = tags.get(row['tag_name'])
collection.file_format = row['file_format']
if not (channel := channels.get(collection.id) if collection.id else None):
channel = Channel(collection=collection)
session.add(channel)
channel.url = row['url']
for name in CHANNEL_KEPT_IF_MISSING:
if row[name] is not None:
setattr(channel, name, row[name])
session.flush()
def delete(self, session: Session, keys: Set[str]):
media_directory = get_media_directory()
orphan_ids = [int(i.removeprefix(ORPHAN_KEY)) for i in keys if i.startswith(ORPHAN_KEY)]
collections = session.query(Collection).filter(
Collection.kind == 'channel',
Collection.directory.in_([str(media_directory / i) for i in keys if not i.startswith(ORPHAN_KEY)])
| Collection.id.in_(orphan_ids)).all()
channels = {i.collection_id: i for i in session.query(Channel).filter(
Channel.collection_id.in_([i.id for i in collections]))}
for collection in collections:
if channel := channels.get(collection.id):
logger.warning(f'Deleting {channel} because it is not in the config.')
channel.delete_with_videos(commit=False)
else:
logger.warning(f'Deleting orphaned Collection {collection.name} (id={collection.id})')
session.delete(collection)
session.flush()
class ChannelDownloadsSection(ConfigSection):
"""The Downloads listed under each Channel, keyed by (Channel directory, URL). An existing Download keeps its own
frequency (download_manager.yaml's); the file's is used to create one."""
name = 'downloads'
def file_rows(self, config: dict) -> Tuple[Dict[tuple, dict], Set[tuple]]:
rows, invalid = dict(), set()
for data in config.get('channels') or []:
try:
key = _channel_key(data)
except Exception:
continue
downloads = data.get('downloads') or []
if not isinstance(downloads, list):
channel_import_logger.error(f'Skipping the Downloads of Channel {key!r}, they are not a list')
# Counted as skipped; the Channel keeps its Downloads (see `ChannelsConfig.prepare`).
invalid.add((key, '\0downloads'))
continue
for position, download in enumerate(downloads):
if isinstance(download, dict) and download.get('url') and isinstance(download['url'], str):
rows[(key, download['url'])] = dict(frequency=download.get('frequency'))
else:
channel_import_logger.error(f'Skipping invalid Download of Channel {key!r}: {download!r}')
# Counted as skipped; the Channel keeps its Downloads (see `ChannelsConfig.prepare`).
invalid.add((key, download if isinstance(download, str) else f'\0{position}'))
return rows, invalid
def present(self, config: dict, file_rows: dict, invalid_keys: set) -> bool:
# A Channel listed with an empty list of Downloads has none (see `ChannelsConfig.prepare` for entries without
# a list).
return bool(config.get('channels'))
def db_rows(self, session: Session) -> Dict[tuple, dict]:
query = session.query(Collection.directory, Download.url, Download.frequency) \
.join(Collection, Collection.id == Download.collection_id) \
.filter(Collection.kind == 'channel', Collection.directory.isnot(None))
return {(media_relative(directory), url): dict(frequency=frequency) for directory, url, frequency in query}
def same(self, file_row: dict, db_row: dict) -> bool:
return True
def upsert(self, session: Session, rows: Dict[tuple, dict]):
media_directory = get_media_directory()
channels = {media_relative(i.directory): i for i in session.query(Channel).join(Collection).filter(
Collection.directory.in_([str(media_directory / i) for i, _ in rows]))}
for (key, url), row in rows.items():
if not (channel := channels.get(key)):
channel_import_logger.warning(f'Cannot find Channel {key!r} for Download {url}')
continue
if existing := Download.get_by_url(session, url):
existing.collection_id = channel.collection_id
elif row['frequency']:
channel.get_or_create_download(session, url, row['frequency'], reset_attempts=True, commit=False)
else:
logger.error(f'Refusing to create Download for Channel without frequency: {url}')
session.flush()
def delete(self, session: Session, keys: Set[tuple]):
media_directory = get_media_directory()
query = session.query(Download, Collection.directory) \
.join(Collection, Collection.id == Download.collection_id) \
.filter(Collection.kind == 'channel', Download.url.in_([url for _, url in keys]))
for download, directory in query.all():
if (media_relative(directory), download.url) in keys:
logger.warning(f'Deleting {download} because it is no longer in its Channel')
download.delete(add_to_skip_list=False, commit=False)
@dataclass
class ChannelsConfigValidator:
version: int = None
channels: list[ChannelDictValidator] = dataclasses.field(default_factory=list)
class ChannelsConfig(ConfigFile):
file_name = 'channels.yaml'
sections = (ChannelsSection(), ChannelDownloadsSection())
default_config = dict(
channels=[dict(
name='WROLPi',
url='https://www.youtube.com/channel/UC4t8bw1besFTyjW7ZBCOIrw/videos',
directory='videos/wrolpi',
download_frequency=604800,
)],
version=0,
)
validator = ChannelsConfigValidator
@property
def channels(self) -> dict:
return self._config['channels']
@channels.setter
def channels(self, value: dict):
self.update({'channels': value})
def import_config(self, file: pathlib.Path = None, send_events=False):
self._config_outdated = False
super().import_config(file, send_events)
def prepare(self, diffs: dict):
channels, downloads = diffs['channels'], diffs['downloads']
db_rows = self.import_db_rows['channels']
# A Tag that no longer exists (it was removed) is dropped, and the config is saved without it.
rows = [*channels.added.values(), *channels.changed.values()]
unknown = unknown_tag_names({i['tag_name'] for i in rows if i['tag_name']})
for row in rows:
if row['tag_name'] in unknown:
channel_import_logger.warning(f'Removing unknown Tag {row["tag_name"]!r} from a Channel')
row['tag_name'] = None
self._config_outdated = True
# A Channel whose directory changed (its name did not) is moved, not deleted and re-created.
removed_by_name = dict()
for key in channels.removed:
removed_by_name.setdefault(db_rows[key]['name'], []).append(key)
moves = dict()
for key, row in list(channels.added.items()):
if len(old := removed_by_name.get(row['name']) or []) == 1:
channels.removed.discard(old[0])
del channels.added[key]
channels.changed[old[0]] = dict(row, directory=key)
moves[old[0]] = key
# An invalid entry keeps its Channel (found by the entry's name when its directory is invalid), and the
# Channel's Downloads; so does an entry without a list of Downloads (an old config), or with an unreadable
# Download.
entries = [i for i in self._config.get('channels') or [] if isinstance(i, dict)]
invalid_names = {i.get('name') for i in entries if i.get('name')} - {i['name'] for i in self._channel_rows()}
for key in [i for i in channels.removed if db_rows[i]['name'] in invalid_names]:
channels.removed.discard(key)
kept = set(db_rows) - set(self._channel_rows_by_key()) - channels.removed - set(moves)
kept |= {key for key, data in self._channel_entries() if 'downloads' not in data}
kept |= {directory for directory, _ in ChannelDownloadsSection().file_rows(self._config)[1]}
# A moved Channel's Downloads are deleted after the move, under its new directory; one its new entry lists
# is the same Download.
for directory, url in [i for i in downloads.removed if i[0] in moves]:
downloads.removed.discard((directory, url))
if (moves[directory], url) in downloads.added:
del downloads.added[(moves[directory], url)]
else:
downloads.removed.add((moves[directory], url))
downloads.removed = {i for i in downloads.removed if i[0] not in kept}
# A new Download needs a frequency.
if missing := {url for (_, url), row in downloads.added.items() if not row['frequency']}:
with get_db_session() as session:
missing -= {i for i, in session.query(Download.url).filter(Download.url.in_(list(missing)))}
for key in [i for i in downloads.added if i[1] in missing]:
channel_import_logger.error(f'Refusing to create Download for Channel without frequency: {key[1]}')
del downloads.added[key]
self.import_skipped += 1
def _channel_entries(self) -> List[Tuple[str, dict]]:
"""The valid entries of the file, keyed by directory."""
entries = list()
for data in self._config.get('channels') or []:
try:
entries.append((_channel_key(data), data))
except Exception:
pass
return entries
def _channel_rows_by_key(self) -> dict:
return ChannelsSection().file_rows(self._config)[0]
def _channel_rows(self) -> list:
return list(self._channel_rows_by_key().values())
def after_import(self, diffs: dict):
"""Claim the Videos of new and changed Channels, find their missing source ids, and link Downloads."""
changed = {row.get('directory', key) for key, row in
{**diffs['channels'].added, **diffs['channels'].changed}.items()}
if changed:
media_directory = get_media_directory()
with get_db_session() as session:
updated = session.query(Channel.id, Channel.url, Channel.source_id, Channel.download_missing_data) \
.join(Collection, Collection.id == Channel.collection_id) \
.filter(Collection.directory.in_([str(media_directory / i) for i in changed])).all()
from modules.videos import claim_videos_for_channels
claim_videos_for_channels([i.id for i in updated])
# A fetched source id changes channels.yaml, which cannot be saved while it has skipped entries.
for id_, url, source_id, download_missing_data in [] if self.import_skipped else updated:
if not source_id and url and download_missing_data and flags.have_internet.is_set() \
and download_manager.can_download:
channel_import_logger.info(f'Fetching channel source id for Channel id={id_}')
background_task(fetch_channel_source_id(id_))
if self.import_skipped:
# Linking changes channels.yaml, which cannot be saved while it has skipped entries: the database would be
# newer than the file, and the next startup would save over them.
channel_import_logger.warning('Not linking Downloads to Channels while channels.yaml has skipped entries')
else:
with get_db_session() as session:
links = channel_download_links(session)
if links:
with get_db_session(commit=True) as session:
_link_downloads(session, links)
self._config_outdated = True
if self._config_outdated:
self.background_dump.activate_switch()
if diffs['downloads']:
# The import's Downloads changed download_manager.yaml; saved after the import's write commits.
save_downloads_config.activate_switch()
def dump_config(self, file: pathlib.Path = None, send_events=False, overwrite=False):
"""Dump all channels from database to config file."""
from wrolpi.collections import Collection
channel_import_logger.info('Dumping channels to config')
with get_db_session() as session:
version = self.db_version_for_dump(session)
# Get all channels, ordered by collection name
channels = session.query(Channel).join(Collection).order_by(Collection.name).all()
# Use config_view to export each channel
channels_data = [channel.config_view() for channel in channels]
self._config['channels'] = channels_data
self._config['version'] = version
channel_import_logger.info(f'Dumping {len(channels_data)} channels to config')
self.save(file, send_events, overwrite)
def preview_backup_import(self, backup_date: str, mode: str) -> dict:
backup_file = self._get_backup_file(backup_date)
backup_data = self.read_config_file(backup_file)
current_data = self.read_config_file() if self.get_file().is_file() else dict(channels=[], version=0)
backup_channels = backup_data.get('channels', [])
current_channels = current_data.get('channels', [])
current_dirs = {ch.get('directory') for ch in current_channels if ch.get('directory')}
backup_dirs = {ch.get('directory') for ch in backup_channels if ch.get('directory')}
add = []
remove = []
unchanged = 0
for ch in backup_channels:
d = ch.get('directory')
if d and d not in current_dirs:
add.append(dict(type='channel', name=ch.get('name', ''), directory=d))
elif d:
unchanged += 1
if mode == 'overwrite':
backup_dir_to_name = {ch.get('directory'): ch.get('name', '') for ch in current_channels}
for d in sorted(current_dirs - backup_dirs):
remove.append(dict(type='channel', name=backup_dir_to_name.get(d, ''), directory=d))
return dict(mode=mode, add=add, remove=remove, unchanged=unchanged)
def import_backup(self, backup_date: str, mode: str, send_events: bool = False):
self._preserve_current_config()
import shutil
backup_file = self._get_backup_file(backup_date)
config_file = self.get_file()
if mode == 'overwrite':
config_file.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(backup_file, config_file)
elif mode == 'merge':
backup_data = self.read_config_file(backup_file)
current_data = self.read_config_file() if config_file.is_file() else dict(channels=[], version=0)
current_dirs = {ch.get('directory') for ch in current_data.get('channels', []) if ch.get('directory')}
merged_channels = list(current_data.get('channels', []))
for ch in backup_data.get('channels', []):
if ch.get('directory') and ch['directory'] not in current_dirs:
merged_channels.append(ch)
merged_data = dict(
channels=merged_channels,
version=current_data.get('version', 0) + 1,
)
config_file.parent.mkdir(parents=True, exist_ok=True)
self.write_config_data(merged_data, config_file)
# Importing a backup is the user's choice: apply it whatever the versions.
self.import_forced(send_events=send_events)
CHANNELS_CONFIG: ChannelsConfig = ChannelsConfig()
# Only the columns channels.yaml saves (not a refresh's counts, info_json, ...); its Collection saves the rest.
register_ownership(ChannelsConfig.file_name, Channel,
columns={'url', 'source_id', 'generate_posters', 'calculate_duration', 'download_missing_data',
'collection_id'})
TEST_CHANNELS_CONFIG: ChannelsConfig = None
def get_channels_config() -> ChannelsConfig:
global TEST_CHANNELS_CONFIG
if PYTEST and not TEST_CHANNELS_CONFIG:
logger.warning('Test did not initialize the channels config')
return
if TEST_CHANNELS_CONFIG:
return TEST_CHANNELS_CONFIG
global CHANNELS_CONFIG
return CHANNELS_CONFIG
@contextlib.contextmanager
def set_test_channels_config():
global TEST_CHANNELS_CONFIG
TEST_CHANNELS_CONFIG = ChannelsConfig()
TEST_CHANNELS_CONFIG.initialize()
yield TEST_CHANNELS_CONFIG
TEST_CHANNELS_CONFIG = None
@dataclass
class VideoDownloaderConfigYtDlpOptionsValidator:
continue_dl: bool
file_name_format: str
nooverwrites: bool
quiet: bool
merge_output_format: str
writeautomaticsub: bool
writeinfojson: bool
writesubtitles: bool
writethumbnail: bool
youtube_include_dash_manifest: bool
def __post_init__(self):
from modules.videos.downloader import preview_filename
try:
preview_filename(self.file_name_format)
except Exception as e:
raise ValueError(f'file_name_format is invalid: {str(e)}')
@dataclass
class VideoDownloaderConfigValidator:
video_resolutions: List[str] = field(default_factory=lambda: ['1080p', '720p', '480p', 'maximum'])
version: int = None
yt_dlp_options: VideoDownloaderConfigYtDlpOptionsValidator = field(default_factory=dict)
yt_dlp_extra_args: str = ''
download_missing_video_comments: bool = False
user_agent: str = ''
sleep_requests: float = 0.75
audio_only: bool = False
audio_format: str = 'mp3'
video_codecs: List[str] = field(default_factory=list)
audio_codecs: List[str] = field(default_factory=list)
transcode: bool = False
strict_codecs: bool = False
def __post_init__(self):
allowed_fields = {i.name for i in dataclasses.fields(VideoDownloaderConfigYtDlpOptionsValidator)}
yt_dlp_options = {k: v for k, v in dict(self.yt_dlp_options).items() if k in allowed_fields}
VideoDownloaderConfigYtDlpOptionsValidator(**yt_dlp_options)
self.yt_dlp_options = yt_dlp_options
class VideoDownloaderConfig(ConfigFile):
file_name = 'videos_downloader.yaml'
default_config = dict(
video_resolutions=['1080p', '720p', '480p', 'maximum'],
version=0,
yt_dlp_options=dict(
continue_dl=True,
file_name_format='%(uploader)s_%(upload_date)s_%(id)s_%(title)s.%(ext)s',
merge_output_format='mp4',
nooverwrites=True,
quiet=False,
writeautomaticsub=True,
writeinfojson=True,
writesubtitles=True,
writethumbnail=True,
youtube_include_dash_manifest=False,
),
yt_dlp_extra_args='',
download_missing_video_comments=False,
user_agent='',
sleep_requests=0.75,
audio_only=False,
audio_format='mp3',
video_codecs=[],
audio_codecs=[],
transcode=False,
strict_codecs=False,
)
validator = VideoDownloaderConfigValidator
@property
def continue_dl(self) -> bool:
return self._config['yt_dlp_options']['continue_dl']
@property
def video_resolutions(self) -> List[str]:
return self._config['video_resolutions']
@video_resolutions.setter
def video_resolutions(self, value: List[str]):
self.update({'video_resolutions': value})
@property
def file_name_format(self) -> str:
return self._config['yt_dlp_options']['file_name_format']
@property
def nooverwrites(self) -> bool:
return self._config['yt_dlp_options']['nooverwrites']
@property
def merge_output_format(self) -> str:
return self._config['yt_dlp_options']['merge_output_format']
@property
def video_format(self) -> str:
"""Alias for merge_output_format, matching the name used in Download.settings."""
return self.merge_output_format
@property
def quiet(self) -> bool:
return self._config['yt_dlp_options']['quiet']
@property
def writeautomaticsub(self) -> bool:
return self._config['yt_dlp_options']['writeautomaticsub']
@property
def writeinfojson(self) -> bool:
return self._config['yt_dlp_options']['writeinfojson']
@property
def writesubtitles(self) -> bool:
return self._config['yt_dlp_options']['writesubtitles']
@property
def writethumbnail(self) -> bool:
return self._config['yt_dlp_options']['writethumbnail']
@property
def youtube_include_dash_manifest(self) -> bool:
return self._config['yt_dlp_options']['youtube_include_dash_manifest']
@property
def yt_dlp_options(self) -> dict:
return self._config['yt_dlp_options']
@yt_dlp_options.setter
def yt_dlp_options(self, value: dict):
self.update({'yt_dlp_options': value})
@property
def yt_dlp_extra_args(self) -> str:
return self._config['yt_dlp_extra_args']
@yt_dlp_extra_args.setter
def yt_dlp_extra_args(self, value: str):
self.update({**self._config, 'yt_dlp_extra_args': value})
@property
def download_missing_video_comments(self) -> bool:
return self._config['download_missing_video_comments']
@download_missing_video_comments.setter
def download_missing_video_comments(self, value: bool):
self.update({**self._config, 'download_missing_video_comments': value})
@property
def user_agent(self) -> str:
return self._config.get('user_agent', '')
@user_agent.setter
def user_agent(self, value: str):
self.update({**self._config, 'user_agent': value})
@property
def sleep_requests(self) -> float:
try:
return float(self._config.get('sleep_requests', 0.75))
except (TypeError, ValueError):
return 0.75
@sleep_requests.setter
def sleep_requests(self, value: float):
self.update({**self._config, 'sleep_requests': value})
@property
def audio_only(self) -> bool:
return self._config.get('audio_only', False)
@audio_only.setter
def audio_only(self, value: bool):
self.update({**self._config, 'audio_only': value})
@property
def audio_format(self) -> str:
return self._config.get('audio_format', 'mp3')
@audio_format.setter
def audio_format(self, value: str):
self.update({**self._config, 'audio_format': value})
@property
def video_codecs(self) -> List[str]:
return self._config.get('video_codecs', [])
@video_codecs.setter
def video_codecs(self, value: List[str]):
self.update({**self._config, 'video_codecs': value})
@property
def audio_codecs(self) -> List[str]:
return self._config.get('audio_codecs', [])
@audio_codecs.setter
def audio_codecs(self, value: List[str]):
self.update({**self._config, 'audio_codecs': value})
@property
def transcode(self) -> bool:
return self._config.get('transcode', False)
@transcode.setter
def transcode(self, value: bool):
self.update({**self._config, 'transcode': value})
@property
def strict_codecs(self) -> bool:
return self._config.get('strict_codecs', False)
@strict_codecs.setter
def strict_codecs(self, value: bool):
self.update({**self._config, 'strict_codecs': value})
VIDEOS_DOWNLOADER_CONFIG: VideoDownloaderConfig = VideoDownloaderConfig()
TEST_VIDEOS_DOWNLOADER_CONFIG: VideoDownloaderConfig = None
def get_videos_downloader_config() -> VideoDownloaderConfig:
global TEST_VIDEOS_DOWNLOADER_CONFIG
if isinstance(TEST_VIDEOS_DOWNLOADER_CONFIG, VideoDownloaderConfig):
return TEST_VIDEOS_DOWNLOADER_CONFIG
global VIDEOS_DOWNLOADER_CONFIG
return VIDEOS_DOWNLOADER_CONFIG
def set_test_downloader_config(enabled: bool):
global TEST_VIDEOS_DOWNLOADER_CONFIG
TEST_VIDEOS_DOWNLOADER_CONFIG = VideoDownloaderConfig() if enabled else None
# Names of settings that can be overridden at channel/RSS or video level.
# Actual default values come from VideoDownloaderConfig.
INHERITABLE_VIDEO_SETTINGS = {
'video_resolutions', 'video_format', 'writesubtitles', 'writeautomaticsub',
'writethumbnail', 'writeinfojson', 'yt_dlp_extra_args', 'sleep_requests',
'user_agent', 'continue_dl', 'nooverwrites', 'audio_only', 'audio_format',
'video_codecs', 'audio_codecs', 'transcode', 'strict_codecs',
}
# Subset of inheritable settings allowed at channel/RSS level.
# continue_dl and nooverwrites are global+video only.
CHANNEL_INHERITABLE_SETTINGS = INHERITABLE_VIDEO_SETTINGS - {'continue_dl', 'nooverwrites'}
def _extract_global_inheritable(config: VideoDownloaderConfig) -> dict:
"""Read current global config values for all inheritable settings, flattened."""
return {
'video_resolutions': config.video_resolutions,
'video_format': config.merge_output_format,
'writesubtitles': config.writesubtitles,
'writeautomaticsub': config.writeautomaticsub,
'writethumbnail': config.writethumbnail,
'writeinfojson': config.writeinfojson,
'yt_dlp_extra_args': config.yt_dlp_extra_args,
'sleep_requests': config.sleep_requests,
'user_agent': config.user_agent,
'continue_dl': config.continue_dl,
'nooverwrites': config.nooverwrites,
'audio_only': config.audio_only,
'audio_format': config.audio_format,
'video_codecs': config.video_codecs,
'audio_codecs': config.audio_codecs,
'transcode': config.transcode,
'strict_codecs': config.strict_codecs,
}
def get_effective_video_settings(download_settings: dict = None) -> dict:
"""Merge global defaults with download-level overrides.
At download time: effective = {**global, **download_settings}
Only inheritable keys present in download_settings override the global value.
"""
config = get_videos_downloader_config()
effective = _extract_global_inheritable(config)
if download_settings:
for key in INHERITABLE_VIDEO_SETTINGS:
if key in download_settings:
effective[key] = download_settings[key]
# Safety net for already-persisted downloads that may contain string values
# for numeric settings (from before the DownloadSettings coercion fix).
# This is the single chokepoint so downstream code can assume correct types.
if effective.get('sleep_requests') not in (None, ''):
try:
effective['sleep_requests'] = float(effective['sleep_requests'])
except (TypeError, ValueError):
# Remove corrupt value so the global config default (already populated
# earlier via _extract_global_inheritable) takes effect.
del effective['sleep_requests']
return effective
def get_yt_dlp_http_headers() -> dict:
"""Get HTTP headers for yt-dlp, including user-agent if configured.
This should be used when cookies are being used to ensure the user-agent
matches the browser that exported the cookies."""
config = get_videos_downloader_config()
if config.user_agent:
return {'User-Agent': config.user_agent}
return {}
def get_yt_dlp_sleep_opts() -> dict:
"""Get sleep options for yt-dlp to avoid rate limiting.
This adds a delay between HTTP requests during metadata extraction,
helping avoid bot detection. Does not affect video file download speed."""
config = get_videos_downloader_config()
try:
if float(config.sleep_requests) > 0:
return {'sleep_requests': config.sleep_requests}
except (TypeError, ValueError):
pass
return {}
@register_switch_handler('save_channels_config')
def save_channels_config():
"""Save the Channels in the DB to the config, with the DB's version (see `ChannelsConfig.dump_config`)."""
with get_db_session() as session:
has_channels = session.query(Channel.id).first() is not None
channels_config = get_channels_config()
if PYTEST and not channels_config:
logger.warning('Refusing to save channels config because test did not initialize a test config!')
return
if not has_channels and channels_config.channels:
logger.warning('Refusing to save empty channels config because channels exist in the config file!')
return
channels_config.dump_config()
logger.info('save_channels_config completed')
save_channels_config: ActivateSwitchMethod
channel_import_logger = logger.getChild('channel_import')
async def fetch_channel_source_id(channel_id: int):
# If we can download from a channel, we must have its source_id.
try:
with get_db_session() as session:
channel = Channel.find_by_id(session, channel_id)
source_id, url = channel.source_id, channel.url
if source_id:
return
# Fetched outside any transaction: this asks the remote site.
source_id = get_channel_source_id(url)
if not source_id:
channel_import_logger.warning(f'Unable to fetch source_id for {url}')
return
with get_db_session(commit=True) as session:
Channel.find_by_id(session, channel_id).source_id = source_id
# Dumped with the database's version (see wrolpi.config_version).
if config := get_channels_config():
config.background_dump.activate_switch()
except Exception as e:
logger.error(f'Failed to get Channel source id of id={channel_id}', exc_info=e)
def import_channels_config():
"""Import channel settings to the DB. Existing channels will be updated."""
if PYTEST and not get_channels_config():
logger.warning('Skipping import_channels_config for this test')
return
channel_import_logger.info('Importing channels config')
get_channels_config().import_config()
channel_import_logger.info('Importing channels config completed')
def channel_download_links(session: Session) -> Dict[int, int]:
"""The recurring Downloads without a Collection which belong to a Channel: {Download id: Collection id}.
A Download belongs to the Channel whose directory is its destination, whose URL (or RSS feed) is its URL, or whose
source id is the channel id of its info json."""
channels = session.query(Channel).options(joinedload(Channel.collection)).all()
by_directory = {str(i.directory): i.collection_id for i in channels if i.directory}
by_url = {i.url: i.collection_id for i in channels if i.url}
for channel in channels:
if rss_url := channel.get_rss_url():
by_url.setdefault(rss_url, channel.collection_id)
by_source_id = {i.source_id: i.collection_id for i in channels if i.source_id}
links = dict()
downloads = session.query(Download.id, Download.url, Download.destination, Download.settings, Download.info_json) \
.filter(Download.frequency.isnot(None), Download.collection_id.is_(None))
for id_, url, destination, settings, info_json in downloads:
destination = str(destination) if destination else (settings or dict()).get('destination')
collection_id = (by_directory.get(destination) if destination else None) or by_url.get(url)
if not collection_id:
info_json = info_json or dict()
if source_id := info_json.get('channel_id') or info_json.get('id'):
collection_id = by_source_id.get(source_id)
if collection_id:
links[id_] = collection_id
return links
def _link_downloads(session: Session, links: Dict[int, int]):
for ids in chunks(list(links), 500):
for download in session.query(Download).filter(Download.id.in_(ids)):
download.collection_id = links[download.id]
session.flush()
def link_channel_and_downloads(session: Session):
"""Associate any Download related to a Channel via its Collection (see `channel_download_links`)."""
if links := channel_download_links(session):
_link_downloads(session, links)
session.commit()
YDL = YoutubeDL(dict(cachedir=YTDLP_CACHE_DIR))
ydl_logger = YDL.params['logger'] = logger.getChild('youtube-dl')
YDL.add_default_info_extractors()
def get_channel_source_id(url: str) -> str:
"""Get the source ID for a channel URL."""
if cookies_unlocked():
logger.info(f'Using encrypted cookies for channel source id: {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
ydl = YoutubeDL(ydl_opts)
ydl.params['logger'] = logger.getChild('youtube-dl')
ydl.add_default_info_extractors()
channel_info = ydl.extract_info(url, download=False, process=False)
return channel_info.get('channel_id') or channel_info['uploader_id']
logger.info(f'No cookies available for channel source id: {url}')
ydl_opts = dict(
cachedir=YTDLP_CACHE_DIR,
**get_yt_dlp_sleep_opts(),
)
ydl = YoutubeDL(ydl_opts)
ydl.params['logger'] = logger.getChild('youtube-dl')
ydl.add_default_info_extractors()
channel_info = ydl.extract_info(url, download=False, process=False)
return channel_info.get('channel_id') or channel_info['uploader_id']
async def get_statistics(session: Session):
with session_curs(session) as curs:
curs.execute('''
SELECT
-- total videos
COUNT(v.id) AS "videos",
-- total videos downloaded over the past week/month/year
COUNT(v.id) FILTER (WHERE published_datetime >= datetime('now', '-7 days')) AS "week",
COUNT(v.id) FILTER (WHERE published_datetime >= datetime('now', '-1 month')) AS "month",
COUNT(v.id) FILTER (WHERE published_datetime >= datetime('now', '-1 year')) AS "year",
-- sum of all video lengths in seconds
COALESCE(SUM(fg.length), 0) AS "sum_duration",
-- sum of all video file sizes
COALESCE(SUM(fg.size), 0) AS "sum_size",
-- largest video
COALESCE(MAX(fg.size), 0) AS "max_size",
-- Videos may or may not have comments.
COUNT(v.id) FILTER ( WHERE v.have_comments = TRUE ) AS "have_comments",
COUNT(v.id) FILTER ( WHERE v.have_comments = FALSE AND v.comments_failed = FALSE
and fg.censored = false and
fg.url is not null) AS "missing_comments",
COUNT(v.id) FILTER ( WHERE v.comments_failed = TRUE ) AS "failed_comments",
COUNT(v.id) FILTER ( WHERE fg.censored = TRUE ) as "censored_videos"
FROM video v
LEFT JOIN file_group fg on v.file_group_id = fg.id
''')
video_stats = dict(curs.fetchone())
# Get the total videos downloaded every month for the past two years.
# (Not parameterized, so `%` needs no escaping.) Like the old Postgres query, months
# without any videos are absent from the results.
curs.execute('''
SELECT strftime('%Y-%m-01 00:00:00', fg.published_datetime) AS "month",
COUNT(video.id) AS "count",
SUM(fg.size) AS "size"
FROM video
LEFT JOIN file_group fg on video.file_group_id = fg.id
WHERE fg.published_datetime IS NOT NULL
AND fg.published_datetime >= datetime('now', 'start of month', '-24 months')
AND fg.published_datetime < datetime('now', 'start of month')
GROUP BY 1
ORDER BY 1
''')
monthly_videos = [dict(i) for i in curs.fetchall()]
for monthly_video in monthly_videos:
# The frontend expects a datetime, not the TEXT that SQLite stores.
monthly_video['month'] = datetime.fromisoformat(monthly_video['month'])
historical_stats = dict(monthly_videos=monthly_videos)
historical_stats['average_count'] = (sum(i['count'] for i in monthly_videos) // len(monthly_videos)) \
if monthly_videos else 0
historical_stats['average_size'] = (sum(i['size'] for i in monthly_videos) // len(monthly_videos)) \
if monthly_videos else 0
curs.execute('''
SELECT COUNT(c.id) AS "channels",
COUNT(c.id) FILTER ( WHERE col.tag_id IS NOT NULL ) AS "tagged_channels"
FROM channel c
LEFT JOIN collection col on col.id = c.collection_id
LEFT JOIN tag t on t.id = col.tag_id
''')
channel_stats = dict(curs.fetchone())
ret = dict(statistics=dict(
videos=video_stats,
channels=channel_stats,
historical=historical_stats,
))
return ret
NAME_PARSER = re.compile(
r'(?:(.*?)_)?((?:\d+?)|(?:NA))_(?:(.{5,25})_)?(.*)'
# The file suffix. Every media suffix a Video FileGroup can lead with belongs here:
# audio is Channel content too, and a suffix missing from this list leaves the file
# without a published date, sorted to the end of its Channel.
r'\.(jpg|webp|flv|mp4|m4v|mkv|webm|mp3|flac|opus|ogg|m4a|wav|aac|wma'
r'|part|info\.json|description|..\.srt|..\.vtt)',
re.IGNORECASE)
def parse_video_file_name(video_path: pathlib.Path) -> \
Tuple[Optional[str], Optional[str], Optional[str], Optional[str]]:
"""
A Video's file name can have data in it, this attempts to extract what may be there.
{published_date} is assumed to always be %Y%m%d
Examples:
NA_{published_date}_{source_id}_{title}{ext}
{channel_name}_NA_{source_id}_{title}{ext}
{channel_name}_{published_date}_{source_id}_{title}{ext} # Typical name format from WROLPi.
{published_date}_{title}{ext}
{title}{ext}
"""
video_str = str(video_path)
if match := NAME_PARSER.match(video_str):
channel, date, source_id, title, _ = match.groups()
channel = None if channel == 'NA' else channel
date = None if date == 'NA' else date
title = title.strip()
if date == 'NA':
return channel, None, source_id, title
if date is None or (len(date) == 8 and date.isdigit()):
return channel, date, source_id, title
# Return the stem as a last resort
title, _ = split_path_stem_and_suffix(video_path)
return None, None, None, title.strip()
def get_video_metadata(video) -> dict:
"""Extract video metadata with fallback chain.
Fallback strategy:
1. Primary: video.channel.name for uploader (when video belongs to a channel)
2. Secondary: info_json + database fields (video.source_id, file_group.published_datetime)
3. Tertiary: parse_video_file_name() - ALL-OR-NOTHING, only if it provides channel AND date AND source_id
Returns:
dict with keys: upload_date (datetime|None), uploader (str), source_id (str), title (str)
"""
file_group = video.file_group
info_json = video.get_info_json() or {}
# Primary sources
upload_date = file_group.published_datetime
if not upload_date and info_json.get('upload_date'):
upload_date = dates.strpdate(info_json['upload_date'])
# Prioritize WROLPi channel name when video belongs to a channel
# This ensures reorganized files use the user's channel name, not the YouTube uploader name
if video.channel and video.channel.name:
uploader = video.channel.name
else:
uploader = info_json.get('uploader') or info_json.get('channel') or ''
source_id = video.source_id or info_json.get('id') or ''
title = file_group.title or info_json.get('title') or info_json.get('fulltitle') or 'untitled'
# Fallback: Parse filename - ALL-OR-NOTHING
# Only use if it provides ALL three fields: channel, date, source_id
if file_group.primary_path and not (upload_date and source_id and uploader):
parsed_channel, parsed_date, parsed_source_id, _ = parse_video_file_name(file_group.primary_path)
if parsed_channel and parsed_date and parsed_source_id:
if not upload_date:
upload_date = dates.strpdate(parsed_date)
if not source_id:
source_id = parsed_source_id
if not uploader:
uploader = parsed_channel
return {
'upload_date': upload_date,
'uploader': uploader,
'source_id': source_id,
'title': title,
}
def find_orphaned_video_files(directory: pathlib.Path) -> Generator[pathlib.Path, None, None]:
"""Finds all files which should be associated with a video file, but a video file does not match their stem.
Example:
video1.mp4, video1.info.json # Video file with info json.
video2.info.json # Orphaned video json.
"""
with get_db_curs() as curs:
if not directory.is_dir():
raise UnknownDirectory()
directory_str = str(directory).rstrip('/')
# Query returns directory and files - files may contain relative or absolute paths
# Use indexed directory column for efficient lookup
# (Not parameterized, so `%` needs no escaping.)
curs.execute(f'''
SELECT directory, files
FROM file_group
WHERE
mimetype NOT LIKE 'video%'
AND (directory = '{directory_str}' OR directory LIKE '{directory_str}/%')
''')
for row in curs.fetchall():
fg_directory = pathlib.Path(row[0]) if row[0] else None
files = row[1]
if isinstance(files, str):
# Raw cursor reads return the JSON column as TEXT.
files = json.loads(files)
for file_info in files:
path = pathlib.Path(file_info['path'])
if not path.is_absolute() and fg_directory:
# Resolve relative path using directory
path = fg_directory / path
yield path
def format_videos_destination(channel_name: str = None, channel_tag: str = None, channel_url: str = None) \
-> pathlib.Path:
"""Return the directory where Videos should be downloaded according to the WROLPi Config.
Note: When you have a Collection object, prefer using Collection.format_destination()
or Collection.get_or_set_directory() instead, as they provide unified behavior
across both channel and domain collections.
@warning: Directory may or may not exist."""
videos_destination = get_wrolpi_config().videos_destination
channel_domain = ''
if channel_url:
try:
channel_domain = extract_domain(channel_url)
except Exception as e:
logger.error(f'Failed to extract domain from Channel URL: {channel_url}', exc_info=e)
if PYTEST:
raise
if channel_name and not isinstance(channel_name, str):
raise RuntimeError('channel name must be string')
if channel_tag and not isinstance(channel_tag, str):
raise RuntimeError('channel tag must be string')
name = escape_file_name(channel_name) if channel_name else ''
variables = dict(
channel_name=name,
channel_tag=channel_tag or '',
channel_domain=channel_domain,
)
try:
videos_destination = videos_destination % variables
except KeyError as e:
msg = f'Cannot download to the "videos_destination" from Settings: {videos_destination}'
raise FileNotFoundError(msg) from e
videos_destination = get_media_directory() / videos_destination.lstrip('/')
return videos_destination
def format_video_filename(
video: Video,
file_name_format: str = None,
) -> str:
"""Single source of truth for video filenames.
Formats a video filename using the configured file_name_format template.
Used by both the video downloader (initial download) and collection reorganization.
Args:
video: Video model with metadata (upload_date, uploader, title, source_id)
file_name_format: Optional override, defaults to config value
Returns:
Formatted filename (may include subdirectory path like "2025/01/video.mp4")
Example:
With format '%(upload_year)s/%(upload_month)s/%(uploader)s_%(upload_date)s_%(id)s_%(title)s.%(ext)s':
Returns '2025/01/WROLPi_20250115_abc123_My Video Title.mp4'
"""
from modules.videos.downloader import convert_wrolpi_filename_format
config = get_videos_downloader_config()
template = file_name_format or config.file_name_format
# Get metadata with fallback chain
metadata = get_video_metadata(video)
upload_date = metadata['upload_date']
uploader = metadata['uploader']
source_id = metadata['source_id']
title = metadata['title']
# Get video file extension
video_path = video.video_path
ext = video_path.suffix.lstrip('.') if video_path else config.merge_output_format or 'mp4'
variables = dict(
uploader=escape_file_name(uploader) if uploader else '',
channel=escape_file_name(uploader) if uploader else '', # Alias
upload_date=upload_date.strftime('%Y%m%d') if upload_date else '',
id=source_id,
title=escape_file_name(title) if title else 'untitled',
ext=ext,
# WROLPi-specific variables for subdirectory organization
upload_year=str(upload_date.year) if upload_date else '',
upload_month=f'{upload_date.month:02d}' if upload_date else '',
)
# Convert WROLPi-specific variables to yt-dlp syntax then apply
# (This handles %(upload_year)s -> %(upload_date>%Y)s conversion)
converted_template = convert_wrolpi_filename_format(template)
# For the converted variables (like %(upload_date>%Y)s), we need to add them too
variables['upload_date>%Y'] = str(upload_date.year) if upload_date else ''
variables['upload_date>%m'] = f'{upload_date.month:02d}' if upload_date else ''
try:
result = converted_template % variables
# Trim the filename portion if it's too long (may include subdirectory paths)
if '/' in result:
# Has subdirectories - only trim the filename part
parts = result.rsplit('/', 1)
subdir = parts[0]
filename = trim_file_name(parts[1])
return f'{subdir}/{filename}'
else:
return trim_file_name(result)
except KeyError as e:
logger.error(f'Invalid variable in video file_name_format: {e}')
# Fallback to a simple format
fallback_title = escape_file_name(title) if title else 'untitled'
return trim_file_name(f'{fallback_title}.{ext}')