refactor: simplify music metadata processing by removing unused code and improving album signal detection

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
+This module only consumes fields already supplied by yt-dlp or retained on
-MeTube's queued playlist entry. It intentionally performs no external lookup.
+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:
+def _first_positive_int(*values: Any) -> Optional[int]:
-    if not isinstance(value, str) or not value.strip():
+    return next((number for value in values if (number := _positive_int(value))), None)
-        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:
+def _has_album_signal(info: dict[str, Any], source_entry: dict[str, Any]) -> bool:
-    """Recognize only strong YouTube Music album signals."""
+    """Use only extractor-owned fields to identify album-level metadata."""
-    entry = entry if isinstance(entry, dict) else {}
+    return any(
-    for key in ('playlist_id', 'playlist'):
+        _has_value(entry.get(key))
-        playlist_id = entry.get(key)
+        for entry in (info, source_entry)
-        if isinstance(playlist_id, str) and playlist_id.startswith('OLAK5uy_'):
+        for key in ('album', 'track_number')
-            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]:
+def _is_music_audio(info: dict[str, Any], source_entry: dict[str, Any]) -> bool:
-    """Return structured track artists without guessing at comma separators."""
+    return _has_album_signal(info, source_entry) or any(
-    artists = info.get('artists')
+        _has_value(entry.get(key))
-    if not isinstance(artists, (list, tuple)):
+        for entry in (info, source_entry)
-        return []
+        for key in ('track', 'artists')
-
+    )
-    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:
@@ -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)
+        if _has_album_signal(info, self._source_entry):
-
             number, inline_total = _track_position(info.get('track_number'))
-        total = inline_total
+            if number is None:
-        if confirmed_album:
+                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'))
+
+            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'] = number
+                info['track_number'] = f'{number}/{total}' if total is not None else 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')
+                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:
+        if _is_music_audio(info, self._source_entry):
-            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

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 music_metadata import MusicMetadataPreProcessor
-
-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):
+def _preprocess(source_entry, info):
-    processor = MusicMetadataPreProcessor(
+    processor = MusicMetadataPreProcessor(source_entry=source_entry)
-        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']},
+        {'title': 'Track', 'album': 'Example Album'},
     )
 
-    assert result['track_number'] == 3
+    assert result['track_number'] == '3/12'
-    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_index': 3, 'playlist_count': 12},
-        {
-            'playlist': 'OLAK5uy_example',
-            'playlist_index': 3,
-            'playlist_count': 12,
-        },
         {'track_number': 7, 'album': 'Official Album'},
     )
 
-    assert result['track_number'] == 7
+    assert result['track_number'] == '7/12'
-    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_count': 12},
-        {'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():
+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({}, {'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 'thumbnail' 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'},
         {
+            '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'},
+        {'track': 'Track', 'thumbnails': thumbnails.copy()},
-        {'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
-    )

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<prefix>)%\((?P<has_key>(?P<key>{}))\)(?P<format>[-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]
+        self.assertEqual(fake_ydl.add_post_processor.call_count, 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()

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):