refactor channel playlist index task backend

This commit is contained in:
simon
2023-03-18 12:59:16 +07:00
parent d533c7acfe
commit b2fa92a8c3
5 changed files with 16 additions and 56 deletions

View File

@@ -96,7 +96,7 @@ class DownloadPostProcess:
def validate_playlists(self): def validate_playlists(self):
"""look for playlist needing to update""" """look for playlist needing to update"""
for id_c, channel_id in enumerate(self.download.channels): for id_c, channel_id in enumerate(self.download.channels):
channel = YoutubeChannel(channel_id) channel = YoutubeChannel(channel_id, task=self.download.task)
overwrites = self.pending.channel_overwrites.get(channel_id, False) overwrites = self.pending.channel_overwrites.get(channel_id, False)
if overwrites and overwrites.get("index_playlists"): if overwrites and overwrites.get("index_playlists"):
# validate from remote # validate from remote

View File

@@ -13,7 +13,6 @@ from home.src.ta.ta_redis import RedisArchivist, RedisQueue
from home.src.ta.urlparser import Parser from home.src.ta.urlparser import Parser
from home.tasks import ( from home.tasks import (
download_pending, download_pending,
index_channel_playlists,
kill_dl, kill_dl,
re_sync_thumbs, re_sync_thumbs,
rescan_filesystem, rescan_filesystem,
@@ -62,7 +61,6 @@ class PostData:
"db-restore": self._db_restore, "db-restore": self._db_restore,
"fs-rescan": self._fs_rescan, "fs-rescan": self._fs_rescan,
"delete-playlist": self._delete_playlist, "delete-playlist": self._delete_playlist,
"find-playlists": self._find_playlists,
} }
return exec_map[self.to_exec] return exec_map[self.to_exec]
@@ -232,9 +230,3 @@ class PostData:
YoutubePlaylist(playlist_id).delete_videos_playlist() YoutubePlaylist(playlist_id).delete_videos_playlist()
return {"success": True} return {"success": True}
def _find_playlists(self):
"""add all playlists of a channel"""
channel_id = self.exec_val
index_channel_playlists.delay(channel_id)
return {"success": True}

View File

@@ -18,7 +18,6 @@ from home.src.es.connect import ElasticWrap, IndexPaginate
from home.src.index.generic import YouTubeItem from home.src.index.generic import YouTubeItem
from home.src.index.playlist import YoutubePlaylist from home.src.index.playlist import YoutubePlaylist
from home.src.ta.helper import clean_string, requests_headers from home.src.ta.helper import clean_string, requests_headers
from home.src.ta.ta_redis import RedisArchivist
class ChannelScraper: class ChannelScraper:
@@ -167,12 +166,12 @@ class YoutubeChannel(YouTubeItem):
es_path = False es_path = False
index_name = "ta_channel" index_name = "ta_channel"
yt_base = "https://www.youtube.com/channel/" yt_base = "https://www.youtube.com/channel/"
msg = "message:playlistscan"
def __init__(self, youtube_id): def __init__(self, youtube_id, task=False):
super().__init__(youtube_id) super().__init__(youtube_id)
self.es_path = f"{self.index_name}/_doc/{youtube_id}" self.es_path = f"{self.index_name}/_doc/{youtube_id}"
self.all_playlists = False self.all_playlists = False
self.task = task
def build_json(self, upload=False, fallback=False): def build_json(self, upload=False, fallback=False):
"""get from es or from youtube""" """get from es or from youtube"""
@@ -324,34 +323,27 @@ class YoutubeChannel(YouTubeItem):
print(f"{self.youtube_id}: index all playlists") print(f"{self.youtube_id}: index all playlists")
self.get_from_es() self.get_from_es()
channel_name = self.json_data["channel_name"] channel_name = self.json_data["channel_name"]
mess_dict = { self.task.send_progress([f"{channel_name}: Looking for Playlists"])
"status": self.msg,
"level": "info",
"title": "Looking for playlists",
"message": f"{channel_name}: Scanning channel in progress",
}
RedisArchivist().set_message(self.msg, mess_dict, expire=True)
self.get_all_playlists() self.get_all_playlists()
if not self.all_playlists: if not self.all_playlists:
print(f"{self.youtube_id}: no playlists found.") print(f"{self.youtube_id}: no playlists found.")
return return
all_youtube_ids = self.get_all_video_ids() all_youtube_ids = self.get_all_video_ids()
total = len(self.all_playlists)
for idx, playlist in enumerate(self.all_playlists): for idx, playlist in enumerate(self.all_playlists):
self._notify_single_playlist(idx, playlist) self._notify_single_playlist(idx, total)
self._index_single_playlist(playlist, all_youtube_ids) self._index_single_playlist(playlist, all_youtube_ids)
print("add playlist: " + playlist[1])
def _notify_single_playlist(self, idx, playlist): def _notify_single_playlist(self, idx, total):
"""send notification""" """send notification"""
channel_name = self.json_data["channel_name"] channel_name = self.json_data["channel_name"]
mess_dict = { message = [
"status": self.msg, f"{channel_name}: Scanning channel for playlists",
"level": "info", f"Progress: {idx + 1}/{total}",
"title": f"{channel_name}: Scanning channel for playlists", ]
"message": f"Progress: {idx + 1}/{len(self.all_playlists)}", self.task.send_progress(message, progress=(idx + 1) / total)
}
RedisArchivist().set_message(self.msg, mess_dict, expire=True)
print("add playlist: " + playlist[1])
@staticmethod @staticmethod
def _index_single_playlist(playlist, all_youtube_ids): def _index_single_playlist(playlist, all_youtube_ids):

View File

@@ -315,19 +315,10 @@ def subscribe_to(url_str):
SubscriptionHandler(url_str).subscribe() SubscriptionHandler(url_str).subscribe()
@shared_task(name="index_playlists") @shared_task(bind=True, name="index_playlists", base=BaseTask)
def index_channel_playlists(channel_id): def index_channel_playlists(self, channel_id):
"""add all playlists of channel to index""" """add all playlists of channel to index"""
channel = YoutubeChannel(channel_id) channel = YoutubeChannel(channel_id, task=self)
# notify
key = "message:playlistscan"
mess_dict = {
"status": key,
"level": "info",
"title": "Looking for playlists",
"message": f"{channel_id}: Channel scan in progress",
}
RedisArchivist().set_message(key, mess_dict, expire=True)
channel.index_channel_playlists() channel.index_channel_playlists()

View File

@@ -619,21 +619,6 @@ class ChannelIdView(ChannelIdBaseView):
to_append = {"term": {"player.watched": {"value": False}}} to_append = {"term": {"player.watched": {"value": False}}}
self.data["query"]["bool"]["must"].append(to_append) self.data["query"]["bool"]["must"].append(to_append)
@staticmethod
def post(request, channel_id):
"""handle post request"""
print(f"handle post from {channel_id}")
channel_overwrite_form = ChannelOverwriteForm(request.POST)
if channel_overwrite_form.is_valid():
overwrites = channel_overwrite_form.cleaned_data
print(f"{channel_id}: set overwrites {overwrites}")
channel_overwrites(channel_id, overwrites=overwrites)
if overwrites.get("index_playlists") == "1":
index_channel_playlists.delay(channel_id)
sleep(1)
return redirect("channel_id", channel_id, permanent=True)
class ChannelIdLiveView(ChannelIdView): class ChannelIdLiveView(ChannelIdView):
"""resolves to /channel/<channel-id>/streams/ """resolves to /channel/<channel-id>/streams/