Merge pull request #3995 from KoWming/v2

- 新增支持设定消息发送的时间范围
- 探索标签页支持拖动排序
- 修复演员头像不显示的问题
- 修复站点流控不生效的问题
- 修复短时间内重复保存设定后定时任务消失的问题
- 修复工作流执行数据叠加的问题

app/actions/__init__.py

@@ -1,6 +1,8 @@
 from abc import ABC, abstractmethod
+from typing import Union
 
 from app.chain import ChainBase
+from app.db.systemconfig_oper import SystemConfigOper
 from app.schemas import ActionContext, ActionParams
 
 
@@ -13,27 +15,35 @@ class BaseAction(ABC):
     工作流动作基类
     """
 
+    # 动作ID
+    _action_id = None
     # 完成标志
     _done_flag = False
     # 执行信息
     _message = ""
+    # 缓存键值
+    _cache_key = "WorkflowCache-%s"
+
+    def __init__(self, action_id: str):
+        self._action_id = action_id
+        self.systemconfigoper = SystemConfigOper()
 
     @classmethod
     @property
     @abstractmethod
-    def name(cls) -> str:
+    def name(cls) -> str:  # noqa
         pass
 
     @classmethod
     @property
     @abstractmethod
-    def description(cls) -> str:
+    def description(cls) -> str:  # noqa
         pass
 
     @classmethod
     @property
     @abstractmethod
-    def data(cls) -> dict:
+    def data(cls) -> dict:  # noqa
         pass
 
     @property
@@ -65,6 +75,29 @@ class BaseAction(ABC):
         self._message = message
         self._done_flag = True
 
+    def check_cache(self, workflow_id: int, key: str) -> bool:
+        """
+        检查是否处理过
+        """
+        workflow_key = self._cache_key % workflow_id
+        workflow_cache = self.systemconfigoper.get(workflow_key) or {}
+        action_cache = workflow_cache.get(self._action_id) or []
+        return key in action_cache
+
+    def save_cache(self, workflow_id: int, data: Union[list, str]):
+        """
+        保存缓存
+        """
+        workflow_key = self._cache_key % workflow_id
+        workflow_cache = self.systemconfigoper.get(workflow_key) or {}
+        action_cache = workflow_cache.get(self._action_id) or []
+        if isinstance(data, list):
+            action_cache.extend(data)
+        else:
+            action_cache.append(data)
+        workflow_cache[self._action_id] = action_cache
+        self.systemconfigoper.set(workflow_key, workflow_cache)
+
     @abstractmethod
     def execute(self, workflow_id: int, params: ActionParams, context: ActionContext) -> ActionContext:
         """

app/actions/add_download.py

@@ -15,9 +15,10 @@ class AddDownloadParams(ActionParams):
     """
     添加下载资源参数
     """
-    downloader: Optional[str] = Field(None, description="下载器")
-    save_path: Optional[str] = Field(None, description="保存路径")
-    only_lack: Optional[bool] = Field(False, description="仅下载缺失的资源")
+    downloader: Optional[str] = Field(default=None, description="下载器")
+    save_path: Optional[str] = Field(default=None, description="保存路径")
+    labels: Optional[str] = Field(default=None, description="标签（,分隔）")
+    only_lack: Optional[bool] = Field(default=False, description="仅下载缺失的资源")
 
 
 class AddDownloadAction(BaseAction):
@@ -29,24 +30,26 @@ class AddDownloadAction(BaseAction):
     _added_downloads = []
     _has_error = False
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.downloadchain = DownloadChain()
         self.mediachain = MediaChain()
+        self._added_downloads = []
+        self._has_error = False
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "添加下载"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "根据资源列表添加下载任务"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return AddDownloadParams().dict()
 
     @property
@@ -58,23 +61,29 @@ class AddDownloadAction(BaseAction):
         将上下文中的torrents添加到下载任务中
         """
         params = AddDownloadParams(**params)
+        _started = False
         for t in context.torrents:
             if global_vars.is_workflow_stopped(workflow_id):
                 break
+            # 检查缓存
+            cache_key = f"{t.torrent_info.site}-{t.torrent_info.title}"
+            if self.check_cache(workflow_id, cache_key):
+                logger.info(f"{t.torrent_info.title} 已添加过下载，跳过")
+                continue
             if not t.meta_info:
-                t.meta_info = MetaInfo(title=t.title, subtitle=t.description)
+                t.meta_info = MetaInfo(title=t.torrent_info.title, subtitle=t.torrent_info.description)
             if not t.media_info:
                 t.media_info = self.mediachain.recognize_media(meta=t.meta_info)
             if not t.media_info:
                 self._has_error = True
-                logger.warning(f"{t.title} 未识别到媒体信息，无法下载")
+                logger.warning(f"{t.torrent_info.title} 未识别到媒体信息，无法下载")
                 continue
             if params.only_lack:
                 exists_info = self.downloadchain.media_exists(t.media_info)
                 if exists_info:
                     if t.media_info.type == MediaType.MOVIE:
                         # 电影
-                        logger.warning(f"{t.title} 媒体库中已存在，跳过")
+                        logger.warning(f"{t.torrent_info.title} 媒体库中已存在，跳过")
                         continue
                     else:
                         # 电视剧
@@ -90,19 +99,23 @@ class AddDownloadAction(BaseAction):
                                     logger.warning(f"{t.meta_info.title} 第 {t.meta_info.begin_season} 季第 {t.meta_info.episode_list} 集已存在，跳过")
                                     continue
 
+            _started = True
             did = self.downloadchain.download_single(context=t,
                                                      downloader=params.downloader,
-                                                     save_path=params.save_path)
+                                                     save_path=params.save_path,
+                                                     label=params.labels)
             if did:
                 self._added_downloads.append(did)
-            else:
-                self._has_error = True
+                # 保存缓存
+                self.save_cache(workflow_id, cache_key)
 
         if self._added_downloads:
             logger.info(f"已添加 {len(self._added_downloads)} 个下载任务")
             context.downloads.extend(
                 [DownloadTask(download_id=did, downloader=params.downloader) for did in self._added_downloads]
             )
+        elif _started:
+            self._has_error = True
 
         self.job_done(f"已添加 {len(self._added_downloads)} 个下载任务")
         return context

app/actions/add_subscribe.py

@@ -22,24 +22,26 @@ class AddSubscribeAction(BaseAction):
     _added_subscribes = []
     _has_error = False
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.subscribechain = SubscribeChain()
         self.subscribeoper = SubscribeOper()
+        self._added_subscribes = []
+        self._has_error = False
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "添加订阅"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "根据媒体列表添加订阅"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return AddSubscribeParams().dict()
 
     @property
@@ -50,15 +52,22 @@ class AddSubscribeAction(BaseAction):
         """
         将medias中的信息添加订阅，如果订阅不存在的话
         """
+        _started = False
         for media in context.medias:
             if global_vars.is_workflow_stopped(workflow_id):
                 break
+            # 检查缓存
+            cache_key = f"{media.type}-{media.title}-{media.year}-{media.season}"
+            if self.check_cache(workflow_id, cache_key):
+                logger.info(f"{media.title} {media.year} 已添加过订阅，跳过")
+                continue
             mediainfo = MediaInfo()
             mediainfo.from_dict(media.dict())
             if self.subscribechain.exists(mediainfo):
                 logger.info(f"{media.title} 已存在订阅")
                 continue
             # 添加订阅
+            _started = True
             sid, message = self.subscribechain.add(mtype=mediainfo.type,
                                                    title=mediainfo.title,
                                                    year=mediainfo.year,
@@ -69,13 +78,15 @@ class AddSubscribeAction(BaseAction):
                                                    username=settings.SUPERUSER)
             if sid:
                 self._added_subscribes.append(sid)
-            else:
-                self._has_error = True
+                # 保存缓存
+                self.save_cache(workflow_id, cache_key)
 
         if self._added_subscribes:
             logger.info(f"已添加 {len(self._added_subscribes)} 个订阅")
             for sid in self._added_subscribes:
                 context.subscribes.append(self.subscribeoper.get(sid))
+        elif _started:
+            self._has_error = True
 
         self.job_done(f"已添加 {len(self._added_subscribes)} 个订阅")
         return context

app/actions/fetch_downloads.py

@@ -18,23 +18,24 @@ class FetchDownloadsAction(BaseAction):
 
     _downloads = []
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.chain = ActionChain()
+        self._downloads = []
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "获取下载任务"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "获取下载队列中的任务状态"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return FetchDownloadsParams().dict()
 
     @property

app/actions/fetch_medias.py

@@ -17,9 +17,9 @@ class FetchMediasParams(ActionParams):
     """
     获取媒体数据参数
     """
-    source_type: Optional[str] = Field("ranking", description="来源")
-    sources: Optional[List[str]] = Field([], description="榜单")
-    api_path: Optional[str] = Field(None, description="API路径")
+    source_type: Optional[str] = Field(default="ranking", description="来源")
+    sources: Optional[List[str]] = Field(default=[], description="榜单")
+    api_path: Optional[str] = Field(default=None, description="API路径")
 
 
 class FetchMediasAction(BaseAction):
@@ -28,12 +28,14 @@ class FetchMediasAction(BaseAction):
     """
 
     _inner_sources = []
-
     _medias = []
+    _has_error = False
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
 
