mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
387282b445
Moving upload destination input out of drop Segment. Disabling uploads when downloads are disabled. Inverting labels for checkboxes in inverted modal. Deleting file_groups which no longer contain the exclusive primary path. Adding debug logging for all API errors. Checking that upload output is not a directory. Expanding nginx request size. Adding docs to upload endpoint. Using dot-circle icon for more image mimetypes. Using Form.Field to make upload destination required. Adding border to inverted Segments. Many files can be uploaded in chunks. Files can be uploaded, progress is displayed. Showing white border around share QR code. Displaying WROLPi colors for download icons. Creating button to clear DirectorySearch. Excluding channel/domain directories so other directories can be returned. Ordering search results. Ignoring when no channel is found only if we have a destination. Limiting directory search results. Displaying categories when searching destination directories. Handling when live_chat.json file exists. Allowing refresh of directory directly. Logging which download was successful. Removing timeout. Creating index for directory names. Sleeping to catch cancel. Arbitrary files can be downloaded using the new FileDownloader.
614 lines
23 KiB
Python
614 lines
23 KiB
Python
import contextlib
|
|
import datetime
|
|
import functools
|
|
import html
|
|
import pathlib
|
|
import re
|
|
import warnings
|
|
from typing import Tuple, Optional, Generator
|
|
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
from yt_dlp import YoutubeDL
|
|
|
|
from wrolpi import before_startup, dates
|
|
from wrolpi.captions import extract_captions
|
|
from wrolpi.common import ConfigFile, get_media_directory, sanitize_link, register_refresh_cleanup, limit_concurrent
|
|
from wrolpi.dates import Seconds
|
|
from wrolpi.db import get_db_curs, get_db_session, optional_session
|
|
from wrolpi.errors import UnknownDirectory
|
|
from wrolpi.files.models import FileGroup
|
|
from wrolpi.tags import Tag
|
|
from wrolpi.vars import PYTEST
|
|
from .common import is_valid_poster, convert_image, \
|
|
generate_video_poster, logger, REQUIRED_OPTIONS, ConfigError, \
|
|
get_no_channel_directory, get_video_duration
|
|
from .models import Channel, Video
|
|
|
|
logger = logger.getChild(__name__)
|
|
|
|
DEFAULT_DOWNLOAD_FREQUENCY = Seconds.week
|
|
|
|
|
|
class VideoInfoJSON(object):
|
|
channel_source_id = None
|
|
duration = None
|
|
title = None
|
|
url = None
|
|
view_count = None
|
|
|
|
|
|
def process_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():
|
|
title = info_json.get('fulltitle') or info_json.get('title')
|
|
video_info_json.title = html.unescape(title) if title else None
|
|
|
|
video_info_json.duration = info_json.get('duration') or None
|
|
video_info_json.view_count = info_json.get('view_count') or None
|
|
video_info_json.url = info_json.get('webpage_url') or info_json.get('url') or None
|
|
video_info_json.channel_source_id = info_json.get('channel_id') or None
|
|
|
|
return video_info_json
|
|
|
|
|
|
@optional_session
|
|
@functools.lru_cache
|
|
def get_channel_id_by_source_id(session: Session, source_id: str) -> Optional[int]:
|
|
"""Return the id of the Channel which matches the provided `source_id`.
|
|
|
|
These are cached results."""
|
|
channel = session.query(Channel).filter_by(source_id=source_id).one_or_none()
|
|
return channel.id if channel else None
|
|
|
|
|
|
EXTRACT_SUBTITLES = False
|
|
|
|
|
|
def validate_video(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.
|
|
"""
|
|
info_json_path = video.info_json_path
|
|
json_data_missing = bool(video.file_group.title) and bool(video.duration) and bool(video.view_count) \
|
|
and bool(video.url)
|
|
if info_json_path and json_data_missing is False:
|
|
# These properties can be found in the info json.
|
|
video_info_json = process_video_info_json(video)
|
|
video.file_group.title = video_info_json.title
|
|
video.duration = video_info_json.duration
|
|
video.url = video_info_json.url
|
|
# 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:
|
|
video.channel_id = get_channel_id_by_source_id(source_id=video_info_json.channel_source_id)
|
|
|
|
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.upload_date 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.
|
|
_, upload_date, source_id, title = parse_video_file_name(video_path)
|
|
upload_date = upload_date or video.upload_date
|
|
if upload_date and isinstance(upload_date, str):
|
|
upload_date = dates.strpdate(upload_date)
|
|
video.file_group.title = video.file_group.title or html.unescape(title)
|
|
video.upload_date = upload_date
|
|
video.source_id = video.source_id or source_id
|
|
|
|
if channel_generate_poster:
|
|
# 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.duration = duration
|
|
|
|
if video_path and not video.duration:
|
|
# Duration was not retrieved during poster generation.
|
|
video.duration = get_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).
|
|
old.unlink()
|
|
old = new
|
|
|
|
if not is_valid_poster(old):
|
|
# Poster is not valid, convert it and place it in the new location.
|
|
try:
|
|
convert_image(old, new)
|
|
if not new.is_file():
|
|
raise FileNotFoundError(f'Failed to convert poster: {new}')
|
|
if old != new:
|
|
# Only remove the old one if we are not converting in-place.
|
|
old.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
|
|
|
|
|
|
class ChannelsConfig(ConfigFile):
|
|
file_name = 'channels.yaml'
|
|
default_config = dict(
|
|
channels={
|
|
'wrolpi': dict(
|
|
name='WROLPi',
|
|
url='https://www.youtube.com/channel/UC4t8bw1besFTyjW7ZBCOIrw/videos',
|
|
directory='videos/wrolpi',
|
|
download_frequency=604800,
|
|
)
|
|
},
|
|
favorites=dict(),
|
|
)
|
|
|
|
@property
|
|
def channels(self) -> dict:
|
|
return self._config['channels']
|
|
|
|
@channels.setter
|
|
def channels(self, value: dict):
|
|
self.update({'channels': value})
|
|
|
|
@property
|
|
def favorites(self) -> dict:
|
|
warnings.warn('Favorites have bee moved to Tags', DeprecationWarning)
|
|
return self._config['favorites']
|
|
|
|
@favorites.setter
|
|
def favorites(self, value: dict):
|
|
warnings.warn('Favorites have bee moved to Tags', DeprecationWarning)
|
|
self.update({'favorites': value})
|
|
|
|
|
|
CHANNELS_CONFIG: ChannelsConfig = ChannelsConfig()
|
|
TEST_CHANNELS_CONFIG: ChannelsConfig = None
|
|
|
|
|
|
def get_channels_config() -> ChannelsConfig:
|
|
global TEST_CHANNELS_CONFIG
|
|
if isinstance(TEST_CHANNELS_CONFIG, ConfigFile):
|
|
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()
|
|
yield
|
|
TEST_CHANNELS_CONFIG = None
|
|
|
|
|
|
class VideoDownloaderConfig(ConfigFile):
|
|
file_name = 'videos_downloader.yaml'
|
|
default_config = dict(
|
|
continue_dl=True,
|
|
dateafter='19900101',
|
|
file_name_format='%(uploader)s_%(upload_date)s_%(id)s_%(title)s.%(ext)s',
|
|
nooverwrites=True,
|
|
quiet=False,
|
|
writeautomaticsub=True,
|
|
writeinfojson=True,
|
|
writesubtitles=True,
|
|
writethumbnail=True,
|
|
youtube_include_dash_manifest=False,
|
|
)
|
|
|
|
@property
|
|
def continue_dl(self) -> bool:
|
|
return self._config['continue_dl']
|
|
|
|
@continue_dl.setter
|
|
def continue_dl(self, value: bool):
|
|
self.update({'continue_dl': value})
|
|
|
|
@property
|
|
def dateafter(self) -> str:
|
|
return self._config['dateafter']
|
|
|
|
@dateafter.setter
|
|
def dateafter(self, value: str):
|
|
self.update({'dateafter': value})
|
|
|
|
@property
|
|
def file_name_format(self) -> str:
|
|
return self._config['file_name_format']
|
|
|
|
@file_name_format.setter
|
|
def file_name_format(self, value: str):
|
|
self.update({'file_name_format': value})
|
|
|
|
@property
|
|
def nooverwrites(self) -> bool:
|
|
return self._config['nooverwrites']
|
|
|
|
@nooverwrites.setter
|
|
def nooverwrites(self, value: bool):
|
|
self.update({'nooverwrites': value})
|
|
|
|
@property
|
|
def quiet(self) -> bool:
|
|
return self._config['quiet']
|
|
|
|
@quiet.setter
|
|
def quiet(self, value: bool):
|
|
self.update({'quiet': value})
|
|
|
|
@property
|
|
def writeautomaticsub(self) -> bool:
|
|
return self._config['writeautomaticsub']
|
|
|
|
@writeautomaticsub.setter
|
|
def writeautomaticsub(self, value: bool):
|
|
self.update({'writeautomaticsub': value})
|
|
|
|
@property
|
|
def writeinfojson(self) -> bool:
|
|
return self._config['writeinfojson']
|
|
|
|
@writeinfojson.setter
|
|
def writeinfojson(self, value: bool):
|
|
self.update({'writeinfojson': value})
|
|
|
|
@property
|
|
def writesubtitles(self) -> bool:
|
|
return self._config['writesubtitles']
|
|
|
|
@writesubtitles.setter
|
|
def writesubtitles(self, value: bool):
|
|
self.update({'writesubtitles': value})
|
|
|
|
@property
|
|
def writethumbnail(self) -> bool:
|
|
return self._config['writethumbnail']
|
|
|
|
@writethumbnail.setter
|
|
def writethumbnail(self, value: bool):
|
|
self.update({'writethumbnail': value})
|
|
|
|
@property
|
|
def youtube_include_dash_manifest(self) -> bool:
|
|
return self._config['youtube_include_dash_manifest']
|
|
|
|
@youtube_include_dash_manifest.setter
|
|
def youtube_include_dash_manifest(self, value: bool):
|
|
self.update({'youtube_include_dash_manifest': value})
|
|
|
|
|
|
VIDEO_DOWNLOADER_CONFIG: VideoDownloaderConfig = VideoDownloaderConfig()
|
|
TEST_VIDEO_DOWNLOADER_CONFIG: VideoDownloaderConfig = None
|
|
|
|
|
|
def get_downloader_config() -> VideoDownloaderConfig:
|
|
global TEST_VIDEO_DOWNLOADER_CONFIG
|
|
if isinstance(TEST_VIDEO_DOWNLOADER_CONFIG, VideoDownloaderConfig):
|
|
return TEST_VIDEO_DOWNLOADER_CONFIG
|
|
|
|
global VIDEO_DOWNLOADER_CONFIG
|
|
return VIDEO_DOWNLOADER_CONFIG
|
|
|
|
|
|
def set_test_downloader_config(enabled: bool):
|
|
global TEST_VIDEO_DOWNLOADER_CONFIG
|
|
if enabled:
|
|
TEST_VIDEO_DOWNLOADER_CONFIG = VideoDownloaderConfig()
|
|
else:
|
|
TEST_VIDEO_DOWNLOADER_CONFIG = None
|
|
|
|
|
|
def get_channels_config_from_db(session: Session) -> dict:
|
|
"""Create a dictionary that contains all the Channels from the DB."""
|
|
channels = session.query(Channel).order_by(Channel.directory).all()
|
|
channels = sorted((i.config_view() for i in channels), key=lambda i: i['directory'])
|
|
return dict(channels=channels)
|
|
|
|
|
|
@optional_session()
|
|
def save_channels_config(session: Session = None):
|
|
"""Get the Channel information from the DB, save it to the config."""
|
|
config = get_channels_config_from_db(session)
|
|
channels_config = get_channels_config()
|
|
channels_config.update(config)
|
|
|
|
|
|
channel_import_logger = logger.getChild('channel_import')
|
|
|
|
|
|
@before_startup
|
|
@register_refresh_cleanup
|
|
@limit_concurrent(1)
|
|
def import_channels_config():
|
|
"""Import channel settings to the DB. Existing channels will be updated."""
|
|
if PYTEST and not TEST_CHANNELS_CONFIG:
|
|
channel_import_logger.warning(
|
|
f'Not importing channels during this test. Use `test_channels_config` fixture if you would '
|
|
f'like to call this.')
|
|
return
|
|
|
|
channel_import_logger.info('Importing videos config')
|
|
try:
|
|
config = get_channels_config()
|
|
channels, favorites = config.channels, config.favorites
|
|
|
|
save_config = False
|
|
with get_db_session(commit=True) as session:
|
|
for data in channels:
|
|
if isinstance(data, str):
|
|
# Outdated style config.
|
|
# TODO remove this after beta.
|
|
data = channels[data]
|
|
# A Channel's directory is saved (in config) relative to the media directory.
|
|
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}"')
|
|
|
|
channel = session.query(Channel).filter(
|
|
or_(
|
|
Channel.directory == str(directory),
|
|
Channel.directory == str(data['directory']),
|
|
Channel.directory == str(directory.relative_to(get_media_directory())),
|
|
),
|
|
).one_or_none()
|
|
if not channel:
|
|
# Channel not yet in the DB, add it.
|
|
channel = Channel(directory=directory)
|
|
session.add(channel)
|
|
|
|
# Only name and directory are required
|
|
channel.name = data['name']
|
|
channel.directory = directory
|
|
|
|
# A URL should not be an empty string
|
|
data['url'] = data['url'] or None
|
|
|
|
# Copy existing channel data, update all values from the config. This is necessary to clear out
|
|
# values not in the config.
|
|
full_data = channel.dict()
|
|
full_data.update(data)
|
|
full_data['skip_download_videos'] = list(set(data.get('skip_download_videos', {})))
|
|
channel.update(full_data)
|
|
|
|
if not channel.source_id and channel.url:
|
|
# If we can download from a channel, we must have its source_id.
|
|
channel.source_id = get_channel_source_id(channel.url)
|
|
save_config = True
|
|
if not channel.source_id:
|
|
channel_import_logger.warning(f'Unable to fetch source_id for {channel.url}')
|
|
else:
|
|
session.commit()
|
|
|
|
channel_import_logger.debug(f'Updated {repr(channel.name)}'
|
|
f' url={channel.url}'
|
|
f' source_id={channel.source_id}'
|
|
f' directory={channel.directory}'
|
|
f' download_frequency={channel.download_frequency}'
|
|
)
|
|
|
|
if save_config:
|
|
# Information about the channel was fetched, store it.
|
|
save_channels_config()
|
|
|
|
with get_db_session(commit=True) as session:
|
|
channels_by_link = {sanitize_link(i.name): i for i in session.query(Channel)}
|
|
for directory_, favorites in favorites.items():
|
|
if directory_ != 'NO CHANNEL':
|
|
# A Channel's directory is saved (in the config) relative to the media directory.
|
|
directory = get_media_directory() / directory_
|
|
channel: Channel = session.query(Channel).filter_by(directory=directory).one_or_none()
|
|
if not channel:
|
|
# Directory may be the outdated "link".
|
|
# TODO remove these old favorites after beta.
|
|
channel = channels_by_link.get(directory_)
|
|
if not channel:
|
|
channel_import_logger.warning(f'Cannot find channel {directory=} for favorites!')
|
|
continue
|
|
channel_dir = channel.directory
|
|
else:
|
|
channel_dir = get_no_channel_directory()
|
|
|
|
favorites_tag = Tag.find_by_name('Favorite')
|
|
if not favorites_tag:
|
|
channel_import_logger.warning(f'Cannot tag favorite videos without Favorite tag!')
|
|
return
|
|
|
|
# Set favorite Videos of this Channel.
|
|
for video_path, data in favorites.items():
|
|
# Favorite in the config is the name of the video_path. Add the channel directory onto this
|
|
# video_path, so we can match the complete path for the Video.
|
|
video_path = channel_dir / video_path
|
|
file_group: FileGroup = FileGroup.find_by_path(video_path, session)
|
|
if file_group and file_group.tag_files:
|
|
# Video is already tagged or favorited.
|
|
pass
|
|
elif file_group:
|
|
file_group.add_tag(favorites_tag, session)
|
|
else:
|
|
channel_import_logger.warning(f'Cannot find video to favorite: {video_path}')
|
|
except Exception as e:
|
|
channel_import_logger.warning('Failed to load channels config!', exc_info=e)
|
|
if PYTEST:
|
|
# Do not interrupt startup, only raise during testing.
|
|
raise
|
|
|
|
|
|
YDL = YoutubeDL()
|
|
YDL.params['logger'] = logger.getChild('youtube-dl')
|
|
YDL.add_default_info_extractors()
|
|
|
|
|
|
def get_channel_source_id(url: str) -> str:
|
|
channel_info = YDL.extract_info(url, download=False, process=False)
|
|
return channel_info.get('uploader_id') or channel_info['channel_id']
|
|
|
|
|
|
async def get_statistics():
|
|
with get_db_curs() 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 upload_date >= current_date - interval '1 week') AS "week",
|
|
COUNT(v.id) FILTER (WHERE upload_date >= current_date - interval '1 month') AS "month",
|
|
COUNT(v.id) FILTER (WHERE upload_date >= current_date - interval '1 year') AS "year",
|
|
-- sum of all video lengths in seconds
|
|
COALESCE(SUM(v.duration), 0) AS "sum_duration",
|
|
-- sum of all video file sizes
|
|
COALESCE(SUM(fg.size), 0)::BIGINT AS "sum_size",
|
|
-- largest video
|
|
COALESCE(MAX(fg.size), 0) AS "max_size"
|
|
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.
|
|
curs.execute('''
|
|
SELECT
|
|
DATE_TRUNC('month', months.a),
|
|
COUNT(video.id)::BIGINT,
|
|
SUM(size)::BIGINT AS "size"
|
|
FROM
|
|
generate_series(
|
|
date_trunc('month', current_date) - interval '2 years',
|
|
date_trunc('month', current_date) - interval '1 month',
|
|
'1 month'::interval) AS months(a),
|
|
video
|
|
LEFT JOIN file_group fg on video.file_group_id = fg.id
|
|
WHERE
|
|
video.upload_date >= date_trunc('month', months.a)
|
|
AND video.upload_date < date_trunc('month', months.a) + interval '1 month'
|
|
AND video.upload_date IS NOT NULL
|
|
GROUP BY
|
|
1
|
|
ORDER BY
|
|
1
|
|
''')
|
|
monthly_videos = [dict(i) for i in curs.fetchall()]
|
|
|
|
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(id) AS "channels"
|
|
FROM
|
|
channel
|
|
''')
|
|
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,15})_)?(.*)\.'
|
|
r'(jpg|webp|flv|mp4|part|info\.json|description|webm|..\.srt|..\.vtt)')
|
|
|
|
|
|
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.
|
|
|
|
Example: {channel_name}_{upload_date}_{source_id}_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 = pathlib.Path(video_path).stem.strip()
|
|
return None, None, None, 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(directory).rstrip('/')
|
|
|
|
curs.execute(f'''
|
|
SELECT files
|
|
FROM file_group
|
|
WHERE
|
|
mimetype NOT LIKE 'video%'
|
|
AND primary_path LIKE '{directory}/%'
|
|
''')
|
|
results = (pathlib.Path(j['path']) for i in curs.fetchall() for j in i[0])
|
|
yield from results
|