Rework source images to depend on indexing instead

[tcely]

Currently there is duplication.

For new sources we are essentially running two indexing tasks.

tubesync/tubesync/sync/models/source.py

Lines 484 to 488 in 7ee8c96

	@property
	def get_image_url(self):
	return get_youtube_image_info(self.url)

tubesync/tubesync/sync/youtube.py

Lines 83 to 148 in 7ee8c96

	def get_image_info(url):
	avatar_url = None
	banner_url = None
	thumbnail_url = None
	opts = get_yt_opts()
	opts.update({
	'skip_download': True,
	'simulate': True,
	'logger': log,
	'extract_flat': True, # Change to False to get detailed info
	'check_formats': False,
	})
	with yt_dlp.YoutubeDL(opts) as y:
	try:
	response = y.extract_info(url, download=False)
	except yt_dlp.utils.DownloadError as e:
	raise YouTubeError(f'Failed to extract info for "{url}": {e}') from e
	else:
	max_height = 0
	for thumbnail in response['thumbnails']:
	thumbnail_height = thumbnail.get('height')
	try:
	thumbnail_height = int(thumbnail_height)
	except (TypeError, ValueError,):
	thumbnail_height = int()
	if thumbnail['id'] == 'avatar_uncropped':
	avatar_url = thumbnail['url']
	elif thumbnail['id'] == 'banner_uncropped':
	banner_url = thumbnail['url']
	elif thumbnail_height > max_height:
	max_height = thumbnail_height
	thumbnail_url = thumbnail['url']
	try:
	entry_type = response['entries'][0].get('_type')
	except IndexError:
	# an empty entries list
	pass
	else:
	if 'url' == entry_type:
	del response['entries']
	elif 'playlist' == entry_type:
	for playlist in response['entries']:
	del playlist['entries']
	from .models import Metadata
	t = Metadata.objects.defer('value').filter(
	source__isnull=True,
	media__isnull=True,
	).get_or_create(
	key=response['id'],
	site=response['extractor_key'],
	)
	md = t[0]
	field_defaults = {
	f.attname: f.get_default()
	for f in md._meta.fields
	if f.has_default()
	}
	if 'retrieved' in field_defaults:
	md.retrieved = field_defaults['retrieved']
	md.value = response
	md.save()
	return avatar_url, banner_url, thumbnail_url

We should instead index the source, moving the information needed for downloading images into the database immediately, then rework the download of source images to wait for the data to become available from the database.

tubesync/tubesync/sync/tasks.py