+        self._medias = []
+        self._has_error = False
         self.__inner_sources = [
             {
                 "func": RecommendChain().tmdb_trending,
@@ -100,22 +102,22 @@ class FetchMediasAction(BaseAction):
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "获取媒体数据"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "获取榜单等媒体数据列表"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return FetchMediasParams().dict()
 
     @property
     def success(self) -> bool:
-        return True if self._medias else False
+        return not self._has_error
 
     def __get_source(self, source: str):
         """
@@ -131,37 +133,41 @@ class FetchMediasAction(BaseAction):
         获取媒体数据，填充到medias
         """
         params = FetchMediasParams(**params)
-        if params.source_type == "ranking":
-            for name in params.sources:
-                if global_vars.is_workflow_stopped(workflow_id):
-                    break
-                source = self.__get_source(name)
-                if not source:
-                    continue
-                logger.info(f"获取媒体数据 {source} ...")
-                results = []
-                if source.get("func"):
-                    results = source['func']()
-                else:
-                    # 调用内部API获取数据
-                    api_url = f"http://127.0.0.1:{settings.PORT}/api/v1/{source['api_path']}?token={settings.API_TOKEN}"
-                    res = RequestUtils(timeout=15).post_res(api_url)
-                    if res:
-                        results = res.json()
-                if results:
-                    logger.info(f"{name} 获取到 {len(results)} 条数据")
-                    self._medias.extend([MediaInfo(**r) for r in results])
-                else:
-                    logger.error(f"{name} 获取数据失败")
-        else:
-            # 调用内部API获取数据
-            api_url = f"http://127.0.0.1:{settings.PORT}{params.api_path}?token={settings.API_TOKEN}"
-            res = RequestUtils(timeout=15).post_res(api_url)
-            if res:
-                results = res.json()
-                if results:
-                    logger.info(f"{params.api_path} 获取到 {len(results)} 条数据")
-                    self._medias.extend([MediaInfo(**r) for r in results])
+        try:
+            if params.source_type == "ranking":
+                for name in params.sources:
+                    if global_vars.is_workflow_stopped(workflow_id):
+                        break
+                    source = self.__get_source(name)
+                    if not source:
+                        continue
+                    logger.info(f"获取媒体数据 {source} ...")
+                    results = []
+                    if source.get("func"):
+                        results = source['func']()
+                    else:
+                        # 调用内部API获取数据
+                        api_url = f"http://127.0.0.1:{settings.PORT}/api/v1/{source['api_path']}?token={settings.API_TOKEN}"
+                        res = RequestUtils(timeout=15).post_res(api_url)
+                        if res:
+                            results = res.json()
+                    if results:
+                        logger.info(f"{name} 获取到 {len(results)} 条数据")
+                        self._medias.extend([MediaInfo(**r) for r in results])
+                    else:
+                        logger.error(f"{name} 获取数据失败")
+            else:
+                # 调用内部API获取数据
+                api_url = f"http://127.0.0.1:{settings.PORT}{params.api_path}?token={settings.API_TOKEN}"
+                res = RequestUtils(timeout=15).post_res(api_url)
+                if res:
+                    results = res.json()
+                    if results:
+                        logger.info(f"{params.api_path} 获取到 {len(results)} 条数据")
+                        self._medias.extend([MediaInfo(**r) for r in results])
+        except Exception as e:
+            logger.error(f"获取媒体数据失败: {e}")
+            self._has_error = True
 
         if self._medias:
             context.medias.extend(self._medias)

app/actions/fetch_rss.py

@@ -15,12 +15,13 @@ class FetchRssParams(ActionParams):
     """
     获取RSS资源列表参数
     """
-    url: str = Field(None, description="RSS地址")
-    proxy: Optional[bool] = Field(False, description="是否使用代理")
-    timeout: Optional[int] = Field(15, description="超时时间")
-    content_type: Optional[str] = Field(None, description="Content-Type")
-    referer: Optional[str] = Field(None, description="Referer")
-    ua: Optional[str] = Field(None, description="User-Agent")
+    url: str = Field(default=None, description="RSS地址")
+    proxy: Optional[bool] = Field(default=False, description="是否使用代理")
+    timeout: Optional[int] = Field(default=15, description="超时时间")
+    content_type: Optional[str] = Field(default=None, description="Content-Type")
+    referer: Optional[str] = Field(default=None, description="Referer")
+    ua: Optional[str] = Field(default=None, description="User-Agent")
+    match_media: Optional[str] = Field(default=None, description="匹配媒体信息")
 
 
 class FetchRssAction(BaseAction):
@@ -31,24 +32,26 @@ class FetchRssAction(BaseAction):
     _rss_torrents = []
     _has_error = False
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.rsshelper = RssHelper()
         self.chain = ActionChain()
+        self._rss_torrents = []
+        self._has_error = False
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "获取RSS资源"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "订阅RSS地址获取资源"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return FetchRssParams().dict()
 
     @property
@@ -98,10 +101,12 @@ class FetchRssAction(BaseAction):
                 pubdate=item["pubdate"].strftime("%Y-%m-%d %H:%M:%S") if item.get("pubdate") else None,
             )
             meta = MetaInfo(title=torrentinfo.title, subtitle=torrentinfo.description)
-            mediainfo = self.chain.recognize_media(meta)
-            if not mediainfo:
-                logger.warning(f"{torrentinfo.title} 未识别到媒体信息")
-                continue
+            mediainfo = None
+            if params.match_media:
+                mediainfo = self.chain.recognize_media(meta)
+                if not mediainfo:
+                    logger.warning(f"{torrentinfo.title} 未识别到媒体信息")
+                    continue
             self._rss_torrents.append(Context(meta_info=meta, media_info=mediainfo, torrent_info=torrentinfo))
 
         if self._rss_torrents:

app/actions/fetch_torrents.py

@@ -15,12 +15,13 @@ class FetchTorrentsParams(ActionParams):
     """
     获取站点资源参数
     """
-    search_type: Optional[str] = Field("keyword", description="搜索类型")
-    name: Optional[str] = Field(None, description="资源名称")
-    year: Optional[str] = Field(None, description="年份")
-    type: Optional[str] = Field(None, description="资源类型 (电影/电视剧)")
-    season: Optional[int] = Field(None, description="季度")
-    sites: Optional[List[int]] = Field([], description="站点列表")
+    search_type: Optional[str] = Field(default="keyword", description="搜索类型")
+    name: Optional[str] = Field(default=None, description="资源名称")
+    year: Optional[str] = Field(default=None, description="年份")
+    type: Optional[str] = Field(default=None, description="资源类型 (电影/电视剧)")
+    season: Optional[int] = Field(default=None, description="季度")
+    sites: Optional[List[int]] = Field(default=[], description="站点列表")
+    match_media: Optional[bool] = Field(default=False, description="匹配媒体信息")
 
 
 class FetchTorrentsAction(BaseAction):
@@ -30,23 +31,24 @@ class FetchTorrentsAction(BaseAction):
 
     _torrents = []
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.searchchain = SearchChain()
+        self._torrents = []
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "搜索站点资源"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "搜索站点种子资源列表"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return FetchTorrentsParams().dict()
 
     @property
@@ -71,10 +73,11 @@ class FetchTorrentsAction(BaseAction):
                 if params.season and torrent.meta_info.begin_season != params.season:
                     continue
                 # 识别媒体信息
-                torrent.media_info = self.searchchain.recognize_media(torrent.meta_info)
-                if not torrent.media_info:
-                    logger.warning(f"{torrent.torrent_info.title} 未识别到媒体信息")
-                    continue
+                if params.match_media:
+                    torrent.media_info = self.searchchain.recognize_media(torrent.meta_info)
+                    if not torrent.media_info:
+                        logger.warning(f"{torrent.torrent_info.title} 未识别到媒体信息")
+                        continue
                 self._torrents.append(torrent)
         else:
             # 搜索媒体列表
@@ -88,8 +91,8 @@ class FetchTorrentsAction(BaseAction):
                 for torrent in torrents:
                     self._torrents.append(torrent)
 
-                # 随机休眠 10-60秒
-                sleep_time = random.randint(10, 60)
+                # 随机休眠 5-30秒
+                sleep_time = random.randint(5, 30)
                 logger.info(f"随机休眠 {sleep_time} 秒 ...")
                 time.sleep(sleep_time)
 

app/actions/filter_medias.py

@@ -4,6 +4,7 @@ from pydantic import Field
 
 from app.actions import BaseAction
 from app.core.config import global_vars
+from app.log import logger
 from app.schemas import ActionParams, ActionContext
 
 
@@ -11,10 +12,9 @@ class FilterMediasParams(ActionParams):
     """
     过滤媒体数据参数
     """
-    type: Optional[str] = Field(None, description="媒体类型 (电影/电视剧)")
-    category: Optional[str] = Field(None, description="媒体类别 (二级分类)")
-    vote: Optional[int] = Field(0, description="评分")
-    year: Optional[str] = Field(None, description="年份")
+    type: Optional[str] = Field(default=None, description="媒体类型 (电影/电视剧)")
+    vote: Optional[int] = Field(default=0, description="评分")
+    year: Optional[str] = Field(default=None, description="年份")
 
 
 class FilterMediasAction(BaseAction):
@@ -24,19 +24,23 @@ class FilterMediasAction(BaseAction):
 
     _medias = []
 
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
+        self._medias = []
+
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "过滤媒体数据"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "对媒体数据列表进行过滤"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return FilterMediasParams().dict()
 
     @property
@@ -53,16 +57,15 @@ class FilterMediasAction(BaseAction):
                 break
             if params.type and media.type != params.type:
                 continue
-            if params.category and media.category != params.category:
-                continue
             if params.vote and media.vote_average < params.vote:
                 continue
             if params.year and media.year != params.year:
                 continue
             self._medias.append(media)
 
-        if self._medias:
-            context.medias = self._medias
+        logger.info(f"过滤后剩余 {len(self._medias)} 条媒体数据")
+
+        context.medias = self._medias
 
         self.job_done(f"过滤后剩余 {len(self._medias)} 条媒体数据")
         return context

app/actions/filter_torrents.py

@@ -5,6 +5,7 @@ from pydantic import Field
 from app.actions import BaseAction, ActionChain
 from app.core.config import global_vars
 from app.helper.torrent import TorrentHelper
+from app.log import logger
 from app.schemas import ActionParams, ActionContext
 
 
@@ -12,13 +13,13 @@ class FilterTorrentsParams(ActionParams):
     """
     过滤资源数据参数
     """
-    rule_groups: Optional[List[str]] = Field([], description="规则组")
-    quality: Optional[str] = Field(None, description="资源质量")
-    resolution: Optional[str] = Field(None, description="资源分辨率")
-    effect: Optional[str] = Field(None, description="特效")
-    include: Optional[str] = Field(None, description="包含规则")
-    exclude: Optional[str] = Field(None, description="排除规则")
-    size: Optional[str] = Field(None, description="资源大小范围（MB）")
+    rule_groups: Optional[List[str]] = Field(default=[], description="规则组")
+    quality: Optional[str] = Field(default=None, description="资源质量")
+    resolution: Optional[str] = Field(default=None, description="资源分辨率")
+    effect: Optional[str] = Field(default=None, description="特效")
+    include: Optional[str] = Field(default=None, description="包含规则")
+    exclude: Optional[str] = Field(default=None, description="排除规则")
+    size: Optional[str] = Field(default=None, description="资源大小范围（MB）")
 
 
 class FilterTorrentsAction(BaseAction):
@@ -28,24 +29,25 @@ class FilterTorrentsAction(BaseAction):
 
     _torrents = []
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.torrenthelper = TorrentHelper()
         self.chain = ActionChain()
+        self._torrents = []
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "过滤资源"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "对资源列表数据进行过滤"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return FilterTorrentsParams().dict()
 
     @property
@@ -78,6 +80,8 @@ class FilterTorrentsAction(BaseAction):
                 ):
                     self._torrents.append(torrent)
 
+        logger.info(f"过滤后剩余 {len(self._torrents)} 个资源")
+
         context.torrents = self._torrents
 
         self.job_done(f"过滤后剩余 {len(self._torrents)} 个资源")

app/actions/scan_file.py

@@ -15,8 +15,8 @@ class ScanFileParams(ActionParams):
     整理文件参数
     """
     # 存储
-    storage: Optional[str] = Field("local", description="存储")
-    directory: Optional[str] = Field(None, description="目录")
+    storage: Optional[str] = Field(default="local", description="存储")
+    directory: Optional[str] = Field(default=None, description="目录")
 
 
 class ScanFileAction(BaseAction):
@@ -27,23 +27,25 @@ class ScanFileAction(BaseAction):
     _fileitems = []
     _has_error = False
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.storagechain = StorageChain()
+        self._fileitems = []
+        self._has_error = False
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "扫描目录"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "扫描目录文件到队列"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return ScanFileParams().dict()
 
     @property
@@ -68,7 +70,14 @@ class ScanFileAction(BaseAction):
                 break
             if not file.extension or f".{file.extension.lower()}" not in settings.RMT_MEDIAEXT:
                 continue
+            # 检查缓存
+            cache_key = f"{file.path}"
+            if self.check_cache(workflow_id, cache_key):
+                logger.info(f"{file.path} 已处理过，跳过")
+                continue
             self._fileitems.append(fileitem)
+            # 保存缓存
+            self.save_cache(workflow_id, cache_key)
 
         if self._fileitems:
             context.fileitems.extend(self._fileitems)

app/actions/scrape_file.py

@@ -24,24 +24,26 @@ class ScrapeFileAction(BaseAction):
     _scraped_files = []
     _has_error = False
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.storagechain = StorageChain()
         self.mediachain = MediaChain()
+        self._scraped_files = []
+        self._has_error = False
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "刮削文件"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "刮削媒体信息和图片"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return ScrapeFileParams().dict()
 
     @property
@@ -52,6 +54,8 @@ class ScrapeFileAction(BaseAction):
         """
         刮削fileitems中的所有文件
         """
+        # 失败次数
+        _failed_count = 0
         for fileitem in context.fileitems:
             if global_vars.is_workflow_stopped(workflow_id):
                 break
@@ -59,14 +63,24 @@ class ScrapeFileAction(BaseAction):
                 continue
             if not self.storagechain.exists(fileitem):
                 continue
+            # 检查缓存
+            cache_key = f"{fileitem.path}"
+            if self.check_cache(workflow_id, cache_key):
+                logger.info(f"{fileitem.path} 已刮削过，跳过")
+                continue
             meta = MetaInfoPath(Path(fileitem.path))
             mediainfo = self.mediachain.recognize_media(meta)
             if not mediainfo:
-                self._has_error = True
+                _failed_count += 1
                 logger.info(f"{fileitem.path} 未识别到媒体信息，无法刮削")
                 continue
             self.mediachain.scrape_metadata(fileitem=fileitem, meta=meta, mediainfo=mediainfo)
             self._scraped_files.append(fileitem)
+            # 保存缓存
+            self.save_cache(workflow_id, cache_key)
 
-        self.job_done(f"成功刮削了 {len(self._scraped_files)} 个文件")
+        if not self._scraped_files and _failed_count:
+            self._has_error = True
+
+        self.job_done(f"成功刮削 {len(self._scraped_files)} 个文件，失败 {_failed_count} 个")
         return context

app/actions/send_event.py

@@ -18,17 +18,17 @@ class SendEventAction(BaseAction):
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "发送事件"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "发送任务执行事件"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return SendEventParams().dict()
 
     @property

app/actions/send_message.py

@@ -4,14 +4,15 @@ from pydantic import Field
 
 from app.actions import BaseAction, ActionChain
 from app.schemas import ActionParams, ActionContext, Notification
+from core.config import settings
 
 
 class SendMessageParams(ActionParams):
     """
     发送消息参数
     """
-    client: Optional[List[str]] = Field([], description="消息渠道")
-    userid: Optional[Union[str, int]] = Field(None, description="用户ID")
+    client: Optional[List[str]] = Field(default=[], description="消息渠道")
+    userid: Optional[Union[str, int]] = Field(default=None, description="用户ID")
 
 
 class SendMessageAction(BaseAction):
@@ -19,23 +20,23 @@ class SendMessageAction(BaseAction):
     发送消息
     """
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.chain = ActionChain()
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "发送消息"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "发送任务执行消息"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return SendMessageParams().dict()
 
     @property
@@ -57,7 +58,7 @@ class SendMessageAction(BaseAction):
                 index += 1
             # 发送消息
             if not params.client:
-                params.client = [None]
+                params.client = [""]
             for client in params.client:
                 self.chain.post_message(
                     Notification(
@@ -65,7 +66,7 @@ class SendMessageAction(BaseAction):
                         userid=params.userid,
                         title="【工作流执行结果】",
                         text=msg_text,
-                        link="#/workflow"
+                        link=settings.MP_DOMAIN("#/workflow")
                     )
                 )
 

app/actions/transfer_file.py

