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

Merge branch 'feat-flat-queue' into testing

This commit is contained in:
Simon 2025-07-10 12:29:15 +07:00
parent ae40df1b6b 29be11cf75
commit 25c9fd99b1
19 changed files with 557 additions and 272 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -510,12 +510,14 @@ class ChannelFullScan:
self.to_update = []
for video in all_local_videos:
video_id = video["youtube_id"]
remote_match = [i for i in all_remote_videos if i[0] == video_id]
remote_match = [
i for i in all_remote_videos if i["id"] == video_id
]
if not remote_match:
print(f"{video_id}: no remote match found")
continue
expected_type = remote_match[0][-1]
expected_type = remote_match[0]["vid_type"]
if video["vid_type"] != expected_type:
self.to_update.append(
{

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

@ -301,7 +301,7 @@ class YoutubeChannel(YouTubeItem):
+ "/playlists?view=1&sort=dd&shelf_id=0"
)
obs = {"skip_download": True, "extract_flat": True}
playlists = YtWrap(obs, self.config).extract(url)
playlists, _ = YtWrap(obs, self.config).extract(url)
if not playlists:
self.all_playlists = []
return

2
backend/common/src/es_connect.py
View File

@ -56,7 +56,7 @@ class ElasticWrap:
return response.json(), response.status_code
def post(
self, data: bool | dict = False, ndjson: bool = False
self, data: bool | dict | str = False, ndjson: bool = False
) -> tuple[dict, int]:
"""post data to es"""

18
backend/common/src/helper.py
View File

@ -103,8 +103,11 @@ def requests_headers() -> dict[str, str]:
return {"User-Agent": template}
def date_parser(timestamp: int | str) -> str:
def date_parser(timestamp: int | str | None) -> str | None:
"""return formatted date string"""
if timestamp is None:
return None
if isinstance(timestamp, int):
date_obj = datetime.fromtimestamp(timestamp, tz=timezone.utc)
elif isinstance(timestamp, str):
@ -152,9 +155,13 @@ def is_shorts(youtube_id: str) -> bool:
"""check if youtube_id is a shorts video, bot not it it's not a shorts"""
shorts_url = f"https://www.youtube.com/shorts/{youtube_id}"
cookies = {"SOCS": "CAI"}
response = requests.head(
shorts_url, cookies=cookies, headers=requests_headers(), timeout=10
)
try:
response = requests.head(
shorts_url, cookies=cookies, headers=requests_headers(), timeout=10
)
except requests.exceptions.RequestException:
# assume video on error
return False
return response.status_code == 200
@ -184,11 +191,12 @@ def get_duration_sec(file_path: str) -> int:
return duration_sec
def get_duration_str(seconds: int) -> str:
def get_duration_str(seconds: int | float) -> str:
"""Return a human-readable duration string from seconds."""
if not seconds:
return "NA"
seconds = int(seconds)
units = [("y", 31536000), ("d", 86400), ("h", 3600), ("m", 60), ("s", 1)]
duration_parts = []

5
backend/common/src/index_generic.py
View File

@ -26,6 +26,7 @@ class YouTubeItem:
self.youtube_id = youtube_id
self.es_path = f"{self.index_name}/_doc/{youtube_id}"
self.config = AppConfig().config
self.error = None
self.youtube_meta = False
self.json_data = False
@ -43,7 +44,9 @@ class YouTubeItem:
obs_request["extractor_args"] = {"youtube": {"lang": langs_list}}
url = self.build_yt_url()
self.youtube_meta = YtWrap(obs_request, self.config).extract(url)
self.youtube_meta, self.error = YtWrap(
obs_request, self.config
).extract(url)
def get_from_es(self):
"""get indexed data from elastic search"""

14
backend/common/src/search_processor.py
View File

@ -165,15 +165,17 @@ class SearchProcess:
def _process_download(self, download_dict):
"""run on single download item"""
video_id = download_dict["youtube_id"]
cache_root = EnvironmentSettings().get_cache_root()
vid_thumb_url = ThumbManager(video_id).vid_thumb_path()
published = date_parser(download_dict["published"])
vid_thumb_url = None
if download_dict.get("vid_thumb_url"):
video_id = download_dict["youtube_id"]
cache_root = EnvironmentSettings().get_cache_root()
relative_path = ThumbManager(video_id).vid_thumb_path()
vid_thumb_url = f"{cache_root}/{relative_path}"
download_dict.update(
{
"vid_thumb_url": f"{cache_root}/{vid_thumb_url}",
"published": published,
"vid_thumb_url": vid_thumb_url,
"published": date_parser(download_dict["published"]),
}
)
return dict(sorted(download_dict.items()))

4
backend/common/src/urlparser.py
View File

@ -124,9 +124,9 @@ class Parser:
"extract_flat": True,
"playlistend": 0,
}
url_info = YtWrap(obs_request).extract(url)
url_info, error = YtWrap(obs_request).extract(url)
if not url_info:
raise ValueError(f"failed to retrieve content from URL: {url}")
raise ValueError(f"failed to retrieve URL: {error}")
channel_id = url_info.get("channel_id", False)
if channel_id:

