mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 02:48:38 -04:00
ad0ec88f27
* Remove a file's Tags Directory links when WROLPi deletes it When WROLPi deleted a tagged file (a forced delete, duplicate Video cleanup, or a poster conversion), the source was unlinked but its Tags Directory hardlink was not. The link became the only copy, so the tags sync refused to delete it and warned on every sync. The links of a file are now removed before WROLPi removes it, and only if the link is the same file as its source. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Remove Tags Directory links on path deletes; keep a valid poster's link The file browser deletes by path (`lib.delete`), which unlinks the sources before any FileGroup is deleted, so its tagged files still left orphaned links. It now removes the links of the files it is about to delete, or of every FileGroup in a deleted directory. A poster is no longer unlinked from the Tags Directory unless it is actually removed: a valid JPEG that is not named .jpg is kept with its link, and a converted poster's links are removed only once the conversion succeeds. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1350 lines
56 KiB
Python
1350 lines
56 KiB
Python
import contextlib
|
|
import dataclasses
|
|
import html
|
|
import json
|
|
import pathlib
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Generator, Type, List
|
|
from typing import Tuple, Optional
|
|
|
|
import pytz
|
|
from sqlalchemy.orm import Session
|
|
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.dates import Seconds, from_timestamp
|
|
from wrolpi.db import get_db_curs, get_db_session, session_curs
|
|
from wrolpi.downloader import Download, download_manager
|
|
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.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
|
|
|
|
|
|
@dataclass
|
|
class ChannelsConfigValidator:
|
|
version: int = None
|
|
channels: list[ChannelDictValidator] = dataclasses.field(default_factory=list)
|
|
|
|
|
|
class ChannelsConfig(ConfigFile):
|
|
file_name = 'channels.yaml'
|
|
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):
|
|
"""Import channels from config file into database using batch operations."""
|
|
file = file or self.get_file()
|
|
|
|
# If config file doesn't exist, mark import as successful (nothing to import)
|
|
if not file.is_file():
|
|
channel_import_logger.info('No channels config file, skipping import')
|
|
self.successful_import = True
|
|
return
|
|
|
|
super().import_config()
|
|
try:
|
|
channels = self.channels
|
|
# Empty channels list = never delete DB records
|
|
if not channels:
|
|
channel_import_logger.info('No channels in config, preserving existing DB channels')
|
|
self.successful_import = True
|
|
return
|
|
|
|
channel_directories = [i['directory'] for i in channels]
|
|
if len(channel_directories) != len(set(channel_directories)):
|
|
raise RuntimeError('Refusing to import channels config because it contains duplicate directories!')
|
|
|
|
# Validate all channels have required options
|
|
for data in channels:
|
|
directory = get_media_directory() / data['directory']
|
|
for option in (i for i in REQUIRED_OPTIONS if i not in data):
|
|
raise ConfigError(f'Channel "{directory}" is required to have "{option}"')
|
|
|
|
updated_channel_ids = set()
|
|
|
|
with get_db_session(commit=True) as session:
|
|
# Prepare data with kind='channel' for all entries
|
|
channel_data_list = []
|
|
for data in channels:
|
|
data = data.copy()
|
|
data['kind'] = 'channel'
|
|
channel_data_list.append(data)
|
|
|
|
# Batch create/update Collections
|
|
from wrolpi.collections import Collection
|
|
collections = Collection.batch_from_config(session, channel_data_list)
|
|
|
|
# Pre-fetch existing Channels by collection_id in ONE query
|
|
collection_ids = [c.id for c in collections]
|
|
existing_channels_list = session.query(Channel).filter(
|
|
Channel.collection_id.in_(collection_ids)
|
|
).all()
|
|
existing_channels = {ch.collection_id: ch for ch in existing_channels_list}
|
|
|
|
# Pre-fetch all tags for channel update in ONE query
|
|
tag_names = {d.get('tag_name') for d in channels if d.get('tag_name')}
|
|
tags_by_name = {}
|
|
if tag_names:
|
|
from wrolpi.tags import Tag
|
|
tags = session.query(Tag).filter(Tag.name.in_(tag_names)).all()
|
|
tags_by_name = {t.name: t for t in tags}
|
|
|
|
# Pre-fetch existing downloads for update in ONE query
|
|
download_urls = set()
|
|
for data in channels:
|
|
for dl in data.get('downloads', []):
|
|
if isinstance(dl, dict):
|
|
download_urls.add(dl['url'])
|
|
existing_downloads = {}
|
|
if download_urls:
|
|
from wrolpi.downloader import Download
|
|
downloads = session.query(Download).filter(Download.url.in_(download_urls)).all()
|
|
existing_downloads = {d.url: d for d in downloads}
|
|
|
|
# Process channels
|
|
new_channels = []
|
|
for collection, data in zip(collections, channel_data_list):
|
|
channel = existing_channels.get(collection.id)
|
|
|
|
if not channel:
|
|
# Create new Channel
|
|
channel = Channel(
|
|
collection_id=collection.id,
|
|
url=data.get('url'),
|
|
source_id=data.get('source_id'),
|
|
)
|
|
new_channels.append(channel)
|
|
existing_channels[collection.id] = channel
|
|
|
|
# Batch add new channels
|
|
if new_channels:
|
|
session.add_all(new_channels)
|
|
session.flush()
|
|
|
|
# Now update all channels (both new and existing)
|
|
for collection, data in zip(collections, channel_data_list):
|
|
channel = existing_channels[collection.id]
|
|
# Call batch_update with pre-fetched data
|
|
channel.batch_update(data, tags_by_name, existing_downloads)
|
|
updated_channel_ids.add(channel.id)
|
|
|
|
# Background task for missing source_id
|
|
if not channel.source_id and channel.url and flags.have_internet.is_set():
|
|
if download_manager.can_download and channel.download_missing_data:
|
|
channel_import_logger.info(f'Fetching channel source id for {channel}')
|
|
background_task(fetch_channel_source_id(channel.id))
|
|
|
|
channel_import_logger.debug(f'Updated {repr(channel.name)}'
|
|
f' url={channel.url}'
|
|
f' source_id={channel.source_id}'
|
|
f' directory={channel.directory}'
|
|
)
|
|
|
|
with get_db_session(commit=True) as session:
|
|
# Delete any Channels that were deleted from the config, along with their Collections.
|
|
orphaned_collection_ids = []
|
|
for channel in session.query(Channel):
|
|
if channel.id not in updated_channel_ids:
|
|
logger.warning(f'Deleting {channel} because it is not in the config.')
|
|
orphaned_collection_ids.append(channel.collection_id)
|
|
channel.delete_with_videos()
|
|
|
|
# Delete orphaned Collections (kind='channel') that no longer have a Channel.
|
|
# This prevents Collections from appearing in the UI with null channel_id.
|
|
from wrolpi.collections import Collection
|
|
for collection_id in orphaned_collection_ids:
|
|
collection = session.query(Collection).filter_by(id=collection_id).one_or_none()
|
|
if collection:
|
|
logger.warning(f'Deleting orphaned Collection {collection.name} (id={collection.id})')
|
|
session.delete(collection)
|
|
|
|
# Also clean up any pre-existing orphaned Collections (kind='channel')
|
|
# that have no associated Channel. This handles orphans created before
|
|
# this cleanup logic was added.
|
|
with get_db_session(commit=True) as session:
|
|
from wrolpi.collections import Collection
|
|
# Find all Collections with kind='channel' that no Channel references
|
|
orphaned_collections = session.query(Collection).outerjoin(
|
|
Channel, Channel.collection_id == Collection.id
|
|
).filter(
|
|
Collection.kind == 'channel',
|
|
Channel.id.is_(None)
|
|
).all()
|
|
|
|
for collection in orphaned_collections:
|
|
logger.warning(f'Deleting pre-existing orphaned Collection {collection.name} (id={collection.id})')
|
|
session.delete(collection)
|
|
|
|
# Assign Videos to Channels using bulk SQL
|
|
from modules.videos import claim_videos_for_channels
|
|
claim_videos_for_channels(list(updated_channel_ids))
|
|
|
|
# Create any missing Downloads. Associate Downloads with any necessary Channels.
|
|
with get_db_session() as session:
|
|
link_channel_and_downloads(session)
|
|
|
|
self.successful_import = True
|
|
except Exception as e:
|
|
self.successful_import = False
|
|
message = f'Failed to import {self.get_relative_file()} config!'
|
|
if send_events:
|
|
channel_import_logger.warning(message, exc_info=e)
|
|
Events.send_config_import_failed(message)
|
|
raise
|
|
|
|
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:
|
|
# 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
|
|
|
|
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)
|
|
|
|
self.import_config(send_events=send_events)
|
|
|
|
|
|
CHANNELS_CONFIG: ChannelsConfig = ChannelsConfig()
|
|
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})
|
|
|
|
def import_config(self, file: pathlib.Path = None, send_events=False):
|
|
super().import_config(file, send_events)
|
|
self.successful_import = True
|
|
|
|
|
|
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 {}
|
|
|
|
|
|
def get_channels_config_from_db(session: Session) -> dict:
|
|
"""Create a dictionary that contains all the Channels from the DB."""
|
|
from wrolpi.collections import Collection
|
|
channels = session.query(Channel).join(Collection).order_by(Collection.directory).all()
|
|
channels = sorted((i.config_view() for i in channels), key=lambda i: i['directory'] if i['directory'] else '')
|
|
return dict(channels=channels)
|
|
|
|
|
|
@register_switch_handler('save_channels_config')
|
|
def save_channels_config():
|
|
"""Get the Channel information from the DB, save it to the config."""
|
|
with get_db_session() as session:
|
|
config = get_channels_config_from_db(session)
|
|
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 config.get('channels') and channels_config.channels:
|
|
logger.warning('Refusing to save empty channels config because channels exist in the config file!')
|
|
return
|
|
|
|
channels_config.update(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:
|
|
from modules.videos.channel.lib import get_channel
|
|
channel = Channel.find_by_id(session, channel_id)
|
|
|
|
channel.source_id = channel.source_id or get_channel_source_id(channel.url)
|
|
if not channel.source_id:
|
|
channel_import_logger.warning(f'Unable to fetch source_id for {channel.url}')
|
|
else:
|
|
session.commit()
|
|
save_channels_config.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 link_channel_and_downloads(session: Session, channel_: Type[Base] = Channel, download_: Type[Base] = Download):
|
|
"""Associate any Download related to a Channel via its Collection.
|
|
|
|
Downloads are linked to Collections (via collection_id) rather than directly to Channels.
|
|
This function finds Channels and links their Downloads to the Channel's Collection.
|
|
"""
|
|
|
|
def get_destination(d):
|
|
"""Get destination from download.destination column or settings['destination']."""
|
|
if d.destination:
|
|
return str(d.destination)
|
|
return (d.settings or dict()).get('destination')
|
|
|
|
# Only Downloads with a frequency can be a Collection Download.
|
|
downloads = list(session.query(download_).filter(download_.frequency.isnot(None)).all())
|
|
# Download.url is unique and cannot be null.
|
|
downloads_by_url = {i.url: i for i in downloads}
|
|
# Many Downloads may share the same destination (from either column or settings).
|
|
downloads_with_destination = [(d, get_destination(d)) for d in downloads if get_destination(d)]
|
|
channels = session.query(channel_).all()
|
|
|
|
need_commit = False
|
|
for channel in channels:
|
|
directory = str(channel.directory)
|
|
for download, destination in downloads_with_destination:
|
|
if destination == directory and not download.collection_id:
|
|
download.collection_id = channel.collection_id
|
|
need_commit = True
|
|
|
|
download = downloads_by_url.get(channel.url)
|
|
if download and not download.collection_id:
|
|
download.collection_id = channel.collection_id
|
|
need_commit = True
|
|
|
|
# Get any Downloads for a Channel's RSS feed.
|
|
rss_url = channel.get_rss_url()
|
|
if rss_url and (download := downloads_by_url.get(rss_url)):
|
|
if not download.collection_id:
|
|
download.collection_id = channel.collection_id
|
|
need_commit = True
|
|
|
|
# Associate any Download which shares a Channel's URL.
|
|
for download in downloads:
|
|
channel = channel_.get_by_url(session, download.url)
|
|
if channel and not download.collection_id:
|
|
download.collection_id = channel.collection_id
|
|
need_commit = True
|
|
|
|
# Associate any Download which has info_json channel_id matching a Channel's source_id.
|
|
for download in downloads:
|
|
if download.collection_id:
|
|
continue
|
|
info_json = download.info_json or {}
|
|
channel_source_id = info_json.get('channel_id') or info_json.get('id')
|
|
if channel_source_id:
|
|
channel = channel_.get_by_source_id(session, channel_source_id)
|
|
if channel:
|
|
download.collection_id = channel.collection_id
|
|
need_commit = True
|
|
|
|
if need_commit:
|
|
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}')
|