import pathlib from pathlib import Path from typing import List, Optional, Union from sqlalchemy import or_, func, desc from sqlalchemy.orm import Query, Session from wrolpi import flags from wrolpi.collections import Collection from wrolpi.common import logger, \ get_media_directory, wrol_mode_check, background_task from wrolpi.db import get_db_session, session_curs from wrolpi.downloader import save_downloads_config, download_manager, Download from wrolpi.errors import APIError, ValidationError, FileWorkerConflict from wrolpi.events import Events from wrolpi.tags import Tag from .. import schema from ..common import check_for_channel_conflicts from ..errors import UnknownChannel from ..lib import save_channels_config, get_channels_config from ..models import Channel, Video logger = logger.getChild(__name__) def check_channels_config_imported(): """Raise ConfigNotImported until channels.yaml is imported; the import would undo a change made before it.""" # A test may have no channels config. if config := get_channels_config(): config.check_imported() async def get_minimal_channels(session: Session) -> List[dict]: """ Get the minimum amount of information necessary about all channels. """ with session_curs(session) as curs: # Get all channels, even if they don't have videos. Also get the minimum frequency download because this is the # one that will consume the most resources. stmt = ''' SELECT c.id, col.name AS "name", col.directory, c.url, t.name AS "tag_name", c.video_count, c.total_size, c.minimum_frequency FROM channel AS c INNER JOIN collection col ON col.id = c.collection_id LEFT JOIN tag t ON t.id = col.tag_id ''' curs.execute(stmt) logger.debug(stmt) channels = sorted([dict(i) for i in curs.fetchall()], key=lambda i: i['name'].lower()) return channels COD = Union[dict, Channel] def get_channel(session: Session, *, channel_id: int = None, source_id: str = None, url: str = None, directory: str = None, name: str = None, return_dict: bool = True) -> COD: """ Attempt to find a Channel using the provided params. The params are in order of reliability, except that directory beats source_id/url: a Video's files must live in their Channel's directory, so the directory a file is in reflects the user's intent and beats yt-dlp metadata. Raises UnknownChannel if no channel is found. """ channel: COD = None # noqa # Try to find the channel by the most reliable methods first. # Load what `Channel.dict()` reads: the Collection (name/directory/tag) and its Downloads. Overwrite any # copy already in the Session so the Channel reflects the database. query = session.query(Channel).options(*Channel.json_options()).populate_existing() if channel_id: channel = query.filter_by(id=channel_id).one_or_none() if not channel and directory: directory = Path(directory) if not directory.is_absolute(): directory = get_media_directory() / directory directory = directory.absolute() # Walk up the directory tree to find a matching channel # This allows finding channels for files in subdirectories (e.g., year subdirs) media_dir = get_media_directory() current = directory while current != media_dir and current != current.parent: channel = query.join(Collection).filter(Collection.directory == str(current)).one_or_none() if channel: break current = current.parent if not channel and source_id: channel = query.filter_by(source_id=source_id).one_or_none() if not channel and url: channel = query.filter_by(url=url).one_or_none() if not channel and url: # Try to extract YouTube channel ID from the URL and match by source_id. from modules.videos.common import get_youtube_channel_id yt_channel_id = get_youtube_channel_id(url) if yt_channel_id: channel = query.filter_by(source_id=yt_channel_id).one_or_none() if not channel and name: channel = query.join(Collection).filter(Collection.name == name).one_or_none() if not channel: raise UnknownChannel(f'No channel matches {channel_id=} {source_id=} {url=} {directory=}') logger.debug(f'Found {channel=} using {channel_id=} {source_id=} {url=} {directory=}') if return_dict: statistics = channel.get_statistics() channel = channel.dict() channel['statistics'] = statistics return channel async def update_channel(session: Session, *, data: schema.ChannelPutRequest, channel_id: int) -> Channel: """Update a Channel's DB record""" check_channels_config_imported() channel = Channel.find_by_id(session, channel_id) # Verify that the URL/Name/directory aren't taken check_for_channel_conflicts( session, id_=channel.id, url=data.url, name=data.name, directory=data.directory, ) data = data.__dict__ old_directory = pathlib.Path(channel.directory) new_directory = get_media_directory() / data.pop('directory') # Apply the changes now that we've OK'd them channel.update(data) session.commit() if new_directory != old_directory: if not new_directory.is_dir(): new_directory.mkdir(parents=True) # Always use background task with its own session to avoid # DetachedInstanceError when the original session is closed. background_task(_background_move_channel(channel.id, new_directory)) save_channels_config.activate_switch() return channel def create_channel(session: Session, data: schema.ChannelPostRequest, return_dict: bool = True) -> Union[Channel, dict]: """ Create a new Channel. Check for conflicts with existing Channels. """ check_channels_config_imported() try: # Verify that the URL/Name/directory aren't taken check_for_channel_conflicts( session, url=data.url or None, name=data.name, directory=data.directory, source_id=data.source_id, ) except APIError as e: raise ValidationError() from e directory = pathlib.Path(data.directory) directory = get_media_directory() / directory if not directory.is_absolute() else directory directory.mkdir(parents=True, exist_ok=True) # Create Collection first collection = Collection( name=data.name, kind='channel', directory=directory, downloads=[], # A new Collection has none; nothing to load. ) session.add(collection) session.flush([collection]) # Create Channel linked to Collection channel = Channel(collection=collection) session.add(channel) session.flush([channel]) # Apply the changes now that we've OK'd them channel.update(data.__dict__) session.commit() save_channels_config.activate_switch() return channel.dict() if return_dict else channel def delete_channel(session: Session, *, channel_id: int): check_channels_config_imported() channel = get_channel(session, channel_id=channel_id, return_dict=False) channel_dict = channel.dict() channel.delete_with_videos() save_channels_config.activate_switch() return channel_dict async def search_channels_by_name(session: Session, name: str, limit: int = 5, order_by_video_count: bool = False) -> List[Channel]: name = name or '' name_no_spaces = ''.join(name.split(' ')) # SQLite sorts by byte value; emulate the old locale-aware (en_US.UTF-8) ordering: compare # names case-insensitively and without spaces, ties broken with lowercase names first. name_order = (func.lower(func.replace(Collection.name, ' ', '')).asc(), Collection.name.desc()) if order_by_video_count: stmt = session.query(Channel, func.count(Video.id).label('video_count')) \ .options(*Channel.json_options()) \ .join(Collection) \ .filter(or_( Collection.name.ilike(f'%{name}%'), Collection.name.ilike(f'%{name_no_spaces}%'), )) \ .outerjoin(Video, Video.channel_id == Channel.id) \ .group_by(Channel.id, Collection.id, Collection.name) \ .order_by(desc('video_count'), *name_order) \ .limit(limit) channels = [i[0] for i in stmt] else: stmt = session.query(Channel) \ .options(*Channel.json_options()) \ .join(Collection) \ .filter(or_( Collection.name.ilike(f'%{name}%'), Collection.name.ilike(f'%{name_no_spaces}%'), )) \ .order_by(*name_order) \ .limit(limit) channels = stmt.all() return channels @wrol_mode_check async def create_channel_download(channel_id: int, url: str, frequency: int, settings: dict): """Create Download for Channel.""" with get_db_session(commit=True) as session: channel = Channel.find_by_id(session, channel_id) download = channel.get_or_create_download(session, url, frequency, reset_attempts=True) download.settings = settings save_channels_config.activate_switch() save_downloads_config.activate_switch() return download @wrol_mode_check async def update_channel_download(channel_id: int, download_id: int, url: str, frequency: int, settings: dict): """Fetch Channel's Download, update its properties.""" with get_db_session(commit=True) as session: # Ensure that the Channel exists. Channel.find_by_id(session, channel_id) download = Download.find_by_id(session, download_id) download.url = url download.frequency = frequency download.settings = settings session.commit() download_manager.remove_from_skip_list(url) save_channels_config.activate_switch() save_downloads_config.activate_switch() return download async def _background_move_channel(channel_id: int, target_directory: pathlib.Path): """Helper to run channel move in background with its own database session. This allows the move to continue even if the original HTTP request is cancelled (e.g., user closes browser tab), and avoids DetachedInstanceError by creating a fresh session for the background task. """ try: with get_db_session(commit=True) as session: channel = Channel.find_by_id(session, channel_id) if not channel: logger.error(f'_background_move_channel: channel {channel_id} not found') Events.send_file_move_failed(f'Channel move failed: channel not found') return await channel.move_channel(target_directory, session, send_events=True) except Exception as e: logger.error(f'_background_move_channel: failed for channel_id={channel_id}', exc_info=e) Events.send_file_move_failed(f'Channel move failed: {e}') @wrol_mode_check async def tag_channel(session: Session, tag_name: str | None, directory: pathlib.Path | None, channel_id: int): """Add a Tag to a Channel, or remove a Tag from a Channel if no `tag_name` is provided. Move the Channel to the new directory, if provided.""" check_channels_config_imported() if directory and flags.file_worker_busy.is_set(): raise FileWorkerConflict('Refusing to move channel while FileWorker is busy') channel = Channel.find_by_id(session, channel_id) # May also clear the tag if `tag_name` is None. channel.set_tag(tag_name) channel.flush() session.commit() # Only move Channel when requested. if directory: # Move to newly defined directory only if necessary. directory.mkdir(parents=True, exist_ok=True) if channel.directory != directory: # Always use background task with its own session to avoid # DetachedInstanceError when the original session is closed. background_task(_background_move_channel(channel.id, directory)) else: save_channels_config.activate_switch() else: logger.info(f'Tagging {channel} with {tag_name}') # Always save config when tag changes, even without directory save_channels_config.activate_switch() def search_channels_query(session: Session, tag_names: List[str] = None, search_str: str = None) -> Optional[Query]: """Channels whose name contains `search_str` and which have any of the Tags. None when there is nothing to search by.""" search_str = (search_str or '').strip() if not tag_names and not search_str: return None query = session.query(Channel).join(Collection, Collection.id == Channel.collection_id) if search_str: name_no_spaces = ''.join(search_str.split(' ')) query = query.filter(or_( Collection.name.ilike(f'%{search_str}%'), Collection.name.ilike(f'%{name_no_spaces}%'), )) if tag_names: query = query.join(Tag, Tag.id == Collection.tag_id).filter(Tag.name.in_(tag_names)) return query async def search_channels(session: Session, tag_names: List[str] = None, search_str: str = None) -> List[Channel]: """Search Channels by their name and/or Tags.""" query = search_channels_query(session, tag_names, search_str) if query is None: return [] return query.options(*Channel.json_options()).order_by(func.lower(Collection.name)).all()