RomanNum3ral
/
tubearchivist
mirror of https://github.com/tubearchivist/tubearchivist.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refac channel query building, better handle overwrites, #1047, #1048

This commit is contained in:
Simon 2025-09-03 15:06:30 +07:00
parent 2d19d85b75
commit 4adfea6e5c
No known key found for this signature in database
GPG Key ID: 2C15AA5E89985DD4
8 changed files with 120 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
backend/appsettings/src/reindex.py
View File

@ -500,11 +500,14 @@ class ChannelFullScan:
self.config = config
self.to_update = False
def scan(self):
def scan(self) -> None:
"""match local with remote"""
print(f"{self.channel_id}: start full scan")
all_local_videos = self._get_all_local()
all_remote_videos = self._get_all_remote()
all_remote_videos = get_last_channel_videos(
self.channel_id, self.config, limit=None
)
self.to_update = []
for video in all_local_videos:
video_id = video["youtube_id"]
@ -526,14 +529,6 @@ class ChannelFullScan:
self.update()
def _get_all_remote(self):
"""get all channel videos"""
all_remote_videos = get_last_channel_videos(
self.channel_id, self.config, limit=False
)
return all_remote_videos
def _get_all_local(self):
"""get all local indexed channel_videos"""
channel = YoutubeChannel(self.channel_id)

7
backend/channel/src/index.py
View File

@ -110,11 +110,6 @@ class YoutubeChannel(YouTubeItem):
"""get channel tabs"""
tabs = VideoTypeEnum.values_known()
config_cp = self.config.copy()
config_cp["subscriptions"] = {
"channel_size": 1,
"live_channel_size": 1,
"shorts_channel_size": 1,
}
tabs = []
for query_filter in VideoTypeEnum:
if query_filter == VideoTypeEnum.UNKNOWN:
@ -123,7 +118,7 @@ class YoutubeChannel(YouTubeItem):
videos = get_last_channel_videos(
channel_id=self.youtube_id,
config=config_cp,
limit=True,
limit=1,
query_filter=query_filter,
)
if videos:

182
backend/channel/src/remote_query.py
View File