Lines 506 to 689 in 7ee8c96

	@db_task(delay=30, priority=80, queue=Val(TaskQueue.LIMIT))
	def index_source(source_id):
	'''
	Indexes media available from a Source object.
	'''
	db.reset_queries()
	cleanup_completed_tasks()
	# deleting expired media should happen any time an index task is requested
	cleanup_old_media()
	try:
	source = Source.objects.get(pk=source_id)
	except Source.DoesNotExist as e:
	# Task triggered but the Source has been deleted, delete the task
	raise CancelExecution(_('no such source'), retry=False) from e
	# An inactive Source would return an empty list for videos anyway
	if not source.is_active:
	return False
	indexing_lock = huey_lock_task(
	f'source:{source.uuid}',
	queue=Val(TaskQueue.FS),
	)
	# be sure that this is locked
	if not indexing_lock.acquired:
	indexing_lock.acquired = True
	# update the target schedule column
	# ruff: ignore[B018]
	source.task_run_at_dt
	update_model(source, target_schedule=source.target_schedule)
	# Reset any errors
	source.has_failed = False
	# Index the source
	videos = source.index_media()
	if not videos:
	source.has_failed = True
	update_model(source, has_failed=source.has_failed)
	indexing_lock.acquired = False
	raise NoMediaException(f'Source "{source}" (ID: {source_id}) returned no '
	f'media to index, is the source key valid? Check the '
	f'source configuration is correct and that the source '
	f'is reachable')
	# Got some media, update the last crawl timestamp
	source.last_crawl = timezone.now()
	update_model(
	source,
	has_failed=source.has_failed,
	last_crawl=source.last_crawl,
	)
	num_videos = len(videos)
	log.info(f'Found {num_videos} media items for source: {source}')
	tvn_format = '{:,}' + f'/{num_videos:,}'
	db_batch_data = queue(list(), maxlen=50)
	db_fields_data = frozenset((
	'retrieved',
	'site',
	'value',
	))
	db_batch_media = queue(list(), maxlen=10)
	db_fields_media = frozenset((
	'duration',
	'published',
	'title',
	))
	fields = lambda f, m: m.get_metadata_field(f)
	task = get_source_index_task(source_id)
	if task:
	task._verbose_name = remove_enclosed(
	task.verbose_name, '[', ']', ' ',
	valid='0123456789/,',
	end=task.verbose_name.find('Index'),
	)
	vn = 0
	video_keys = set()
	while len(videos) > 0:
	vn += 1
	video = videos.popleft()
	# Create or update each video as a Media object
	key = video.get(source.key_field, None)
	if not key:
	# Video has no unique key (ID), it can't be indexed
	continue
	video_keys.add(key)
	if len(db_batch_data) == db_batch_data.maxlen:
	save_db_batch(Metadata.objects, db_batch_data, db_fields_data)
	if len(db_batch_media) == db_batch_media.maxlen:
	save_db_batch(Media.objects, db_batch_media, db_fields_media)
	update_task_status(task, tvn_format.format(vn))
	media_defaults = dict()
	# create a dummy instance to use its functions
	media = Media(source=source, key=key)
	media_defaults['duration'] = float(video.get(fields('duration', media), None) or 0) or None
	media_defaults['title'] = str(video.get(fields('title', media), ''))[:200]
	site = video.get(fields('ie_key', media), None)
	timestamp = video.get(fields('timestamp', media), None)
	try:
	published_dt = media.ts_to_dt(timestamp)
	except AssertionError:
	pass
	else:
	if published_dt:
	media_defaults['published'] = published_dt
	# Retrieve or create the actual media instance
	media, new_media = source.media_source.only(
	'uuid',
	'source',
	'key',
	*db_fields_media,
	).get_or_create(defaults=media_defaults, source=source, key=key)
	db_batch_media.append(media)
	data, new_data = source.videos.defer('value').filter(
	media__isnull=True,
	).get_or_create(source=source, key=key)
	if site:
	data.site = site
	data.retrieved = source.last_crawl
	data.value = { k: v for k,v in video.items() if v is not None }
	db_batch_data.append(data)
	migrating_lock = huey_lock_task(
	f'index_media:{media.uuid}',
	queue=Val(TaskQueue.FS),
	)
	if not migrating_lock.acquired:
	migrating_lock.acquired = True
	migrate_to_metadata(str(media.pk))
	if not new_media:
	# update the existing media
	for key, value in media_defaults.items():
	setattr(media, key, value)
	log.debug(f'Indexed media: {vn}: {source} / {media}')
	else:
	# log the new media instances
	log.info(f'Indexed new media: {source} / {media}')
	log.info(f'Scheduling tasks to download thumbnail for: {media.key}')
	thumbnail_fmt = 'https://i.ytimg.com/vi/{}/{}default.jpg'
	for num, prefix in enumerate(reversed(('hq', 'sd', 'maxres',))):
	thumbnail_url = thumbnail_fmt.format(
	media.key,
	prefix,
	)
	download_media_image.schedule(
	(str(media.pk), thumbnail_url,),
	priority=10+(5*num),
	delay=65-(30*num),
	)
	priority = download_media_metadata.settings.get('default_priority', 50)
	if source.download_media:
	priority += 5
	else:
	priority -= 5
	log.info(f'Scheduling task to download metadata for: {media.url}')
	TaskHistory.schedule(
	download_media_metadata,
	str(media.pk),
	priority=priority,
	remove_duplicates=True,
	vn_fmt=_('Downloading metadata for: "{}": {}'),
	vn_args=(media.key, media.name,),
	)
	# Reset task.verbose_name to the saved value
	update_task_status(task, None)
	# Update any remaining items in the batches
	save_db_batch(Metadata.objects, db_batch_data, db_fields_data)
	save_db_batch(Media.objects, db_batch_media, db_fields_media)
	# Cleanup of media no longer available from the source
	cleanup_removed_media(str(source.pk), video_keys)
	# Clear references to indexed data
	videos = video = None
	db_batch_data.clear()
	db_batch_media.clear()
	# Let the checking task run
	indexing_lock.acquired = False
	# Create the checking task
	TaskHistory.schedule(
	save_all_media_for_source,
	str(source.pk),
	remove_duplicates=True,
	vn_fmt = _('Checking all media for "{}"'),
	vn_args=(
	source.name,
	),
	)
	return True

tubesync/tubesync/sync/models/source.py

Lines 559 to 603 in 7ee8c96

	def get_index(self, url_type, /):
	indexer = self.INDEXERS.get(self.source_type, None)
	if not callable(indexer):
	raise Exception(f'Source type f"{self.source_type}" has no indexer')
	days = None
	if self.download_cap_date:
	days = timezone.timedelta(seconds=self.download_cap).days
	entries = list()
	try:
	response = indexer(self.get_index_url(url_type), days=days)
	except DownloadError as e:
	if str(e).endswith(f': This channel does not have a {url_type} tab'):
	return entries
	raise
	else:
	if not isinstance(response, dict):
	return entries
	entries = response.get('entries', list())
	return entries
	def index_media(self):
	'''
	Index the media source returning a queue of media metadata as dicts.
	'''
	entries = queue(list(), getattr(settings, 'MAX_ENTRIES_PROCESSING', 0) or None)
	if self.index_videos:
	videos = self.get_index('videos')
	entries.extend(reversed(videos))
	# Playlists do something different that I have yet to figure out
	if not self.is_playlist:
	if self.index_streams:
	streams = self.get_index('streams')
	if entries.maxlen is None or 0 == len(entries):
	entries.extend(reversed(streams))
	else:
	# share the queue between streams and videos
	allowed_streams = max(
	entries.maxlen // 2,
	entries.maxlen - len(entries),
	)
	entries.extend(reversed(streams[: allowed_streams]))
	return entries

