from app.workflow.actions import BaseAction, ActionChain from app.core.config import global_vars from app.schemas import ActionParams, ActionContext from app.log import logger class FetchDownloadsParams(ActionParams): """ 获取下载任务参数 """ pass class FetchDownloadsAction(BaseAction): """ 获取下载任务 """ contract = { "inputs": [{"name": "downloads", "label": "下载任务", "kind": "list"}], "outputs": [{"name": "downloads", "label": "下载任务", "kind": "list", "merge": "replace"}], "concurrency_key": "download", } def __init__(self, action_id: str): super().__init__(action_id) self._downloads = [] @classmethod @property def name(cls) -> str: # noqa return "获取下载任务" @classmethod @property def description(cls) -> str: # noqa return "获取下载队列中的任务状态" @classmethod @property def data(cls) -> dict: # noqa return FetchDownloadsParams().model_dump() @property def success(self) -> bool: return self.done def execute(self, workflow_id: int, params: dict, context: ActionContext) -> ActionContext: """ 更新downloads中的下载任务状态 """ self._downloads = context.downloads or [] if not self._downloads: self.job_done("无下载任务") return context for download in self._downloads: if global_vars.is_workflow_stopped(workflow_id): break logger.info(f"获取下载任务 {download.download_id} 状态 ...") torrents = ActionChain().list_torrents( hashs=[download.download_id], downloader=download.downloader, ) if not torrents: download.completed = True continue for t in torrents: download.path = t.path if t.progress >= 100: logger.info(f"下载任务 {download.download_id} 已完成") download.completed = True else: logger.info(f"下载任务 {download.download_id} 未完成") download.completed = False if all([d.completed for d in self._downloads]): self.job_done("下载任务已全部完成") return context