@@ -18,7 +18,7 @@ class TransferFileParams(ActionParams):
     整理文件参数
     """
     # 来源
-    source: Optional[str] = Field("downloads", description="来源")
+    source: Optional[str] = Field(default="downloads", description="来源")
 
 
 class TransferFileAction(BaseAction):
@@ -29,25 +29,27 @@ class TransferFileAction(BaseAction):
     _fileitems = []
     _has_error = False
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, action_id: str):
+        super().__init__(action_id)
         self.transferchain = TransferChain()
         self.storagechain = StorageChain()
         self.transferhis = TransferHistoryOper()
+        self._fileitems = []
+        self._has_error = False
 
     @classmethod
     @property
-    def name(cls) -> str:
+    def name(cls) -> str: # noqa
         return "整理文件"
 
     @classmethod
     @property
-    def description(cls) -> str:
+    def description(cls) -> str: # noqa
         return "整理队列中的文件"
 
     @classmethod
     @property
-    def data(cls) -> dict:
+    def data(cls) -> dict: # noqa
         return TransferFileParams().dict()
 
     @property
@@ -68,6 +70,8 @@ class TransferFileAction(BaseAction):
             return True
 
         params = TransferFileParams(**params)
+        # 失败次数
+        _failed_count = 0
         if params.source == "downloads":
             # 从下载任务中整理文件
             for download in context.downloads:
@@ -76,6 +80,11 @@ class TransferFileAction(BaseAction):
                 if not download.completed:
                     logger.info(f"下载任务 {download.download_id} 未完成")
                     continue
+                # 检查缓存
+                cache_key = f"{download.download_id}"
+                if self.check_cache(workflow_id, cache_key):
+                    logger.info(f"{download.path} 已整理过，跳过")
+                    continue
                 fileitem = self.storagechain.get_file_item(storage="local", path=Path(download.path))
                 if not fileitem:
                     logger.info(f"文件 {download.path} 不存在")
@@ -87,16 +96,22 @@ class TransferFileAction(BaseAction):
                 logger.info(f"开始整理文件 {download.path} ...")
                 state, errmsg = self.transferchain.do_transfer(fileitem, background=False)
                 if not state:
-                    self._has_error = True
+                    _failed_count += 1
                     logger.error(f"整理文件 {download.path} 失败: {errmsg}")
                     continue
                 logger.info(f"整理文件 {download.path} 完成")
                 self._fileitems.append(fileitem)
+                self.save_cache(workflow_id, cache_key)
         else:
             # 从 fileitems 中整理文件
             for fileitem in copy.deepcopy(context.fileitems):
                 if not check_continue():
                     break
+                # 检查缓存
+                cache_key = f"{fileitem.path}"
+                if self.check_cache(workflow_id, cache_key):
+                    logger.info(f"{fileitem.path} 已整理过，跳过")
+                    continue
                 transferd = self.transferhis.get_by_src(fileitem.path, storage=fileitem.storage)
                 if transferd:
                     # 已经整理过的文件不再整理
@@ -105,16 +120,20 @@ class TransferFileAction(BaseAction):
                 state, errmsg = self.transferchain.do_transfer(fileitem, background=False,
                                                                continue_callback=check_continue)
                 if not state:
-                    self._has_error = True
+                    _failed_count += 1
                     logger.error(f"整理文件 {fileitem.path} 失败: {errmsg}")
                     continue
                 logger.info(f"整理文件 {fileitem.path} 完成")
                 # 从 fileitems 中移除已整理的文件
                 context.fileitems.remove(fileitem)
                 self._fileitems.append(fileitem)
+                # 记录已整理的文件
+                self.save_cache(workflow_id, cache_key)
 
         if self._fileitems:
             context.fileitems.extend(self._fileitems)
+        elif _failed_count:
+            self._has_error = True
 
-        self.job_done()
+        self.job_done(f"整理成功 {len(self._fileitems)} 个文件，失败 {_failed_count} 个")
         return context

app/api/endpoints/search.py

@@ -29,7 +29,7 @@ def search_by_id(mediaid: str,
                  mtype: str = None,
                  area: str = "title",
                  title: str = None,
-                 year: int = None,
+                 year: str = None,
                  season: str = None,
                  sites: str = None,
                  _: schemas.TokenPayload = Depends(verify_token)) -> Any:

app/api/endpoints/system.py

@@ -24,7 +24,7 @@ from app.db.models import User
 from app.db.systemconfig_oper import SystemConfigOper
 from app.db.user_oper import get_current_active_superuser
 from app.helper.mediaserver import MediaServerHelper
-from app.helper.message import MessageHelper
+from app.helper.message import MessageHelper, MessageQueueManager
 from app.helper.progress import ProgressHelper
 from app.helper.rule import RuleHelper
 from app.helper.sites import SitesHelper
@@ -479,6 +479,7 @@ def reload_module(_: User = Depends(get_current_active_superuser)):
     """
     重新加载模块（仅管理员）
     """
+    MessageQueueManager().init_config()
     ModuleManager().reload()
     Scheduler().init()
     Monitor().init()

app/api/endpoints/workflow.py

@@ -9,6 +9,7 @@ from app.core.config import global_vars
 from app.core.workflow import WorkFlowManager
 from app.db import get_db
 from app.db.models.workflow import Workflow
+from app.db.systemconfig_oper import SystemConfigOper
 from app.db.user_oper import get_current_active_user
 from app.chain.workflow import WorkflowChain
 from app.scheduler import Scheduler
@@ -84,8 +85,12 @@ def delete_workflow(workflow_id: int,
     workflow = Workflow.get(db, workflow_id)
     if not workflow:
         return schemas.Response(success=False, message="工作流不存在")
+    # 删除定时任务
     Scheduler().remove_workflow_job(workflow)
+    # 删除工作流
     Workflow.delete(db, workflow_id)
+    # 删除缓存
+    SystemConfigOper().delete(f"WorkflowCache-{workflow_id}")
     return schemas.Response(success=True, message="删除成功")
 
 
@@ -112,8 +117,9 @@ def start_workflow(workflow_id: int,
     workflow = Workflow.get(db, workflow_id)
     if not workflow:
         return schemas.Response(success=False, message="工作流不存在")
+    # 添加定时任务
     Scheduler().update_workflow_job(workflow)
-    global_vars.workflow_resume(workflow_id)
+    # 更新状态
     workflow.update_state(db, workflow_id, "W")
     return schemas.Response(success=True)
 
@@ -128,7 +134,29 @@ def pause_workflow(workflow_id: int,
     workflow = Workflow.get(db, workflow_id)
     if not workflow:
         return schemas.Response(success=False, message="工作流不存在")
+    # 删除定时任务
     Scheduler().remove_workflow_job(workflow)
+    # 停止工作流
     global_vars.stop_workflow(workflow_id)
+    # 更新状态
     workflow.update_state(db, workflow_id, "P")
     return schemas.Response(success=True)
+
+
+@router.post("/{workflow_id}/reset", summary="重置工作流", response_model=schemas.Response)
+def reset_workflow(workflow_id: int,
+                   db: Session = Depends(get_db),
+                   _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any:
+    """
+    重置工作流
+    """
+    workflow = Workflow.get(db, workflow_id)
+    if not workflow:
+        return schemas.Response(success=False, message="工作流不存在")
+    # 停止工作流
+    global_vars.stop_workflow(workflow_id)
+    # 重置工作流
+    workflow.reset(db, workflow_id)
+    # 删除缓存
+    SystemConfigOper().delete(f"WorkflowCache-{workflow_id}")
+    return schemas.Response(success=True)

app/chain/__init__.py

@@ -16,7 +16,7 @@ from app.core.meta import MetaBase
 from app.core.module import ModuleManager
 from app.db.message_oper import MessageOper
 from app.db.user_oper import UserOper
-from app.helper.message import MessageHelper
+from app.helper.message import MessageHelper, MessageQueueManager
 from app.helper.service import ServiceConfigHelper
 from app.log import logger
 from app.schemas import TransferInfo, TransferTorrent, ExistMediaInfo, DownloadingTorrent, CommingMessage, Notification, \
@@ -38,6 +38,9 @@ class ChainBase(metaclass=ABCMeta):
         self.eventmanager = EventManager()
         self.messageoper = MessageOper()
         self.messagehelper = MessageHelper()
+        self.messagequeue = MessageQueueManager(
+            send_callback=self.run_module
+        )
         self.useroper = UserOper()
 
     @staticmethod
@@ -347,7 +350,7 @@ class ChainBase(metaclass=ABCMeta):
                                torrent_list=torrent_list, mediainfo=mediainfo)
 
     def download(self, content: Union[Path, str], download_dir: Path, cookie: str,
-                 episodes: Set[int] = None, category: str = None,
+                 episodes: Set[int] = None, category: str = None, label: str = None,
                  downloader: str = None
                  ) -> Optional[Tuple[Optional[str], Optional[str], Optional[str], str]]:
         """
@@ -357,11 +360,12 @@ class ChainBase(metaclass=ABCMeta):
         :param cookie:  cookie
         :param episodes:  需要下载的集数
         :param category:  种子分类
+        :param label:  标签
         :param downloader:  下载器
         :return: 下载器名称、种子Hash、种子文件布局、错误原因
         """
         return self.run_module("download", content=content, download_dir=download_dir,
-                               cookie=cookie, episodes=episodes, category=category,
+                               cookie=cookie, episodes=episodes, category=category, label=label,
                                downloader=downloader)
 
     def download_added(self, context: Context, download_dir: Path, torrent_path: Path = None) -> None:
@@ -490,11 +494,6 @@ class ChainBase(metaclass=ABCMeta):
         :param message:  消息体
         :return: 成功或失败
         """
-        logger.info(f"发送消息：channel={message.channel}，"
-                    f"source={message.source},"
-                    f"title={message.title}, "
-                    f"text={message.text}，"
-                    f"userid={message.userid}")
         # 保存原消息
         self.messagehelper.put(message, role="user", title=message.title)
         self.messageoper.add(**message.dict())
@@ -544,13 +543,13 @@ class ChainBase(metaclass=ABCMeta):
                     # 按设定发送
                     self.eventmanager.send_event(etype=EventType.NoticeMessage,
                                                  data={**send_message.dict(), "type": send_message.mtype})
-                    self.run_module("post_message", message=send_message)
+                    self.messagequeue.send_message("post_message", message=send_message)
                 if not send_orignal:
                     return
         # 发送消息事件
         self.eventmanager.send_event(etype=EventType.NoticeMessage, data={**message.dict(), "type": message.mtype})
         # 按原消息发送
-        self.run_module("post_message", message=message)
+        self.messagequeue.send_message("post_message", message=message)
 
     def post_medias_message(self, message: Notification, medias: List[MediaInfo]) -> None:
         """
@@ -562,7 +561,7 @@ class ChainBase(metaclass=ABCMeta):
         note_list = [media.to_dict() for media in medias]
         self.messagehelper.put(message, role="user", note=note_list, title=message.title)
         self.messageoper.add(**message.dict(), note=note_list)
-        return self.run_module("post_medias_message", message=message, medias=medias)
+        return self.messagequeue.send_message("post_medias_message", message=message, medias=medias)
 
     def post_torrents_message(self, message: Notification, torrents: List[Context]) -> None:
         """
@@ -574,7 +573,7 @@ class ChainBase(metaclass=ABCMeta):
         note_list = [torrent.torrent_info.to_dict() for torrent in torrents]
         self.messagehelper.put(message, role="user", note=note_list, title=message.title)
         self.messageoper.add(**message.dict(), note=note_list)
-        return self.run_module("post_torrents_message", message=message, torrents=torrents)
+        return self.messagequeue.send_message("post_torrents_message", message=message, torrents=torrents)
 
     def metadata_img(self, mediainfo: MediaInfo, season: int = None, episode: int = None) -> Optional[dict]:
         """

app/chain/download.py

@@ -209,7 +209,8 @@ class DownloadChain(ChainBase):
                         save_path: str = None,
                         userid: Union[str, int] = None,
                         username: str = None,
-                        media_category: str = None) -> Optional[str]:
+                        media_category: str = None,
+                        label: str = None) -> Optional[str]:
         """
         下载及发送通知
         :param context: 资源上下文
@@ -222,6 +223,7 @@ class DownloadChain(ChainBase):
         :param userid: 用户ID
         :param username: 调用下载的用户名/插件名
         :param media_category: 自定义媒体类别
+        :param label: 自定义标签
         """
         # 发送资源下载事件，允许外部拦截下载
         event_data = ResourceDownloadEventData(
@@ -310,6 +312,7 @@ class DownloadChain(ChainBase):
                                                 episodes=episodes,
                                                 download_dir=download_dir,
                                                 category=_media.category,
+                                                label=label,
                                                 downloader=downloader or _site_downloader)
         if result:
             _downloader, _hash, _layout, error_msg = result

app/chain/mediaserver.py

@@ -1,5 +1,5 @@
 import threading
-from typing import List, Union, Optional, Generator
+from typing import List, Union, Optional, Generator, Any
 
 from app.chain import ChainBase
 from app.core.cache import cached
@@ -27,8 +27,8 @@ class MediaServerChain(ChainBase):
         """
         return self.run_module("mediaserver_librarys", server=server, username=username, hidden=hidden)
 
-    def items(self, server: str, library_id: Union[str, int], start_index: int = 0, limit: Optional[int] = -1) \
-            -> Optional[Generator]:
+    def items(self, server: str, library_id: Union[str, int],
+              start_index: int = 0, limit: Optional[int] = -1) -> Generator[Any, None, None]:
         """
         获取媒体服务器项目列表，支持分页和不分页逻辑，默认不分页获取所有数据
 

app/chain/search.py

@@ -312,11 +312,6 @@ class SearchChain(ChainBase):
         for indexer in self.siteshelper.get_indexers():
             # 检查站点索引开关
             if not sites or indexer.get("id") in sites:
-                # 站点流控
-                state, msg = self.siteshelper.check(indexer.get("domain"))
-                if state:
-                    logger.warn(msg)
-                    continue
                 indexer_sites.append(indexer)
         if not indexer_sites:
             logger.warn('未开启任何有效站点，无法搜索资源')

app/chain/site.py

@@ -52,6 +52,7 @@ class SiteChain(ChainBase):
             "1ptba.com": self.__indexphp_test,
             "star-space.net": self.__indexphp_test,
             "yemapt.org": self.__yema_test,
+            "hddolby.com": self.__hddolby_test,
         }
 
     def refresh_userdata(self, site: dict = None) -> Optional[SiteUserData]:
@@ -251,6 +252,32 @@ class SiteChain(ChainBase):
         site.url = f"{site.url}index.php"
         return self.__test(site)
 
+    @staticmethod
+    def __hddolby_test(site: Site) -> Tuple[bool, str]:
+        """
+        判断站点是否已经登陆：hddolby
+        """
+        url = f"{site.url}api/v1/user/data"
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json, text/plain, */*",
+            "x-api-key": site.apikey,
+        }
+        res = RequestUtils(
+            headers=headers,
+            proxies=settings.PROXY if site.proxy else None,
+            timeout=site.timeout or 15
+        ).get_res(url=url)
+        if res is None:
+            return False, "无法打开网站！"
+        if res.status_code == 200:
+            user_info = res.json()
+            if user_info and user_info.get("status") == 0:
+                return True, "连接成功"
+            return False, "APIKEY已过期"
+        else:
+            return False, f"错误：{res.status_code} {res.reason}"
+
     @staticmethod
     def __parse_favicon(url: str, cookie: str, ua: str) -> Tuple[str, Optional[str]]:
         """