@ -1,128 +1,130 @@
"""build queries for video extraction from channel subscriptions"""
from appsettings.src.config import AppConfigType
from download.src.yt_dlp_base import YtWrap
from video.src.constants import VideoTypeEnum
class VideoQueryBuilder:
"""Build queries for yt-dlp."""
"""
Build queries for yt-dlp.
limit:
- None: no limit
- bool: limit lookup from overwrite or config if True
- int: limit as int direct
"""
def __init__(self, config: dict, channel_overwrites: dict | None = None):
MAPPING = {
VideoTypeEnum.VIDEOS: {
"config_key": "channel_size",
"overwrite_key": "subscriptions_channel_size",
},
VideoTypeEnum.SHORTS: {
"config_key": "shorts_channel_size",
"overwrite_key": "subscriptions_shorts_channel_size",
},
VideoTypeEnum.STREAMS: {
"config_key": "live_channel_size",
"overwrite_key": "subscriptions_live_channel_size",
},
}
def __init__(
self,
config: AppConfigType,
channel_overwrites: dict | None = None,
limit: None | bool | int = True,
):
self.config = config
self.channel_overwrites = channel_overwrites or {}
self.limit = limit
def build_queries(
self,
video_type: VideoTypeEnum | list[VideoTypeEnum] | None,
limit: bool = True,
vid_types: list[VideoTypeEnum] = VideoTypeEnum.known(),
) -> list[tuple[VideoTypeEnum, int | None]]:
"""Build queries for all or specific video type."""
query_methods = {
VideoTypeEnum.VIDEOS: self.videos_query,
VideoTypeEnum.STREAMS: self.streams_query,
VideoTypeEnum.SHORTS: self.shorts_query,
}
"""build queries"""
queries: list[tuple[VideoTypeEnum, int | None]] = []
for vid_type in vid_types:
if vid_type not in self.MAPPING:
continue
if video_type and video_type != VideoTypeEnum.UNKNOWN:
# build query for specific type/s
if not isinstance(video_type, list):
video_type = [video_type]
queries = []
for video_type_item in video_type:
query_method = query_methods.get(video_type_item)
if not query_method:
continue
query = query_method(limit)
if query[1] != 0:
queries.append(query)
return queries
# Build and return queries for all video types
queries = []
for build_query in query_methods.values():
query = build_query(limit)
if query[1] != 0:
query = self.build_query_type(vid_type)
if query:
queries.append(query)
return queries
def videos_query(self, limit: bool) -> tuple[VideoTypeEnum, int | None]:
"""Build query for videos."""
return self._build_generic_query(
video_type=VideoTypeEnum.VIDEOS,
overwrite_key="subscriptions_channel_size",
config_key="channel_size",
limit=limit,
)
def streams_query(self, limit: bool) -> tuple[VideoTypeEnum, int | None]:
"""Build query for streams."""
return self._build_generic_query(
video_type=VideoTypeEnum.STREAMS,
overwrite_key="subscriptions_live_channel_size",
config_key="live_channel_size",
limit=limit,
)
def shorts_query(self, limit: bool) -> tuple[VideoTypeEnum, int | None]:
"""Build query for shorts."""
return self._build_generic_query(
video_type=VideoTypeEnum.SHORTS,
overwrite_key="subscriptions_shorts_channel_size",
config_key="shorts_channel_size",
limit=limit,
)
def _build_generic_query(
def build_query_type(
self,
video_type: VideoTypeEnum,
overwrite_key: str,
config_key: str,
limit: bool,
) -> tuple[VideoTypeEnum, int | None]:
"""Generic query for video page scraping."""
app_config_size = self.config["subscriptions"].get(config_key)
if not limit or app_config_size is None:
# treat None as unlimited
return (video_type, None)
vid_type: VideoTypeEnum,
) -> tuple[VideoTypeEnum, int | None] | None:
"""build query for vid_type"""
if self.limit is None:
return (vid_type, None)
if (
overwrite_key in self.channel_overwrites
and self.channel_overwrites[overwrite_key] is not None
):
overwrite = self.channel_overwrites[overwrite_key]
return (video_type, overwrite)
if isinstance(self.limit, bool):
if self.limit is False:
return (vid_type, None)
if app_config_size:
return (video_type, app_config_size)
overwrite_key = self.MAPPING[vid_type]["overwrite_key"]
overwrite = self.channel_overwrites.get(overwrite_key)
if overwrite == 0:
return None
return (video_type, 0)
if overwrite:
return (vid_type, overwrite)
config_key = self.MAPPING[vid_type]["config_key"]
app_config = self.config["subscriptions"].get(config_key)
if app_config == 0:
return None
if app_config:
return (vid_type, app_config) # type: ignore
return (vid_type, None)
if isinstance(self.limit, int):
return (vid_type, self.limit)
return (vid_type, None)
def get_last_channel_videos(
channel_id,
config,
limit=None,
query_filter=None,
channel_overwrites=None,
):
channel_id: str,
config: AppConfigType,
limit: None | bool | int = None,
query_filter: VideoTypeEnum | list[VideoTypeEnum] | None = None,
) -> list[dict]:
"""get a list of last videos from channel"""
query_handler = VideoQueryBuilder(config, channel_overwrites)
queries = query_handler.build_queries(query_filter)
last_videos = []
builder = VideoQueryBuilder(config, limit=limit)
queries = []
if query_filter is None:
queries = builder.build_queries()
elif isinstance(query_filter, list):
queries = builder.build_queries(vid_types=query_filter)
else:
query = builder.build_query_type(vid_type=query_filter)
if query:
queries.append(query)
last_videos: list[dict] = []
if not query:
return last_videos
for vid_type_enum, limit_amount in queries:
obs = {
obs: dict[str, bool | str] = {
"skip_download": True,
"extract_flat": True,
}
vid_type = vid_type_enum.value
if limit is not None:
obs.update({"playlist_items": f":{limit_amount}:1"})
obs["playlist_items"] = f":{limit_amount}:1"
url = f"https://www.youtube.com/channel/{channel_id}/{vid_type}"
channel_query, _ = YtWrap(obs, config).extract(url)

33
backend/channel/tests/test_src/test_remote_query.py
View File

@ -40,7 +40,7 @@ def overwrites():
def test_build_all_queries_with_limit(default_config, empty_overwrites):
"""default, empty overwrite"""
builder = VideoQueryBuilder(default_config, empty_overwrites)
result = builder.build_queries(None, limit=True)
result = builder.build_queries()
expected = [
(VideoTypeEnum.VIDEOS, 5),
(VideoTypeEnum.STREAMS, 3),
@ -51,8 +51,8 @@ def test_build_all_queries_with_limit(default_config, empty_overwrites):
def test_build_all_queries_without_limit(default_config, empty_overwrites):
"""limit disabled"""
builder = VideoQueryBuilder(default_config, empty_overwrites)
result = builder.build_queries(None, limit=False)
builder = VideoQueryBuilder(default_config, empty_overwrites, limit=False)
result = builder.build_queries()
expected = [
(VideoTypeEnum.VIDEOS, None),
(VideoTypeEnum.STREAMS, None),
@ -64,8 +64,8 @@ def test_build_all_queries_without_limit(default_config, empty_overwrites):
def test_build_specific_query(default_config, empty_overwrites):
"""single vid_type"""
builder = VideoQueryBuilder(default_config, empty_overwrites)
result = builder.build_queries(VideoTypeEnum.VIDEOS)
assert result == [(VideoTypeEnum.VIDEOS, 5)]
result = builder.build_query_type(VideoTypeEnum.VIDEOS)
assert result == (VideoTypeEnum.VIDEOS, 5)
def test_build_multiple_queries(default_config, empty_overwrites):
@ -77,21 +77,10 @@ def test_build_multiple_queries(default_config, empty_overwrites):
assert result == [(VideoTypeEnum.VIDEOS, 5), (VideoTypeEnum.SHORTS, 2)]
def test_build_unknown_queries(default_config, empty_overwrites):
"""vid_type unknown"""
builder = VideoQueryBuilder(default_config, empty_overwrites)
result = builder.build_queries(VideoTypeEnum.UNKNOWN)
assert result == [
(VideoTypeEnum.VIDEOS, 5),
(VideoTypeEnum.STREAMS, 3),
(VideoTypeEnum.SHORTS, 2),
]
def test_overwrite_applied(default_config, overwrites):
"""with overwrite from channel config"""
builder = VideoQueryBuilder(default_config, overwrites)
result = builder.build_queries(None, limit=True)
result = builder.build_queries()
expected = [
(VideoTypeEnum.VIDEOS, 10), # Overwritten
# STREAMS is overwritten to 0, should be excluded
@ -102,16 +91,16 @@ def test_overwrite_applied(default_config, overwrites):
def test_no_limit_ignores_config_and_overwrites(default_config, overwrites):
"""no limit single vid_type"""
builder = VideoQueryBuilder(default_config, overwrites)
result = builder.build_queries([VideoTypeEnum.STREAMS], limit=False)
builder = VideoQueryBuilder(default_config, overwrites, limit=False)
result = builder.build_queries([VideoTypeEnum.STREAMS])
assert result == [(VideoTypeEnum.STREAMS, None)]
def test_zero_query_not_included(default_config):
"""overwrite to zero to disable"""
overwrites = {"subscriptions_live_channel_size": 0}
builder = VideoQueryBuilder(default_config, overwrites)
result = builder.build_queries([VideoTypeEnum.STREAMS], limit=True)
builder = VideoQueryBuilder(default_config, overwrites, limit=True)
result = builder.build_queries([VideoTypeEnum.STREAMS])
assert not result # Should be skipped due to 0
@ -124,5 +113,5 @@ def test_invalid_video_type_is_ignored(default_config):
INVALID = "invalid"
result = builder.build_queries([FakeEnum.INVALID], limit=True)
result = builder.build_queries([FakeEnum.INVALID])
assert not result

2
backend/download/src/queue.py
View File

@ -191,7 +191,7 @@ class PendingList(PendingIndex):
return to_add
def _parse_channel(self, entry):
def _parse_channel(self, entry) -> None:
"""parse channel"""
url = entry["url"]
vid_type = entry["vid_type"]

8
backend/download/src/subscriptions.py
View File

@ -63,15 +63,15 @@ class ChannelSubscription:
queries = VideoQueryBuilder(
config=self.config,
channel_overwrites=channel.get("channel_overwrites", {}),
).build_queries(video_type=enums)
).build_queries(vid_types=enums)
for query in queries:
for vid_type, limit in queries:
all_channel_urls.append(
ParsedURLType(
type="channel",
url=channel["channel_id"],
vid_type=query[0],
limit=query[1],
vid_type=vid_type,
limit=limit,
)
)

1
backend/download/views.py
View File

@ -124,6 +124,7 @@ class DownloadApiListView(ApiBaseView):
pending = [i["youtube_id"] for i in to_add if i["status"] == "pending"]
url_str = " ".join(pending)
print(f"url_str: {url_str}")
task = extrac_dl.delay(
url_str, auto_start=auto_start, flat=flat, force=force
)

5
backend/video/src/constants.py
View File

@ -24,6 +24,11 @@ class VideoTypeEnum(enum.Enum):
"""values known"""
return [i.value for i in cls if i.value != "unknown"]
@classmethod
def known(cls):
"""known members"""
return [i for i in cls if i.value != "unknown"]
class SortEnum(enum.Enum):
"""all sort by options"""