9
backend/download/serializers.py
View File

@ -15,11 +15,11 @@ class DownloadItemSerializer(serializers.Serializer):
channel_indexed = serializers.BooleanField()
channel_name = serializers.CharField()
duration = serializers.CharField()
published = serializers.CharField()
published = serializers.CharField(allow_null=True)
status = serializers.ChoiceField(choices=["pending", "ignore"])
timestamp = serializers.IntegerField()
timestamp = serializers.IntegerField(allow_null=True)
title = serializers.CharField()
vid_thumb_url = serializers.CharField()
vid_thumb_url = serializers.CharField(allow_null=True)
vid_type = serializers.ChoiceField(choices=VideoTypeEnum.values())
youtube_id = serializers.CharField()
message = serializers.CharField(required=False)
@ -47,6 +47,8 @@ class DownloadListQuerySerializer(
)
channel = serializers.CharField(required=False, help_text="channel ID")
page = serializers.IntegerField(required=False)
q = serializers.CharField(required=False, help_text="Search Query")
error = serializers.BooleanField(required=False, allow_null=True)
class DownloadListQueueDeleteQuerySerializer(serializers.Serializer):
@ -76,6 +78,7 @@ class AddToDownloadQuerySerializer(serializers.Serializer):
"""add to queue query serializer"""
autostart = serializers.BooleanField(required=False)
flat = serializers.BooleanField(required=False)
class BulkUpdateDowloadQuerySerializer(serializers.Serializer):

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