app/chain/subscribe.py

@@ -1262,7 +1262,7 @@ class SubscribeChain(ChainBase, metaclass=Singleton):
         订阅相关的下载和文件信息
         """
         if not subscribe:
-            return
+            return None
 
         # 返回订阅数据
         subscribe_info = schemas.SubscrbieInfo()

app/chain/transfer.py

@@ -606,7 +606,7 @@ class TransferChain(ChainBase, metaclass=Singleton):
                 logger.error(f"整理队列处理出现错误：{e} - {traceback.format_exc()}")
 
     def __handle_transfer(self, task: TransferTask,
-                          callback: Optional[Callable] = None) -> Tuple[bool, str]:
+                          callback: Optional[Callable] = None) -> Optional[Tuple[bool, str]]:
         """
         处理整理任务
         """
@@ -671,9 +671,17 @@ class TransferChain(ChainBase, metaclass=Singleton):
 
             # 获取集数据
             if task.mediainfo.type == MediaType.TV and not task.episodes_info:
+                # 判断注意season为0的情况
+                season_num = task.mediainfo.season
+                if season_num is None and task.meta.season_seq:
+                    if task.meta.season_seq.isdigit():
+                        season_num = int(task.meta.season_seq)
+                # 默认值1
+                if season_num is None:
+                    season_num = 1
                 task.episodes_info = self.tmdbchain.tmdb_episodes(
                     tmdbid=task.mediainfo.tmdb_id,
-                    season=task.mediainfo.season or task.meta.begin_season or 1
+                    season=season_num
                 )
 
             # 查询整理目标目录

app/chain/workflow.py

@@ -66,6 +66,7 @@ class WorkflowExecutor:
 
         # 初始上下文
         if workflow.current_action and workflow.context:
+            logger.info(f"工作流已执行动作：{workflow.current_action}")
             # Base64解码
             decoded_data = base64.b64decode(workflow.context["content"])
             # 反序列化数据
@@ -73,7 +74,9 @@ class WorkflowExecutor:
         else:
             self.context = ActionContext()
 
-        # 初始化队列：入度为0的节点
+        # 恢复工作流
+        global_vars.workflow_resume(self.workflow.id)
+        # 初始化队列，添加入度为0的节点
         for action_id in self.actions:
             if self.indegree[action_id] == 0:
                 self.queue.append(action_id)
@@ -91,7 +94,7 @@ class WorkflowExecutor:
                 if not self.success:
                     break
                 if not self.queue:
-                    sleep(1)
+                    sleep(0.1)
                     continue
                 # 取出队首节点
                 node_id = self.queue.popleft()

app/core/config.py

@@ -363,7 +363,7 @@ class Settings(BaseSettings, ConfigModel, LogConfigModel):
                 raise ValueError(f"配置项 '{field_name}' 的值 '{value}' 无法转换成正确的类型") from e
             logger.error(
                 f"配置项 '{field_name}' 的值 '{value}' 无法转换成正确的类型，使用默认值 '{default}'，错误信息: {e}")
-            return default, True
+        return default, True
 
     @validator('*', pre=True, always=True)
     def generic_type_validator(cls, value: Any, field):  # noqa

app/core/module.py

@@ -121,7 +121,7 @@ class ModuleManager(metaclass=Singleton):
         获取实现了同一方法的模块列表
         """
         if not self._running_modules:
-            return []
+            return
         for _, module in self._running_modules.items():
             if hasattr(module, method) \
                     and ObjectUtils.check_method(getattr(module, method)):
@@ -132,7 +132,7 @@ class ModuleManager(metaclass=Singleton):
         获取指定类型的模块列表
         """
         if not self._running_modules:
-            return []
+            return
         for _, module in self._running_modules.items():
             if hasattr(module, 'get_type') \
                     and module.get_type() == module_type:
@@ -143,7 +143,7 @@ class ModuleManager(metaclass=Singleton):
         获取指定子类型的模块
         """
         if not self._running_modules:
-            return []
+            return
         for _, module in self._running_modules.items():
             if hasattr(module, 'get_subtype') \
                     and module.get_subtype() == module_subtype:

app/core/security.py

@@ -4,7 +4,8 @@ import hmac
 import json
 import os
 import traceback
-from datetime import datetime, timedelta
+import datetime
+from datetime import timedelta
 from typing import Any, Union, Annotated, Optional
 
 import jwt
@@ -69,13 +70,13 @@ def create_access_token(
     if expires_delta is not None:
         if expires_delta.total_seconds() <= 0:
             raise ValueError("过期时间必须为正数")
-        expire = datetime.utcnow() + expires_delta
+        expire = datetime.datetime.now(datetime.UTC) + expires_delta
     else:
-        expire = datetime.utcnow() + default_expire
+        expire = datetime.datetime.now(datetime.UTC) + default_expire
 
     to_encode = {
         "exp": expire,
-        "iat": datetime.utcnow(),
+        "iat": datetime.datetime.now(datetime.UTC),
         "sub": str(userid),
         "username": username,
         "super_user": super_user,
@@ -102,7 +103,7 @@ def __set_or_refresh_resource_token_cookie(request: Request, response: Response,
             decoded_token = jwt.decode(resource_token, settings.RESOURCE_SECRET_KEY, algorithms=[ALGORITHM])
             exp = decoded_token.get("exp")
             if exp:
-                remaining_time = datetime.utcfromtimestamp(exp) - datetime.utcnow()
+                remaining_time = datetime.datetime.fromtimestamp(exp, tz=datetime.UTC) - datetime.datetime.now(datetime.UTC)
                 # 根据剩余时长提前刷新令牌
                 if remaining_time < timedelta(seconds=(settings.RESOURCE_ACCESS_TOKEN_EXPIRE_SECONDS / 3)):
                     raise jwt.ExpiredSignatureError

app/core/workflow.py

@@ -63,8 +63,10 @@ class WorkFlowManager(metaclass=Singleton):
         if not context:
             context = ActionContext()
         if action.type in self._actions:
+            # 实例化之前，清理掉类对象的数据
+
             # 实例化
-            action_obj = self._actions[action.type]()
+            action_obj = self._actions[action.type](action.id)
             # 执行
             logger.info(f"执行动作: {action.id} - {action.name}")
             try:

app/db/models/workflow.py

@@ -88,6 +88,7 @@ class Workflow(Base):
             "state": 'W',
             "result": None,
             "current_action": None,
+            "run_count": 0,
         })
         return True
 
@@ -95,7 +96,7 @@ class Workflow(Base):
     @db_update
     def update_current_action(db, wid: int, action_id: str, context: dict):
         db.query(Workflow).filter(Workflow.id == wid).update({
-            "current_action": f"{Workflow.current_action},{action_id}" if Workflow.current_action else action_id,
+            "current_action": Workflow.current_action + f",{action_id}" if Workflow.current_action else action_id,
             "context": context
         })
         return True

app/helper/browser.py

@@ -1,4 +1,4 @@
-from typing import Callable, Any
+from typing import Callable, Any, Optional
 
 from playwright.sync_api import sync_playwright, Page
 from cf_clearance import sync_cf_retry, sync_stealth
@@ -61,7 +61,7 @@ class PlaywrightHelper:
                         ua: str = None,
                         proxies: dict = None,
                         headless: bool = False,
-                        timeout: int = 20) -> str:
+                        timeout: int = 20) -> Optional[str]:
         """
         获取网页源码
         :param url: 网页地址

app/helper/message.py

@@ -1,9 +1,152 @@
+from __future__ import annotations
+
 import json
 import queue
+import threading
 import time
-from typing import Optional, Any, Union
+from datetime import datetime
+from typing import Any, Union
+from typing import List, Optional, Callable
 
-from app.utils.singleton import Singleton
+from app.core.config import global_vars
+from app.db.systemconfig_oper import SystemConfigOper
+from app.schemas.types import SystemConfigKey
+from app.utils.singleton import Singleton, SingletonClass
+from app.log import logger
+
+
+class MessageQueueManager(metaclass=SingletonClass):
+    """
+    消息发送队列管理器
+    """
+
+    schedule_periods: List[tuple[int, int, int, int]] = []
+
+    def __init__(
+            self,
+            send_callback: Optional[Callable] = None,
+            check_interval: int = 10
+    ) -> None:
+        """
+        消息队列管理器初始化
+
+        :param send_callback: 实际发送消息的回调函数
+        :param check_interval: 时间检查间隔（秒）
+        """
+        self.init_config()
+
+        self.queue: queue.Queue[Any] = queue.Queue()
+        self.send_callback = send_callback
+        self.check_interval = check_interval
+
+        self._running = True
+        self.thread = threading.Thread(target=self._monitor_loop, daemon=True)
+        self.thread.start()
+
+    def init_config(self):
+        """
+        初始化配置
+        """
+        self.schedule_periods = self._parse_schedule(
+            SystemConfigOper().get(SystemConfigKey.NotificationSendTime)
+        )
+
+    @staticmethod
+    def _parse_schedule(periods: Union[list, dict]) -> List[tuple[int, int, int, int]]:
+        """
+        将字符串时间格式转换为分钟数元组
+        """
+        parsed = []
+        if not periods:
+            return parsed
+        if not isinstance(periods, list):
+            periods = [periods]
+        for period in periods:
+            if not period:
+                continue
+            start_h, start_m = map(int, period['start'].split(':'))
+            end_h, end_m = map(int, period['end'].split(':'))
+            parsed.append((start_h, start_m, end_h, end_m))
+        return parsed
+
+    @staticmethod
+    def _time_to_minutes(time_str: str) -> int:
+        """
+        将 'HH:MM' 格式转换为分钟数
+        """
+        hours, minutes = map(int, time_str.split(':'))
+        return hours * 60 + minutes
+
+    def _is_in_scheduled_time(self, current_time: datetime) -> bool:
+        """
+        检查当前时间是否在允许发送的时间段内
+        """
+        if not self.schedule_periods:
+            return True
+        current_minutes = current_time.hour * 60 + current_time.minute
+        for period in self.schedule_periods:
+            s_h, s_m, e_h, e_m = period
+            start = s_h * 60 + s_m
+            end = e_h * 60 + e_m
+
+            if start <= end:
+                if start <= current_minutes <= end:
+                    return True
+            else:
+                if current_minutes >= start or current_minutes <= end:
+                    return True
+        return False
+
+    def send_message(self, *args, **kwargs) -> None:
+        """
+        发送消息（立即发送或加入队列）
+        """
+        if self._is_in_scheduled_time(datetime.now()):
+            self._send(*args, **kwargs)
+        else:
+            self.queue.put({
+                "args": args,
+                "kwargs": kwargs
+            })
+            logger.info(f"消息已加入队列，当前队列长度：{self.queue.qsize()}")
+
+    def _send(self, *args, **kwargs) -> None:
+        """
+        实际发送消息（可通过回调函数自定义）
+        """
+        if self.send_callback:
+            try:
+                logger.info(f"发送消息：{kwargs}")
+                self.send_callback(*args, **kwargs)
+            except Exception as e:
+                logger.error(f"发送消息错误：{str(e)}")
+
+    def _monitor_loop(self) -> None:
+        """
+        后台线程循环检查时间并处理队列
+        """
+        while self._running:
+            current_time = datetime.now()
+            if self._is_in_scheduled_time(current_time):
+                while not self.queue.empty():
+                    if global_vars.is_system_stopped:
+                        break
+                    if not self._is_in_scheduled_time(datetime.now()):
+                        break
+                    try:
+                        message = self.queue.get_nowait()
+                        self._send(*message['args'], **message['kwargs'])
+                        logger.info(f"队列剩余消息：{self.queue.qsize()}")
+                    except queue.Empty:
+                        break
+            time.sleep(self.check_interval)
+
+    def stop(self) -> None:
+        """
+        停止队列管理器
+        """
+        self._running = False
+        self.thread.join()
 
 
 class MessageHelper(metaclass=Singleton):

app/helper/plugin.py

@@ -448,58 +448,6 @@ class PluginHelper(metaclass=Singleton):
         if plugin_dir.exists():
             shutil.rmtree(plugin_dir, ignore_errors=True)
 
-    @staticmethod
-    def __pip_uninstall_and_install_with_fallback(requirements_file: Path) -> Tuple[bool, str]:
-        """
-        先卸载 requirements.txt 中的依赖，再按照自动降级策略重新安装，不使用 PIP 缓存
-
-        :param requirements_file: 依赖的 requirements.txt 文件路径
-        :return: (是否成功, 错误信息)
-        """
-        # 读取 requirements.txt 文件中的依赖列表
-        try:
-            with open(requirements_file, "r", encoding="utf-8") as f:
-                dependencies = [line.strip() for line in f if line.strip() and not line.startswith("#")]
-        except Exception as e:
-            return False, f"无法读取 requirements.txt 文件：{str(e)}"
-
-        # 1. 先卸载所有依赖包
-        for dep in dependencies:
-            pip_uninstall_command = ["pip", "uninstall", "-y", dep]
-            logger.debug(f"尝试卸载依赖：{dep}，命令：{' '.join(pip_uninstall_command)}")
-            success, message = SystemUtils.execute_with_subprocess(pip_uninstall_command)
-            if success:
-                logger.debug(f"依赖 {dep} 卸载成功，输出：{message}")
-            else:
-                error_message = f"卸载依赖 {dep} 失败，错误信息：{message}"
-                logger.error(error_message)
-
-        # 2. 重新安装所有依赖，使用自动降级策略
-        strategies = []
-
-        # 添加策略到列表中
-        if settings.PIP_PROXY:
-            strategies.append(("镜像站",
-                               ["pip", "install", "-r", str(requirements_file),
-                                "-i", settings.PIP_PROXY, "--no-cache-dir"]))
-        if settings.PROXY_HOST:
-            strategies.append(("代理",
-                               ["pip", "install", "-r", str(requirements_file),
-                                "--proxy", settings.PROXY_HOST, "--no-cache-dir"]))
-        strategies.append(("直连", ["pip", "install", "-r", str(requirements_file), "--no-cache-dir"]))
-
-        # 遍历策略进行安装
-        for strategy_name, pip_command in strategies:
-            logger.debug(f"[PIP] 尝试使用策略：{strategy_name} 安装依赖，命令：{' '.join(pip_command)}")
-            success, message = SystemUtils.execute_with_subprocess(pip_command)
-            if success:
-                logger.debug(f"[PIP] 策略：{strategy_name} 安装依赖成功，输出：{message}")
-                return True, message
-            else:
-                logger.error(f"[PIP] 策略：{strategy_name} 安装依赖失败，错误信息：{message}")
-
-        return False, "[PIP] 所有策略均安装依赖失败，请检查网络连接或 PIP 配置"
-
     @staticmethod
     def __pip_install_with_fallback(requirements_file: Path) -> Tuple[bool, str]:
         """

