From edf101faa0c7b3ed2526894fb31843ca8be115af Mon Sep 17 00:00:00 2001 From: Your GitHub Name Date: Thu, 16 Jul 2026 19:31:57 -0700 Subject: [PATCH 1/2] feat: add music metadata processing and writing functionality --- app/music_metadata.py | 275 +++++++++++++++++++++++++++ app/tests/test_music_metadata.py | 306 +++++++++++++++++++++++++++++++ app/tests/test_ytdl_utils.py | 20 +- app/ytdl.py | 10 + 4 files changed, 609 insertions(+), 2 deletions(-) create mode 100644 app/music_metadata.py create mode 100644 app/tests/test_music_metadata.py diff --git a/app/music_metadata.py b/app/music_metadata.py new file mode 100644 index 0000000..e36e7a5 --- /dev/null +++ b/app/music_metadata.py @@ -0,0 +1,275 @@ +"""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. +""" + +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: + if isinstance(value, str): + return bool(value.strip()) + if isinstance(value, (list, tuple)): + return any(_has_value(item) for item in value) + return value is not None + + +def _positive_int(value: Any) -> Optional[int]: + if isinstance(value, bool): + return None + try: + number = int(value) + except (TypeError, ValueError): + return None + return number if number > 0 else None + + +def _track_position(value: Any) -> tuple[Optional[int], Optional[int]]: + """Return a track number and optional total from a scalar or ``n/total``.""" + if isinstance(value, str) and '/' in value: + number, total = value.split('/', 1) + return _positive_int(number.strip()), _positive_int(total.strip()) + 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 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 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 prefer_square_thumbnail(info: dict[str, Any]) -> None: + """Move the largest known square thumbnail to yt-dlp's preferred slot.""" + thumbnails = info.get('thumbnails') + if not isinstance(thumbnails, list) or len(thumbnails) < 2: + return + + candidates: list[tuple[int, int]] = [] + for index, thumbnail in enumerate(thumbnails): + if not isinstance(thumbnail, dict): + continue + width = _positive_int(thumbnail.get('width')) + height = _positive_int(thumbnail.get('height')) + if width is not None and width == height: + candidates.append((width * height, index)) + if not candidates: + return + + _, selected_index = max(candidates) + selected = thumbnails.pop(selected_index) + thumbnails.append(selected) + if selected.get('url'): + info['thumbnail'] = selected['url'] + + +class MusicMetadataPreProcessor(PostProcessor): + """Enrich a fully extracted audio info-dict from its queued source entry.""" + + def __init__(self, downloader=None, *, source_url=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 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) + if not _has_value(info.get('album')): + album = 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): + 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 new file mode 100644 index 0000000..512416c --- /dev/null +++ b/app/tests/test_music_metadata.py @@ -0,0 +1,306 @@ +"""Tests for conservative audio metadata enrichment and tag mappings.""" + +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, +) + + +def _preprocess(source_url, source_entry, info): + processor = MusicMetadataPreProcessor( + source_url=source_url, + 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']}, + ) + + assert result['track_number'] == 3 + assert result[_TRACK_NUMBER_KEY] == 3 + assert result[_TRACK_TOTAL_KEY] == 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, + }, + {'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['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}, + {'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(): + result = _preprocess( + 'https://music.youtube.com/watch?v=track', + { + 'playlist': 'PLexample', + 'playlist_index': 3, + 'playlist_count': 12, + 'playlist_title': 'Example Playlist', + }, + {'title': 'Track'}, + ) + + 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(): + 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, + ) + + assert result['thumbnails'] == thumbnails + assert 'album' not in result + assert 'track_number' not in result + + +def test_music_audio_prefers_largest_existing_square_thumbnail(): + result = _preprocess( + 'https://music.youtube.com/watch?v=track', + {'playlist': 'PLexample'}, + { + 'thumbnails': [ + {'url': 'small-square.jpg', 'width': 200, 'height': 200}, + {'url': 'large-square.jpg', 'width': 1000, 'height': 1000}, + {'url': 'landscape.jpg', 'width': 1280, 'height': 720}, + ], + }, + ) + + assert result['thumbnails'][-1]['url'] == 'large-square.jpg' + assert result['thumbnail'] == 'large-square.jpg' + + +def test_landscape_only_music_artwork_keeps_existing_order(): + thumbnails = [ + {'url': 'small.jpg', 'width': 640, 'height': 360}, + {'url': 'large.jpg', 'width': 1280, 'height': 720}, + ] + result = _preprocess( + 'https://music.youtube.com/watch?v=track', + {'playlist': 'PLexample'}, + {'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 72c716e..6406259 100644 --- a/app/tests/test_ytdl_utils.py +++ b/app/tests/test_ytdl_utils.py @@ -31,6 +31,10 @@ class _PostProcessor: self._downloader = downloader +class _PostProcessingError(Exception): + pass + + fake_impersonate.ImpersonateTarget = _ImpersonateTarget fake_networking.impersonate = fake_impersonate fake_postprocessor_common.PostProcessor = _PostProcessor @@ -39,6 +43,7 @@ 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 @@ -52,6 +57,8 @@ sys.modules.setdefault("yt_dlp.utils", fake_utils) from ytdl import ( Download, DownloadInfo, + MusicMetadataPreProcessor, + MusicMetadataWriterPostProcessor, _compact_persisted_entry, _convert_srt_to_txt_file, _AlbumArtistPostProcessor, @@ -178,9 +185,18 @@ class AlbumArtistRegistrationTests(unittest.TestCase): result = download._make_youtube_dl({'quiet': True}) self.assertIs(result, fake_ydl) - postprocessor, = fake_ydl.add_post_processor.call_args.args + album_artist_call = fake_ydl.add_post_processor.call_args_list[0] + postprocessor, = album_artist_call.args self.assertIsInstance(postprocessor, _AlbumArtistPostProcessor) - self.assertEqual(fake_ydl.add_post_processor.call_args.kwargs, {'when': 'pre_process'}) + self.assertEqual(album_artist_call.kwargs, {'when': 'pre_process'}) + metadata_pre_call = fake_ydl.add_post_processor.call_args_list[1] + 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'}) def test_video_download_does_not_register_postprocessor(self): download = _make_test_download() diff --git a/app/ytdl.py b/app/ytdl.py index 20a0f4b..819a59d 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -23,6 +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 datetime import datetime from state_store import AtomicJsonStore, from_json_compatible, read_legacy_shelf, to_json_compatible from subscriptions import _entry_id @@ -597,6 +598,15 @@ class Download: ydl = yt_dlp.YoutubeDL(params=params) if getattr(self.info, 'download_type', '') == 'audio': ydl.add_post_processor(_AlbumArtistPostProcessor(ydl), when='pre_process') + 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): From f3d670e288243a5a8c74fc6cf8d9f51f280bb619 Mon Sep 17 00:00:00 2001 From: Your GitHub Name Date: Fri, 17 Jul 2026 10:43:39 -0700 Subject: [PATCH 2/2] refactor: simplify music metadata processing by removing unused code and improving album signal detection --- app/music_metadata.py | 235 +++++----------------------- app/tests/test_music_metadata.py | 259 +++++-------------------------- app/tests/test_ytdl_utils.py | 11 +- app/ytdl.py | 4 +- 4 files changed, 80 insertions(+), 429 deletions(-) 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):