diff --git a/tubearchivist/api/views.py b/tubearchivist/api/views.py index 22e0ec9a..85b44b0e 100644 --- a/tubearchivist/api/views.py +++ b/tubearchivist/api/views.py @@ -711,7 +711,8 @@ class TaskIDView(ApiBaseView): message = {"message": "task can not be stopped"} return Response(message, status=400) - TaskCommand().stop(task_id) + message_key = self._build_message_key(task_conf, task_id) + TaskCommand().stop(task_id, message_key) if command == "kill": if not task_conf.get("api-stop"): message = {"message": "task can not be killed"} @@ -721,6 +722,10 @@ class TaskIDView(ApiBaseView): return Response({"message": "command sent"}) + def _build_message_key(self, task_conf, task_id): + """build message key to forward command to notification""" + return f"message:{task_conf.get('group')}:{task_id.split('-')[0]}" + class RefreshView(ApiBaseView): """resolves to /api/refresh/ diff --git a/tubearchivist/home/src/download/subscriptions.py b/tubearchivist/home/src/download/subscriptions.py index dd8a6bf8..b006f845 100644 --- a/tubearchivist/home/src/download/subscriptions.py +++ b/tubearchivist/home/src/download/subscriptions.py @@ -127,13 +127,14 @@ class ChannelSubscription: continue if self.task: + if self.task.is_stopped(): + self.task.send_progress(["Received Stop signal."]) + break + 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 @@ -291,7 +292,8 @@ class SubscriptionScanner: self.missing_videos = [] self.scan_channels() - self.scan_playlists() + if self.task and not self.task.is_stopped(): + self.scan_playlists() return self.missing_videos diff --git a/tubearchivist/home/src/ta/task_manager.py b/tubearchivist/home/src/ta/task_manager.py index a47a6273..7c3d124c 100644 --- a/tubearchivist/home/src/ta/task_manager.py +++ b/tubearchivist/home/src/ta/task_manager.py @@ -5,7 +5,7 @@ functionality: """ from home import tasks as ta_tasks -from home.src.ta.ta_redis import TaskRedis +from home.src.ta.ta_redis import RedisArchivist, TaskRedis class TaskManager: @@ -92,11 +92,12 @@ class TaskCommand: return message - def stop(self, task_id): + def stop(self, task_id, message_key): """ send stop signal to task_id, needs to be implemented in task to take effect """ + print(f"[task][{task_id}]: received STOP signal.") handler = TaskRedis() task = handler.get_single(task_id) @@ -104,7 +105,9 @@ class TaskCommand: raise ValueError handler.set_command(task_id, "STOP") + RedisArchivist().set_message(message_key, "STOP", path=".command") def kill(self, task_id): """send kill signal to task_id""" + print(f"[task][{task_id}]: received KILL signal.") ta_tasks.app.control.revoke(task_id, terminate=True) diff --git a/tubearchivist/home/tasks.py b/tubearchivist/home/tasks.py index 8ef59251..d61f9b47 100644 --- a/tubearchivist/home/tasks.py +++ b/tubearchivist/home/tasks.py @@ -149,7 +149,8 @@ class BaseTask(Task): message.update({"level": level, "id": task_id}) task_result = TaskManager().get_task(task_id) if task_result: - message.update({"command": task_result.get("command", False)}) + command = task_result.get("command", False) + message.update({"command": command}) key = f"message:{message.get('group')}:{task_id.split('-')[0]}" return message, key