app/log.py

@@ -246,12 +246,12 @@ class LoggerManager:
         else:
             # 使用默认日志文件
             logfile = self._default_log_file
-
-        # 获取调用者的模块的logger
-        _logger = self._loggers.get(logfile)
-        if not _logger:
-            _logger = self.__setup_logger(log_file=logfile)
-            self._loggers[logfile] = _logger
+        with LoggerManager._lock:  # 添加锁
+            # 获取调用者的模块的logger
+            _logger = self._loggers.get(logfile)
+            if not _logger:
+                _logger = self.__setup_logger(log_file=logfile)
+                self._loggers[logfile] = _logger
         # 调用logger的方法打印日志
         if hasattr(_logger, method):
             log_method = getattr(_logger, method)

app/modules/emby/emby.py

@@ -3,7 +3,7 @@ import re
 import traceback
 from datetime import datetime
 from pathlib import Path
-from typing import List, Optional, Union, Dict, Generator, Tuple
+from typing import List, Optional, Union, Dict, Generator, Tuple, Any
 
 from requests import Response
 
@@ -13,6 +13,7 @@ from app.log import logger
 from app.schemas.types import MediaType
 from app.utils.http import RequestUtils
 from app.utils.url import UrlUtils
+from schemas import MediaServerItem
 
 
 class Emby:
@@ -545,7 +546,7 @@ class Emby:
             return False
         return False
 
-    def refresh_library_by_items(self, items: List[schemas.RefreshMediaItem]) -> bool:
+    def refresh_library_by_items(self, items: List[schemas.RefreshMediaItem]) -> Optional[bool]:
         """
         按类型、名称、年份来刷新媒体库
         :param items: 已识别的需要刷新媒体库的媒体信息列表
@@ -668,8 +669,8 @@ class Emby:
             logger.error(f"连接/Users/{self.user}/Items/{itemid}出错：" + str(e))
         return None
 
-    def get_items(self, parent: Union[str, int], start_index: int = 0, limit: Optional[int] = -1) \
-            -> Optional[Generator]:
+    def get_items(self, parent: Union[str, int], start_index: int = 0,
+                  limit: Optional[int] = -1) -> Generator[MediaServerItem | None | Any, Any, None]:
         """
         获取媒体服务器项目列表，支持分页和不分页逻辑，默认不分页获取所有数据
 

app/modules/filemanager/storages/alist.py

@@ -201,12 +201,12 @@ class Alist(StorageBase, metaclass=Singleton):
 
         if resp is None:
             logging.warning(f"请求获取目录 {fileitem.path} 的文件列表失败，无法连接alist服务")
-            return
+            return None
         if resp.status_code != 200:
             logging.warning(
                 f"请求获取目录 {fileitem.path} 的文件列表失败，状态码：{resp.status_code}"
             )
-            return
+            return None
 
         result = resp.json()
 
@@ -214,7 +214,7 @@ class Alist(StorageBase, metaclass=Singleton):
             logging.warning(
                 f'获取目录 {fileitem.path} 的文件列表失败，错误信息：{result["message"]}'
             )
