diff --git a/modules/videos/__init__.py b/modules/videos/__init__.py index 89d9192e..4a376848 100644 --- a/modules/videos/__init__.py +++ b/modules/videos/__init__.py @@ -7,7 +7,7 @@ from sqlalchemy import or_ from sqlalchemy.orm import Session from modules.videos.common import get_or_create_ffprobe_json -from modules.videos.models import Video +from modules.videos.models import Video, AUDIO_PLAYLIST_MIMETYPES from wrolpi.common import logger, limit_concurrent, register_modeler, register_refresh_cleanup from wrolpi.db import get_db_curs, get_db_session from wrolpi.files.models import FileGroup @@ -21,16 +21,48 @@ __all__ = ['video_modeler'] VIDEO_PROCESSING_LIMIT = 20 +def _model_video(session: Session, file_group: FileGroup, video: Video | None, probed: dict) -> Video: + """Create/update and validate the Video row for one FileGroup.""" + if not video: + video = Video(file_group=file_group, file_group_id=file_group.id) + session.add(video) + session.flush([video]) + if not Session.object_session(video): + session.add(video) + video.flush() + # Store the ffprobe data gathered outside the write transaction. + if (result := probed.get(file_group.id)) is not None: + video.ffprobe_json, ffprobe_file = result + if ffprobe_file: + # Track the .ffprobe.json cache file that was just written. + file_group.append_files(ffprobe_file) + video.flush(session) + # Validate and index subtitles. (Poster generation happens here when a Channel + # asks for it; it is the remaining slow work inside this transaction.) + video.validate(session) + return video + + @register_modeler async def video_modeler(progress_callback: Callable[[int], None] = None): total_processed = 0 + # Ids that failed to model this run. A failure leaves no Video row, so the `Video.id IS + # NULL` gate below would re-select it forever; failures are retried on the next refresh. + failed_ids: set = set() while True: # Read the batch; nothing is claimed yet, so the write lock stays free while ffprobe runs. with get_db_session() as session: - batch: List[Tuple[int, pathlib.Path]] = list(session.query(FileGroup.id, FileGroup.primary_path).filter( - FileGroup.indexed != True, + query = session.query(FileGroup.id, FileGroup.primary_path) \ + .outerjoin(Video, Video.file_group_id == FileGroup.id) \ + .filter( or_(FileGroup.mimetype.like('video/%'), FileGroup.mimetype.like('audio/%')), - ).limit(VIDEO_PROCESSING_LIMIT).all()) + FileGroup.mimetype.notin_(AUDIO_PLAYLIST_MIMETYPES), + # Also model indexed FileGroups that never got a Video row. + or_(Video.id.is_(None), FileGroup.indexed != True), + ) + if failed_ids: + query = query.filter(FileGroup.id.notin_(failed_ids)) + batch: List[Tuple[int, pathlib.Path]] = list(query.limit(VIDEO_PROCESSING_LIMIT).all()) if not batch: break @@ -56,31 +88,14 @@ async def video_modeler(progress_callback: Callable[[int], None] = None): .outerjoin(Video, Video.file_group_id == FileGroup.id)) for file_group, video in file_groups: - video_id = None try: - if not video: - video = Video(file_group=file_group, file_group_id=file_group.id) - session.add(video) - session.flush([video]) - video_id = video.id - if not Session.object_session(video): - session.add(video) - video.flush() - # Store the ffprobe data gathered above. - if (result := probed.get(file_group.id)) is not None: - video.ffprobe_json, ffprobe_file = result - if ffprobe_file: - # Track the .ffprobe.json cache file that was just written. - file_group.append_files(ffprobe_file) - video.flush(session) - # Validate and index subtitles. (Poster generation happens here when a Channel - # asks for it; it is the remaining slow work inside this transaction.) - video.validate(session) + _model_video(session, file_group, video, probed) except Exception as e: + # Before the re-raise, so no failure path can re-select this id forever. + failed_ids.add(file_group.id) if PYTEST: raise - i = video.file_group.primary_path if video.file_group else video_id - logger.error(f'Unable to model Video: {str(i)}', exc_info=e) + logger.error(f'Unable to model Video: {file_group.primary_path}', exc_info=e) file_group.indexed = True @@ -162,15 +177,16 @@ _UNCLAIMED_VIDEOS_SQL = ''' @limit_concurrent(1) def video_cleanup(): logger.info('Claiming Videos for their Channels') - # Read the FileGroups that are no longer video/audio, then unmodel them in short transactions. + # Read the FileGroups that are no longer video/audio (audio playlists never were), + # then unmodel them in short transactions. with get_db_curs() as curs: curs.execute(''' SELECT id FROM file_group WHERE model = 'video' - AND mimetype NOT LIKE 'video/%' - AND mimetype NOT LIKE 'audio/%' - ''') + AND ((mimetype NOT LIKE 'video/%' AND mimetype NOT LIKE 'audio/%') + OR mimetype IN (SELECT value FROM json_each(:playlists))) + ''', dict(playlists=json.dumps(AUDIO_PLAYLIST_MIMETYPES))) stale_ids = [i['id'] for i in curs.fetchall()] for chunk in _chunks(stale_ids): ids = json.dumps(chunk) diff --git a/modules/videos/models.py b/modules/videos/models.py index 90221542..3320f401 100644 --- a/modules/videos/models.py +++ b/modules/videos/models.py @@ -20,7 +20,12 @@ from wrolpi.vars import PYTEST, VIDEO_INFO_JSON_KEYS_TO_CLEAN logger = logger.getChild(__name__) -__all__ = ['Video', 'Channel'] +__all__ = ['Video', 'Channel', 'AUDIO_PLAYLIST_MIMETYPES'] + +# Playlists that detect as audio/* but are text lists of URLs, not media. Detection varies by +# libmagic version (.pls is text/plain on some, audio/x-scpls on others); the denylist covers +# whatever comes back audio/*. +AUDIO_PLAYLIST_MIMETYPES = ('audio/x-mpegurl', 'audio/mpegurl', 'audio/x-scpls', 'audio/scpls') class Video(ModelHelper, Base): @@ -100,7 +105,11 @@ class Video(ModelHelper, Base): @staticmethod def can_model(file_group: FileGroup) -> bool: - return file_group.mimetype.startswith('video/') + # Audio in a Channel is Channel content; playlists are not media. Must agree with the + # batch video_modeler's query. + if file_group.mimetype in AUDIO_PLAYLIST_MIMETYPES: + return False + return file_group.mimetype.startswith('video/') or file_group.mimetype.startswith('audio/') @staticmethod def do_model(session: Session, file_group: FileGroup) -> 'Video': diff --git a/modules/videos/test/test_audio_modeling.py b/modules/videos/test/test_audio_modeling.py new file mode 100644 index 00000000..443ba53e --- /dev/null +++ b/modules/videos/test/test_audio_modeling.py @@ -0,0 +1,148 @@ +"""Uploaded audio files must become Videos, and audio playlists must not. + +The upload flow (`upsert_file` -> `FileGroup.do_model` -> `Video.can_model`) and the refresh +flow (the batch `video_modeler`) must agree on what a Video is: video/* and audio/*, except +audio playlists. A FileGroup either path leaves behind must be picked up by the other. +""" +import asyncio +import shutil +from unittest.mock import patch + +import pytest + +from modules.videos import video_modeler, video_cleanup, VIDEO_PROCESSING_LIMIT +from modules.videos.models import Video, AUDIO_PLAYLIST_MIMETYPES +from wrolpi.files.lib import upsert_file +from wrolpi.files.models import FileGroup +from wrolpi.vars import PROJECT_DIR + + +@pytest.mark.asyncio +async def test_uploaded_audio_becomes_a_channel_video(async_client, test_session, channel_factory): + """An audio file uploaded into a Channel's directory is a Video of that Channel.""" + channel = channel_factory(name='The Tech Prepper') + audio_path = channel.directory / 'interview.mp3' + shutil.copy(PROJECT_DIR / 'test/big_buck_bunny.mp3', audio_path) + + await upsert_file(audio_path) + + file_group = test_session.query(FileGroup).filter_by(primary_path=str(audio_path)).one() + assert file_group.mimetype.startswith('audio/'), 'the file under test must be audio' + assert file_group.model == 'video', 'uploaded audio must be modeled as a Video' + video = test_session.query(Video).filter_by(file_group_id=file_group.id).one() + assert video.channel_id == channel.id, 'the Video must belong to the Channel it was uploaded into' + + +@pytest.mark.asyncio +async def test_audio_playlist_is_not_a_video(async_client, test_session, test_directory): + """A playlist with an audio/* mimetype is a text list of URLs, not media. + + Neither the upload flow (do_model) nor the batch modeler may turn one into a Video. The + mimetype is forced rather than detected: libmagic versions disagree (.m3u is + audio/x-mpegurl on some, text/plain on others), and the denylist must hold wherever an + audio/* playlist type comes back. + """ + playlist = test_directory / 'stations.m3u' + playlist.write_text('#EXTM3U\n#EXTINF:0,WRMI\nhttp://example.org/stream\n') + file_group = FileGroup.from_paths(test_session, playlist) + file_group.mimetype = 'audio/x-mpegurl' + test_session.commit() + + # The upload path's decision point. + file_group.do_model(test_session) + test_session.commit() + assert file_group.model is None, 'a playlist must not be modeled by an upload' + + # The refresh path: the modeler selects any audio/* FileGroup without a Video row, + # so only the playlist denylist keeps this one out. + await video_modeler() + assert test_session.query(Video).count() == 0, 'a playlist must not create a Video' + + +def test_can_model_rejects_every_playlist_mimetype(test_session, test_directory): + """Every mimetype on the playlist denylist is refused, whatever libmagic detected it from. + + Direct assignment rather than detection: libmagic versions disagree on .pls (text/plain + here, audio/x-scpls elsewhere), and the denylist must hold wherever audio/* comes back. + """ + audio = test_directory / 'stations.pls' + audio.write_text('[playlist]\nFile1=http://example.org/stream\nNumberOfEntries=1\n') + file_group = FileGroup.from_paths(test_session, audio) + for mimetype in AUDIO_PLAYLIST_MIMETYPES: + file_group.mimetype = mimetype + assert not Video.can_model(file_group), f'{mimetype} must not model as a Video' + file_group.mimetype = 'audio/flac' + assert Video.can_model(file_group), 'non-playlist audio must model as a Video' + + +@pytest.mark.asyncio +async def test_video_modeler_heals_indexed_filegroup_without_video(async_client, test_session, + channel_factory): + """An audio/video FileGroup that is already indexed but has no Video row is still modeled.""" + channel = channel_factory(name='The Tech Prepper') + audio_path = channel.directory / 'stuck.mp3' + shutil.copy(PROJECT_DIR / 'test/big_buck_bunny.mp3', audio_path) + + # The stuck production state: indexed, no model, no Video row. + file_group = FileGroup.from_paths(test_session, audio_path) + file_group.indexed = True + file_group.model = None + test_session.commit() + + await video_modeler() + + test_session.expire_all() + video = test_session.query(Video).filter_by(file_group_id=file_group.id).one() + assert file_group.model == 'video' + assert video.channel_id == channel.id, 'the healed Video must be claimed by its Channel' + + +@pytest.mark.asyncio +async def test_video_cleanup_unmodels_playlists(async_client, test_session, test_directory): + """A playlist wrongly modeled as a Video is unmodeled by cleanup, for every denylisted type.""" + file_groups = [] + for index, mimetype in enumerate(AUDIO_PLAYLIST_MIMETYPES): + playlist = test_directory / f'stations_{index}.m3u' + playlist.write_text('#EXTM3U\n#EXTINF:0,WRMI\nhttp://example.org/stream\n') + file_group = FileGroup.from_paths(test_session, playlist) + # The wrong historical state: a Video row and model='video' on a playlist. + file_group.mimetype = mimetype + file_group.model = 'video' + file_group.indexed = True + test_session.add(Video(file_group=file_group, file_group_id=file_group.id)) + file_groups.append(file_group) + test_session.commit() + + video_cleanup() + + test_session.expire_all() + assert test_session.query(Video).count() == 0, 'every playlist Video must be deleted' + assert all(fg.model is None for fg in file_groups), 'every playlist must be unmodeled' + + +@pytest.mark.asyncio +async def test_video_modeler_terminates_on_persistent_failure(async_client, test_session, test_directory): + """video_modeler must terminate even when modeling keeps failing before a Video row exists. + + The `Video.id IS NULL` gate keeps matching a file that fails to model; only the per-run + `failed_ids` guard stops a full batch of such files from being re-selected forever. + """ + audio_dir = test_directory / 'audio' + audio_dir.mkdir(parents=True) + num_files = VIDEO_PROCESSING_LIMIT + 5 + for i in range(num_files): + path = audio_dir / f'stuck_{i:03d}.mp3' + shutil.copy(PROJECT_DIR / 'test/big_buck_bunny.mp3', path) + for path in sorted(audio_dir.iterdir()): + fg = FileGroup.from_paths(test_session, path) + fg.indexed = True # Already indexed, no Video row: only the new gate selects these. + test_session.commit() + + # _model_video always raises without persisting a Video row. PYTEST re-raises inside the + # loop's per-item handler; disable that to exercise the failure path and the termination + # guard, not the re-raise. + with patch('modules.videos._model_video', side_effect=RuntimeError('boom')), \ + patch('modules.videos.PYTEST', False): + await asyncio.wait_for(video_modeler(), timeout=10) + + assert test_session.query(Video).count() == 0, 'every attempt failed, yet the loop terminated'