implement bulk add subscriptions in appsettings

backend/common/src/helper.py

@@ -292,6 +292,50 @@ def get_channel_overwrites() -> dict[str, dict[str, Any]]:
     return overwrites
 
 
+def get_channels(
+    subscribed_only: bool, source: list[str] | None = None
+) -> list[dict]:
+    """get a list of all channels"""
+    data = {
+        "sort": [{"channel_name.keyword": {"order": "asc"}}],
+    }
+    if subscribed_only:
+        query = {"term": {"channel_subscribed": {"value": True}}}
+    else:
+        query = {"match_all": {}}
+
+    data["query"] = query  # type: ignore
+
+    if source:
+        data["_source"] = source  # type: ignore
+
+    all_channels = IndexPaginate("ta_channel", data).get_results()
+
+    return all_channels
+
+
+def get_playlists(
+    subscribed_only: bool, source: list[str] | None = None
+) -> list[dict]:
+    """get list of playlists"""
+
+    data = {
+        "sort": [{"playlist_channel.keyword": {"order": "desc"}}],
+    }
+
+    must_list = [{"term": {"playlist_active": {"value": True}}}]
+    if subscribed_only:
+        must_list.append({"term": {"playlist_subscribed": {"value": True}}})
+
+    data = {"query": {"bool": {"must": must_list}}}  # type: ignore
+    if source:
+        data["_source"] = source  # type: ignore
+
+    all_playlists = IndexPaginate("ta_playlist", data).get_results()
+
+    return all_playlists
+
+
 def calc_is_watched(duration: float, position: float) -> bool:
     """considered watched based on duration position"""
 

backend/common/src/urlparser.py

@@ -4,6 +4,7 @@ Functionality:
 - identify vid_type if possible
 """
 
+from typing import Literal, NotRequired, TypedDict
 from urllib.parse import parse_qs, urlparse
 
 from common.src.ta_redis import RedisArchivist
@@ -11,19 +12,28 @@ from download.src.yt_dlp_base import YtWrap
 from video.src.constants import VideoTypeEnum
 
 
+class ParsedURLType(TypedDict):
+    """represents single parsed url"""
+
+    type: Literal["video", "channel", "playlist"]
+    url: str
+    vid_type: VideoTypeEnum
+    limit: NotRequired[int | None]
+
+
 class Parser:
     """
     take a multi line string and detect valid youtube ids
     channel handle lookup is cached, can be disabled for unittests
     """
 
-    def __init__(self, url_str, use_cache=True):
+    def __init__(self, url_str: str, use_cache: bool = True):
         self.url_list = [i.strip() for i in url_str.split()]
         self.use_cache = use_cache
 
-    def parse(self):
+    def parse(self) -> list[ParsedURLType]:
         """parse the list"""
-        ids = []
+        ids: list[ParsedURLType] = []
         for url in self.url_list:
             parsed = urlparse(url)
             if parsed.netloc: