✨ Add /c/{channel} for channel browsing

- use htmx for infinite scroll & lazy loading
- update dependencies
- refactor existing code

tests/test_web.py

@@ -1,8 +1,7 @@
 import pytest
+from bs4 import BeautifulSoup
 from httpx import BasicAuth
 
-from yourss.youtube.utils import bs_parse
-
 
 @pytest.mark.anyio
 async def test_default(client):
@@ -58,7 +57,7 @@ async def test_page_content(client):
     resp = await client.get("/" + ",".join(names))
     assert resp.status_code == 200
 
-    soup = bs_parse(resp.text)
+    soup = BeautifulSoup(resp.text, features="html.parser")
     assert len(soup.find_all("div", class_="yourss-filterable")) == 45
 
     # when no valid feed given, should return 404

tests/test_youtube.py

@@ -1,28 +1,23 @@
 from http.cookiejar import CookieJar
 
 import pytest
+from bs4 import BeautifulSoup
 from httpx import AsyncClient
 
-from yourss.youtube import (
-    YoutubeMetadata,
-    YoutubeRssApi,
-    YoutubeWebApi,
-)
-from yourss.youtube.scrapper import VideoScrapper
-from yourss.youtube.utils import bs_parse
+from yourss.youtube import PageScrapper, VideoScrapper, YoutubeApi
 
 
 @pytest.mark.asyncio(loop_scope="module")
 async def test_rgpd():