@ -4,16 +4,18 @@ Functionality:
- linked with ta_dowload index
"""
import json
from datetime import datetime
from appsettings.src.config import AppConfig
from channel.src.index import YoutubeChannel
from common.src.es_connect import ElasticWrap, IndexPaginate
from common.src.helper import get_duration_str, is_shorts, rand_sleep
from download.src.subscriptions import ChannelSubscription
from download.src.thumbnails import ThumbManager
from download.src.yt_dlp_base import YtWrap
from playlist.src.index import YoutubePlaylist
from video.src.constants import VideoTypeEnum
from video.src.index import YoutubeVideo
class PendingIndex:
@ -195,22 +197,30 @@ class PendingList(PendingIndex):
"check_formats": None,
}
def __init__(self, youtube_ids=False, task=False):
def __init__(
self, youtube_ids=False, task=False, auto_start=False, flat=False
):
super().__init__()
self.config = AppConfig().config
self.youtube_ids = youtube_ids
self.task = task
self.auto_start = auto_start
self.flat = flat
self.to_skip = False
self.missing_videos = False
def parse_url_list(self, auto_start=False):
"""extract youtube ids from list"""
self.missing_videos = []
def parse_url_list(self):
"""extract youtube ids from list"""
self.get_download()
self.get_indexed()
self.get_channels()
total = len(self.youtube_ids)
for idx, entry in enumerate(self.youtube_ids):
self._process_entry(entry, auto_start=auto_start)
self._process_entry(entry, idx, total)
if self.task and self.task.is_stopped():
break
rand_sleep(self.config)
if not self.task:
continue
@ -219,102 +229,312 @@ class PendingList(PendingIndex):
progress=(idx + 1) / total,
)
def _process_entry(self, entry, auto_start=False):
def _process_entry(self, entry: dict, idx_url: int, total_url: int):
"""process single entry from url list"""
vid_type = self._get_vid_type(entry)
if entry["type"] == "video":
self._add_video(entry["url"], vid_type, auto_start=auto_start)
self._add_video(
entry["url"], entry["vid_type"], idx_url, total_url
)
elif entry["type"] == "channel":
self._parse_channel(entry["url"], vid_type)
self._parse_channel(entry["url"], entry["vid_type"])
elif entry["type"] == "playlist":
self._parse_playlist(entry["url"])
else:
raise ValueError(f"invalid url_type: {entry}")
@staticmethod
def _get_vid_type(entry):
"""add vid type enum if available"""
vid_type_str = entry.get("vid_type")
if not vid_type_str:
return VideoTypeEnum.UNKNOWN
return VideoTypeEnum(vid_type_str)
def _add_video(self, url, vid_type, auto_start=False):
def _add_video(self, url, vid_type, idx, total):
"""add video to list"""
if auto_start and url in set(
if self.auto_start and url in set(
i["youtube_id"] for i in self.all_pending
):
PendingInteract(youtube_id=url, status="priority").update_status()
return
if url not in self.missing_videos and url not in self.to_skip:
self.missing_videos.append((url, vid_type))
else:
if url in self.missing_videos or url in self.to_skip:
print(f"{url}: skipped adding already indexed video to download.")
else:
to_add = self._parse_video(
url,
vid_type,
url_type="video",
idx=idx,
total=total,
)
if to_add:
self.missing_videos.append(to_add)
def _parse_channel(self, url, vid_type):
"""add all videos of channel to list"""
"""parse channel"""
query_filter = getattr(VideoTypeEnum, vid_type.upper())
video_results = ChannelSubscription().get_last_youtube_videos(
url, limit=False, query_filter=vid_type
url, limit=False, query_filter=query_filter
)
for video_id, _, vid_type in video_results:
self._add_video(video_id, vid_type)
def _parse_playlist(self, url):
"""add all videos of playlist to list"""
playlist = YoutubePlaylist(url)
is_active = playlist.update_playlist()
if not is_active:
message = f"{playlist.youtube_id}: failed to extract metadata"
print(message)
raise ValueError(message)
entries = playlist.json_data["playlist_entries"]
to_add = [i["youtube_id"] for i in entries if not i["downloaded"]]
if not to_add:
if not video_results:
print(f"{url}: no videos to add from channel, skipping")
return
for video_id in to_add:
# match vid_type later
self._add_video(video_id, VideoTypeEnum.UNKNOWN)
channel_handler = YoutubeChannel(url)
channel_handler.build_json(upload=False)
if not channel_handler.json_data:
print(f"{url}: channel metadata extraction failed, skipping")
return
def add_to_pending(self, status="pending", auto_start=False):
"""add missing videos to pending list"""
self.get_channels()
total = len(video_results)
for idx, video_data in enumerate(video_results):
video_id = video_data["id"]
if video_id in self.to_skip:
continue
total = len(self.missing_videos)
videos_added = []
for idx, (youtube_id, vid_type) in enumerate(self.missing_videos):
if self.task and self.task.is_stopped():
break
print(f"{youtube_id}: [{idx + 1}/{total}]: add to queue")
self._notify_add(idx, total)
video_details = self.get_youtube_details(youtube_id, vid_type)
if not video_details:
rand_sleep(self.config)
if self.flat:
if not video_data.get("channel"):
channel_name = channel_handler.json_data["channel_name"]
video_data["channel"] = channel_name
if not video_data.get("channel_id"):
channel_id = channel_handler.json_data["channel_id"]
video_data["channel_id"] = channel_id
to_add = self._parse_entry(
youtube_id=video_id,
video_data=video_data,
url_type="channel",
idx=idx,
total=total,
)
else:
to_add = self._parse_video(
video_id,
vid_type,
url_type="channel",
idx=idx,
total=total,
)
if to_add:
self.missing_videos.append(to_add)
def _parse_playlist(self, url):
"""fast parse playlist"""
playlist = YoutubePlaylist(url)
playlist.get_from_youtube()
if not playlist.youtube_meta:
print(f"{url}: playlist metadata extraction failed, skipping")
return
video_results = playlist.youtube_meta["entries"]
total = len(video_results)
for idx, video_data in enumerate(video_results):
video_id = video_data["id"]
if video_id in self.to_skip:
continue
video_details.update(
if self.task and self.task.is_stopped():
break
if self.flat:
if not video_data.get("channel"):
video_data["channel"] = playlist.youtube_meta["channel"]
if not video_data.get("channel_id"):
channel_id = playlist.youtube_meta["channel_id"]
video_data["channel_id"] = channel_id
to_add = self._parse_entry(
video_id,
video_data,
url_type="playlist",
idx=idx,
total=total,
)
else:
to_add = self._parse_video(
video_id,
vid_type=None,
url_type="playlist",
idx=idx,
total=total,
)
if to_add:
self.missing_videos.append(to_add)
def _parse_video(self, url, vid_type, url_type, idx, total):
"""parse video"""
video = YoutubeVideo(youtube_id=url)
video.get_from_youtube()
if not video.youtube_meta:
print(f"{url}: video metadata extraction failed, skipping")
if self.task:
self.task.send_progress(
message_lines=[
"Video extraction failed.",
f"{video.error}",
],
level="error",
)
return None
video.youtube_meta["vid_type"] = vid_type
to_add = self._parse_entry(
youtube_id=url,
video_data=video.youtube_meta,
url_type=url_type,
idx=idx,
total=total,
)
ThumbManager(item_id=url).download_video_thumb(to_add["vid_thumb_url"])
rand_sleep(self.config)
return to_add
def _parse_entry(
self,
youtube_id: str,
video_data: dict,
url_type: str,
idx: int,
total: int,
) -> dict | None:
"""parse entry"""
if video_data.get("id") != youtube_id:
# skip premium videos with different id or redirects
print(f"{youtube_id}: skipping redirect, id not matching")
return None
if video_data.get("live_status") in ["is_upcoming", "is_live"]:
print(f"{youtube_id}: skip is_upcoming or is_live")
return None
to_add = {
"youtube_id": video_data["id"],
"title": video_data["title"],
"vid_thumb_url": self.__extract_thumb(video_data),
"duration": get_duration_str(video_data.get("duration", 0)),
"published": self.__extract_published(video_data),
"timestamp": int(datetime.now().timestamp()),
"vid_type": self.__extract_vid_type(video_data),
"channel_name": video_data["channel"],
"channel_id": video_data["channel_id"],
"channel_indexed": video_data["channel_id"] in self.all_channels,
}
self._notify_progress(url_type, video_data["title"], idx, total)
return to_add
def __extract_thumb(self, video_data) -> str | None:
"""extract thumb"""
if "thumbnail" in video_data:
return video_data["thumbnail"]
return None
def __extract_published(self, video_data) -> str | int | None:
"""build published date or timestamp"""
timestamp = video_data.get("timestamp")
if timestamp:
return timestamp
upload_date = video_data.get("upload_date")
if not upload_date:
return None
upload_date_time = datetime.strptime(upload_date, "%Y%m%d")
published = upload_date_time.strftime("%Y-%m-%d")
return published
def __extract_vid_type(self, video_data) -> str:
"""build vid type"""
if "vid_type" in video_data:
return video_data["vid_type"]
if video_data.get("live_status") == "was_live":
return VideoTypeEnum.STREAMS.value
if video_data.get("width", 0) > video_data.get("height", 0):
return VideoTypeEnum.VIDEOS.value
duration = video_data.get("duration")
if duration and isinstance(duration, int):
if duration > 3 * 60:
return VideoTypeEnum.VIDEOS.value
if is_shorts(video_data["id"]):
return VideoTypeEnum.SHORTS.value
return VideoTypeEnum.VIDEOS.value
def add_to_pending(self, status="pending") -> int:
"""add missing videos to pending list"""
total = len(self.missing_videos)
if not self.missing_videos:
self._notify_empty()
return 0
self._notify_start(total)
bulk_list = []
for video_entry in self.missing_videos:
video_entry.update(
{
"status": status,
"auto_start": auto_start,
"auto_start": self.auto_start,
}
)
video_id = video_entry["youtube_id"]
action = {"index": {"_index": "ta_download", "_id": video_id}}
bulk_list.append(json.dumps(action))
bulk_list.append(json.dumps(video_entry))
url = video_details["vid_thumb_url"]
ThumbManager(youtube_id).download_video_thumb(url)
es_url = f"ta_download/_doc/{youtube_id}"
_, _ = ElasticWrap(es_url).put(video_details)
videos_added.append(youtube_id)
# add last newline
bulk_list.append("\n")
query_str = "\n".join(bulk_list)
_, status_code = ElasticWrap("_bulk").post(query_str, ndjson=True)
if status_code != 200:
self._notify_fail(status_code)
else:
self._notify_done(total)
if idx != total:
rand_sleep(self.config)
return len(self.missing_videos)
return videos_added
def _notify_progress(self, url_type, name, idx, total):
"""notify extraction progress"""
if not self.task:
return
def _notify_add(self, idx, total):
if self.flat:
second_line = f"Bulk processing {total} items."
else:
second_line = f"Processing video {idx + 1}/{total}."
self.task.send_progress(
message_lines=[
f"Extracting '{name}' from {url_type.title()}.",
second_line,
],
progress=(idx + 1) / total,
)
def _notify_empty(self):
"""notify nothing to add"""
if not self.task:
return
self.task.send_progress(
message_lines=[
"Extracting videos completed.",
"No new videos found to add.",
]
)
def _notify_start(self, total):
"""send notification for adding videos to download queue"""
if not self.task:
return
@ -322,82 +542,31 @@ class PendingList(PendingIndex):
self.task.send_progress(
message_lines=[
"Adding new videos to download queue.",
f"Extracting items {idx + 1}/{total}",
],
progress=(idx + 1) / total,
f"Bulk adding {total} videos",
]
)
def get_youtube_details(self, youtube_id, vid_type=VideoTypeEnum.VIDEOS):
"""get details from youtubedl for single pending video"""
vid = YtWrap(self.yt_obs, self.config).extract(youtube_id)
if not vid:
return False
def _notify_done(self, total):
"""send done notification"""
if not self.task:
return
if vid.get("id") != youtube_id:
# skip premium videos with different id
print(f"{youtube_id}: skipping premium video, id not matching")
return False
# stop if video is streaming live now
if vid["live_status"] in ["is_upcoming", "is_live"]:
print(f"{youtube_id}: skip is_upcoming or is_live")
return False
self.task.send_progress(
message_lines=[
"Adding new videos to the queue completed.",
f"Added {total} videos.",
]
)
if vid["live_status"] == "was_live":
vid_type = VideoTypeEnum.STREAMS
else:
if self._check_shorts(vid):
vid_type = VideoTypeEnum.SHORTS
else:
vid_type = VideoTypeEnum.VIDEOS
def _notify_fail(self, status_code):
"""failed to add"""
if not self.task:
return
if not vid.get("channel"):
print(f"{youtube_id}: skip video not part of channel")
return False
return self._parse_youtube_details(vid, vid_type)
@staticmethod
def _check_shorts(vid):
"""check if vid is shorts video"""
if vid["width"] > vid["height"]:
return False
duration = vid.get("duration")
if duration and isinstance(duration, int):
if duration > 3 * 60:
return False
return is_shorts(vid["id"])
def _parse_youtube_details(self, vid, vid_type=VideoTypeEnum.VIDEOS):
"""parse response"""
vid_id = vid.get("id")
# build dict
youtube_details = {
"youtube_id": vid_id,
"channel_name": vid["channel"],
"vid_thumb_url": vid["thumbnail"],
"title": vid["title"],
"channel_id": vid["channel_id"],
"duration": get_duration_str(vid["duration"]),
"published": self._build_published(vid),
"timestamp": int(datetime.now().timestamp()),
"vid_type": vid_type.value,
"channel_indexed": vid["channel_id"] in self.all_channels,
}
return youtube_details
@staticmethod
def _build_published(vid):
"""build published date or timestamp"""
timestamp = vid["timestamp"]
if timestamp:
return timestamp
upload_date = vid["upload_date"]
upload_date_time = datetime.strptime(upload_date, "%Y%m%d")
published = upload_date_time.strftime("%Y-%m-%d")
return published
self.task.send_progress(
message_lines=[
"Adding extracted videos failed.",
f"Status code: {status_code}",
],
level="error",
)

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

@ -61,16 +61,13 @@ class ChannelSubscription:
obs["playlistend"] = limit_amount
url = f"https://www.youtube.com/channel/{channel_id}/{vid_type}"
channel_query = YtWrap(obs, self.config).extract(url)
channel_query, _ = YtWrap(obs, self.config).extract(url)
if not channel_query:
continue
last_videos.extend(
[
(i["id"], i["title"], vid_type)
for i in channel_query["entries"]
]
)
for entry in channel_query["entries"]:
entry["vid_type"] = vid_type
last_videos.append(entry)
return last_videos
@ -93,7 +90,9 @@ class ChannelSubscription:
if last_videos:
ids_to_add = is_missing([i[0] for i in last_videos])
for video_id, _, vid_type in last_videos:
for video_entry in last_videos:
video_id = video_entry["id"]
vid_type = video_entry["vid_type"]
if video_id in ids_to_add:
missing_videos.append((video_id, vid_type))

29
backend/download/src/yt_dlp_base.py
View File

@ -80,30 +80,33 @@ class YtWrap:
return True, True
def extract(self, url):
"""make extract request"""
def extract(self, url) -> tuple[dict | None, str | None]:
"""
make extract request
returns response, error
"""
with yt_dlp.YoutubeDL(self.obs) as ydl:
try:
response = ydl.extract_info(url)
except cookiejar.LoadError as err:
print(f"cookie file is invalid: {err}")
return False
return None, str(err)
except yt_dlp.utils.ExtractorError as err:
print(f"{url}: failed to extract: {err}, continue...")
return False
return None, str(err)
except yt_dlp.utils.DownloadError as err:
if "This channel does not have a" in str(err):
return False
return None, None
print(f"{url}: failed to get info from youtube: {err}")
if "Temporary failure in name resolution" in str(err):
raise ConnectionError("lost the internet, abort!") from err
return False
return None, str(err)
self._validate_cookie()
return response
return response, None
def _validate_cookie(self):
"""check cookie and write it back for next use"""
@ -146,7 +149,7 @@ class CookieHandler:
AppConfig().update_config({"downloads": {"cookie_import": False}})
print("[cookie]: revoked")
def validate(self):
def validate(self) -> bool:
"""validate cookie using the liked videos playlist"""
validation = RedisArchivist().get_message_dict("cookie:valid")
if validation:
@ -159,8 +162,8 @@ class CookieHandler:
"extract_flat": True,
}
validator = YtWrap(obs_request, self.config)
response = bool(validator.extract("LL"))
self.store_validation(response)
response, error = validator.extract("LL")
self.store_validation(bool(response))
# update in redis to avoid expiring
modified = validator.obs["cookiefile"].getvalue().strip("\x00")
@ -173,15 +176,15 @@ class CookieHandler:
"status": "message:download",
"level": "error",
"title": "Cookie validation failed, exiting...",
"message": "",
"message": error,
}
RedisArchivist().set_message(
"message:download", mess_dict, expire=4
)
print("[cookie]: validation failed, exiting...")
print(f"[cookie]: validation success: {response}")
return response
print(f"[cookie]: validation success: {bool(response)}")
return bool(response)
@staticmethod
def store_validation(response):

15
backend/download/views.py
View File

@ -73,6 +73,16 @@ class DownloadApiListView(ApiBaseView):
{"term": {"vid_type": {"value": vid_type_filter}}}
)
search_query = validated_data.get("q")
if search_query:
must_list.append({"match_phrase_prefix": {"title": search_query}})
if validated_data.get("error") is not None:
operator = "must" if validated_data["error"] else "must_not"
must_list.append(
{"bool": {operator: [{"exists": {"field": "message"}}]}}
)
self.data["query"] = {"bool": {"must": must_list}}
self.get_document_list(request)
@ -107,12 +117,13 @@ class DownloadApiListView(ApiBaseView):
validated_query = query_serializer.validated_data
auto_start = validated_query.get("autostart")
print(f"auto_start: {auto_start}")
flat = validated_query.get("flat", False)
print(f"auto_start: {auto_start}, flat: {flat}")
to_add = validated_data["data"]
pending = [i["youtube_id"] for i in to_add if i["status"] == "pending"]
url_str = " ".join(pending)
task = extrac_dl.delay(url_str, auto_start=auto_start)
task = extrac_dl.delay(url_str, auto_start=auto_start, flat=flat)
message = {
"message": "add to queue task started",

26
backend/task/tasks.py
View File

@ -39,10 +39,10 @@ class BaseTask(Task):
RedisArchivist().set_message(key, message, expire=20)
def on_success(self, retval, task_id, args, kwargs):
"""callback task completed successfully"""
"""callback task completed"""
print(f"{task_id} success callback")
message, key = self._build_message()
message.update({"messages": ["Task completed successfully"]})
message.update({"messages": ["Task completed"]})
RedisArchivist().set_message(key, message, expire=5)
def before_start(self, task_id, args, kwargs):
@ -58,9 +58,11 @@ class BaseTask(Task):
task_title = TASK_CONFIG.get(self.name).get("title")
Notifications(self.name).send(task_id, task_title)
def send_progress(self, message_lines, progress=False, title=False):
def send_progress(
self, message_lines, progress=False, title=False, level="info"
):
"""send progress message"""
message, key = self._build_message()
message, key = self._build_message(level=level)
message.update(
{
"messages": message_lines,
@ -147,7 +149,9 @@ def download_pending(self, auto_only=False):
@shared_task(name="extract_download", bind=True, base=BaseTask)
def extrac_dl(self, youtube_ids, auto_start=False, status="pending"):
def extrac_dl(
self, youtube_ids, auto_start=False, flat=False, status="pending"
):
"""parse list passed and add to pending"""
TaskManager().init(self)
if isinstance(youtube_ids, str):
@ -155,17 +159,17 @@ def extrac_dl(self, youtube_ids, auto_start=False, status="pending"):
else:
to_add = youtube_ids
pending_handler = PendingList(youtube_ids=to_add, task=self)
pending_handler.parse_url_list(auto_start=auto_start)
videos_added = pending_handler.add_to_pending(
status=status, auto_start=auto_start
pending_handler = PendingList(
youtube_ids=to_add, task=self, auto_start=auto_start, flat=flat
)
pending_handler.parse_url_list()
videos_added = pending_handler.add_to_pending(status=status)
if auto_start:
download_pending.delay(auto_only=True)
if videos_added:
return f"added {len(videos_added)} Videos to Queue"
return f"added {videos_added} Videos to Queue"
return None
@ -259,7 +263,7 @@ def rescan_filesystem(self):
handler = Scanner(task=self)
handler.scan()
handler.apply()
ThumbValidator(task=self).validate()
thumbnail_check.delay()
@shared_task(bind=True, name="thumbnail_check", base=BaseTask)

4
backend/video/src/comments.py
View File

@ -79,7 +79,9 @@ class Comments:
def get_yt_comments(self):
"""get comments from youtube"""
yt_obs = self.build_yt_obs()
info_json = YtWrap(yt_obs, config=self.config).extract(self.youtube_id)
info_json, _ = YtWrap(yt_obs, config=self.config).extract(
self.youtube_id
)
if not info_json:
return False, False

4
backend/video/src/index.py
View File

@ -14,6 +14,7 @@ from common.src.es_connect import ElasticWrap
from common.src.helper import get_duration_sec, get_duration_str, randomizor
from common.src.index_generic import YouTubeItem
from django.conf import settings
from download.src.thumbnails import ThumbManager
from playlist.src import index as ta_playlist
from ryd_client import ryd_client
from user.src.user_config import UserConfig
@ -409,5 +410,8 @@ def index_new_video(youtube_id, video_type=VideoTypeEnum.VIDEOS):
raise ValueError("failed to get metadata for " + youtube_id)
video.check_subtitles()
url = video.json_data["vid_thumb_url"]
ThumbManager(item_id=video.youtube_id).download_video_thumb(url=url)
video.upload_to_es()
return video.json_data

12
frontend/src/api/actions/updateDownloadQueue.ts
View File

@ -1,6 +1,6 @@
import APIClient from '../../functions/APIClient';
const updateDownloadQueue = async (youtubeIdStrings: string, autostart: boolean) => {
const updateDownloadQueue = async (youtubeIdStrings: string, autostart: boolean, flat: boolean) => {
const urls = [];
const containsMultiple = youtubeIdStrings.includes('\n');
@ -16,12 +16,12 @@ const updateDownloadQueue = async (youtubeIdStrings: string, autostart: boolean)
urls.push({ youtube_id: youtubeIdStrings, status: 'pending' });
}
let params = '';
if (autostart) {
params = '?autostart=true';
}
const searchParams = new URLSearchParams();
if (autostart) searchParams.append('autostart', 'true');
if (flat) searchParams.append('flat', 'true');
const endpoint = `/api/download/${searchParams.toString() ? `?${searchParams.toString()}` : ''}`;
return APIClient(`/api/download/${params}`, {
return APIClient(endpoint, {
method: 'POST',
body: { data: [...urls] },
});

4
frontend/src/api/loader/loadDownloadQueue.ts
View File

@ -5,13 +5,17 @@ const loadDownloadQueue = async (
page: number,
channelId: string | null,
vid_type: string | null,
errorFilterFromUrl: string | null,
showIgnored: boolean,
search: string,
) => {
const searchParams = new URLSearchParams();
if (page) searchParams.append('page', page.toString());
if (channelId) searchParams.append('channel', channelId);
if (vid_type) searchParams.append('vid_type', vid_type);
if (search) searchParams.append('q', encodeURIComponent(search));
if (errorFilterFromUrl !== null) searchParams.append('error', errorFilterFromUrl);
searchParams.append('filter', showIgnored ? 'ignore' : 'pending');
const endpoint = `/api/download/${searchParams.toString() ? `?${searchParams.toString()}` : ''}`;

5
frontend/src/components/DownloadListItem.tsx
View File

@ -57,8 +57,9 @@ const DownloadListItem = ({ download, setRefresh }: DownloadListItemProps) => {
</div>
<p>
Published: {formatDate(download.published)} | Duration: {download.duration} |{' '}
{download.youtube_id}
{download.published && <span>Published: {formatDate(download.published)} | </span>}
<span>Duration: {download.duration} | </span>
<span>{download.youtube_id}</span>
</p>
{download.message && <p className="danger-zone">{download.message}</p>}

204
frontend/src/pages/Download.tsx
View File

@ -33,9 +33,9 @@ type Download = {
channel_name: string;
duration: string;
message?: string;
published: string;
published: string | null;
status: string;
timestamp: number;
timestamp: number | null;
title: string;
vid_thumb_url: string;
vid_type: string;
@ -58,10 +58,14 @@ const Download = () => {
const channelFilterFromUrl = searchParams.get('channel');
const ignoredOnlyParam = searchParams.get('ignored');
const vidTypeFilterFromUrl = searchParams.get('vid-type');
const errorFilterFromUrl = searchParams.get('error');
const [refresh, setRefresh] = useState(false);
const [showHiddenForm, setShowHiddenForm] = useState(false);
const [addAsAutoStart, setAddAsAutoStart] = useState(false);
const [addAsFlat, setAddAsFlat] = useState(false);
const [showQueueActions, setShowQueueActions] = useState(false);
const [searchInput, setSearchInput] = useState('');
const [showDeleteConfirm, setShowDeleteConfirm] = useState(false);
const [downloadPending, setDownloadPending] = useState(false);
const [rescanPending, setRescanPending] = useState(false);
@ -115,7 +119,9 @@ const Download = () => {
currentPage,
channelFilterFromUrl,
vidTypeFilterFromUrl,
errorFilterFromUrl,
showIgnored,
searchInput,
);
const { data: channelResponseData } = videosResponse ?? {};
const videoCount = channelResponseData?.paginate?.total_hits;
@ -134,7 +140,14 @@ const Download = () => {
useEffect(() => {
setRefresh(true);
}, [channelFilterFromUrl, vidTypeFilterFromUrl, currentPage, showIgnored]);
}, [
channelFilterFromUrl,
vidTypeFilterFromUrl,
errorFilterFromUrl,
currentPage,
showIgnored,
searchInput,
]);
useEffect(() => {
(async () => {
@ -215,6 +228,46 @@ const Download = () => {
{showHiddenForm && (
<div className="show-form">
<div>
<div className="toggle">
<div className="toggleBox">
<input
id="hide_watched"
type="checkbox"
checked={addAsFlat}
onChange={() => setAddAsFlat(!addAsFlat)}
/>
{addAsFlat ? (
<label htmlFor="" className="onbtn">
On
</label>
) : (
<label htmlFor="" className="ofbtn">
Off
</label>
)}
</div>
<span>Fast add</span>
</div>
<div className="toggle">
<div className="toggleBox">
<input
id="hide_watched"
type="checkbox"
checked={addAsAutoStart}
onChange={() => setAddAsAutoStart(!addAsAutoStart)}
/>
{addAsAutoStart ? (
<label htmlFor="" className="onbtn">
On
</label>
) : (
<label htmlFor="" className="ofbtn">
Off
</label>
)}
</div>
<span>Auto Download</span>
</div>
<textarea
value={downloadQueueText}
onChange={e => setDownloadQueueText(e.target.value)}
@ -226,24 +279,13 @@ const Download = () => {
label="Add to queue"
onClick={async () => {
if (downloadQueueText.trim()) {
await updateDownloadQueue(downloadQueueText, false);
await updateDownloadQueue(downloadQueueText, addAsAutoStart, addAsFlat);
setDownloadQueueText('');
setRefresh(true);
setShowHiddenForm(false);
}
}}
/>{' '}
<Button
label="Download now"
onClick={async () => {
if (downloadQueueText.trim()) {
await updateDownloadQueue(downloadQueueText, true);
setDownloadQueueText('');
setRefresh(true);
setShowHiddenForm(false);
}
}}
/>
</div>
</div>
)}
@ -279,59 +321,8 @@ const Download = () => {
</div>
<div className="view-icons">
<Button onClick={() => setShowQueueActions(!showQueueActions)}>
{showQueueActions ? 'Hide Actions' : 'Show Actions'}
{showQueueActions ? 'Hide Advanced' : 'Show Advanced'}
</Button>
<select
name="vid_type_filter"
id="vid_type_filter"
value={vidTypeFilterFromUrl || 'all'}
onChange={async event => {
const value = event.currentTarget.value;
const params = searchParams;
if (value !== 'all') {
params.set('vid-type', value);
} else {
params.delete('vid-type');
}
setSearchParams(params);
}}
>
<option value="all">all types</option>
<option value="videos">Videos</option>
<option value="streams">Streams</option>
<option value="shorts">Shorts</option>
</select>
{channelAggsList && channelAggsList.length > 1 && (
<select
name="channel_filter"
id="channel_filter"
value={channelFilterFromUrl || 'all'}
onChange={async event => {
const value = event.currentTarget.value;
const params = searchParams;
if (value !== 'all') {
params.set('channel', value);
} else {
params.delete('channel');
}
setSearchParams(params);
}}
>
<option value="all">all channels</option>
{channelAggsList.map(channel => {
const [name, id] = channel.key;
const count = channel.doc_count;
return (
<option key={id} value={id}>
{name} ({count})
</option>
);
})}
</select>
)}
{isGridView && (
<div className="grid-count">
@ -394,6 +385,85 @@ const Download = () => {
</h3>
{showQueueActions && (
<div className="settings-group">
<h3>Search & Filter</h3>
<select
name="vid_type_filter"
id="vid_type_filter"
value={vidTypeFilterFromUrl || 'all'}
onChange={async event => {
const value = event.currentTarget.value;
const params = searchParams;
if (value !== 'all') {
params.set('vid-type', value);
} else {
params.delete('vid-type');
}
setSearchParams(params);
}}
>
<option value="all">all types</option>
<option value="videos">Videos</option>
<option value="streams">Streams</option>
<option value="shorts">Shorts</option>
</select>
{channelAggsList && channelAggsList.length > 1 && (
<select
name="channel_filter"
id="channel_filter"
value={channelFilterFromUrl || 'all'}
onChange={async event => {
const value = event.currentTarget.value;
const params = searchParams;
if (value !== 'all') {
params.set('channel', value);
} else {
params.delete('channel');
}
setSearchParams(params);
}}
>
<option value="all">all channels</option>
{channelAggsList.map(channel => {
const [name, id] = channel.key;
const count = channel.doc_count;
return (
<option key={id} value={id}>
{name} ({count})
</option>
);
})}
</select>
)}
<select
name="error_filter"
id="error_filter"
value={errorFilterFromUrl || 'all'}
onChange={async event => {
const value = event.currentTarget.value;
const params = searchParams;
if (value !== 'all') {
params.set('error', value);
} else {
params.delete('error');
}
setSearchParams(params);
}}
>
<option value="all">all error state</option>
<option value="true">has error</option>
<option value="false">has no error</option>
</select>
<input
type="text"
placeholder="Search..."
value={searchInput}
onChange={e => setSearchInput(e.target.value)}
/>
{searchInput && <Button onClick={() => setSearchInput('')}>Clear</Button>}
<h3>Bulk actions</h3>
<p>
Applied filtered by status <i>'{showIgnoredFilter}'</i>