From 12a49843affed83aa790397781dc6600da297426 Mon Sep 17 00:00:00 2001 From: Woodyy07 Date: Mon, 8 Jun 2026 12:19:37 +0200 Subject: [PATCH] Add YouTube Music artist subscription support When a music.youtube.com/channel/... URL is subscribed, use ytmusicapi to enumerate the artist's album and single releases instead of yt-dlp flat-extract, which only returns individual topic video uploads. Each release is tracked as an OLAK5uy_ playlist ID in seen_ids, so new albums and singles are detected and queued automatically on each check. - Add extract_ytmusic_artist_releases() using ytmusicapi.get_artist() and get_artist_albums() for full pagination - Route YTMusic URLs in add_subscription() and _check_one_unlocked() - Add ytmusicapi dependency - Add 31 tests covering URL detection, extraction, and manager integration --- app/subscriptions.py | 193 ++++++++-- app/tests/test_ytmusic_subscriptions.py | 489 ++++++++++++++++++++++++ pyproject.toml | 1 + uv.lock | 14 + 4 files changed, 658 insertions(+), 39 deletions(-) create mode 100644 app/tests/test_ytmusic_subscriptions.py diff --git a/app/subscriptions.py b/app/subscriptions.py index 0b5774a..1f28100 100644 --- a/app/subscriptions.py +++ b/app/subscriptions.py @@ -31,6 +31,9 @@ _MEDIA_HINT_FIELDS = ( "live_status", "availability", ) +_YTMUSIC_ARTIST_URL_RE = re.compile( + r"https?://music\.youtube\.com/channel/([A-Za-z0-9_-]+)" +) def _impersonate_opt(ytdl_options: dict) -> dict: @@ -132,6 +135,88 @@ def _is_subscriber_only_entry(entry: dict) -> bool: return str(entry.get("availability") or "") == "subscriber_only" +def _is_ytmusic_artist_url(url: str) -> bool: + """Return True if the URL is a YouTube Music artist channel URL.""" + return bool(_YTMUSIC_ARTIST_URL_RE.match(url or "")) + + +def _ytmusic_channel_id(url: str) -> Optional[str]: + m = _YTMUSIC_ARTIST_URL_RE.match(url or "") + return m.group(1) if m else None + + +def extract_ytmusic_artist_releases(url: str) -> tuple[Optional[dict], list[dict]]: + """Return (info_dict, entries) for a YouTube Music artist channel URL.""" + try: + from ytmusicapi import YTMusic # soft dependency + except ImportError: + log.error( + "ytmusicapi is not installed; cannot handle YouTube Music artist URLs. " + "Add 'ytmusicapi' to your dependencies." + ) + return None, [] + + channel_id = _ytmusic_channel_id(url) + if not channel_id: + log.warning("Could not extract channel ID from URL: %s", url) + return None, [] + + ytm = YTMusic() + try: + artist = ytm.get_artist(channel_id) + except Exception as exc: + log.warning("ytmusicapi.get_artist(%s) failed: %s", channel_id, exc) + return None, [] + + artist_name = artist.get("name") or url + info: dict = { + "_type": "channel", + "title": artist_name, + } + + entries: list[dict] = [] + + def _collect(section_key: str) -> None: + section = artist.get(section_key) or {} + browse_id = section.get("browseId") + params = section.get("params") + + if browse_id and params: + # get_artist_albums uses "playlistId" + try: + results = ytm.get_artist_albums(browse_id, params) or [] + except Exception as exc: + log.warning( + "ytmusicapi.get_artist_albums(%s, %s) failed: %s", + channel_id, section_key, exc, + ) + results = section.get("results") or [] + else: + # get_artist() uses "audioPlaylistId" + results = section.get("results") or [] + + for release in results: + # field name differs between get_artist() and get_artist_albums() + playlist_id = release.get("playlistId") or release.get("audioPlaylistId") + if not playlist_id: + continue + playlist_url = f"https://music.youtube.com/playlist?list={playlist_id}" + entries.append({ + "id": playlist_id, + "url": playlist_url, + "webpage_url": playlist_url, + "title": release.get("title") or playlist_id, + }) + + _collect("albums") + _collect("singles") + + log.info( + "YTMusic artist %s: found %d release(s)", artist_name, len(entries) + ) + return info, entries + + def coerce_optional_bool(value: Any, *, default: bool = False, field_name: str = "value") -> bool: """Parse optional JSON booleans for subscription settings.""" if value is None: @@ -512,10 +597,15 @@ class SubscriptionManager: try: scan_first = max(int(getattr(self.config, "SUBSCRIPTION_SCAN_PLAYLIST_END", 50)), 1) - try: - info, entries = extract_flat_playlist(self.config, url, scan_first) - except yt_dlp.utils.YoutubeDLError as exc: - return {"status": "error", "msg": str(exc)} + if _is_ytmusic_artist_url(url): + info, entries = extract_ytmusic_artist_releases(url) + if not info: + return {"status": "error", "msg": "Could not resolve YouTube Music artist URL"} + else: + try: + info, entries = extract_flat_playlist(self.config, url, scan_first) + except yt_dlp.utils.YoutubeDLError as exc: + return {"status": "error", "msg": str(exc)} if not info: return {"status": "error", "msg": "Could not resolve URL"} @@ -532,7 +622,11 @@ class SubscriptionManager: or url ) - seen_entries = [ent for ent in entries if _is_media_entry(ent)] + seen_entries = ( + entries if _is_ytmusic_artist_url(url) + else [ent for ent in entries if _is_media_entry(ent)] + ) + all_ids: list[str] = [] for ent in seen_entries: if ent.get("live_status") == "is_upcoming": @@ -680,41 +774,62 @@ class SubscriptionManager: sid = sub.id scan = int(getattr(self.config, "SUBSCRIPTION_SCAN_PLAYLIST_END", 50)) log.info("Checking subscription: %s", sub.name) - try: - info, entries = extract_flat_playlist(self.config, sub.url, scan) - except yt_dlp.utils.YoutubeDLError as exc: - async with self._lock: - cur = self._subs.get(sid) - if cur: - previous = copy.deepcopy(cur) - cur.error = str(exc) - try: - self._save_locked() - except Exception: - self._subs[sid] = previous - raise - sub = cur - log.warning("Subscription check failed for %s: %s", sub.name, exc) - await self.notifier.subscription_updated(sub) - return - entries = [ent for ent in entries if _is_media_entry(ent)] + if _is_ytmusic_artist_url(sub.url): + # playlist IDs instead of yt-dlp flat-extract on the channel. + info, entries = extract_ytmusic_artist_releases(sub.url) + if not info: + async with self._lock: + cur = self._subs.get(sid) + if cur: + previous = copy.deepcopy(cur) + cur.error = "Could not resolve YouTube Music artist" + try: + self._save_locked() + except Exception: + self._subs[sid] = previous + raise + sub = cur + log.warning("YTMusic subscription check failed for %s", sub.name) + await self.notifier.subscription_updated(sub) + return + else: + try: + info, entries = extract_flat_playlist(self.config, sub.url, scan) + except yt_dlp.utils.YoutubeDLError as exc: + async with self._lock: + cur = self._subs.get(sid) + if cur: + previous = copy.deepcopy(cur) + cur.error = str(exc) + try: + self._save_locked() + except Exception: + self._subs[sid] = previous + raise + sub = cur + log.warning("Subscription check failed for %s: %s", sub.name, exc) + await self.notifier.subscription_updated(sub) + return + entries = [ent for ent in entries if _is_media_entry(ent)] - etype = (info or {}).get("_type") or "video" - if etype == "video" or not entries: - async with self._lock: - cur = self._subs.get(sid) - if cur: - previous = copy.deepcopy(cur) - cur.error = VIDEO_ONLY_MSG - try: - self._save_locked() - except Exception: - self._subs[sid] = previous - raise - sub = cur - log.warning("Subscription %s no longer resolves to a subscribable feed", sub.name) - await self.notifier.subscription_updated(sub) - return + etype = (info or {}).get("_type") or "video" + if etype == "video" or not entries: + async with self._lock: + cur = self._subs.get(sid) + if cur: + previous = copy.deepcopy(cur) + cur.error = VIDEO_ONLY_MSG + try: + self._save_locked() + except Exception: + self._subs[sid] = previous + raise + sub = cur + log.warning( + "Subscription %s no longer resolves to a subscribable feed", sub.name + ) + await self.notifier.subscription_updated(sub) + return async with self._lock: cur = self._subs.get(sid) diff --git a/app/tests/test_ytmusic_subscriptions.py b/app/tests/test_ytmusic_subscriptions.py new file mode 100644 index 0000000..de209fc --- /dev/null +++ b/app/tests/test_ytmusic_subscriptions.py @@ -0,0 +1,489 @@ +from __future__ import annotations + +import sys +import tempfile +import types +import unittest +from unittest.mock import MagicMock, patch + +# Replicate the yt_dlp fake setup from test_subscriptions.py so this file +# can run independently without yt_dlp installed. +fake_yt_dlp = types.ModuleType("yt_dlp") +fake_networking = types.ModuleType("yt_dlp.networking") +fake_impersonate = types.ModuleType("yt_dlp.networking.impersonate") + + +class _ImpersonateTarget: + @staticmethod + def from_str(value): + return value + + +fake_impersonate.ImpersonateTarget = _ImpersonateTarget +fake_networking.impersonate = fake_impersonate +fake_yt_dlp.networking = fake_networking +fake_yt_dlp.utils = types.SimpleNamespace(YoutubeDLError=Exception) +sys.modules.setdefault("yt_dlp", fake_yt_dlp) +sys.modules.setdefault("yt_dlp.networking", fake_networking) +sys.modules.setdefault("yt_dlp.networking.impersonate", fake_impersonate) + +from subscriptions import ( + SubscriptionManager, + _is_ytmusic_artist_url, + _ytmusic_channel_id, + extract_ytmusic_artist_releases, +) + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + +_YTMUSIC_URL = "https://music.youtube.com/channel/UCtest123" +_ARTIST_INFO = {"_type": "channel", "title": "Test Artist"} +_ALBUM_1 = { + "id": "OLAK5uy_album1", + "url": "https://music.youtube.com/playlist?list=OLAK5uy_album1", + "webpage_url": "https://music.youtube.com/playlist?list=OLAK5uy_album1", + "title": "Album 1", +} +_ALBUM_2 = { + "id": "OLAK5uy_album2", + "url": "https://music.youtube.com/playlist?list=OLAK5uy_album2", + "webpage_url": "https://music.youtube.com/playlist?list=OLAK5uy_album2", + "title": "Album 2", +} +_SINGLE_1 = { + "id": "OLAK5uy_single1", + "url": "https://music.youtube.com/playlist?list=OLAK5uy_single1", + "webpage_url": "https://music.youtube.com/playlist?list=OLAK5uy_single1", + "title": "Single 1", +} + + +class _Config: + def __init__(self, state_dir: str): + self.STATE_DIR = state_dir + self.SUBSCRIPTION_SCAN_PLAYLIST_END = 50 + self.SUBSCRIPTION_MAX_SEEN_IDS = 50000 + self.DOWNLOAD_DIR = state_dir + self.TEMP_DIR = state_dir + self.YTDL_OPTIONS = {} + + +class _Queue: + def __init__(self): + self.entries = [] + self.fail = False + + async def add(self, *args, **kwargs): + return None + + async def add_entry(self, entry, *args, **kwargs): + if self.fail: + return {"status": "error", "msg": "queue failed"} + self.entries.append((entry, args, kwargs)) + return {"status": "ok"} + + +class _Notifier: + async def subscription_added(self, sub): + return None + + async def subscription_updated(self, sub): + return None + + async def subscription_removed(self, sub_id): + return None + + async def subscriptions_all(self, subs): + return None + + +async def _add_ytmusic_sub(mgr, releases, *, url=_YTMUSIC_URL): + """Add a YTMusic subscription with mocked extract_ytmusic_artist_releases.""" + with patch( + "subscriptions.extract_ytmusic_artist_releases", + return_value=(_ARTIST_INFO, releases), + ): + return await mgr.add_subscription( + url, + check_interval_minutes=60, + download_type="audio", + codec="auto", + format="opus", + quality="best", + folder="", + custom_name_prefix="", + auto_start=True, + playlist_item_limit=0, + split_by_chapters=False, + chapter_template="", + subtitle_language="en", + subtitle_mode="prefer_manual", + ) + + +# --------------------------------------------------------------------------- +# URL detection helpers +# --------------------------------------------------------------------------- + +class YTMusicUrlHelperTests(unittest.TestCase): + def test_accepts_music_youtube_channel_url(self): + self.assertTrue(_is_ytmusic_artist_url("https://music.youtube.com/channel/UCtest123")) + + def test_accepts_http_variant(self): + self.assertTrue(_is_ytmusic_artist_url("http://music.youtube.com/channel/UCtest123")) + + def test_rejects_regular_youtube_channel(self): + self.assertFalse(_is_ytmusic_artist_url("https://www.youtube.com/channel/UCtest123")) + + def test_rejects_ytmusic_playlist_url(self): + self.assertFalse(_is_ytmusic_artist_url("https://music.youtube.com/playlist?list=OLAK5uy_xxx")) + + def test_rejects_ytmusic_watch_url(self): + self.assertFalse(_is_ytmusic_artist_url("https://music.youtube.com/watch?v=xxx")) + + def test_rejects_empty_string(self): + self.assertFalse(_is_ytmusic_artist_url("")) + + def test_rejects_none_coerced(self): + self.assertFalse(_is_ytmusic_artist_url("")) + + def test_channel_id_extracted_correctly(self): + self.assertEqual( + _ytmusic_channel_id("https://music.youtube.com/channel/UCtest123"), + "UCtest123", + ) + + def test_channel_id_returns_none_for_non_matching_url(self): + self.assertIsNone(_ytmusic_channel_id("https://www.youtube.com/channel/UCtest123")) + + def test_channel_id_returns_none_for_empty(self): + self.assertIsNone(_ytmusic_channel_id("")) + + +# --------------------------------------------------------------------------- +# extract_ytmusic_artist_releases +# --------------------------------------------------------------------------- + +class ExtractYTMusicReleasesTests(unittest.TestCase): + def _make_ytm(self, artist_response, albums_response=None, singles_response=None): + """Build a mock YTMusic instance.""" + ytm = MagicMock() + ytm.get_artist.return_value = artist_response + ytm.get_artist_albums.side_effect = lambda browse_id, params: ( + albums_response if "album" in browse_id.lower() or singles_response is None + else singles_response + ) + return ytm + + def test_info_dict_has_channel_type_and_artist_title(self): + ytm = MagicMock() + ytm.get_artist.return_value = {"name": "Test Artist", "albums": {}, "singles": {}} + with patch("ytmusicapi.YTMusic", return_value=ytm): + info, _ = extract_ytmusic_artist_releases(_YTMUSIC_URL) + self.assertEqual(info["_type"], "channel") + self.assertEqual(info["title"], "Test Artist") + self.assertNotIn("channel", info) # only "title", not redundant "channel" key + + def test_calls_get_artist_albums_when_browse_id_and_params_present(self): + ytm = MagicMock() + ytm.get_artist.return_value = { + "name": "Artist", + "albums": {"browseId": "MPADUCxxx", "params": "abc"}, + "singles": {}, + } + ytm.get_artist_albums.return_value = [ + {"playlistId": "OLAK5uy_album1", "title": "Album 1"}, + ] + with patch("ytmusicapi.YTMusic", return_value=ytm): + _, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + ytm.get_artist_albums.assert_called_once_with("MPADUCxxx", "abc") + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0]["id"], "OLAK5uy_album1") + + def test_falls_back_to_preview_results_when_no_browse_id(self): + ytm = MagicMock() + ytm.get_artist.return_value = { + "name": "Artist", + "albums": { + "results": [{"audioPlaylistId": "OLAK5uy_preview", "title": "Preview Album"}] + }, + "singles": {}, + } + with patch("ytmusicapi.YTMusic", return_value=ytm): + _, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + ytm.get_artist_albums.assert_not_called() + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0]["id"], "OLAK5uy_preview") + + def test_uses_audio_playlist_id_fallback(self): + """get_artist() results use audioPlaylistId; should still be picked up.""" + ytm = MagicMock() + ytm.get_artist.return_value = { + "name": "Artist", + "albums": { + "results": [{"audioPlaylistId": "OLAK5uy_audio", "title": "Album"}] + }, + "singles": {}, + } + with patch("ytmusicapi.YTMusic", return_value=ytm): + _, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + self.assertEqual(entries[0]["id"], "OLAK5uy_audio") + + def test_get_artist_albums_failure_falls_back_to_preview_results(self): + ytm = MagicMock() + ytm.get_artist.return_value = { + "name": "Artist", + "albums": { + "browseId": "MPADUCxxx", + "params": "abc", + "results": [{"audioPlaylistId": "OLAK5uy_fallback", "title": "Fallback"}], + }, + "singles": {}, + } + ytm.get_artist_albums.side_effect = Exception("network error") + with patch("ytmusicapi.YTMusic", return_value=ytm): + _, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0]["id"], "OLAK5uy_fallback") + + def test_collects_both_albums_and_singles(self): + ytm = MagicMock() + ytm.get_artist.return_value = { + "name": "Artist", + "albums": { + "browseId": "MPADUCalbums", + "params": "p1", + }, + "singles": { + "browseId": "MPADUCsingles", + "params": "p2", + }, + } + ytm.get_artist_albums.side_effect = [ + [{"playlistId": "OLAK5uy_album1", "title": "Album"}], + [{"playlistId": "OLAK5uy_single1", "title": "Single"}], + ] + with patch("ytmusicapi.YTMusic", return_value=ytm): + _, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + ids = [e["id"] for e in entries] + self.assertIn("OLAK5uy_album1", ids) + self.assertIn("OLAK5uy_single1", ids) + self.assertEqual(len(entries), 2) + + def test_entry_url_points_to_ytmusic_playlist(self): + ytm = MagicMock() + ytm.get_artist.return_value = { + "name": "Artist", + "albums": { + "results": [{"audioPlaylistId": "OLAK5uy_xxx", "title": "Album"}] + }, + "singles": {}, + } + with patch("ytmusicapi.YTMusic", return_value=ytm): + _, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + expected_url = "https://music.youtube.com/playlist?list=OLAK5uy_xxx" + self.assertEqual(entries[0]["url"], expected_url) + self.assertEqual(entries[0]["webpage_url"], expected_url) + + def test_skips_releases_without_playlist_id(self): + ytm = MagicMock() + ytm.get_artist.return_value = { + "name": "Artist", + "albums": { + "results": [ + {"title": "No ID release"}, + {"audioPlaylistId": "OLAK5uy_valid", "title": "Valid"}, + ] + }, + "singles": {}, + } + with patch("ytmusicapi.YTMusic", return_value=ytm): + _, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0]["id"], "OLAK5uy_valid") + + def test_get_artist_failure_returns_none_and_empty(self): + ytm = MagicMock() + ytm.get_artist.side_effect = Exception("API error") + with patch("ytmusicapi.YTMusic", return_value=ytm): + info, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + self.assertIsNone(info) + self.assertEqual(entries, []) + + def test_ytmusicapi_not_installed_returns_none_and_empty(self): + with patch.dict("sys.modules", {"ytmusicapi": None}): + info, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + self.assertIsNone(info) + self.assertEqual(entries, []) + + def test_empty_artist_returns_empty_entries(self): + ytm = MagicMock() + ytm.get_artist.return_value = {"name": "Artist", "albums": {}, "singles": {}} + with patch("ytmusicapi.YTMusic", return_value=ytm): + info, entries = extract_ytmusic_artist_releases(_YTMUSIC_URL) + self.assertIsNotNone(info) + self.assertEqual(entries, []) + + +# --------------------------------------------------------------------------- +# SubscriptionManager integration +# --------------------------------------------------------------------------- + +class YTMusicSubscriptionManagerTests(unittest.IsolatedAsyncioTestCase): + async def test_add_subscription_uses_extract_ytmusic_not_flat_playlist(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = SubscriptionManager(_Config(tmp), _Queue(), _Notifier()) + with patch("subscriptions.extract_flat_playlist") as flat, \ + patch("subscriptions.extract_ytmusic_artist_releases", + return_value=(_ARTIST_INFO, [_ALBUM_1])) as ytm: + await _add_ytmusic_sub(mgr, [_ALBUM_1]) + flat.assert_not_called() + + async def test_add_subscription_ytmusic_marks_existing_releases_seen(self): + with tempfile.TemporaryDirectory() as tmp: + queue = _Queue() + mgr = SubscriptionManager(_Config(tmp), queue, _Notifier()) + result = await _add_ytmusic_sub(mgr, [_ALBUM_1, _ALBUM_2]) + self.assertEqual(result["status"], "ok") + sub = mgr.list_all()[0] + self.assertIn("OLAK5uy_album1", sub.seen_ids) + self.assertIn("OLAK5uy_album2", sub.seen_ids) + self.assertEqual(queue.entries, []) # no downloads queued on subscribe + + async def test_add_subscription_ytmusic_uses_artist_name(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = SubscriptionManager(_Config(tmp), _Queue(), _Notifier()) + result = await _add_ytmusic_sub(mgr, [_ALBUM_1]) + self.assertEqual(result["subscription"]["name"], "Test Artist") + + async def test_add_subscription_ytmusic_extract_failure_returns_error(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = SubscriptionManager(_Config(tmp), _Queue(), _Notifier()) + with patch( + "subscriptions.extract_ytmusic_artist_releases", + return_value=(None, []), + ): + result = await mgr.add_subscription( + _YTMUSIC_URL, + check_interval_minutes=60, + download_type="audio", + codec="auto", + format="opus", + quality="best", + folder="", + custom_name_prefix="", + auto_start=True, + playlist_item_limit=0, + split_by_chapters=False, + chapter_template="", + subtitle_language="en", + subtitle_mode="prefer_manual", + ) + self.assertEqual(result["status"], "error") + self.assertEqual(mgr.list_all(), []) + + async def test_check_now_ytmusic_queues_new_release(self): + with tempfile.TemporaryDirectory() as tmp: + queue = _Queue() + mgr = SubscriptionManager(_Config(tmp), queue, _Notifier()) + # Subscribe with album1 already seen + result = await _add_ytmusic_sub(mgr, [_ALBUM_1]) + sub_id = result["subscription"]["id"] + # Check: album1 still seen, album2 is new + with patch( + "subscriptions.extract_ytmusic_artist_releases", + return_value=(_ARTIST_INFO, [_ALBUM_1, _ALBUM_2]), + ): + await mgr.check_now([sub_id]) + queued_urls = [e["webpage_url"] for e, _, _ in queue.entries] + self.assertIn(_ALBUM_2["webpage_url"], queued_urls) + self.assertNotIn(_ALBUM_1["webpage_url"], queued_urls) + + async def test_check_now_ytmusic_does_not_requeue_seen_release(self): + with tempfile.TemporaryDirectory() as tmp: + queue = _Queue() + mgr = SubscriptionManager(_Config(tmp), queue, _Notifier()) + result = await _add_ytmusic_sub(mgr, [_ALBUM_1]) + sub_id = result["subscription"]["id"] + # Check: same releases, nothing new + with patch( + "subscriptions.extract_ytmusic_artist_releases", + return_value=(_ARTIST_INFO, [_ALBUM_1]), + ): + await mgr.check_now([sub_id]) + self.assertEqual(queue.entries, []) + + async def test_check_now_ytmusic_updates_seen_ids_after_queueing(self): + with tempfile.TemporaryDirectory() as tmp: + queue = _Queue() + mgr = SubscriptionManager(_Config(tmp), queue, _Notifier()) + result = await _add_ytmusic_sub(mgr, [_ALBUM_1]) + sub_id = result["subscription"]["id"] + with patch( + "subscriptions.extract_ytmusic_artist_releases", + return_value=(_ARTIST_INFO, [_ALBUM_1, _ALBUM_2]), + ): + await mgr.check_now([sub_id]) + sub = mgr.list_all()[0] + self.assertIn("OLAK5uy_album1", sub.seen_ids) + self.assertIn("OLAK5uy_album2", sub.seen_ids) + self.assertIsNone(sub.error) + + async def test_check_now_ytmusic_sets_error_on_extract_failure(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = SubscriptionManager(_Config(tmp), _Queue(), _Notifier()) + result = await _add_ytmusic_sub(mgr, [_ALBUM_1]) + sub_id = result["subscription"]["id"] + with patch( + "subscriptions.extract_ytmusic_artist_releases", + return_value=(None, []), + ): + await mgr.check_now([sub_id]) + sub = mgr.list_all()[0] + self.assertIsNotNone(sub.error) + + async def test_check_now_ytmusic_does_not_call_extract_flat_playlist(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = SubscriptionManager(_Config(tmp), _Queue(), _Notifier()) + result = await _add_ytmusic_sub(mgr, [_ALBUM_1]) + sub_id = result["subscription"]["id"] + with patch("subscriptions.extract_flat_playlist") as flat, \ + patch("subscriptions.extract_ytmusic_artist_releases", + return_value=(_ARTIST_INFO, [_ALBUM_1])): + await mgr.check_now([sub_id]) + flat.assert_not_called() + + async def test_regular_channel_url_still_uses_extract_flat_playlist(self): + with tempfile.TemporaryDirectory() as tmp: + queue = _Queue() + mgr = SubscriptionManager(_Config(tmp), queue, _Notifier()) + with patch( + "subscriptions.extract_flat_playlist", + return_value=( + {"_type": "channel", "title": "Channel"}, + [{"id": "v1", "title": "One", "webpage_url": "https://example.com/v1"}], + ), + ) as flat: + await mgr.add_subscription( + "https://www.youtube.com/channel/UCregular", + check_interval_minutes=60, + download_type="video", + codec="auto", + format="any", + quality="best", + folder="", + custom_name_prefix="", + auto_start=True, + playlist_item_limit=0, + split_by_chapters=False, + chapter_template="", + subtitle_language="en", + subtitle_mode="prefer_manual", + ) + flat.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/pyproject.toml b/pyproject.toml index 710988b..5b46a40 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ dependencies = [ "mutagen", "curl-cffi", "watchfiles", + "ytmusicapi", ] [dependency-groups] diff --git a/uv.lock b/uv.lock index 7d2c324..72604a4 100644 --- a/uv.lock +++ b/uv.lock @@ -473,6 +473,7 @@ dependencies = [ { name = "python-socketio" }, { name = "watchfiles" }, { name = "yt-dlp", extra = ["curl-cffi", "default", "deno"] }, + { name = "ytmusicapi" }, ] [package.dev-dependencies] @@ -491,6 +492,7 @@ requires-dist = [ { name = "python-socketio", specifier = ">=5.0,<6.0" }, { name = "watchfiles" }, { name = "yt-dlp", extras = ["default", "curl-cffi", "deno"] }, + { name = "ytmusicapi" }, ] [package.metadata.requires-dev] @@ -1094,3 +1096,15 @@ sdist = { url = "https://files.pythonhosted.org/packages/d3/e6/cceb9530e8f4e5940 wheels = [ { url = "https://files.pythonhosted.org/packages/e3/bd/520769863744b669440a924271a6159ddd82ad5ae26b4ac4d4b69e9f8d44/yt_dlp_ejs-0.8.0-py3-none-any.whl", hash = "sha256:79300e5fca7f937a1eeede11f0456862c1b41107ce1d726871e0207424f4bdb4", size = 53443, upload-time = "2026-03-17T22:49:17.736Z" }, ] + +[[package]] +name = "ytmusicapi" +version = "1.12.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "requests" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/8a/16/f9737ae3a565abaf2e3d106507272f85e43abff6886fb514a22db3a8adaf/ytmusicapi-1.12.1.tar.gz", hash = "sha256:4e33f424db311ece1b9e5f29a51ade46b0697f6e2f284f8a40bcbd7219772755", size = 529510, upload-time = "2026-06-05T16:36:18.225Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/53/6d/f7156e7ab9953d09e14ef538de4e92c484169fac19d207b19b9d9eb9c845/ytmusicapi-1.12.1-py3-none-any.whl", hash = "sha256:2a53f6df504b97b93a4d41c804b699b46190207cb9d864fcf4edeb338b1f15d9", size = 108567, upload-time = "2026-06-05T16:36:16.523Z" }, +]