-    api = YoutubeWebApi()
+    api = YoutubeApi()
 
     url = "/@jonnygiger"
 
     resp = await api.get_html(url)
     assert resp.status_code == 200
     assert (
         len(
-            bs_parse(resp.text).find_all(
+            BeautifulSoup(resp.text, features="html.parser").find_all(
                 "form",
                 attrs={"method": "POST", "action": "https://consent.youtube.com/save"},
             )
@@ -34,7 +29,7 @@ async def test_rgpd():
     assert resp.status_code == 200
     assert (
         len(
-            bs_parse(resp.text).find_all(
+            BeautifulSoup(resp.text, features="html.parser").find_all(
                 "form",
                 attrs={"method": "POST", "action": "https://consent.youtube.com/save"},
             )
@@ -45,51 +40,49 @@ async def test_rgpd():
 
 @pytest.mark.asyncio(loop_scope="module")
 async def test_rss_channel():
-    api = YoutubeRssApi()
+    api = YoutubeApi()
 
     feed = await api.get_channel_rss("UCVooVnzQxPSTXTMzSi1s6uw")
     assert feed.title == "Jonny Giger"
 
 
 @pytest.mark.asyncio(loop_scope="module")
 async def test_rss_playlist():
-    api = YoutubeRssApi()
+    api = YoutubeApi()
 
     feed = await api.get_playlist_rss("PLw-vK1_d04zZCal3yMX_T23h5nDJ2toTk")
     assert feed.title == "IMPOSSIBLE TRICKS OF RODNEY MULLEN"
 
 
 @pytest.mark.asyncio(loop_scope="module")
 async def test_metadata_channel():
-    api = YoutubeWebApi(AsyncClient(cookies=CookieJar()))
+    api = YoutubeApi(AsyncClient(cookies=CookieJar()))
 
     resp = await api.get_homepage("UCVooVnzQxPSTXTMzSi1s6uw")
-    meta = YoutubeMetadata.from_response(resp)
+    page = PageScrapper.from_response(resp)
+    meta = page.get_metadata()
     assert meta.title == "Jonny Giger"
     assert meta.channel_id == "UCVooVnzQxPSTXTMzSi1s6uw"
-    assert (
-        meta.url.geturl() == "https://www.youtube.com/channel/UCVooVnzQxPSTXTMzSi1s6uw"
-    )
+    assert meta.url == "https://www.youtube.com/channel/UCVooVnzQxPSTXTMzSi1s6uw"
     assert meta.avatar_url is not None
 
 
 @pytest.mark.asyncio(loop_scope="module")
 async def test_metadata_user():
-    api = YoutubeWebApi(AsyncClient(cookies=CookieJar()))
+    api = YoutubeApi(AsyncClient(cookies=CookieJar()))
 
     resp = await api.get_homepage("@jonnygiger")
-    meta = YoutubeMetadata.from_response(resp)
+    page = PageScrapper.from_response(resp)
+    meta = page.get_metadata()
     assert meta.title == "Jonny Giger"
     assert meta.channel_id == "UCVooVnzQxPSTXTMzSi1s6uw"
-    assert (
-        meta.url.geturl() == "https://www.youtube.com/channel/UCVooVnzQxPSTXTMzSi1s6uw"
-    )
+    assert meta.url == "https://www.youtube.com/channel/UCVooVnzQxPSTXTMzSi1s6uw"
     assert meta.avatar_url is not None
 
 
 @pytest.mark.asyncio(loop_scope="module")
 async def test_scrap_videos():
-    scrapper = VideoScrapper(YoutubeWebApi())
+    scrapper = VideoScrapper()
 
     page_iterator = scrapper.iter_videos("UCVooVnzQxPSTXTMzSi1s6uw")
     page1 = await anext(page_iterator)

yourss/async_utils.py

@@ -3,45 +3,40 @@
 
 from .youtube import (
     Feed,
-    YoutubeMetadata,
-    YoutubeRssApi,
-    YoutubeWebApi,
+    YoutubeApi,
     is_channel_id,
     is_playlist_id,
     is_user,
 )
+from .youtube.scrapper import PageScrapper
 
 
-async def _fetch_feed(
-    name: str, *, rss_api: YoutubeRssApi, web_api: YoutubeWebApi
-) -> Feed:
+async def _fetch_feed(name: str, *, api: YoutubeApi) -> Feed:
     if is_playlist_id(name):
-        return await rss_api.get_playlist_rss(name)
+        return await api.get_playlist_rss(name)
 
     # if given id is a name, get the channel id
     if is_user(name):
-        meta = YoutubeMetadata.from_response(await web_api.get_homepage(name))
+        page = PageScrapper.from_response(await api.get_homepage(name))
+        meta = page.get_metadata()
         name = meta.channel_id
 
     # check valid channel id
     if not is_channel_id(name):
         raise ValueError(f"Invalid channel id: {name}")
 
-    return await rss_api.get_channel_rss(name)
+    return await api.get_channel_rss(name)
 
 
 async def afetch_feeds(
-    names: List[str], *, rss_api: YoutubeRssApi, web_api: YoutubeWebApi
+    names: List[str], *, api: YoutubeApi
 ) -> Dict[str, Feed | BaseException]:
     return {
         name: task
         for name, task in zip(
             names,
             await asyncio.gather(
-                *[
-                    _fetch_feed(name, rss_api=rss_api, web_api=web_api)
-                    for name in names
-                ],
+                *[_fetch_feed(name, api=api) for name in names],
                 return_exceptions=True,
             ),
         )

yourss/routers/proxy.py

@@ -1,36 +1,30 @@
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, HTTPException
 from fastapi.responses import RedirectResponse
-from httpx import AsyncClient
 from pydantic import PositiveInt
 from starlette.status import HTTP_404_NOT_FOUND
 
 from ..youtube import (
-    YoutubeMetadata,
-    YoutubeRssApi,
-    YoutubeWebApi,
+    PageScrapper,
+    YoutubeApi,
     is_channel_id,
     is_playlist_id,
     is_user,
 )
 from .schema import ChannelId, Playlist_Id, UserId
-from .utils import force_https, get_youtube_web_client
+from .utils import force_https
 
 router = APIRouter()
 
 
 @router.get("/rss/{name}", response_class=RedirectResponse)
-async def rss_feed(
-    name: UserId | ChannelId | Playlist_Id,
-    yt_client: AsyncClient = Depends(get_youtube_web_client),
-):
-    api = YoutubeRssApi()
-    webapi = YoutubeWebApi(yt_client)
+async def rss_feed(name: UserId | ChannelId | Playlist_Id):
+    api = YoutubeApi()
 
     feed = None
     # if a user is provided, get the channel id
     if is_user(name):
-        homepage = await webapi.get_homepage(name)
-        meta = YoutubeMetadata.from_response(homepage)
+        homepage = PageScrapper.from_response(await api.get_homepage(name))
+        meta = homepage.get_metadata()
         name = meta.channel_id
 
     if is_channel_id(name):
@@ -46,13 +40,11 @@ async def rss_feed(
 
 
 @router.get("/avatar/{name}", response_class=RedirectResponse)
-async def avatar(
-    name: UserId | ChannelId, yt_client: AsyncClient = Depends(get_youtube_web_client)
-):
-    webapi = YoutubeWebApi(yt_client)
+async def avatar(name: UserId | ChannelId):
+    api = YoutubeApi()
 
-    homepage = await webapi.get_homepage(name)
-    meta = YoutubeMetadata.from_response(homepage)
+    homepage = PageScrapper.from_response(await api.get_homepage(name))
+    meta = homepage.get_metadata()
 
     if (url := meta.avatar_url) is None:
         raise HTTPException(
@@ -62,15 +54,13 @@ async def avatar(
 
 
 @router.get("/home/{name}", response_class=RedirectResponse)
-async def home(
-    name: UserId | ChannelId, yt_client: AsyncClient = Depends(get_youtube_web_client)
-):
-    webapi = YoutubeWebApi(yt_client)
+async def home(name: UserId | ChannelId):
+    api = YoutubeApi()
 
-    homepage = await webapi.get_homepage(name)
-    meta = YoutubeMetadata.from_response(homepage)
+    homepage = PageScrapper.from_response(await api.get_homepage(name))
+    meta = homepage.get_metadata()
 
-    return RedirectResponse(meta.url.geturl())
+    return RedirectResponse(meta.url)
 
 
 @router.get("/thumbnail/{video_id}", response_class=RedirectResponse)

yourss/routers/utils.py

@@ -1,15 +1,7 @@
-from http.cookiejar import CookieJar
-from typing import AsyncGenerator, Callable, List
+from typing import Callable, List
 
-from httpx import AsyncClient
 from starlette.templating import Jinja2Templates, _TemplateResponse
 
-cookiejar = CookieJar()
-
-
-async def get_youtube_web_client() -> AsyncGenerator[AsyncClient, None]:
-    yield AsyncClient(cookies=cookiejar)
-
 
 def force_https(url: str) -> str:
     assert isinstance(url, str)