Merge pull request #3337 from InfinityPacer/feature/subscribe

feat(event): add ResourceDownload event for cancel download

app/api/endpoints/download.py

@@ -51,7 +51,7 @@ def download(
         torrent_info=torrentinfo
     )
     did = DownloadChain().download_single(context=context, username=current_user.name,
-                                          downloader=downloader, save_path=save_path)
+                                          downloader=downloader, save_path=save_path, source="Manual")
     if not did:
         return schemas.Response(success=False, message="任务添加失败")
     return schemas.Response(success=True, data={
@@ -84,7 +84,7 @@ def add(
         torrent_info=torrentinfo
     )
     did = DownloadChain().download_single(context=context, username=current_user.name,
-                                          downloader=downloader, save_path=save_path)
+                                          downloader=downloader, save_path=save_path, source="Manual")
     if not did:
         return schemas.Response(success=False, message="任务添加失败")
     return schemas.Response(success=True, data={

app/chain/download.py

@@ -20,7 +20,7 @@ from app.helper.message import MessageHelper
 from app.helper.torrent import TorrentHelper
 from app.log import logger
 from app.schemas import ExistMediaInfo, NotExistMediaInfo, DownloadingTorrent, Notification
-from app.schemas.event import ResourceSelectionEventData
+from app.schemas.event import ResourceSelectionEventData, ResourceDownloadEventData
 from app.schemas.types import MediaType, TorrentStatus, EventType, MessageChannel, NotificationType, ChainEventType
 from app.utils.http import RequestUtils
 from app.utils.string import StringUtils
@@ -192,7 +192,7 @@ class DownloadChain(ChainBase):
             logger.error(f"下载种子文件失败：{torrent.title} - {torrent_url}")
             self.post_message(Notification(
                 channel=channel,
-                source=source,
+                source=source if channel else None,
                 mtype=NotificationType.Manual,
                 title=f"{torrent.title} 种子下载失败！",
                 text=f"错误信息：{error_msg}\n站点：{torrent.site_name}",
@@ -204,7 +204,8 @@ class DownloadChain(ChainBase):
 
     def download_single(self, context: Context, torrent_file: Path = None,
                         episodes: Set[int] = None,
-                        channel: MessageChannel = None, source: str = None,
+                        channel: MessageChannel = None,
+                        source: str = None,
                         downloader: str = None,
                         save_path: str = None,
                         userid: Union[str, int] = None,
@@ -216,13 +217,38 @@ class DownloadChain(ChainBase):
         :param torrent_file: 种子文件路径
         :param episodes: 需要下载的集数
         :param channel: 通知渠道
-        :param source: 通知来源
+        :param source: 来源（消息通知、Subscribe、Manual等）
         :param downloader: 下载器
         :param save_path: 保存路径
         :param userid: 用户ID
         :param username: 调用下载的用户名/插件名
         :param media_category: 自定义媒体类别
         """
+        # 发送资源下载事件，允许外部拦截下载
+        event_data = ResourceDownloadEventData(
+            context=context,
+            episodes=episodes,
+            channel=channel,
+            origin=source,
+            downloader=downloader,
+            options={
+                "save_path": save_path,
+                "userid": userid,
+                "username": username,
+                "media_category": media_category
+            }
+        )
+        # 触发资源下载事件
+        event = eventmanager.send_event(ChainEventType.ResourceDownload, event_data)
+        if event and event.event_data:
+            event_data: ResourceDownloadEventData = event.event_data
+            # 如果事件被取消，跳过资源下载
+            if event_data.cancel:
+                logger.debug(
+                    f"Resource download canceled by event: {event_data.source},"
+                    f"Reason: {event_data.reason}")
+                return None
+
         _torrent = context.torrent_info
         _media = context.media_info
         _meta = context.meta_info
@@ -366,7 +392,7 @@ class DownloadChain(ChainBase):
             # 只发送给对应渠道和用户
             self.post_message(Notification(
                 channel=channel,
-                source=source,
+                source=source if channel else None,
                 mtype=NotificationType.Manual,
                 title="添加下载任务失败：%s %s"
                       % (_media.title_year, _meta.season_episode),
@@ -394,7 +420,7 @@ class DownloadChain(ChainBase):
         :param no_exists:  缺失的剧集信息
         :param save_path:  保存路径
         :param channel:  通知渠道
-        :param source:  通知来源
+        :param source:  来源（消息通知、订阅、手工下载等）
         :param userid:  用户ID
         :param username: 调用下载的用户名/插件名
         :param media_category: 自定义媒体类别

app/chain/message.py

@@ -111,6 +111,8 @@ class MessageChain(ChainBase):
         info = self.message_parser(source=source, body=body, form=form, args=args)
         if not info:
             return
+        # 更新消息来源
+        source = info.source
         # 渠道
         channel = info.channel
         # 用户ID

app/chain/subscribe.py

@@ -399,6 +399,7 @@ class SubscribeChain(ChainBase, metaclass=Singleton):
                         save_path=subscribe.save_path,
                         media_category=subscribe.media_category,
                         downloader=subscribe.downloader,
+                        source="Subscribe"
                     )
 
                     # 判断是否应完成订阅
@@ -789,7 +790,8 @@ class SubscribeChain(ChainBase, metaclass=Singleton):
                                                                      username=subscribe.username,
                                                                      save_path=subscribe.save_path,
                                                                      media_category=subscribe.media_category,
-                                                                     downloader=subscribe.downloader)
+                                                                     downloader=subscribe.downloader,
+                                                                     source="Subscribe")
                 # 判断是否要完成订阅
                 self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo,
                                              downloads=downloads, lefts=lefts)

app/core/event.py

@@ -347,8 +347,17 @@ class EventManager(metaclass=Singleton):
         if not handlers:
             logger.debug(f"No handlers found for chain event: {event}")
             return False
+
+        # 过滤出启用的处理器
+        enabled_handlers = {handler_id: (priority, handler) for handler_id, (priority, handler) in handlers.items()
+                            if self.__is_handler_enabled(handler)}
+
+        if not enabled_handlers:
+            logger.debug(f"No enabled handlers found for chain event: {event}. Skipping execution.")
+            return False
+
         self.__log_event_lifecycle(event, "Started")
-        for handler_id, (priority, handler) in handlers.items():
+        for handler_id, (priority, handler) in enabled_handlers.items():
             start_time = time.time()
             self.__safe_invoke_handler(handler, event)
             logger.debug(

app/schemas/event.py

@@ -1,9 +1,10 @@
 from pathlib import Path
-from typing import Optional, Dict, Any, List
+from typing import Optional, Dict, Any, List, Set
 
 from pydantic import BaseModel, Field, root_validator
 
 from app.core.context import Context
+from app.schemas import MessageChannel
 
 
 class BaseEventData(BaseModel):
@@ -169,3 +170,35 @@ class ResourceSelectionEventData(BaseModel):
     updated: bool = Field(False, description="是否已更新")
     updated_contexts: Optional[List[Context]] = Field(None, description="已更新的资源上下文列表")
     source: Optional[str] = Field("未知拦截源", description="拦截源")
+
+
+class ResourceDownloadEventData(ChainEventData):
+    """
+    ResourceDownload 事件的数据模型
+
+    Attributes:
+        # 输入参数
+        context (Context): 当前资源上下文
+        episodes (Set[int]): 需要下载的集数
+        channel (MessageChannel): 通知渠道
+        origin (str): 来源（消息通知、Subscribe、Manual等）
+        downloader (str): 下载器
+        options (dict): 其他参数
+
+        # 输出参数
+        cancel (bool): 是否取消下载，默认值为 False
+        source (str): 拦截源，默认值为 "未知拦截源"
+        reason (str): 拦截原因，描述拦截的具体原因
+    """
+    # 输入参数
+    context: Any = Field(None, description="当前资源上下文")
+    episodes: Optional[Set[int]] = Field(None, description="需要下载的集数")
+    channel: Optional[MessageChannel] = Field(None, description="通知渠道")
+    origin: Optional[str] = Field(None, description="来源")
+    downloader: Optional[str] = Field(None, description="下载器")
+    options: Optional[dict] = Field(None, description="其他参数")
+
+    # 输出参数
+    cancel: bool = Field(False, description="是否取消下载")
+    source: str = Field("未知拦截源", description="拦截源")
+    reason: str = Field("", description="拦截原因")

app/schemas/types.py

@@ -72,6 +72,8 @@ class ChainEventType(Enum):
     TransferRename = "transfer.rename"
     # 资源选择
     ResourceSelection = "resource.selection"
+    # 资源下载
+    ResourceDownload = "resource.download"
 
 
 # 系统配置Key字典