diff --git a/app/music_metadata.py b/app/music_metadata.py index e36e7a5..a3b1bef 100644 --- a/app/music_metadata.py +++ b/app/music_metadata.py @@ -1,28 +1,15 @@ """Conservative music metadata enrichment for audio downloads. -This module only consumes metadata already supplied by yt-dlp or retained on -MeTube's queued playlist entry. It intentionally performs no external lookup. +This module only consumes fields already supplied by yt-dlp or retained on +MeTube's queued playlist entry. It intentionally performs no external lookup +or site-specific album detection. """ from __future__ import annotations -import os from typing import Any, Optional -from urllib.parse import urlparse -from mutagen import MutagenError -from mutagen.flac import FLAC -from mutagen.id3 import ID3, ID3NoHeaderError, TRCK, TXXX -from mutagen.mp4 import AtomDataType, MP4, MP4FreeForm -from mutagen.oggopus import OggOpus from yt_dlp.postprocessor.common import PostProcessor -from yt_dlp.utils import PostProcessingError - - -_ARTISTS_KEY = '__metube_track_artists' -_TRACK_NUMBER_KEY = '__metube_track_number' -_TRACK_TOTAL_KEY = '__metube_track_total' -_YOUTUBE_MUSIC_HOSTS = frozenset(('music.youtube.com', 'music.youtube-nocookie.com')) def _has_value(value: Any) -> bool: @@ -51,71 +38,25 @@ def _track_position(value: Any) -> tuple[Optional[int], Optional[int]]: return _positive_int(value), None -def _is_youtube_music_url(value: Any) -> bool: - if not isinstance(value, str) or not value.strip(): - return False - try: - return (urlparse(value).hostname or '').lower() in _YOUTUBE_MUSIC_HOSTS - except ValueError: - return False +def _first_positive_int(*values: Any) -> Optional[int]: + return next((number for value in values if (number := _positive_int(value))), None) -def is_confirmed_music_album(source_url: Any, entry: Any) -> bool: - """Recognize only strong YouTube Music album signals.""" - entry = entry if isinstance(entry, dict) else {} - for key in ('playlist_id', 'playlist'): - playlist_id = entry.get(key) - if isinstance(playlist_id, str) and playlist_id.startswith('OLAK5uy_'): - return True - - candidate_urls = ( - source_url, - entry.get('original_url'), - entry.get('webpage_url'), - entry.get('url'), - ) - for value in candidate_urls: - if not _is_youtube_music_url(value): - continue - try: - path = urlparse(value).path.rstrip('/') - except ValueError: - continue - if path.startswith('/browse/MPRE'): - return True - return False - - -def _is_music_audio(info: dict[str, Any], source_url: Any, confirmed_album: bool) -> bool: - if confirmed_album or _is_youtube_music_url(source_url): - return True - if ( - _is_youtube_music_url(info.get('webpage_url')) - or _is_youtube_music_url(info.get('original_url')) - ): - return True - return _has_value(info.get('track')) or ( - _has_value(info.get('album')) and _has_value(info.get('artists')) +def _has_album_signal(info: dict[str, Any], source_entry: dict[str, Any]) -> bool: + """Use only extractor-owned fields to identify album-level metadata.""" + return any( + _has_value(entry.get(key)) + for entry in (info, source_entry) + for key in ('album', 'track_number') ) -def structured_track_artists(info: dict[str, Any]) -> list[str]: - """Return structured track artists without guessing at comma separators.""" - artists = info.get('artists') - if not isinstance(artists, (list, tuple)): - return [] - - result: list[str] = [] - for value in artists: - if not isinstance(value, str): - continue - # YouTube Music uses a *spaced* middle dot between artist credits. - # Do not split unspaced names such as "half\u00b7alive". - for artist in value.split(' \u00b7 '): - artist = artist.strip() - if artist and artist not in result: - result.append(artist) - return result +def _is_music_audio(info: dict[str, Any], source_entry: dict[str, Any]) -> bool: + return _has_album_signal(info, source_entry) or any( + _has_value(entry.get(key)) + for entry in (info, source_entry) + for key in ('track', 'artists') + ) def prefer_square_thumbnail(info: dict[str, Any]) -> None: @@ -143,133 +84,41 @@ def prefer_square_thumbnail(info: dict[str, Any]) -> None: class MusicMetadataPreProcessor(PostProcessor): - """Enrich a fully extracted audio info-dict from its queued source entry.""" + """Enrich extracted audio metadata using extractor-owned album signals.""" - def __init__(self, downloader=None, *, source_url=None, source_entry=None): + def __init__(self, downloader=None, *, source_entry=None): super().__init__(downloader) - self._source_url = source_url self._source_entry = source_entry if isinstance(source_entry, dict) else {} def run(self, info): - confirmed_album = is_confirmed_music_album(self._source_url, self._source_entry) - - number, inline_total = _track_position(info.get('track_number')) - total = inline_total - if confirmed_album: + if _has_album_signal(info, self._source_entry): + number, inline_total = _track_position(info.get('track_number')) + if number is None: + number, source_inline_total = _track_position( + self._source_entry.get('track_number') + ) + inline_total = inline_total or source_inline_total if number is None: number = _positive_int(self._source_entry.get('playlist_index')) - if number is not None: - info['track_number'] = number - total = total or next(( - value for value in ( - _positive_int(info.get('track_count')), - _positive_int(info.get('track_total')), - _positive_int(self._source_entry.get('playlist_count')), - _positive_int(self._source_entry.get('n_entries')), - ) if value is not None - ), None) + + total = inline_total or _first_positive_int( + info.get('track_count'), + info.get('track_total'), + self._source_entry.get('track_count'), + self._source_entry.get('track_total'), + self._source_entry.get('playlist_count'), + self._source_entry.get('n_entries'), + ) + if number is not None: + info['track_number'] = f'{number}/{total}' if total is not None else number + if not _has_value(info.get('album')): - album = self._source_entry.get('playlist_title') + album = self._source_entry.get('album') or self._source_entry.get( + 'playlist_title' + ) if isinstance(album, str) and album.strip(): info['album'] = album.strip() - if number is not None: - info[_TRACK_NUMBER_KEY] = number - if total is not None: - info[_TRACK_TOTAL_KEY] = total - - artists = structured_track_artists(info) - if len(artists) > 1: - info[_ARTISTS_KEY] = artists - - if _is_music_audio(info, self._source_url, confirmed_album): + if _is_music_audio(info, self._source_entry): prefer_square_thumbnail(info) return [], info - - -def _write_mp3(path: str, artists: list[str], number: Optional[int], total: Optional[int]) -> None: - try: - tags = ID3(path) - except ID3NoHeaderError: - tags = ID3() - if artists: - tags.delall('TXXX:Artists') - tags.delall('TXXX:ARTISTS') - tags.add(TXXX(encoding=3, desc='Artists', text=artists)) - if number is not None: - value = f'{number}/{total}' if total is not None else str(number) - tags.setall('TRCK', [TRCK(encoding=3, text=[value])]) - tags.save(path, v2_version=4) - - -def _write_m4a(path: str, artists: list[str], number: Optional[int], total: Optional[int]) -> None: - audio = MP4(path) - if audio.tags is None: - audio.add_tags() - if artists: - # A list under one key is serialized as one multi-value atom. Duplicate - # atoms are intentionally avoided because TagLib reads only the first. - audio.tags['----:com.apple.iTunes:ARTISTS'] = [ - MP4FreeForm(artist.encode('utf-8'), dataformat=AtomDataType.UTF8) - for artist in artists - ] - if number is not None: - audio.tags['trkn'] = [(number, total or 0)] - audio.save() - - -def _write_vorbis(audio, artists: list[str], number: Optional[int], total: Optional[int]) -> None: - if artists: - audio['ARTISTS'] = artists - if number is not None: - audio['TRACKNUMBER'] = [str(number)] - if total is not None: - value = [str(total)] - audio['TRACKTOTAL'] = value - audio['TOTALTRACKS'] = value - audio.save() - - -def write_music_tags( - path: str, - extension: str, - artists: list[str], - number: Optional[int], - total: Optional[int], -) -> None: - """Write only the supplemental tags needed for music library scanners.""" - extension = extension.lower() - if extension == 'mp3': - _write_mp3(path, artists, number, total) - elif extension == 'm4a': - _write_m4a(path, artists, number, total) - elif extension == 'flac': - _write_vorbis(FLAC(path), artists, number, total) - elif extension == 'opus': - _write_vorbis(OggOpus(path), artists, number, total) - - -class MusicMetadataWriterPostProcessor(PostProcessor): - """Write supplemental tags after yt-dlp has moved the completed audio file.""" - - _SUPPORTED_EXTENSIONS = frozenset(('mp3', 'm4a', 'flac', 'opus')) - - def run(self, info): - artists = info.get(_ARTISTS_KEY) - artists = artists if isinstance(artists, list) else [] - number = _positive_int(info.get(_TRACK_NUMBER_KEY)) - total = _positive_int(info.get(_TRACK_TOTAL_KEY)) - if not artists and number is None: - return [], info - - path = info.get('filepath') - extension = str(info.get('ext') or os.path.splitext(str(path))[1][1:]).lower() - if not isinstance(path, str) or extension not in self._SUPPORTED_EXTENSIONS: - return [], info - try: - write_music_tags(path, extension, artists, number, total) - except (MutagenError, OSError, TypeError, ValueError) as error: - raise PostProcessingError( - f'Unable to write supplemental music metadata to "{path}": {error}' - ) from error - return [], info diff --git a/app/tests/test_music_metadata.py b/app/tests/test_music_metadata.py index 512416c..e3493a5 100644 --- a/app/tests/test_music_metadata.py +++ b/app/tests/test_music_metadata.py @@ -1,105 +1,71 @@ -"""Tests for conservative audio metadata enrichment and tag mappings.""" +"""Tests for conservative audio metadata enrichment.""" from __future__ import annotations -from unittest.mock import MagicMock, patch - -from mutagen.id3 import APIC, ID3, TPE1, TPE2, TRCK, TXXX - -from music_metadata import ( - _ARTISTS_KEY, - _TRACK_NUMBER_KEY, - _TRACK_TOTAL_KEY, - MusicMetadataPreProcessor, - MusicMetadataWriterPostProcessor, - _write_m4a, - _write_vorbis, - is_confirmed_music_album, - structured_track_artists, - write_music_tags, -) +from music_metadata import MusicMetadataPreProcessor -def _preprocess(source_url, source_entry, info): - processor = MusicMetadataPreProcessor( - source_url=source_url, - source_entry=source_entry, - ) +def _preprocess(source_entry, info): + processor = MusicMetadataPreProcessor(source_entry=source_entry) _, result = processor.run(info) return result -def test_album_detection_requires_mpre_or_olak_signal(): - assert is_confirmed_music_album( - 'https://music.youtube.com/watch?v=track', - {'playlist': 'OLAK5uy_example'}, - ) - assert is_confirmed_music_album( - 'https://music.youtube.com/browse/MPREb_example', - {}, - ) - assert not is_confirmed_music_album( - 'https://music.youtube.com/playlist?list=PLexample', - {'playlist': 'PLexample'}, - ) - assert not is_confirmed_music_album( - 'https://www.youtube.com/playlist?list=PLexample', - {'playlist': 'PLexample'}, - ) - - def test_album_uses_existing_order_and_total_when_track_number_is_missing(): result = _preprocess( - 'https://music.youtube.com/watch?v=track', { - 'playlist': 'OLAK5uy_example', 'playlist_index': '03', 'playlist_count': 12, 'playlist_title': 'Example Album', }, - {'title': 'Track', 'artists': ['Artist One']}, + {'title': 'Track', 'album': 'Example Album'}, ) - assert result['track_number'] == 3 - assert result[_TRACK_NUMBER_KEY] == 3 - assert result[_TRACK_TOTAL_KEY] == 12 + assert result['track_number'] == '3/12' assert result['album'] == 'Example Album' def test_official_track_number_wins_over_album_order(): result = _preprocess( - 'https://music.youtube.com/watch?v=track', - { - 'playlist': 'OLAK5uy_example', - 'playlist_index': 3, - 'playlist_count': 12, - }, + {'playlist_index': 3, 'playlist_count': 12}, {'track_number': 7, 'album': 'Official Album'}, ) - assert result['track_number'] == 7 - assert result[_TRACK_NUMBER_KEY] == 7 - assert result[_TRACK_TOTAL_KEY] == 12 + assert result['track_number'] == '7/12' assert result['album'] == 'Official Album' def test_inline_official_track_total_is_preserved(): result = _preprocess( - 'https://music.youtube.com/watch?v=track', - {'playlist': 'OLAK5uy_example', 'playlist_count': 12}, + {'playlist_count': 12}, {'track_number': '4/10', 'album': 'Official Album'}, ) assert result['track_number'] == '4/10' - assert result[_TRACK_NUMBER_KEY] == 4 - assert result[_TRACK_TOTAL_KEY] == 10 -def test_youtube_music_playlist_does_not_infer_album_or_track_number(): +def test_source_track_number_and_total_are_retained_from_flat_extraction(): + result = _preprocess( + {'track_number': 2, 'track_count': 9, 'playlist_index': 4}, + {'title': 'Track'}, + ) + + assert result['track_number'] == '2/9' + + +def test_album_title_falls_back_to_source_playlist_title(): + result = _preprocess( + {'playlist_title': 'Example Album'}, + {'track_number': 4}, + ) + + assert result['album'] == 'Example Album' + assert result['track_number'] == 4 + + +def test_playlist_without_extractor_album_signals_is_not_changed(): result = _preprocess( - 'https://music.youtube.com/watch?v=track', { - 'playlist': 'PLexample', 'playlist_index': 3, 'playlist_count': 12, 'playlist_title': 'Example Playlist', @@ -109,33 +75,24 @@ def test_youtube_music_playlist_does_not_infer_album_or_track_number(): assert 'album' not in result assert 'track_number' not in result - assert _TRACK_NUMBER_KEY not in result - assert _TRACK_TOTAL_KEY not in result -def test_regular_youtube_playlist_and_video_are_not_changed(): +def test_regular_video_artwork_is_not_changed(): thumbnails = [ {'url': 'square.jpg', 'width': 500, 'height': 500}, {'url': 'landscape.jpg', 'width': 1280, 'height': 720}, ] - info = {'title': 'Regular Video', 'thumbnails': thumbnails.copy()} - - result = _preprocess( - 'https://www.youtube.com/watch?v=video', - {'playlist': 'PLexample', 'playlist_index': 2, 'playlist_count': 5}, - info, - ) + result = _preprocess({}, {'title': 'Regular Video', 'thumbnails': thumbnails.copy()}) assert result['thumbnails'] == thumbnails - assert 'album' not in result - assert 'track_number' not in result + assert 'thumbnail' not in result def test_music_audio_prefers_largest_existing_square_thumbnail(): result = _preprocess( - 'https://music.youtube.com/watch?v=track', - {'playlist': 'PLexample'}, + {}, { + 'track': 'Track', 'thumbnails': [ {'url': 'small-square.jpg', 'width': 200, 'height': 200}, {'url': 'large-square.jpg', 'width': 1000, 'height': 1000}, @@ -154,153 +111,9 @@ def test_landscape_only_music_artwork_keeps_existing_order(): {'url': 'large.jpg', 'width': 1280, 'height': 720}, ] result = _preprocess( - 'https://music.youtube.com/watch?v=track', - {'playlist': 'PLexample'}, - {'thumbnails': thumbnails.copy()}, + {}, + {'track': 'Track', 'thumbnails': thumbnails.copy()}, ) assert result['thumbnails'] == thumbnails assert 'thumbnail' not in result - - -def test_structured_artists_split_only_spaced_middle_dot(): - assert structured_track_artists({ - 'artists': ['Artist One \u00b7 Artist Two', 'Earth, Wind & Fire', 'half\u00b7alive'], - }) == ['Artist One', 'Artist Two', 'Earth, Wind & Fire', 'half\u00b7alive'] - assert structured_track_artists({'artist': 'Artist One, Artist Two'}) == [] - - -def test_preprocessor_retains_separate_structured_artists(): - result = _preprocess( - 'https://music.youtube.com/watch?v=track', - {}, - {'artist': 'Artist One, Artist Two', 'artists': ['Artist One', 'Artist Two']}, - ) - - assert result['artist'] == 'Artist One, Artist Two' - assert result[_ARTISTS_KEY] == ['Artist One', 'Artist Two'] - - -def test_mp3_mapping_uses_plural_artists_and_fractional_track(): - tags = MagicMock() - with patch('music_metadata.ID3', return_value=tags): - write_music_tags('track.mp3', 'mp3', ['Artist One', 'Artist Two'], 3, 12) - - artist_frame = tags.add.call_args.args[0] - assert isinstance(artist_frame, TXXX) - assert artist_frame.desc == 'Artists' - assert artist_frame.text == ['Artist One', 'Artist Two'] - track_frame = tags.setall.call_args.args[1][0] - assert isinstance(track_frame, TRCK) - assert track_frame.text == ['3/12'] - tags.save.assert_called_once_with('track.mp3', v2_version=4) - - -def test_mp3_round_trip_preserves_display_artist_artwork_and_album_artist(tmp_path): - path = tmp_path / 'track.mp3' - initial = ID3() - initial.add(TPE1(encoding=3, text=['Artist One, Artist Two'])) - initial.add(TPE2(encoding=3, text=['Existing Album Artist'])) - initial.add(APIC(encoding=3, mime='image/jpeg', type=3, desc='Cover', data=b'cover')) - initial.save(path) - - write_music_tags(str(path), 'mp3', ['Artist One', 'Artist Two'], 3, 12) - - result = ID3(path) - assert result.getall('TPE1')[0].text == ['Artist One, Artist Two'] - assert result.getall('TPE2')[0].text == ['Existing Album Artist'] - assert result.getall('APIC')[0].data == b'cover' - assert result.getall('TXXX:Artists')[0].text == ['Artist One', 'Artist Two'] - assert result.getall('TRCK')[0].text == ['3/12'] - - -class _FakeM4A: - def __init__(self): - self.tags = { - '\xa9ART': ['Artist One, Artist Two'], - 'covr': [b'cover'], - 'aART': ['Existing Album Artist'], - } - self.saved = False - - def add_tags(self): - self.tags = {} - - def save(self): - self.saved = True - - -def test_m4a_mapping_preserves_display_artist_artwork_and_album_artist(): - audio = _FakeM4A() - with patch('music_metadata.MP4', return_value=audio): - _write_m4a('track.m4a', ['Artist One', 'Artist Two'], 3, 12) - - assert audio.tags['\xa9ART'] == ['Artist One, Artist Two'] - assert audio.tags['covr'] == [b'cover'] - assert audio.tags['aART'] == ['Existing Album Artist'] - assert [ - bytes(value) for value in audio.tags['----:com.apple.iTunes:ARTISTS'] - ] == [b'Artist One', b'Artist Two'] - assert audio.tags['trkn'] == [(3, 12)] - assert audio.saved - - -class _FakeVorbis(dict): - def __init__(self): - super().__init__({ - 'ARTIST': ['Artist One, Artist Two'], - 'ALBUMARTIST': ['Existing Album Artist'], - 'METADATA_BLOCK_PICTURE': ['cover'], - }) - self.saved = False - - def save(self): - self.saved = True - - -def test_flac_mapping_preserves_singular_artist_artwork_and_album_artist(): - audio = _FakeVorbis() - _write_vorbis(audio, ['Artist One', 'Artist Two'], 3, 12) - - assert audio['ARTIST'] == ['Artist One, Artist Two'] - assert audio['ALBUMARTIST'] == ['Existing Album Artist'] - assert audio['METADATA_BLOCK_PICTURE'] == ['cover'] - assert audio['ARTISTS'] == ['Artist One', 'Artist Two'] - assert audio['TRACKNUMBER'] == ['3'] - assert audio['TRACKTOTAL'] == ['12'] - assert audio['TOTALTRACKS'] == ['12'] - assert audio.saved - - -def test_after_move_writer_ignores_unsupported_and_empty_metadata(): - processor = MusicMetadataWriterPostProcessor() - empty = {'filepath': 'track.mp3', 'ext': 'mp3'} - unsupported = { - 'filepath': 'track.wav', - 'ext': 'wav', - _TRACK_NUMBER_KEY: 3, - } - - with patch('music_metadata.write_music_tags') as writer: - processor.run(empty) - processor.run(unsupported) - - writer.assert_not_called() - - -def test_after_move_writer_dispatches_enriched_metadata(): - processor = MusicMetadataWriterPostProcessor() - info = { - 'filepath': 'track.flac', - 'ext': 'flac', - _ARTISTS_KEY: ['Artist One', 'Artist Two'], - _TRACK_NUMBER_KEY: 3, - _TRACK_TOTAL_KEY: 12, - } - - with patch('music_metadata.write_music_tags') as writer: - processor.run(info) - - writer.assert_called_once_with( - 'track.flac', 'flac', ['Artist One', 'Artist Two'], 3, 12 - ) diff --git a/app/tests/test_ytdl_utils.py b/app/tests/test_ytdl_utils.py index 6406259..5b20b47 100644 --- a/app/tests/test_ytdl_utils.py +++ b/app/tests/test_ytdl_utils.py @@ -31,10 +31,6 @@ class _PostProcessor: self._downloader = downloader -class _PostProcessingError(Exception): - pass - - fake_impersonate.ImpersonateTarget = _ImpersonateTarget fake_networking.impersonate = fake_impersonate fake_postprocessor_common.PostProcessor = _PostProcessor @@ -43,7 +39,6 @@ fake_postprocessor_common.PostProcessor = _PostProcessor # ``_resolve_outtmpl_fields`` reads via ``match.group('key')``. fake_utils.STR_FORMAT_RE_TMPL = r"(?P)%\((?P(?P{}))\)(?P[-0-9.]*{})" fake_utils.STR_FORMAT_TYPES = "diouxXeEfFgGcrsa" -fake_utils.PostProcessingError = _PostProcessingError fake_yt_dlp.networking = fake_networking fake_yt_dlp.postprocessor = fake_postprocessor fake_yt_dlp.utils = fake_utils @@ -58,7 +53,6 @@ from ytdl import ( Download, DownloadInfo, MusicMetadataPreProcessor, - MusicMetadataWriterPostProcessor, _compact_persisted_entry, _convert_srt_to_txt_file, _AlbumArtistPostProcessor, @@ -193,10 +187,7 @@ class AlbumArtistRegistrationTests(unittest.TestCase): metadata_preprocessor, = metadata_pre_call.args self.assertIsInstance(metadata_preprocessor, MusicMetadataPreProcessor) self.assertEqual(metadata_pre_call.kwargs, {'when': 'pre_process'}) - metadata_writer_call = fake_ydl.add_post_processor.call_args_list[2] - metadata_writer, = metadata_writer_call.args - self.assertIsInstance(metadata_writer, MusicMetadataWriterPostProcessor) - self.assertEqual(metadata_writer_call.kwargs, {'when': 'after_move'}) + self.assertEqual(fake_ydl.add_post_processor.call_count, 2) def test_video_download_does_not_register_postprocessor(self): download = _make_test_download() diff --git a/app/ytdl.py b/app/ytdl.py index 819a59d..25be664 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -23,7 +23,7 @@ from yt_dlp.postprocessor.common import PostProcessor from yt_dlp.utils import STR_FORMAT_RE_TMPL, STR_FORMAT_TYPES import bg_tasks from dl_formats import get_format, get_opts, AUDIO_FORMATS, merge_ytdl_option_layers -from music_metadata import MusicMetadataPreProcessor, MusicMetadataWriterPostProcessor +from music_metadata import MusicMetadataPreProcessor from datetime import datetime from state_store import AtomicJsonStore, from_json_compatible, read_legacy_shelf, to_json_compatible from subscriptions import _entry_id @@ -601,12 +601,10 @@ class Download: ydl.add_post_processor( MusicMetadataPreProcessor( ydl, - source_url=getattr(self.info, 'url', None), source_entry=getattr(self.info, 'entry', None), ), when='pre_process', ) - ydl.add_post_processor(MusicMetadataWriterPostProcessor(ydl), when='after_move') return ydl def _download(self):