-            return
+            return None
 
         return [
             schemas.FileItem(
@@ -259,15 +259,15 @@ class Alist(StorageBase, metaclass=Singleton):
         """
         if resp is None:
             logging.warning(f"请求创建目录 {path} 失败，无法连接alist服务")
-            return
+            return None
         if resp.status_code != 200:
             logging.warning(f"请求创建目录 {path} 失败，状态码：{resp.status_code}")
-            return
+            return None
 
         result = resp.json()
         if result["code"] != 200:
             logging.warning(f'创建目录 {path} 失败，错误信息：{result["message"]}')
-            return
+            return None
 
         return self.get_item(path)
 
@@ -349,15 +349,15 @@ class Alist(StorageBase, metaclass=Singleton):
         """
         if resp is None:
             logging.warning(f"请求获取文件 {path} 失败，无法连接alist服务")
-            return
+            return None
         if resp.status_code != 200:
             logging.warning(f"请求获取文件 {path} 失败，状态码：{resp.status_code}")
-            return
+            return None
 
         result = resp.json()
         if result["code"] != 200:
             logging.debug(f'获取文件 {path} 失败，错误信息：{result["message"]}')
-            return
+            return None
 
         return schemas.FileItem(
             storage=self.schema.value,
@@ -513,15 +513,15 @@ class Alist(StorageBase, metaclass=Singleton):
         """
         if not resp:
             logging.warning(f"请求获取文件 {path} 失败，无法连接alist服务")
-            return
+            return None
         if resp.status_code != 200:
             logging.warning(f"请求获取文件 {path} 失败，状态码：{resp.status_code}")
-            return
+            return None
 
         result = resp.json()
         if result["code"] != 200:
             logging.warning(f'获取文件 {path} 失败，错误信息：{result["message"]}')
-            return
+            return None
 
         if result["data"]["raw_url"]:
             download_url = result["data"]["raw_url"]
@@ -569,7 +569,7 @@ class Alist(StorageBase, metaclass=Singleton):
 
         if resp.status_code != 200:
             logging.warning(f"请求上传文件 {path} 失败，状态码：{resp.status_code}")
-            return
+            return None
 
         new_item = self.get_item(Path(fileitem.path) / path.name)
         if new_item and new_name and new_name != path.name:

app/modules/filter/__init__.py

@@ -52,7 +52,7 @@ class FilterModule(_ModuleBase):
         },
         # 官种
         "GZ": {
-            "include": [r'官方', r'官种'],
+            "include": [r'官方', r'官种', r'官组'],
             "match": ["labels"]
         },
         # 特效字幕
@@ -259,7 +259,7 @@ class FilterModule(_ModuleBase):
 
         return None if not matched else torrent
 
-    def __match_group(self, torrent: TorrentInfo, rule_group: Union[list, str]) -> bool:
+    def __match_group(self, torrent: TorrentInfo, rule_group: Union[list, str]) -> Optional[bool]:
         """
         判断种子是否匹配规则组
         """

app/modules/indexer/__init__.py

@@ -10,6 +10,7 @@ from app.log import logger
 from app.modules import _ModuleBase
 from app.modules.indexer.parser import SiteParserBase
 from app.modules.indexer.spider.haidan import HaiDanSpider
+from app.modules.indexer.spider.hddolby import HddolbySpider
 from app.modules.indexer.spider.mtorrent import MTorrentSpider
 from app.modules.indexer.spider.tnode import TNodeSpider
 from app.modules.indexer.spider.torrentleech import TorrentLeech
@@ -121,6 +122,12 @@ class IndexerModule(_ModuleBase):
                 logger.warn(f"{site.get('name')} 不支持中文搜索")
                 continue
 
+            # 站点流控
+            state, msg = SitesHelper().check(StringUtils.get_url_domain(site.get("domain")))
+            if state:
+                logger.warn(msg)
+                continue
+
             # 去除搜索关键字中的特殊字符
             if search_word:
                 search_word = StringUtils.clear(search_word, replace_word=" ", allow_space=True)
@@ -153,6 +160,12 @@ class IndexerModule(_ModuleBase):
                         keyword=search_word,
                         mtype=mtype
                     )
+                elif site.get('parser') == "HDDolby":
+                    error_flag, result = HddolbySpider(site).search(
+                        keyword=search_word,
+                        mtype=mtype,
+                        page=page
+                    )
                 else:
                     error_flag, result = self.__spider_search(
                         search_word=search_word,

app/modules/indexer/parser/__init__.py

@@ -32,6 +32,7 @@ class SiteSchema(Enum):
     TNode = "TNode"
     MTorrent = "MTorrent"
     Yema = "Yema"
+    HDDolby = "HDDolby"
 
 
 class SiteParserBase(metaclass=ABCMeta):
@@ -155,11 +156,17 @@ class SiteParserBase(metaclass=ABCMeta):
         解析站点信息
         :return:
         """
-        # 获取站点首页html
-        self._index_html = self._get_page_content(url=self._site_url)
-        # 检查是否已经登录
-        if not self._parse_logged_in(self._index_html):
-            return
+        # Cookie模式时，获取站点首页html
+        if self.request_mode == "apikey":
+            if not self.apikey and not self.token:
+                logger.warn(f"{self._site_name} 未设置cookie 或 apikey/token，跳过后续操作")
+                return
+            self._index_html = {}
+        else:
+            # 检查是否已经登录
+            self._index_html = self._get_page_content(url=self._site_url)
+            if not self._parse_logged_in(self._index_html):
+                return
         # 解析站点页面
         self._parse_site_page(self._index_html)
         # 解析用户基础信息
@@ -293,9 +300,13 @@ class SiteParserBase(metaclass=ABCMeta):
         req_headers = None
         proxies = settings.PROXY if self._proxy else None
         if self._ua or headers or self._addition_headers:
-            req_headers = {
-                "User-Agent": f"{self._ua}"
-            }
+
+            if self.request_mode == "apikey":
+                req_headers = {}
+            else:
+                req_headers = {
+                    "User-Agent": f"{self._ua}"
+                }
 
             if headers:
                 req_headers.update(headers)

app/modules/indexer/parser/hddolby.py

@@ -0,0 +1,157 @@
+# -*- coding: utf-8 -*-
+import json
+from typing import Optional, Tuple
+
+from app.modules.indexer.parser import SiteParserBase, SiteSchema
+from app.utils.string import StringUtils
+
+
+class HDDolbySiteUserInfo(SiteParserBase):
+    schema = SiteSchema.HDDolby
+    request_mode = "apikey"
+
+    # 用户级别字典
+    HDDolby_sysRoleList = {
+        "0": "Peasant",
+        "1": "User",
+        "2": "Power User",
+        "3": "Elite User",
+        "4": "Crazy User",
+        "5": "Insane User",
+        "6": "Veteran User",
+        "7": "Extreme User",
+        "8": "Ultimate User",
+        "9": "Nexus Master",
+        "10": "VIP",
+        "11": "Retiree",
+        "12": "Helper",
+        "13": "Seeder",
+        "14": "Transferrer",
+        "15": "Uploader",
+        "16": "Torrent Manager",
+        "17": "Forum Moderator",
+        "18": "Coder",
+        "19": "Moderator",
+        "20": "Administrator",
+        "21": "Sysop",
+        "22": "Staff Leader",
+    }
+
+    def _parse_site_page(self, html_text: str):
+        """
+        获取站点页面地址
+        """
+        # 更换api地址
+        self._base_url = f"https://api.{StringUtils.get_url_domain(self._base_url)}"
+        self._user_traffic_page = None
+        self._user_detail_page = None
+        self._user_basic_page = "api/v1/user/data"
+        self._user_basic_params = {}
+        self._user_basic_headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json, text/plain, */*"
+        }
+        self._sys_mail_unread_page = None
+        self._user_mail_unread_page = None
+        self._mail_unread_params = {}
+        self._torrent_seeding_page = "api/v1/user/peers"
+        self._torrent_seeding_params = {}
+        self._torrent_seeding_headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json, text/plain, */*"
+        }
+        self._addition_headers = {
+            "x-api-key": self.apikey,
+        }
+
+    def _parse_logged_in(self, html_text):
+        """
+        判断是否登录成功, 通过判断是否存在用户信息
+        暂时跳过检测，待后续优化
+        :param html_text:
+        :return:
+        """
+        return True
+
+    def _parse_user_base_info(self, html_text: str):
+        """
+        解析用户基本信息，这里把_parse_user_traffic_info和_parse_user_detail_info合并到这里
+        """
+        if not html_text:
+            return None
+        detail = json.loads(html_text)
+        if not detail or detail.get("status") != 0:
+            return
+        user_infos = detail.get("data")
+        """
+        {
+            "id": "1",
+            "added": "2019-03-03 15:30:36",
+            "last_access": "2025-02-18 19:48:04",
+            "class": "22",
+            "uploaded": "852071699418375",
+            "downloaded": "1885536536176",
+            "seedbonus": "99774808.0",
+            "sebonus": "3739023.7",
+            "unread_messages": "0",
+        }
+        """
+        if not user_infos:
+            return
+        user_info = user_infos[0]
+        self.userid = user_info.get("id")
+        self.username = user_info.get("username")
+        self.user_level = self.HDDolby_sysRoleList.get(user_info.get("class") or "1")
+        self.join_at = user_info.get("added")
+        self.upload = int(user_info.get("uploaded") or '0')
+        self.download = int(user_info.get("downloaded") or '0')
+        self.ratio = round(self.upload / self.download, 2) if self.download else 0
+        self.bonus = float(user_info.get("seedbonus") or "0")
+        self.message_unread = int(user_info.get("unread_messages") or '0')
+
+    def _parse_user_traffic_info(self, html_text: str):
+        """
+        解析用户流量信息
+        """
+        pass
+
+    def _parse_user_detail_info(self, html_text: str):
+        """
+        解析用户详细信息
+        """
+        pass
+
+    def _parse_user_torrent_seeding_info(self, html_text: str, multi_page: bool = False) -> Optional[str]:
+        """
+        解析用户做种信息
+        """
+        if not html_text:
+            return None
+        seeding_info = json.loads(html_text)
+        if not seeding_info or seeding_info.get("status") != 0:
+            return None
+        torrents = seeding_info.get("data", [])
+        page_seeding_size = 0
+        page_seeding_info = []
+        for info in torrents:
+            size = info.get("size")
+            seeder = info.get("seeders") or 1
+            page_seeding_size += size
+            page_seeding_info.append([seeder, size])
+        self.seeding += len(torrents)
+        self.seeding_size += page_seeding_size
+        self.seeding_info.extend(page_seeding_info)
+
+        return None
+
+    def _parse_message_unread_links(self, html_text: str, msg_links: list) -> Optional[str]:
+        """
+        解析未读消息链接，这里直接读出详情
+        """
+        pass
+
+    def _parse_message_content(self, html_text) -> Tuple[Optional[str], Optional[str], Optional[str]]:
+        """
+        解析消息内容
+        """
+        pass

app/modules/indexer/parser/ipt_project.py

@@ -54,7 +54,7 @@ class IptSiteUserInfo(SiteParserBase):
     def _parse_user_torrent_seeding_info(self, html_text: str, multi_page: bool = False) -> Optional[str]:
         html = etree.HTML(html_text)
         if not StringUtils.is_valid_html_element(html):
-            return
+            return None
         # seeding start
         seeding_end_pos = 3
         if html.xpath('//tr/td[text() = "Leechers"]'):

app/modules/indexer/parser/tnode.py

@@ -65,7 +65,7 @@ class TNodeSiteUserInfo(SiteParserBase):
         """
         seeding_info = json.loads(html_text)
         if seeding_info.get("status") != 200:
-            return
+            return None
 
         torrents = seeding_info.get("data", {}).get("torrents", [])
 

app/modules/indexer/spider/hddolby.py

@@ -0,0 +1,211 @@
+from typing import Tuple, List
+
+from app.core.config import settings
+from app.db.systemconfig_oper import SystemConfigOper
+from app.log import logger
+from app.schemas import MediaType
+from app.utils.http import RequestUtils
+from app.utils.string import StringUtils
+
+
+class HddolbySpider:
+    """
+    HDDolby API
+    """
+    _indexerid = None
+    _domain = None
+    _domain_host = None
+    _name = ""
+    _proxy = None
+    _cookie = None
+    _ua = None
+    _apikey = None
+    _size = 40
+    _pageurl = None
+    _timeout = 15
+    _searchurl = None
+
+    # 分类
+    _movie_category = [401, 405]
+    _tv_category = [402, 403, 404, 405]
+
+    # 标签
+    _labels = {
+        "gf": "官方",
+        "gy": "国语",
+        "yy": "粤语",
+        "ja": "日语",
+        "ko": "韩语",
+        "zz": "中文字幕",
+        "jz": "禁转",
+        "xz": "限转",
+        "diy": "DIY",
+        "sf": "首发",
+        "yq": "应求",
+        "m0": "零魔",
+        "yc": "原创",
+        "gz": "官字",
+        "db": "Dolby Vision",
+        "hdr10": "HDR10",
+        "hdrm": "HDR10+",
+        "tx": "特效",
+        "lz": "连载",
+        "wj": "完结",
+        "hdrv": "HDR Vivid",
+        "hlg": "HLG",
+        "hq": "高码率",
+        "hfr": "高帧率",
+    }
+
+    def __init__(self, indexer: dict):
+        self.systemconfig = SystemConfigOper()
+        if indexer:
+            self._indexerid = indexer.get('id')
+            self._domain = indexer.get('domain')
+            self._domain_host = StringUtils.get_url_domain(self._domain)
+            self._name = indexer.get('name')
+            if indexer.get('proxy'):
+                self._proxy = settings.PROXY
+            self._cookie = indexer.get('cookie')
+            self._ua = indexer.get('ua')
+            self._apikey = indexer.get('apikey')
+            self._timeout = indexer.get('timeout') or 15
+            self._searchurl = f"https://api.{self._domain_host}/api/v1/torrent/search"
+            self._pageurl = f"{self._domain}details.php?id=%s&hit=1"
+
+    def search(self, keyword: str, mtype: MediaType = None, page: int = 0) -> Tuple[bool, List[dict]]:
+        """
+        搜索
+        """
+
+        if mtype == MediaType.TV:
+            categories = self._tv_category
+        elif mtype == MediaType.MOVIE:
+            categories = self._movie_category
+        else:
+            categories = list(set(self._movie_category + self._tv_category))
+
+        # 输入参数
+        params = {
+            "keyword": keyword,
+            "page_number": page,
+            "page_size": 100,
+            "categories": categories,
+            "visible": 1,
+        }
+
+        res = RequestUtils(
+            headers={
+                "Content-Type": "application/json",
+                "Accept": "application/json, text/plain, */*",
+                "x-api-key": self._apikey
+            },
+            cookies=self._cookie,
+            proxies=self._proxy,
+            referer=f"{self._domain}",
+            timeout=self._timeout
+        ).post_res(url=self._searchurl, json=params)
+        torrents = []
+        if res and res.status_code == 200:
+            results = res.json().get('data', []) or []
+            for result in results:
+                """
+                {
+                    "id": 120202,
+                    "promotion_time_type": 0,
+                    "promotion_until": "0000-00-00 00:00:00",
+                    "category": 402,
+                    "medium": 6,
+                    "codec": 1,
+                    "standard": 2,
+                    "team": 10,
+                    "audiocodec": 14,
+                    "leechers": 0,
+                    "seeders": 1,
+                    "name": "[DBY] Lost S06 2010 Complete 1080p Netflix WEB-DL AVC DDP5.1-DBTV",
+                    "small_descr": "lost ",
+                    "times_completed": 0,
+                    "size": 33665425886,
+                    "added": "2025-02-18 19:47:56",
+                    "url": 0,
+                    "hr": 0,
+                    "tmdb_type": "tv",
+                    "tmdb_id": 4607,
+                    "imdb_id": null,
+                    "tags": "gf"
+                }
+                """
+                # 类别
+                category_value = result.get('category')
+                if category_value in self._tv_category:
+                    category = MediaType.TV.value
+                elif category_value in self._movie_category:
+                    category = MediaType.MOVIE.value
+                else:
+                    category = MediaType.UNKNOWN.value
+                # 标签
+                torrentLabelIds = result.get('tags', "").split(";") or []
+                torrentLabels = []
+                for labelId in torrentLabelIds:
+                    if self._labels.get(labelId) is not None:
+                        torrentLabels.append(self._labels.get(labelId))
+                # 种子信息
+                torrent = {
+                    'title': result.get('name'),
+                    'description': result.get('small_descr'),
+                    'enclosure': self.__get_download_url(result.get('id'), result.get('downhash')),
+                    'pubdate': result.get('added'),
+                    'size': result.get('size'),
+                    'seeders': result.get('seeders'),
+                    'peers': result.get('leechers'),
+                    'grabs': result.get('times_completed'),
+                    'downloadvolumefactor': self.__get_downloadvolumefactor(result.get('promotion_time_type')),
+                    'uploadvolumefactor': self.__get_uploadvolumefactor(result.get('promotion_time_type')),
+                    'freedate': result.get('promotion_until'),
+                    'page_url': self._pageurl % result.get('id'),
+                    'labels': torrentLabels,
+                    'category': category
+                }
+                torrents.append(torrent)
+        elif res is not None:
+            logger.warn(f"{self._name} 搜索失败，错误码：{res.status_code}")
+            return True, []
+        else:
+            logger.warn(f"{self._name} 搜索失败，无法连接 {self._domain}")
+            return True, []
+        return False, torrents
+
+    @staticmethod
+    def __get_downloadvolumefactor(discount: int) -> float:
+        """
+        获取下载系数
+        """
+        discount_dict = {
+            2: 0,
+            5: 0.5,
+            6: 1,
+            7: 0.3
+        }
+        if discount:
+            return discount_dict.get(discount, 1)
+        return 1
+
+    @staticmethod
+    def __get_uploadvolumefactor(discount: int) -> float:
+        """
+        获取上传系数
+        """
+        discount_dict = {
+            3: 2,
+            4: 2,
+            6: 2
+        }
+        if discount:
+            return discount_dict.get(discount, 1)
+        return 1
+
+    def __get_download_url(self, torrent_id: int, downhash: str) -> str:
+        """
+        获取下载链接，返回base64编码的json字符串及URL
+        """
+        return f"{self._domain}download.php?id={torrent_id}&downhash={downhash}"

app/modules/jellyfin/jellyfin.py

@@ -1,6 +1,6 @@
 import json
 from datetime import datetime
-from typing import List, Union, Optional, Dict, Generator, Tuple
+from typing import List, Union, Optional, Dict, Generator, Tuple, Any
 
 from requests import Response
 
@@ -10,6 +10,7 @@ from app.log import logger
 from app.schemas import MediaType
 from app.utils.http import RequestUtils
 from app.utils.url import UrlUtils
+from schemas import MediaServerItem
 
 
 class Jellyfin:
@@ -548,7 +549,7 @@ class Jellyfin:
             logger.error(f"连接Items/Id/Ancestors出错：" + str(e))
             return None
 
-    def refresh_root_library(self) -> bool:
+    def refresh_root_library(self) -> Optional[bool]:
         """
         通知Jellyfin刷新整个媒体库
         """
@@ -762,7 +763,7 @@ class Jellyfin:
         return None
 
     def get_items(self, parent: Union[str, int], start_index: int = 0, limit: Optional[int] = -1) \
-            -> Optional[Generator]:
+            -> Generator[MediaServerItem | None | Any, Any, None]:
         """
         获取媒体服务器项目列表，支持分页和不分页逻辑，默认不分页获取所有数据
 

app/modules/plex/plex.py

@@ -14,6 +14,7 @@ from app.log import logger
 from app.schemas import MediaType
 from app.utils.http import RequestUtils
 from app.utils.url import UrlUtils
+from schemas import MediaServerItem
 
 
 class Plex:
@@ -367,7 +368,7 @@ class Plex:
             return False
         return self._plex.library.update()
 
-    def refresh_library_by_items(self, items: List[schemas.RefreshMediaItem]) -> bool:
+    def refresh_library_by_items(self, items: List[schemas.RefreshMediaItem]) -> Optional[bool]:
         """
         按路径刷新媒体库 item: target_path
         """
@@ -512,7 +513,7 @@ class Plex:
         )
 
     def get_items(self, parent: Union[str, int], start_index: int = 0, limit: Optional[int] = -1) \
-            -> Optional[Generator]:
+            -> Generator[MediaServerItem | None, Any, None]:
         """
         获取媒体服务器项目列表，支持分页和不分页逻辑，默认不分页获取所有数据
 
@@ -855,7 +856,7 @@ class Plex:
         :param kwargs: 其他请求参数，如headers, cookies, proxies等
         """
         if not self._session:
-            return
+            return None
         try:
             url = UrlUtils.adapt_request_url(host=self._host, endpoint=endpoint)
             kwargs.setdefault("headers", self.__get_request_headers())

app/modules/qbittorrent/__init__.py

@@ -78,7 +78,7 @@ class QbittorrentModule(_ModuleBase, _DownloaderBase[Qbittorrent]):
                 server.reconnect()
 
     def download(self, content: Union[Path, str], download_dir: Path, cookie: str,
-                 episodes: Set[int] = None, category: str = None,
+                 episodes: Set[int] = None, category: str = None, label: str = None,
                  downloader: str = None) -> Optional[Tuple[Optional[str], Optional[str], Optional[str], str]]:
         """
         根据种子文件，选择并添加下载任务
@@ -87,6 +87,7 @@ class QbittorrentModule(_ModuleBase, _DownloaderBase[Qbittorrent]):
         :param cookie:  cookie
         :param episodes:  需要下载的集数
         :param category:  分类
+        :param label:  标签
         :param downloader:  下载器
         :return: 下载器名称、种子Hash、种子文件布局、错误原因
         """
@@ -118,7 +119,9 @@ class QbittorrentModule(_ModuleBase, _DownloaderBase[Qbittorrent]):
 
         # 生成随机Tag
         tag = StringUtils.generate_random_str(10)
-        if settings.TORRENT_TAG:
+        if label:
+            tags = label.split(',') + [tag]
+        elif settings.TORRENT_TAG:
             tags = [tag, settings.TORRENT_TAG]
         else:
             tags = [tag]

app/modules/transmission/__init__.py

@@ -79,7 +79,7 @@ class TransmissionModule(_ModuleBase, _DownloaderBase[Transmission]):
                 server.reconnect()
 
     def download(self, content: Union[Path, str], download_dir: Path, cookie: str,
-                 episodes: Set[int] = None, category: str = None,
+                 episodes: Set[int] = None, category: str = None, label: str = None,
                  downloader: str = None) -> Optional[Tuple[Optional[str], Optional[str], Optional[str], str]]:
         """
         根据种子文件，选择并添加下载任务
@@ -88,6 +88,7 @@ class TransmissionModule(_ModuleBase, _DownloaderBase[Transmission]):
         :param cookie:  cookie
         :param episodes:  需要下载的集数
         :param category:  分类，TR中未使用
+        :param label:  标签
         :param downloader:  下载器
         :return: 下载器名称、种子Hash、种子文件布局、错误原因
         """
@@ -118,8 +119,11 @@ class TransmissionModule(_ModuleBase, _DownloaderBase[Transmission]):
 
         # 如果要选择文件则先暂停
         is_paused = True if episodes else False
+
         # 标签
-        if settings.TORRENT_TAG:
+        if label:
+            labels = label.split(',')
+        elif settings.TORRENT_TAG:
             labels = [settings.TORRENT_TAG]
         else:
             labels = None

app/scheduler.py

@@ -30,6 +30,9 @@ from app.utils.singleton import Singleton
 from app.utils.timer import TimerUtils
 
 
+lock = threading.Lock()
+
+
 class SchedulerChain(ChainBase):
     pass
 
@@ -56,85 +59,6 @@ class Scheduler(metaclass=Singleton):
         """
         初始化定时服务
         """
-        # 各服务的运行状态
-        self._jobs = {
-            "cookiecloud": {
-                "name": "同步CookieCloud站点",
-                "func": SiteChain().sync_cookies,
-                "running": False,
-            },
-            "mediaserver_sync": {
-                "name": "同步媒体服务器",
-                "func": MediaServerChain().sync,
-                "running": False,
-            },
-            "subscribe_tmdb": {
-                "name": "订阅元数据更新",
-                "func": SubscribeChain().check,
-                "running": False,
-            },
-            "subscribe_search": {
-                "name": "订阅搜索补全",
-                "func": SubscribeChain().search,
-                "running": False,
-                "kwargs": {
-                    "state": "R"
-                }
-            },
-            "new_subscribe_search": {
-                "name": "新增订阅搜索",
-                "func": SubscribeChain().search,
-                "running": False,
-                "kwargs": {
-                    "state": "N"
-                }
-            },
-            "subscribe_refresh": {
-                "name": "订阅刷新",
-                "func": SubscribeChain().refresh,
-                "running": False,
-            },
-            "subscribe_follow": {
-                "name": "关注的订阅分享",
-                "func": SubscribeChain().follow,
-                "running": False,
-            },
-            "transfer": {
-                "name": "下载文件整理",
-                "func": TransferChain().process,
-                "running": False,
-            },
-            "clear_cache": {
-                "name": "缓存清理",
-                "func": self.clear_cache,
-                "running": False,
-            },
-            "user_auth": {
-                "name": "用户认证检查",
-                "func": self.user_auth,
-                "running": False,
-            },
-            "scheduler_job": {
-                "name": "公共定时服务",
-                "func": SchedulerChain().scheduler_job,
-                "running": False,
-            },
-            "random_wallpager": {
-                "name": "壁纸缓存",
-                "func": TmdbChain().get_trending_wallpapers,
-                "running": False,
-            },
-            "sitedata_refresh": {
-                "name": "站点数据刷新",
-                "func": SiteChain().refresh_userdatas,
-                "running": False,
-            },
-            "recommend_refresh": {
-                "name": "推荐缓存",
-                "func": RecommendChain().refresh_recommend,
-                "running": False,
-            }
-        }
 
         # 停止定时服务
         self.stop()
@@ -143,221 +67,302 @@ class Scheduler(metaclass=Singleton):
         if settings.DEV:
             return
 
-        # 创建定时服务
-        self._scheduler = BackgroundScheduler(timezone=settings.TZ,
-                                              executors={
-                                                  'default': ThreadPoolExecutor(100)
-                                              })
-
-        # CookieCloud定时同步
-        if settings.COOKIECLOUD_INTERVAL \
-                and str(settings.COOKIECLOUD_INTERVAL).isdigit():
-            self._scheduler.add_job(
-                self.start,
-                "interval",
-                id="cookiecloud",
-                name="同步CookieCloud站点",
-                minutes=int(settings.COOKIECLOUD_INTERVAL),
-                next_run_time=datetime.now(pytz.timezone(settings.TZ)) + timedelta(minutes=1),
-                kwargs={
-                    'job_id': 'cookiecloud'
+        with lock:
+            # 各服务的运行状态
+            self._jobs = {
+                "cookiecloud": {
+                    "name": "同步CookieCloud站点",
+                    "func": SiteChain().sync_cookies,
+                    "running": False,
+                },
+                "mediaserver_sync": {
+                    "name": "同步媒体服务器",
+                    "func": MediaServerChain().sync,
+                    "running": False,
+                },
+                "subscribe_tmdb": {
+                    "name": "订阅元数据更新",
+                    "func": SubscribeChain().check,
+                    "running": False,
+                },
+                "subscribe_search": {
+                    "name": "订阅搜索补全",
+                    "func": SubscribeChain().search,
+                    "running": False,
+                    "kwargs": {
+                        "state": "R"
+                    }
+                },
+                "new_subscribe_search": {
+                    "name": "新增订阅搜索",
+                    "func": SubscribeChain().search,
+                    "running": False,
+                    "kwargs": {
+                        "state": "N"
+                    }
+                },
+                "subscribe_refresh": {
+                    "name": "订阅刷新",
+                    "func": SubscribeChain().refresh,
+                    "running": False,
+                },
+                "subscribe_follow": {
+                    "name": "关注的订阅分享",
+                    "func": SubscribeChain().follow,
+                    "running": False,
+                },
+                "transfer": {
+                    "name": "下载文件整理",
+                    "func": TransferChain().process,
+                    "running": False,
+                },
+                "clear_cache": {
+                    "name": "缓存清理",
+                    "func": self.clear_cache,
+                    "running": False,
+                },
+                "user_auth": {
+                    "name": "用户认证检查",
+                    "func": self.user_auth,
+                    "running": False,
+                },
+                "scheduler_job": {
+                    "name": "公共定时服务",
+                    "func": SchedulerChain().scheduler_job,
+                    "running": False,
+                },
+                "random_wallpager": {
+                    "name": "壁纸缓存",
+                    "func": TmdbChain().get_trending_wallpapers,
+                    "running": False,
+                },
+                "sitedata_refresh": {
+                    "name": "站点数据刷新",
+                    "func": SiteChain().refresh_userdatas,
+                    "running": False,
+                },
+                "recommend_refresh": {
+                    "name": "推荐缓存",
+                    "func": RecommendChain().refresh_recommend,
+                    "running": False,
                 }
-            )
-
-        # 媒体服务器同步
-        if settings.MEDIASERVER_SYNC_INTERVAL \
-                and str(settings.MEDIASERVER_SYNC_INTERVAL).isdigit():
-            self._scheduler.add_job(
-                self.start,
-                "interval",
-                id="mediaserver_sync",
-                name="同步媒体服务器",
-                hours=int(settings.MEDIASERVER_SYNC_INTERVAL),
-                next_run_time=datetime.now(pytz.timezone(settings.TZ)) + timedelta(minutes=5),
-                kwargs={
-                    'job_id': 'mediaserver_sync'
-                }
-            )
-
-        # 新增订阅时搜索（5分钟检查一次）
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="new_subscribe_search",
-            name="新增订阅搜索",
-            minutes=5,
-            kwargs={
-                'job_id': 'new_subscribe_search'
             }
-        )
 
-        # 检查更新订阅TMDB数据（每隔6小时）
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="subscribe_tmdb",
-            name="订阅元数据更新",
-            hours=6,
-            kwargs={
-                'job_id': 'subscribe_tmdb'
-            }
-        )
+            # 创建定时服务
+            self._scheduler = BackgroundScheduler(timezone=settings.TZ,
+                                                  executors={
+                                                      'default': ThreadPoolExecutor(100)
+                                                  })
 
-        # 订阅状态每隔24小时搜索一次
-        if settings.SUBSCRIBE_SEARCH:
-            self._scheduler.add_job(
-                self.start,
-                "interval",
-                id="subscribe_search",
-                name="订阅搜索补全",
-                hours=24,
-                kwargs={
-                    'job_id': 'subscribe_search'
-                }
-            )
-
-        if settings.SUBSCRIBE_MODE == "spider":
-            # 站点首页种子定时刷新模式
-            triggers = TimerUtils.random_scheduler(num_executions=32)
-            for trigger in triggers:
+            # CookieCloud定时同步
+            if settings.COOKIECLOUD_INTERVAL \
+                    and str(settings.COOKIECLOUD_INTERVAL).isdigit():
                 self._scheduler.add_job(
                     self.start,
-                    "cron",
-                    id=f"subscribe_refresh|{trigger.hour}:{trigger.minute}",
-                    name="订阅刷新",
-                    hour=trigger.hour,
-                    minute=trigger.minute,
+                    "interval",
+                    id="cookiecloud",
+                    name="同步CookieCloud站点",
+                    minutes=int(settings.COOKIECLOUD_INTERVAL),
+                    next_run_time=datetime.now(pytz.timezone(settings.TZ)) + timedelta(minutes=1),
+                    kwargs={
+                        'job_id': 'cookiecloud'
+                    }
+                )
+
+            # 媒体服务器同步
+            if settings.MEDIASERVER_SYNC_INTERVAL \
+                    and str(settings.MEDIASERVER_SYNC_INTERVAL).isdigit():
+                self._scheduler.add_job(
+                    self.start,
+                    "interval",
+                    id="mediaserver_sync",
+                    name="同步媒体服务器",
+                    hours=int(settings.MEDIASERVER_SYNC_INTERVAL),
+                    next_run_time=datetime.now(pytz.timezone(settings.TZ)) + timedelta(minutes=5),
+                    kwargs={
+                        'job_id': 'mediaserver_sync'
+                    }
+                )
+
+            # 新增订阅时搜索（5分钟检查一次）
+            self._scheduler.add_job(
+                self.start,
+                "interval",
+                id="new_subscribe_search",
+                name="新增订阅搜索",
+                minutes=5,
+                kwargs={
+                    'job_id': 'new_subscribe_search'
+                }
+            )
+
+            # 检查更新订阅TMDB数据（每隔6小时）
+            self._scheduler.add_job(
+                self.start,
+                "interval",
+                id="subscribe_tmdb",
+                name="订阅元数据更新",
+                hours=6,
+                kwargs={
+                    'job_id': 'subscribe_tmdb'
+                }
+            )
+
+            # 订阅状态每隔24小时搜索一次
+            if settings.SUBSCRIBE_SEARCH:
+                self._scheduler.add_job(
+                    self.start,
+                    "interval",
+                    id="subscribe_search",
+                    name="订阅搜索补全",
+                    hours=24,
+                    kwargs={
+                        'job_id': 'subscribe_search'
+                    }
+                )
+
+            if settings.SUBSCRIBE_MODE == "spider":
+                # 站点首页种子定时刷新模式
+                triggers = TimerUtils.random_scheduler(num_executions=32)
+                for trigger in triggers:
+                    self._scheduler.add_job(
+                        self.start,
+                        "cron",
+                        id=f"subscribe_refresh|{trigger.hour}:{trigger.minute}",
+                        name="订阅刷新",
+                        hour=trigger.hour,
+                        minute=trigger.minute,
+                        kwargs={
+                            'job_id': 'subscribe_refresh'
+                        })
+            else:
+                # RSS订阅模式
+                if not settings.SUBSCRIBE_RSS_INTERVAL \
+                        or not str(settings.SUBSCRIBE_RSS_INTERVAL).isdigit():
+                    settings.SUBSCRIBE_RSS_INTERVAL = 30
+                elif int(settings.SUBSCRIBE_RSS_INTERVAL) < 5:
+                    settings.SUBSCRIBE_RSS_INTERVAL = 5
+                self._scheduler.add_job(
+                    self.start,
+                    "interval",
+                    id="subscribe_refresh",
+                    name="RSS订阅刷新",
+                    minutes=int(settings.SUBSCRIBE_RSS_INTERVAL),
                     kwargs={
                         'job_id': 'subscribe_refresh'
-                    })
-        else:
-            # RSS订阅模式
-            if not settings.SUBSCRIBE_RSS_INTERVAL \
-                    or not str(settings.SUBSCRIBE_RSS_INTERVAL).isdigit():
-                settings.SUBSCRIBE_RSS_INTERVAL = 30
-            elif int(settings.SUBSCRIBE_RSS_INTERVAL) < 5:
-                settings.SUBSCRIBE_RSS_INTERVAL = 5
+                    }
+                )
+
+            # 关注订阅分享（每1小时）
             self._scheduler.add_job(
                 self.start,
                 "interval",
-                id="subscribe_refresh",
-                name="RSS订阅刷新",
-                minutes=int(settings.SUBSCRIBE_RSS_INTERVAL),
+                id="subscribe_follow",
+                name="关注的订阅分享",
+                hours=1,
                 kwargs={
-                    'job_id': 'subscribe_refresh'
+                    'job_id': 'subscribe_follow'
                 }
             )
 
