diff --git a/modules/videos/channel/lib.py b/modules/videos/channel/lib.py index f65e3409..c22cdf4a 100644 --- a/modules/videos/channel/lib.py +++ b/modules/videos/channel/lib.py @@ -70,6 +70,13 @@ def get_channel(session: Session, *, channel_id: int = None, source_id: str = No source_id=source_id).one_or_none() if not channel and url: channel = session.query(Channel).options(joinedload(Channel.collection)).filter_by(url=url).one_or_none() + if not channel and url: + # Try to extract YouTube channel ID from the URL and match by source_id. + from modules.videos.common import get_youtube_channel_id + yt_channel_id = get_youtube_channel_id(url) + if yt_channel_id: + channel = session.query(Channel).options(joinedload(Channel.collection)).filter_by( + source_id=yt_channel_id).one_or_none() if not channel and directory: directory = Path(directory) if not directory.is_absolute(): diff --git a/modules/videos/common.py b/modules/videos/common.py index 9401eb05..29c43fdc 100644 --- a/modules/videos/common.py +++ b/modules/videos/common.py @@ -8,6 +8,7 @@ from datetime import timedelta from decimal import Decimal from pathlib import Path from typing import Optional, Tuple, Union +from urllib.parse import urlparse import PIL from PIL import Image @@ -46,6 +47,18 @@ def get_no_channel_directory() -> pathlib.Path: return directory +def get_youtube_channel_id(url: str) -> Optional[str]: + """Extract YouTube channel ID (UC...) from /channel/ URLs.""" + parsed = urlparse(url) + if parsed.netloc not in ('youtube.com', 'www.youtube.com', 'm.youtube.com'): + return None + if parsed.path.startswith('/channel/'): + parts = parsed.path.split('/') + if len(parts) >= 3 and parts[2]: + return parts[2] + return None + + def check_for_channel_conflicts(session: Session, id_=None, url=None, name=None, directory=None, source_id=None): """ diff --git a/modules/videos/downloader.py b/modules/videos/downloader.py index ad1bc23d..79e60405 100755 --- a/modules/videos/downloader.py +++ b/modules/videos/downloader.py @@ -418,7 +418,8 @@ class ChannelDownloader(Downloader, ABC): if not name: raise ValueError(f'Could not find name') channel_source_id = info.get('channel_id') or info.get('id') - channel = get_or_create_channel(session, channel_source_id, download.url, name, channel_tag_name) + channel_url = info.get('channel_url') or info.get('uploader_url') or download.url + channel = get_or_create_channel(session, channel_source_id, channel_url, name, channel_tag_name) channel.dict() # get all attributes while we have the session. # Link this channel download to the channel's collection. diff --git a/modules/videos/test/test_downloader.py b/modules/videos/test/test_downloader.py index 83756668..4a27e8fb 100644 --- a/modules/videos/test/test_downloader.py +++ b/modules/videos/test/test_downloader.py @@ -266,6 +266,48 @@ async def test_get_or_create_channel(async_client, test_session, test_directory, assert channel.tag_name == one.name and channel.tag == one +@pytest.mark.asyncio +async def test_get_or_create_channel_conflict_recovery(async_client, test_session, test_directory): + """get_or_create_channel should find an existing channel even when source_id and url don't match, + rather than raising ChannelNameConflict.""" + from wrolpi.collections import Collection + + # Create a channel with a specific source_id and url. + collection = Collection(name='My Channel', kind='channel') + test_session.add(collection) + test_session.flush([collection]) + channel = Channel(collection_id=collection.id, source_id='old_source_id', + url='https://www.youtube.com/channel/UCold') + test_session.add(channel) + test_session.commit() + + # Calling with different source_id and url but same name should return the existing channel. + result = get_or_create_channel(test_session, source_id='new_source_id', + url='https://www.youtube.com/@mychannel', name='My Channel') + assert result.id == channel.id + + +@pytest.mark.asyncio +async def test_get_or_create_channel_youtube_channel_url(async_client, test_session, test_directory): + """get_or_create_channel should find an existing channel when the URL contains a YouTube /channel/ ID + that matches the channel's source_id.""" + from wrolpi.collections import Collection + + collection = Collection(name='WROLPi', kind='channel') + test_session.add(collection) + test_session.flush([collection]) + channel = Channel(collection_id=collection.id, source_id='UC4t8bw1besFTyjW7ZBCOIrw', + url='https://www.youtube.com/@wrolpi') + test_session.add(channel) + test_session.commit() + + # A /channel/UCXXX URL should match via source_id extraction. + result = get_or_create_channel(test_session, + url='https://www.youtube.com/channel/UC4t8bw1besFTyjW7ZBCOIrw', + name='WROLPi Channel') + assert result.id == channel.id + + def test_channel_downloader_hidden(video_download_manager): """ ChannelDownloader should not be presented to the User. diff --git a/modules/videos/test/test_video_url_resolver.py b/modules/videos/test/test_video_url_resolver.py index 73ab8670..0a29c2ed 100644 --- a/modules/videos/test/test_video_url_resolver.py +++ b/modules/videos/test/test_video_url_resolver.py @@ -1,5 +1,6 @@ import pytest +from modules.videos.common import get_youtube_channel_id from modules.videos.normalize_video_url import normalize_video_url @@ -20,3 +21,21 @@ from modules.videos.normalize_video_url import normalize_video_url ) def test_normalize_video_url(url: str, expected: str): assert normalize_video_url(url) == expected + + +@pytest.mark.parametrize( + 'url,expected', + [ + ('https://www.youtube.com/channel/UC4t8bw1besFTyjW7ZBCOIrw', 'UC4t8bw1besFTyjW7ZBCOIrw'), + ('https://www.youtube.com/channel/UC4t8bw1besFTyjW7ZBCOIrw/videos', 'UC4t8bw1besFTyjW7ZBCOIrw'), + ('https://www.youtube.com/channel/UC4t8bw1besFTyjW7ZBCOIrw/', 'UC4t8bw1besFTyjW7ZBCOIrw'), + ('https://m.youtube.com/channel/UC4t8bw1besFTyjW7ZBCOIrw', 'UC4t8bw1besFTyjW7ZBCOIrw'), + ('https://www.youtube.com/@wrolpi', None), + ('https://www.youtube.com/@wrolpi/featured', None), + ('https://www.youtube.com/@wrolpi/videos', None), + ('https://www.youtube.com/@wrolpi/playlists', None), + ('https://example.com/channel/something', None), + ] +) +def test_get_youtube_channel_id(url: str, expected): + assert get_youtube_channel_id(url) == expected