rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 10:51:36 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

channels.yaml imports by diffing (#872)

Browse source 
* channels.yaml imports by diffing

channels.yaml declares two sections: Channels keyed by directory
(comparing name, URL, Tag, file format, and the source id and flags,
which keep the database's value when missing), and the Downloads listed
under each Channel, keyed by (directory, URL); an existing Download keeps
its own frequency.  An unchanged import writes nothing (it ran about a
thousand queries and committed several times).

Removed Channels are deleted with their Collection (and Downloads);
a Download removed from a Channel is deleted.  A channel Collection
without a Channel is deleted unless listed (then it gets its Channel).
A Channel whose directory changed, with the same name, is moved.  An
invalid entry is skipped and counted, and keeps its Channel (by name when
its directory is invalid) and that Channel's Downloads; so does an entry
without a list of Downloads.  Duplicate directories still refuse the
import.  Unknown Tags are dropped once tags.yaml is imported.

New Downloads are created in the import's one write (commit=False on
Channel.get_or_create_download, Download.delete and delete_with_videos).
After the import, Videos are claimed and source ids fetched only for new
and changed Channels; Downloads are linked without a query per Download,
and not while entries are skipped.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* channels.yaml: moves, invalid Downloads, and saves after the write

- A moved Channel's Download diff is retargeted to its new directory:
  one its new entry lists is the same Download, one it does not is
  deleted in the same import (the old directory was treated as kept).
- An unreadable Download is counted as skipped, and its Channel keeps
  its Downloads (it was dropped, and the Channel's Download deleted).
- A new Download without a frequency is dropped before the write and
  counted (it re-opened a write on every import).
- No source id is fetched while entries are skipped (its save would be
  refused, leaving the database newer than the file); a fetched one is
  saved with dump_config, which records the database's version.
- Download.delete(commit=False) neither changes the skip list nor starts
  config saves, which would read the database before the import commits
  (and write a removed Download back into channels.yaml).  The import
  saves download_manager.yaml after its write.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
lrnselfreliance 2026-10-25 13:09:51 -06:00 • committed by Roland
parent 57c2a29022
commit 315b60e9eb
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
4 changed files with 631 additions and 225 deletions
Download patch file Download diff file Expand all files Collapse all files

546
modules/videos/lib.py
View file

@ -6,11 +6,11 @@ import pathlib
import re
from dataclasses import dataclass, field
from datetime import datetime
from typing import Generator, Type, List
from typing import Dict, Generator, Type, List, Set
from typing import Tuple, Optional
import pytz
from sqlalchemy.orm import Session
from sqlalchemy.orm import Session, joinedload
from yt_dlp import YoutubeDL
from modules.videos.models import Video
@ -18,14 +18,18 @@ 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.config_version import IMPORT, SAVE, register_ownership
from wrolpi.collections import Collection
from wrolpi.collections.models import validate_collection_directory
from wrolpi.config_import import ConfigSection, 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
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, \
@ -257,6 +261,189 @@ class ChannelDictValidator:
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; never a directory. (The entry's name keeps its Channel, see `prepare`.)
invalid.add(f'\0{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
for position, download in enumerate(data.get('downloads') or []):
if isinstance(download, dict) and download.get('url'):
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
@ -265,6 +452,7 @@ class ChannelsConfigValidator:
class ChannelsConfig(ConfigFile):
file_name = 'channels.yaml'
sections = (ChannelsSection(), ChannelDownloadsSection())
default_config = dict(
channels=[dict(
name='WROLPi',
@ -285,172 +473,116 @@ class ChannelsConfig(ConfigFile):
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()
self._config_outdated = False
super().import_config(file, send_events)
# 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
def prepare(self, diffs: dict):
channels, downloads = diffs['channels'], diffs['downloads']
db_rows = self.import_db_rows['channels']
super().import_config(file)
# 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
decision = self.import_decision()
if decision == SAVE:
self.save_database_instead()
return
# 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
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
# 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]}
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!')
# 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}
# 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 decision == IMPORT else []:
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.
# 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:
link_channel_and_downloads(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
self.record_import_version()
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 _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."""
@ -937,7 +1069,9 @@ async def fetch_channel_source_id(channel_id: int):
with get_db_session(commit=True) as session:
Channel.find_by_id(session, channel_id).source_id = source_id
save_channels_config.activate_switch()
# 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)
@ -953,67 +1087,45 @@ def import_channels_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.
def channel_download_links(session: Session) -> Dict[int, int]:
"""The recurring Downloads without a Collection which belong to a Channel: {Download id: Collection id}.
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
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:
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
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}
download = downloads_by_url.get(channel.url)
if download and not download.collection_id:
download.collection_id = channel.collection_id
need_commit = True
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
# 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
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()
# 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:
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()

15
modules/videos/models.py
View file

@ -731,8 +731,9 @@ class Channel(ModelHelper, Base):
raise FileNotFoundError(f'Cannot create Channel info json because directory is not defined: {self}')
def delete_with_videos(self):
"""Delete all Video records (but not video files) related to this Channel. Then delete the Channel and its Collection."""
def delete_with_videos(self, commit: bool = True):
"""Delete all Video records (but not video files) related to this Channel. Then delete the Channel and its
Collection. `commit`: False leaves committing to the caller's write."""
session = Session.object_session(self)
# Disown the videos.
@ -742,7 +743,7 @@ class Channel(ModelHelper, Base):
for download in self.downloads:
download: Download
download.delete()
download.delete(commit=commit)
# Get the Collection before deleting the Channel
collection = self.collection
@ -1139,11 +1140,12 @@ class Channel(ModelHelper, Base):
raise RuntimeError(f'More than one info_json entry matches {video_source_id}')
def get_or_create_download(self, session: Session, url: str, frequency: int,
reset_attempts: bool = False) -> Download:
reset_attempts: bool = False, commit: bool = True) -> Download:
"""Get a Download record, if it does not exist, create it. Create a Download if necessary
which goes into this Channel's directory.
Delegates to Collection.get_or_create_download() with Channel-specific downloaders.
Delegates to Collection.get_or_create_download() with Channel-specific downloaders. `commit`: False leaves
committing a new Download to the caller's write.
"""
from modules.videos.downloader import ChannelDownloader, VideoDownloader
@ -1151,7 +1153,8 @@ class Channel(ModelHelper, Base):
session, url, frequency,
reset_attempts=reset_attempts,
downloader_name=ChannelDownloader.name,
sub_downloader_name=VideoDownloader.name
sub_downloader_name=VideoDownloader.name,
commit=commit,
)
def get_rss_url(self) -> str | None:

286
modules/videos/test/test_channels_config_import.py Normal file
View file

@ -0,0 +1,286 @@
"""channels.yaml imports by diffing its Channels (and their Downloads) against the database, and writing only the
differences (see wrolpi.config_import). A Channel is keyed by its directory."""
import pytest
import yaml
from modules.videos import lib
from modules.videos.lib import get_channels_config, import_channels_config
from modules.videos.models import Channel
from wrolpi.collections import Collection
from wrolpi.common import get_relative_to_media_directory
from wrolpi.config_version import get_config_version
from wrolpi.downloader import Download, DownloadFrequency
from wrolpi.test.test_config_import import edit_config, statements, writes
@pytest.fixture
async def saved_channels(test_session, test_directory, test_channels_config, test_download_manager_config,
channel_factory, tag_factory, await_switches):
"""Channel `one` (tagged, with a weekly Download) and `two`, saved to the config (which records its
version)."""
await tag_factory('news')
channel_factory(name='one', url='https://example.com/one', tag_name='news',
download_frequency=DownloadFrequency.weekly)
channel_factory(name='two', url='https://example.com/two')
get_channels_config().dump_config()
await await_switches()
assert get_config_version(test_session, 'channels.yaml') is not None
return test_channels_config
def channels(test_session) -> dict:
test_session.expire_all()
return {i.name: i for i in test_session.query(Channel)}
def edit_channel(path, name, **values):
def change(data):
next(i for i in data['channels'] if i['name'] == name).update(values)
edit_config(path, change)
def directory_of(test_session, name) -> str:
return str(get_relative_to_media_directory(channels(test_session)[name].directory))
@pytest.mark.asyncio
async def test_unchanged_import_writes_nothing(test_session, saved_channels):
"""Importing a config the database already matches opens no write transaction, nor does linking Downloads when
there are none to link."""
config = get_channels_config()
with statements() as collected:
import_channels_config()
assert config.successful_import is True
assert writes(collected) == []
@pytest.mark.asyncio
async def test_import_writes_only_the_differences(test_session, saved_channels):
"""A changed Channel is updated, a removed one is deleted (with its Collection), an added one is created; the
next import writes nothing."""
def change(data):
data['channels'] = [i for i in data['channels'] if i['name'] != 'two']
one = data['channels'][0]
one.update(name='One', url='https://example.com/new-one', generate_posters=False)
data['channels'].append(dict(name='three', directory='videos/three', url='https://example.com/three'))
edit_config(saved_channels, change)
config = get_channels_config()
config.import_config()
assert config.successful_import is True
found = channels(test_session)
assert set(found) == {'One', 'three'}
assert (found['One'].url, found['One'].generate_posters, found['One'].tag_name) == \
('https://example.com/new-one', False, 'news')
assert {i.name for i in test_session.query(Collection).filter_by(kind='channel')} == {'One', 'three'}
with statements() as collected:
config.import_config()
assert writes(collected) == []
@pytest.mark.asyncio
async def test_downloads_follow_the_config(test_session, saved_channels):
"""A Download removed from a Channel is deleted; an added one is created (with the file's frequency) in the
import's one write. An existing Download keeps its own frequency."""
edit_channel(saved_channels, 'one', downloads=[])
edit_channel(saved_channels, 'two', downloads=[dict(url='https://example.com/two/feed', frequency=86400)])
config = get_channels_config()
config.import_config()
test_session.expire_all()
assert {(i.url, i.frequency, i.collection_id) for i in test_session.query(Download)} == {
('https://example.com/two/feed', 86400, channels(test_session)['two'].collection_id)}
with statements() as collected:
config.import_config()
assert writes(collected) == []
@pytest.mark.asyncio
async def test_failed_import_creates_nothing(test_session, saved_channels, monkeypatch):
"""A new Download is created in the import's one write: a failure later in the write leaves nothing (it was
committed on its own)."""
edit_channel(saved_channels, 'two', downloads=[dict(url='https://example.com/two/feed', frequency=86400)])
edit_config(saved_channels, lambda data: data.update(channels=[i for i in data['channels'] if i['name'] != 'one']))
def fail(*args, **kwargs):
raise RuntimeError('Write failed')
monkeypatch.setattr(lib.ChannelsSection, 'delete', fail)
with pytest.raises(RuntimeError):
get_channels_config().import_config()
test_session.expire_all()
assert {i.url for i in test_session.query(Download)} == {'https://example.com/one'}
assert set(channels(test_session)) == {'one', 'two'}
@pytest.mark.asyncio
async def test_moved_channel_is_the_same_channel(test_session, saved_channels):
"""A Channel whose directory changed in the config (its name did not) is moved, not deleted and re-created."""
before = channels(test_session)['two'].id
edit_channel(saved_channels, 'two', directory='videos/moved/two')
get_channels_config().import_config()
two = channels(test_session)['two']
assert (two.id, directory_of(test_session, 'two')) == (before, 'videos/moved/two')
@pytest.mark.asyncio
async def test_invalid_entries_are_kept(test_session, saved_channels):
"""An entry without a name, or with an invalid directory, is skipped (counted), and its Channel is kept."""
one_directory = directory_of(test_session, 'one')
def change(data):
next(i for i in data['channels'] if i['name'] == 'one').pop('name')
next(i for i in data['channels'] if i.get('name') == 'two')['directory'] = '../outside'
edit_config(saved_channels, change)
config = get_channels_config()
config.import_config()
assert config.import_skipped == 2
assert set(channels(test_session)) == {'one', 'two'}
assert directory_of(test_session, 'one') == one_directory
@pytest.mark.asyncio
async def test_orphan_collections(test_session, test_directory, saved_channels):
"""A channel Collection without a Channel is deleted when the config does not list it; one the config lists
gets its Channel."""
test_session.add_all([
Collection(name='orphan', kind='channel', directory=test_directory / 'videos/orphan'),
Collection(name='listed', kind='channel', directory=test_directory / 'videos/listed'),
])
test_session.commit()
# Saved: the config lists only Channels.
get_channels_config().dump_config()
edit_config(saved_channels, lambda data: data['channels'].append(
dict(name='listed', directory='videos/listed', url='https://example.com/listed')))
get_channels_config().import_config()
test_session.expire_all()
assert {i.name for i in test_session.query(Collection).filter_by(kind='channel')} == {'one', 'two', 'listed'}
assert channels(test_session)['listed'].url == 'https://example.com/listed'
@pytest.mark.asyncio
async def test_unknown_tag_with_a_skipped_entry_converges(test_session, saved_channels, test_tags_config,
await_switches):
"""Once tags.yaml is imported, a Channel's Tag that no longer exists is dropped; with a skipped entry (the config
cannot be saved) the next import writes nothing."""
from wrolpi.tags import get_tags_config
get_tags_config().dump_config()
get_tags_config().import_config()
edit_channel(saved_channels, 'two', tag_name='gone')
edit_config(saved_channels, lambda data: data['channels'].append(dict(directory='videos/nameless')))
config = get_channels_config()
config.import_config()
await await_switches()
assert channels(test_session)['two'].tag_name is None
with statements() as collected:
config.import_config()
assert writes(collected) == []
@pytest.mark.asyncio
async def test_no_links_while_entries_are_skipped(test_session, saved_channels, await_switches):
"""Linking changes channels.yaml; while an entry is skipped the config cannot be saved, so Downloads are not
linked (the next startup would save over the skipped entry)."""
test_session.add(Download(url='https://example.com/two', downloader='video_channel', frequency=86400))
test_session.commit()
edit_config(saved_channels, lambda data: data['channels'].append(dict(directory='videos/nameless')))
import_channels_config()
await await_switches()
test_session.expire_all()
assert Download.get_by_url(test_session, 'https://example.com/two').collection_id is None
assert get_config_version(test_session, 'channels.yaml') == yaml.safe_load(saved_channels.read_text())['version']
@pytest.mark.asyncio
async def test_moved_channel_downloads(test_session, saved_channels):
"""A moved Channel keeps the Downloads its new entry lists, and loses the ones it does not, in the same import;
the next import writes nothing."""
edit_channel(saved_channels, 'one', directory='videos/moved/one', downloads=[])
edit_channel(saved_channels, 'two', directory='videos/moved/two',
downloads=[dict(url='https://example.com/two/feed', frequency=86400)])
config = get_channels_config()
config.import_config()
test_session.expire_all()
assert {i.url for i in test_session.query(Download)} == {'https://example.com/two/feed'}
assert directory_of(test_session, 'one') == 'videos/moved/one'
with statements() as collected:
config.import_config()
assert writes(collected) == []
@pytest.mark.asyncio
async def test_invalid_download_keeps_the_channel_downloads(test_session, saved_channels):
"""An unreadable Download of a Channel is counted as skipped, and the Channel's Downloads are kept."""
edit_channel(saved_channels, 'one', downloads=[dict(uri='https://example.com/one')])
config = get_channels_config()
config.import_config()
assert config.import_skipped == 1
test_session.expire_all()
assert {i.url for i in test_session.query(Download)} == {'https://example.com/one'}
@pytest.mark.asyncio
async def test_new_download_without_frequency_is_skipped(test_session, saved_channels):
"""A new Download without a frequency cannot be created: it is skipped (counted), and the next import writes
nothing."""
edit_channel(saved_channels, 'two', downloads=[dict(url='https://example.com/two/feed')])
config = get_channels_config()
with statements() as collected:
config.import_config()
assert config.import_skipped == 1
assert writes(collected) == []
@pytest.mark.asyncio
async def test_no_source_id_fetch_while_entries_are_skipped(test_session, saved_channels, monkeypatch):
"""Fetching a Channel's source id changes channels.yaml; while an entry is skipped it is not fetched."""
from wrolpi import flags
scheduled = []
monkeypatch.setattr(lib, 'background_task', lambda coro: scheduled.append(coro.close()))
monkeypatch.setattr(flags.have_internet, 'is_set', lambda: True)
monkeypatch.setattr(type(lib.download_manager), 'can_download', property(lambda self: True))
edit_config(saved_channels, lambda data: data['channels'].extend([
dict(name='three', directory='videos/three', url='https://example.com/three', download_missing_data=True),
dict(directory='videos/nameless'),
]))
get_channels_config().import_config()
assert set(channels(test_session)) == {'one', 'two', 'three'}
assert scheduled == []
@pytest.mark.asyncio
async def test_import_does_not_save_during_its_write(test_session, saved_channels, test_download_manager_config,
monkeypatch):
"""Deleting a Download inside the import's write neither skips its URL nor starts a config save (which would read
the database before the import commits)."""
from wrolpi.downloader import get_download_manager_config
saves = []
monkeypatch.setattr(lib.save_channels_config, 'activate_switch', lambda *a, **kw: saves.append('channels'))
edit_channel(saved_channels, 'one', downloads=[])
get_channels_config().import_config()
assert saves == []
assert 'https://example.com/one' not in (get_download_manager_config().skip_urls or [])

9
wrolpi/downloader.py
View file

@ -377,9 +377,11 @@ class Download(ModelHelper, Base): # noqa
def add_to_skip_list(self):
download_manager.add_to_skip_list(self.url)
def delete(self, add_to_skip_list: bool = True):
def delete(self, add_to_skip_list: bool = True, commit: bool = True):
"""`commit`: False only flushes; the caller's write commits, and saves the configs afterward (a save now
would read the database before that commit), and does not change the skip list."""
# Do not download this automatically again. This saves the config.
if add_to_skip_list:
if add_to_skip_list and commit:
self.add_to_skip_list()
session = Session.object_session(self)
@ -389,6 +391,9 @@ class Download(ModelHelper, Base): # noqa
collection_kind = collection.kind if collection else None
session.delete(self)
if not commit:
session.flush([self])
return
session.commit()
# Save appropriate config based on collection kind. Channels/domains own their configs;