-        # 关注订阅分享（每1小时）
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="subscribe_follow",
-            name="关注的订阅分享",
-            hours=1,
-            kwargs={
-                'job_id': 'subscribe_follow'
-            }
-        )
-
-        # 下载器文件转移（每5分钟）
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="transfer",
-            name="下载文件整理",
-            minutes=5,
-            kwargs={
-                'job_id': 'transfer'
-            }
-        )
-
-        # 后台刷新TMDB壁纸
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="random_wallpager",
-            name="壁纸缓存",
-            minutes=30,
-            next_run_time=datetime.now(pytz.timezone(settings.TZ)) + timedelta(seconds=3),
-            kwargs={
-                'job_id': 'random_wallpager'
-            }
-        )
-
-        # 公共定时服务
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="scheduler_job",
-            name="公共定时服务",
-            minutes=10,
-            kwargs={
-                'job_id': 'scheduler_job'
-            }
-        )
-
-        # 缓存清理服务，每隔24小时
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="clear_cache",
-            name="缓存清理",
-            hours=settings.CACHE_CONF["meta"] / 3600,
-            kwargs={
-                'job_id': 'clear_cache'
-            }
-        )
-
-        # 定时检查用户认证，每隔10分钟
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="user_auth",
-            name="用户认证检查",
-            minutes=10,
-            kwargs={
-                'job_id': 'user_auth'
-            }
-        )
-
-        # 站点数据刷新
-        if settings.SITEDATA_REFRESH_INTERVAL:
+            # 下载器文件转移（每5分钟）
             self._scheduler.add_job(
                 self.start,
                 "interval",
-                id="sitedata_refresh",
-                name="站点数据刷新",
-                minutes=settings.SITEDATA_REFRESH_INTERVAL * 60,
+                id="transfer",
+                name="下载文件整理",
+                minutes=5,
                 kwargs={
-                    'job_id': 'sitedata_refresh'
+                    'job_id': 'transfer'
                 }
             )
 
