make update_subscribed stoppable

tubearchivist/home/src/download/subscriptions.py

@@ -126,10 +126,14 @@ class ChannelSubscription:
             if not self.task:
                 continue
 
-            self.task.send_progress(
-                message_lines=[f"Scanning Channel {idx + 1}/{total}"],
-                progress=(idx + 1) / total,
-            )
+            if self.task:
+                self.task.send_progress(
+                    message_lines=[f"Scanning Channel {idx + 1}/{total}"],
+                    progress=(idx + 1) / total,
+                )
+                if self.task.is_stopped():
+                    self.task.send_progress(["Received Stop signal."])
+                    break
 
         return missing_videos
 
@@ -261,10 +265,14 @@ class PlaylistSubscription:
             if not self.task:
                 continue
 
-            self.task.send_progress(
-                message_lines=[f"Scanning Playlists {idx + 1}/{total}"],
-                progress=(idx + 1) / total,
-            )
+            if self.task:
+                self.task.send_progress(
+                    message_lines=[f"Scanning Playlists {idx + 1}/{total}"],
+                    progress=(idx + 1) / total,
+                )
+                if self.task.is_stopped():
+                    self.task.send_progress(["Received Stop signal."])
+                    break
 
         return missing_videos
 

tubearchivist/home/src/ta/task_manager.py

@@ -40,6 +40,13 @@ class TaskManager:
 
         return bool([i for i in tasks if i.get("status") == "PENDING"])
 
+    def is_stopped(self, task_id):
+        """check if task_id has received STOP command"""
+        task = self.get_task(task_id)
+        print(task)
+
+        return task.get("command") == "STOP"
+
     def get_pending(self, task_name):
         """get all pending tasks of task_name"""
         tasks = self.get_tasks_by_name(task_name)

tubearchivist/home/tasks.py

@@ -53,6 +53,7 @@ class BaseTask(Task):
             "title": "Rescan your Subscriptions",
             "group": "download:scan",
             "api-start": True,
+            "api-stop": True,
         },
         "download_pending": {
             "title": "Downloading",
@@ -146,6 +147,10 @@ class BaseTask(Task):
         key = f"message:{message.get('group')}:{task_id.split('-')[0]}"
         return message, key
 
+    def is_stopped(self):
+        """check if task is stopped"""
+        return TaskManager().is_stopped(self.request.id)
+
 
 @shared_task(name="update_subscribed", bind=True, base=BaseTask)
 def update_subscribed(self):