Trying to get existing channel using more variations of Youtube URLs.

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

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

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.

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.

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