diff --git a/backend/download/src/queue.py b/backend/download/src/queue.py index f0bea7e8..1451e307 100644 --- a/backend/download/src/queue.py +++ b/backend/download/src/queue.py @@ -11,6 +11,7 @@ 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.queue_interact import PendingInteract from download.src.subscriptions import ChannelSubscription from download.src.thumbnails import ThumbManager from playlist.src.index import YoutubePlaylist @@ -90,103 +91,6 @@ class PendingIndex: self.video_overwrites.update({video_id: overwrites}) -class PendingInteract: - """interact with items in download queue""" - - def __init__(self, youtube_id=False, status=False): - self.youtube_id = youtube_id - self.status = status - - def delete_item(self): - """delete single item from pending""" - path = f"ta_download/_doc/{self.youtube_id}" - _, _ = ElasticWrap(path).delete(refresh=True) - - def delete_bulk(self, channel_id: str | None, vid_type: str | None): - """delete all matching item by status""" - must_list = [{"term": {"status": {"value": self.status}}}] - if channel_id: - must_list.append({"term": {"channel_id": {"value": channel_id}}}) - - if vid_type: - must_list.append({"term": {"vid_type": {"value": vid_type}}}) - - data = {"query": {"bool": {"must": must_list}}} - - path = "ta_download/_delete_by_query?refresh=true" - _, _ = ElasticWrap(path).post(data=data) - - def update_bulk( - self, channel_id: str | None, vid_type: str | None, new_status: str - ): - """update status in bulk""" - must_list = [{"term": {"status": {"value": self.status}}}] - if channel_id: - must_list.append({"term": {"channel_id": {"value": channel_id}}}) - - if vid_type: - must_list.append({"term": {"vid_type": {"value": vid_type}}}) - - if new_status == "priority": - source = """ - ctx._source.status = 'pending'; - ctx._source.auto_start = true; - ctx._source.message = null; - """ - else: - source = f"ctx._source.status = '{new_status}'" - - data = { - "query": {"bool": {"must": must_list}}, - "script": {"source": source, "lang": "painless"}, - } - - path = "ta_download/_update_by_query?refresh=true" - _, _ = ElasticWrap(path).post(data) - - def update_status(self): - """update status of pending item""" - if self.status == "priority": - data = { - "doc": { - "status": "pending", - "auto_start": True, - "message": None, - } - } - else: - data = {"doc": {"status": self.status}} - - path = f"ta_download/_update/{self.youtube_id}/?refresh=true" - _, _ = ElasticWrap(path).post(data=data) - - def get_item(self): - """return pending item dict""" - path = f"ta_download/_doc/{self.youtube_id}" - response, status_code = ElasticWrap(path).get() - return response["_source"], status_code - - def get_channel(self): - """ - get channel metadata from queue to not depend on channel to be indexed - """ - data = { - "size": 1, - "query": {"term": {"channel_id": {"value": self.youtube_id}}}, - } - response, _ = ElasticWrap("ta_download/_search").get(data=data) - hits = response["hits"]["hits"] - if not hits: - channel_name = "NA" - else: - channel_name = hits[0]["_source"].get("channel_name", "NA") - - return { - "channel_id": self.youtube_id, - "channel_name": channel_name, - } - - class PendingList(PendingIndex): """manage the pending videos list""" diff --git a/backend/download/src/queue_interact.py b/backend/download/src/queue_interact.py new file mode 100644 index 00000000..260b5e5a --- /dev/null +++ b/backend/download/src/queue_interact.py @@ -0,0 +1,100 @@ +"""interact with queue items""" + +from common.src.es_connect import ElasticWrap + + +class PendingInteract: + """interact with items in download queue""" + + def __init__(self, youtube_id=False, status=False): + self.youtube_id = youtube_id + self.status = status + + def delete_item(self): + """delete single item from pending""" + path = f"ta_download/_doc/{self.youtube_id}" + _, _ = ElasticWrap(path).delete(refresh=True) + + def delete_bulk(self, channel_id: str | None, vid_type: str | None): + """delete all matching item by status""" + must_list = [{"term": {"status": {"value": self.status}}}] + if channel_id: + must_list.append({"term": {"channel_id": {"value": channel_id}}}) + + if vid_type: + must_list.append({"term": {"vid_type": {"value": vid_type}}}) + + data = {"query": {"bool": {"must": must_list}}} + + path = "ta_download/_delete_by_query?refresh=true" + _, _ = ElasticWrap(path).post(data=data) + + def update_bulk( + self, channel_id: str | None, vid_type: str | None, new_status: str + ): + """update status in bulk""" + must_list = [{"term": {"status": {"value": self.status}}}] + if channel_id: + must_list.append({"term": {"channel_id": {"value": channel_id}}}) + + if vid_type: + must_list.append({"term": {"vid_type": {"value": vid_type}}}) + + if new_status == "priority": + source = """ + ctx._source.status = 'pending'; + ctx._source.auto_start = true; + ctx._source.message = null; + """ + else: + source = f"ctx._source.status = '{new_status}'" + + data = { + "query": {"bool": {"must": must_list}}, + "script": {"source": source, "lang": "painless"}, + } + + path = "ta_download/_update_by_query?refresh=true" + _, _ = ElasticWrap(path).post(data) + + def update_status(self): + """update status of pending item""" + if self.status == "priority": + data = { + "doc": { + "status": "pending", + "auto_start": True, + "message": None, + } + } + else: + data = {"doc": {"status": self.status}} + + path = f"ta_download/_update/{self.youtube_id}/?refresh=true" + _, _ = ElasticWrap(path).post(data=data) + + def get_item(self): + """return pending item dict""" + path = f"ta_download/_doc/{self.youtube_id}" + response, status_code = ElasticWrap(path).get() + return response["_source"], status_code + + def get_channel(self): + """ + get channel metadata from queue to not depend on channel to be indexed + """ + data = { + "size": 1, + "query": {"term": {"channel_id": {"value": self.youtube_id}}}, + } + response, _ = ElasticWrap("ta_download/_search").get(data=data) + hits = response["hits"]["hits"] + if not hits: + channel_name = "NA" + else: + channel_name = hits[0]["_source"].get("channel_name", "NA") + + return { + "channel_id": self.youtube_id, + "channel_name": channel_name, + } diff --git a/backend/download/views.py b/backend/download/views.py index bc30d12b..05f3deb5 100644 --- a/backend/download/views.py +++ b/backend/download/views.py @@ -17,7 +17,7 @@ from download.serializers import ( DownloadListSerializer, DownloadQueueItemUpdateSerializer, ) -from download.src.queue import PendingInteract +from download.src.queue_interact import PendingInteract from drf_spectacular.utils import OpenApiResponse, extend_schema from rest_framework.response import Response from task.tasks import download_pending, extrac_dl