tubesync/tubesync/sync/youtube.py

Lines 171 to 279 in 7ee8c96

	def get_media_info(url, /, *, days=None, info_json=None):
	'''
	Extracts information from a YouTube URL and returns it as a dict. For a channel
	or playlist this returns a dict of all the videos on the channel or playlist
	as well as associated metadata.
	'''
	start = None
	if days is not None:
	try:
	days = int(str(days), 10)
	except (TypeError, ValueError):
	days = None
	start = (
	f'yesterday-{days!s}days' if days else None
	)
	opts = get_yt_opts()
	default_opts = yt_dlp.parse_options([]).options
	class NoDefaultValue: pass # a unique Singleton, that may be checked for later
	user_set = lambda k, d, default=NoDefaultValue: d[k] if k in d.keys() else default
	default_paths = user_set('paths', default_opts.__dict__, dict())
	paths = user_set('paths', opts, default_paths)
	if 'temp' in paths:
	temp_dir_obj = TemporaryDirectory(prefix='.yt_dlp-', dir=paths['temp'])
	temp_dir_path = Path(temp_dir_obj.name)
	(temp_dir_path / '.ignore').touch(exist_ok=True)
	paths.update({
	'temp': str(temp_dir_path),
	})
	try:
	info_json_path = Path(info_json).resolve(strict=False)
	except (RuntimeError, TypeError):
	pass
	else:
	paths.update({
	'infojson': user_set('infojson', paths, str(info_json_path))
	})
	default_ea = user_set('extractor_args', default_opts.__dict__, dict())
	extractor_args = user_set('extractor_args', opts, default_ea)
	ea_ytt_dict = extractor_args.get('youtubetab', dict())
	ea_ytt_dict['approximate_date'] = ['true']
	extractor_args['youtubetab'] = ea_ytt_dict
	default_postprocessors = user_set('postprocessors', default_opts.__dict__, list())
	postprocessors = user_set('postprocessors', opts, default_postprocessors)
	postprocessors.append(dict(
	key='Exec',
	when='playlist',
	exec_cmd="/usr/bin/env python3 /app/manage.py full-playlist %(id)q '%(playlist_count)d'",
	))
	cache_directory_path = Path(user_set('cachedir', opts, '/dev/shm'))
	playlist_infojson = 'postprocessor_[%(id)s]_%(n_entries)d_%(playlist_count)d_temp'
	outtmpl = dict(
	default='',
	infojson='%(extractor_key)s/%(id)s.%(ext)s' if paths.get('infojson') else '',
	pl_infojson=f'{cache_directory_path}/infojson/playlist/{playlist_infojson}.%(ext)s',
	)
	for k in OUTTMPL_TYPES.keys():
	outtmpl.setdefault(k, '')
	sleep_interval_requests = getattr(settings, 'YOUTUBE_INFO_SLEEP_REQUESTS', 1)
	opts.update({
	'ignoreerrors': False, # explicitly set this to catch exceptions
	'ignore_no_formats_error': False, # we must fail first to try again with this enabled
	'skip_download': True,
	'simulate': False,
	'logger': log,
	'extract_flat': True,
	'allow_playlist_files': True,
	'check_formats': True,
	'check_thumbnails': False,
	'clean_infojson': False,
	'daterange': yt_dlp.utils.DateRange(start=start),
	'extractor_args': extractor_args,
	'outtmpl': outtmpl,
	'overwrites': True,
	'paths': paths,
	'postprocessors': postprocessors,
	'skip_unavailable_fragments': False,
	'sleep_interval_requests': sleep_interval_requests,
	'verbose': True if settings.DEBUG else False,
	'writeinfojson': True,
	})
	if start:
	log.debug(f'get_media_info: used date range: {opts["daterange"]} for URL: {url}')
	response = {}
	with yt_dlp.YoutubeDL(opts) as y:
	try:
	response = y.extract_info(url, download=False)
	except yt_dlp.utils.DownloadError as e:
	if not _subscriber_only(msg=e.msg):
	raise YouTubeError(f'Failed to extract_info for "{url}": {e}') from e
	# adjust options and try again
	opts.update({'ignore_no_formats_error': True,})
	with yt_dlp.YoutubeDL(opts) as yy:
	try:
	response = yy.extract_info(url, download=False)
	except yt_dlp.utils.DownloadError as ee:
	raise YouTubeError(f'Failed (again) to extract_info for "{url}": {ee}') from ee
	# validate the response is what we expected
	if not _subscriber_only(response=response):
	response = {}
	if not response:
	raise YouTubeError(f'Failed to extract_info for "{url}": No metadata was '
	f'returned by youtube-dl, check for error messages in the '
	f'logs above. This task will be retried later with an '
	f'exponential backoff.')
	return response