-        # 推荐缓存
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="recommend_refresh",
-            name="推荐缓存",
-            hours=24,
-            next_run_time=datetime.now(pytz.timezone(settings.TZ)) + timedelta(seconds=3),
-            kwargs={
-                'job_id': 'recommend_refresh'
-            }
-        )
+            # 后台刷新TMDB壁纸
+            self._scheduler.add_job(
+                self.start,
+                "interval",
+                id="random_wallpager",
+                name="壁纸缓存",
+                minutes=30,
+                next_run_time=datetime.now(pytz.timezone(settings.TZ)) + timedelta(seconds=3),
+                kwargs={
+                    'job_id': 'random_wallpager'
+                }
+            )
 
-        # 初始化工作流服务
-        self.init_workflow_jobs()
-        
-        # 初始化插件服务
-        self.init_plugin_jobs()
+            # 公共定时服务
+            self._scheduler.add_job(
+                self.start,
+                "interval",
+                id="scheduler_job",
+                name="公共定时服务",
+                minutes=10,
+                kwargs={
+                    'job_id': 'scheduler_job'
+                }
+            )
 
-        # 打印服务
-        logger.debug(self._scheduler.print_jobs())
+            # 缓存清理服务，每隔24小时
+            self._scheduler.add_job(
+                self.start,
+                "interval",
+                id="clear_cache",
+                name="缓存清理",
+                hours=settings.CACHE_CONF["meta"] / 3600,
+                kwargs={
+                    'job_id': 'clear_cache'
+                }
+            )
 
-        # 启动定时服务
-        self._scheduler.start()
+            # 定时检查用户认证，每隔10分钟
+            self._scheduler.add_job(
+                self.start,
+                "interval",
+                id="user_auth",
+                name="用户认证检查",
+                minutes=10,
+                kwargs={
+                    'job_id': 'user_auth'
+                }
+            )
+
+            # 站点数据刷新
+            if settings.SITEDATA_REFRESH_INTERVAL:
+                self._scheduler.add_job(
+                    self.start,
+                    "interval",
+                    id="sitedata_refresh",
+                    name="站点数据刷新",
+                    minutes=settings.SITEDATA_REFRESH_INTERVAL * 60,
+                    kwargs={
+                        'job_id': 'sitedata_refresh'
+                    }
+                )
+
+            # 推荐缓存
+            self._scheduler.add_job(
+                self.start,
+                "interval",
+                id="recommend_refresh",
+                name="推荐缓存",
+                hours=24,
+                next_run_time=datetime.now(pytz.timezone(settings.TZ)) + timedelta(seconds=3),
+                kwargs={
+                    'job_id': 'recommend_refresh'
+                }
+            )
+
+            # 初始化工作流服务
+            self.init_workflow_jobs()
+
+            # 初始化插件服务
+            self.init_plugin_jobs()
+
+            # 打印服务
+            logger.debug(self._scheduler.print_jobs())
+
+            # 启动定时服务
+            self._scheduler.start()
 
     def start(self, job_id: str, *args, **kwargs):
         """
@@ -496,7 +501,6 @@ class Scheduler(metaclass=Singleton):
         """
         if not self._scheduler:
             return
-
         # 移除该工作流的全部服务
         self.remove_workflow_job(workflow)
         # 添加工作流服务
@@ -625,17 +629,18 @@ class Scheduler(metaclass=Singleton):
         """
         关闭定时服务
         """
-        try:
-            if self._scheduler:
-                logger.info("正在停止定时任务...")
-                self._event.set()
-                self._scheduler.remove_all_jobs()
-                if self._scheduler.running:
-                    self._scheduler.shutdown()
-                self._scheduler = None
-                logger.info("定时任务停止完成")
-        except Exception as e:
-            logger.error(f"停止定时任务失败：：{str(e)} - {traceback.format_exc()}")
+        with lock:
+            try:
+                if self._scheduler:
+                    logger.info("正在停止定时任务...")
+                    self._event.set()
+                    self._scheduler.remove_all_jobs()
+                    if self._scheduler.running:
+                        self._scheduler.shutdown()
+                    self._scheduler = None
+                    logger.info("定时任务停止完成")
+            except Exception as e:
+                logger.error(f"停止定时任务失败：：{str(e)} - {traceback.format_exc()}")
 
     @staticmethod
     def clear_cache():

app/schemas/context.py

@@ -1,4 +1,4 @@
-from typing import Optional, Dict, List, Union
+from typing import Optional, Dict, List, Union, Any
 
 from pydantic import BaseModel, Field
 
@@ -235,9 +235,9 @@ class Context(BaseModel):
     上下文
     """
     # 元数据
-    meta_info: Optional[MetaInfo] = None
+    meta_info: Optional[Union[MetaInfo, Any]] = None
     # 媒体信息
-    media_info: Optional[MediaInfo] = None
+    media_info: Optional[Union[MediaInfo, Any]] = None
     # 种子信息
     torrent_info: Optional[TorrentInfo] = None
 

app/schemas/download.py

@@ -7,7 +7,7 @@ class DownloadTask(BaseModel):
     """
      下载任务
     """
-    download_id: Optional[str] = Field(None, description="任务ID")
-    downloader: Optional[str] = Field(None, description="下载器")
-    path: Optional[str] = Field(None, description="下载路径")
-    completed: Optional[bool] = Field(False, description="是否完成")
+    download_id: Optional[str] = Field(default=None, description="任务ID")
+    downloader: Optional[str] = Field(default=None, description="下载器")
+    path: Optional[str] = Field(default=None, description="下载路径")
+    completed: Optional[bool] = Field(default=False, description="是否完成")

app/schemas/event.py

@@ -11,7 +11,7 @@ class Event(BaseModel):
     事件模型
     """
     event_type: str = Field(..., description="事件类型")
-    event_data: Optional[dict] = Field({}, description="事件数据")
+    event_data: Optional[dict] = Field(default={}, description="事件数据")
     priority: Optional[int] = Field(0, description="事件优先级")
 
 

app/schemas/types.py

@@ -147,6 +147,8 @@ class SystemConfigKey(Enum):
     UserSiteAuthParams = "UserSiteAuthParams"
     # Follow订阅分享者
     FollowSubscribers = "FollowSubscribers"
+    # 通知发送时间
+    NotificationSendTime = "NotificationSendTime"
 
 
 # 处理进度Key字典

app/schemas/workflow.py

@@ -13,18 +13,18 @@ class Workflow(BaseModel):
     """
     工作流信息
     """
-    id: Optional[int] = Field(None, description="工作流ID")
-    name: Optional[str] = Field(None, description="工作流名称")
-    description: Optional[str] = Field(None, description="工作流描述")
-    timer: Optional[str] = Field(None, description="定时器")
-    state: Optional[str] = Field(None, description="状态")
-    current_action: Optional[str] = Field(None, description="已执行动作")
-    result: Optional[str] = Field(None, description="任务执行结果")
-    run_count: Optional[int] = Field(0, description="已执行次数")
-    actions: Optional[list] = Field([], description="任务列表")
-    flows: Optional[list] = Field([], description="任务流")
-    add_time: Optional[str] = Field(None, description="创建时间")
-    last_time: Optional[str] = Field(None, description="最后执行时间")
+    id: Optional[int] = Field(default=None, description="工作流ID")
+    name: Optional[str] = Field(default=None, description="工作流名称")
+    description: Optional[str] = Field(default=None, description="工作流描述")
+    timer: Optional[str] = Field(default=None, description="定时器")
+    state: Optional[str] = Field(default=None, description="状态")
+    current_action: Optional[str] = Field(default=None, description="已执行动作")
+    result: Optional[str] = Field(default=None, description="任务执行结果")
+    run_count: Optional[int] = Field(default=0, description="已执行次数")
+    actions: Optional[list] = Field(default=[], description="任务列表")
+    flows: Optional[list] = Field(default=[], description="任务流")
+    add_time: Optional[str] = Field(default=None, description="创建时间")
+    last_time: Optional[str] = Field(default=None, description="最后执行时间")
 
     class Config:
         orm_mode = True
@@ -34,51 +34,51 @@ class ActionParams(BaseModel):
     """
     动作基础参数
     """
-    loop: Optional[bool] = Field(False, description="是否需要循环")
-    loop_interval: Optional[int] = Field(0, description="循环间隔 (秒)")
+    loop: Optional[bool] = Field(default=False, description="是否需要循环")
+    loop_interval: Optional[int] = Field(default=0, description="循环间隔 (秒)")
 
 
 class Action(BaseModel):
     """
     动作信息
     """
-    id: Optional[str] = Field(None, description="动作ID")
-    type: Optional[str] = Field(None, description="动作类型 (类名)")
-    name: Optional[str] = Field(None, description="动作名称")
-    description: Optional[str] = Field(None, description="动作描述")
-    position: Optional[dict] = Field({}, description="位置")
-    data: Optional[dict] = Field({}, description="参数")
+    id: Optional[str] = Field(default=None, description="动作ID")
+    type: Optional[str] = Field(default=None, description="动作类型 (类名)")
+    name: Optional[str] = Field(default=None, description="动作名称")
+    description: Optional[str] = Field(default=None, description="动作描述")
+    position: Optional[dict] = Field(default={}, description="位置")
+    data: Optional[dict] = Field(default={}, description="参数")
 
 
 class ActionExecution(BaseModel):
     """
     动作执行情况
     """
-    action: Optional[str] = Field(None, description="当前动作（名称）")
-    result: Optional[bool] = Field(None, description="执行结果")
-    message: Optional[str] = Field(None, description="执行消息")
+    action: Optional[str] = Field(default=None, description="当前动作（名称）")
+    result: Optional[bool] = Field(default=None, description="执行结果")
+    message: Optional[str] = Field(default=None, description="执行消息")
 
 
 class ActionContext(BaseModel):
     """
     动作基础上下文，各动作通用数据
     """
-    content: Optional[str] = Field(None, description="文本类内容")
-    torrents: Optional[List[Context]] = Field([], description="资源列表")
-    medias: Optional[List[MediaInfo]] = Field([], description="媒体列表")
-    fileitems: Optional[List[FileItem]] = Field([], description="文件列表")
-    downloads: Optional[List[DownloadTask]] = Field([], description="下载任务列表")
-    sites: Optional[List[Site]] = Field([], description="站点列表")
-    subscribes: Optional[List[Subscribe]] = Field([], description="订阅列表")
-    execute_history: Optional[List[ActionExecution]] = Field([], description="执行历史")
-    progress: Optional[int] = Field(0, description="执行进度（%）")
+    content: Optional[str] = Field(default=None, description="文本类内容")
+    torrents: Optional[List[Context]] = Field(default=[], description="资源列表")
+    medias: Optional[List[MediaInfo]] = Field(default=[], description="媒体列表")
+    fileitems: Optional[List[FileItem]] = Field(default=[], description="文件列表")
+    downloads: Optional[List[DownloadTask]] = Field(default=[], description="下载任务列表")
+    sites: Optional[List[Site]] = Field(default=[], description="站点列表")
+    subscribes: Optional[List[Subscribe]] = Field(default=[], description="订阅列表")
+    execute_history: Optional[List[ActionExecution]] = Field(default=[], description="执行历史")
+    progress: Optional[int] = Field(default=0, description="执行进度（%）")
 
 
 class ActionFlow(BaseModel):
     """
     工作流流程
     """
-    id: Optional[str] = Field(None, description="流程ID")
-    source: Optional[str] = Field(None, description="源动作")
-    target: Optional[str] = Field(None, description="目标动作")
-    animated: Optional[bool] = Field(True, description="是否动画流程")
+    id: Optional[str] = Field(default=None, description="流程ID")
+    source: Optional[str] = Field(default=None, description="源动作")
+    target: Optional[str] = Field(default=None, description="目标动作")
+    animated: Optional[bool] = Field(default=True, description="是否动画流程")

app/utils/singleton.py

@@ -3,7 +3,7 @@ import abc
 
 class Singleton(abc.ABCMeta, type):
     """
-    类单例模式
+    类单例模式（按参数）
     """
 
     _instances: dict = {}
@@ -19,3 +19,24 @@ class AbstractSingleton(abc.ABC, metaclass=Singleton):
     """
     抽像类单例模式
     """
+    pass
+
+
+class SingletonClass(abc.ABCMeta, type):
+    """
+    类单例模式（按类）
+    """
+
+    _instances: dict = {}
+
+    def __call__(cls, *args, **kwargs):
+        if cls not in cls._instances:
+            cls._instances[cls] = super(SingletonClass, cls).__call__(*args, **kwargs)
+        return cls._instances[cls]
+
+
+class AbstractSingletonClass(abc.ABC, metaclass=SingletonClass):
+    """
+    抽像类单例模式（按类）
+    """
+    pass

version.py

@@ -1,2 +1,2 @@
-APP_VERSION = 'v2.3.2'
-FRONTEND_VERSION = 'v2.3.2'
+APP_VERSION = 'v2.3.4'
+FRONTEND_VERSION = 'v2.3.4'