Merge pull request #5573 from wikrin/refactor-static-methods-conversion

app/agent/tools/impl/query_directory_settings.py

@@ -31,20 +31,20 @@ class QueryDirectorySettingsTool(MoviePilotTool):
         directory_type = kwargs.get("directory_type", "all")
         storage_type = kwargs.get("storage_type", "all")
         name = kwargs.get("name")
-        
+
         parts = ["正在查询目录配置"]
-        
+
         if directory_type != "all":
             type_map = {"download": "下载目录", "library": "媒体库目录"}
             parts.append(f"类型: {type_map.get(directory_type, directory_type)}")
-        
+
         if storage_type != "all":
             storage_map = {"local": "本地存储", "remote": "远程存储"}
             parts.append(f"存储: {storage_map.get(storage_type, storage_type)}")
-        
+
         if name:
             parts.append(f"名称: {name}")
-        
+
         return " | ".join(parts) if len(parts) > 1 else parts[0]
 
     async def run(self, directory_type: Optional[str] = "all",
@@ -53,16 +53,14 @@ class QueryDirectorySettingsTool(MoviePilotTool):
         logger.info(f"执行工具: {self.name}, 参数: directory_type={directory_type}, storage_type={storage_type}, name={name}")
 
         try:
-            directory_helper = DirectoryHelper()
-            
             # 根据目录类型获取目录列表
             if directory_type == "download":
-                dirs = directory_helper.get_download_dirs()
+                dirs = DirectoryHelper.get_download_dirs()
             elif directory_type == "library":
-                dirs = directory_helper.get_library_dirs()
+                dirs = DirectoryHelper.get_library_dirs()
             else:
-                dirs = directory_helper.get_dirs()
+                dirs = DirectoryHelper.get_dirs()
-            
+
             # 按存储类型过滤
             filtered_dirs = []
             for d in dirs:
@@ -91,13 +89,13 @@ class QueryDirectorySettingsTool(MoviePilotTool):
                             continue
                         if d.library_path and d.library_storage == "local":
                             continue
-                
+
                 # 按名称过滤（部分匹配）
                 if name and d.name and name.lower() not in d.name.lower():
                     continue
-                
+
                 filtered_dirs.append(d)
-            
+
             if filtered_dirs:
                 # 转换为字典格式，只保留关键信息
                 simplified_dirs = []
@@ -124,7 +122,7 @@ class QueryDirectorySettingsTool(MoviePilotTool):
                         "library_category_folder": d.library_category_folder
                     }
                     simplified_dirs.append(simplified)
-                
+
                 result_json = json.dumps(simplified_dirs, ensure_ascii=False, indent=2)
                 return result_json
             return "未找到相关目录配置"

app/agent/tools/impl/query_rule_groups.py

@@ -26,17 +26,16 @@ class QueryRuleGroupsTool(MoviePilotTool):
 
     async def run(self, **kwargs) -> str:
         logger.info(f"执行工具: {self.name}")
-        
+
         try:
-            rule_helper = RuleHelper()
+            rule_groups = RuleHelper.get_rule_groups()
-            rule_groups = rule_helper.get_rule_groups()
+
-            
             if not rule_groups:
                 return json.dumps({
                     "message": "未找到任何规则组",
                     "rule_groups": []
                 }, ensure_ascii=False, indent=2)
-            
+
             # 精简字段，过滤掉 rule_string 避免结果过大
             simplified_groups = []
             for group in rule_groups:
@@ -46,14 +45,14 @@ class QueryRuleGroupsTool(MoviePilotTool):
                     "category": group.category
                 }
                 simplified_groups.append(simplified)
-            
+
             result = {
                 "message": f"找到 {len(simplified_groups)} 个规则组",
                 "rule_groups": simplified_groups
             }
-            
+
             return json.dumps(result, ensure_ascii=False, indent=2)
-            
+
         except Exception as e:
             error_message = f"查询规则组失败: {str(e)}"
             logger.error(f"查询规则组失败: {e}", exc_info=True)

app/api/endpoints/dashboard.py

@@ -56,7 +56,7 @@ def storage(_: schemas.TokenPayload = Depends(verify_token)) -> Any:
     查询本地存储空间信息
     """
     total, available = 0, 0
-    dirs = DirectoryHelper().get_dirs()
+    dirs = DirectoryHelper.get_dirs()
     if not dirs:
         return schemas.Storage(total_storage=total, used_storage=total - available)
     storages = set([d.library_storage for d in dirs if d.library_storage])
@@ -93,7 +93,7 @@ def downloader(name: Optional[str] = None, _: schemas.TokenPayload = Depends(ver
     查询下载器信息
     """
     # 下载目录空间
-    download_dirs = DirectoryHelper().get_local_download_dirs()
+    download_dirs = DirectoryHelper.get_local_download_dirs()
     _, free_space = SystemUtils.space_usage([Path(d.download_path) for d in download_dirs])
     # 下载器信息
     downloader_info = schemas.DownloaderInfo()

app/api/endpoints/system.py

@@ -474,7 +474,7 @@ def ruletest(title: str,
         description=subtitle,
     )
     # 查询规则组详情
-    rulegroup = RuleHelper().get_rule_group(rulegroup_name)
+    rulegroup = RuleHelper.get_rule_group(rulegroup_name)
     if not rulegroup:
         return schemas.Response(success=False, message=f"过滤规则组 {rulegroup_name} 不存在！")
 

app/chain/download.py

@@ -230,7 +230,7 @@ class DownloadChain(ChainBase):
             return None
 
         # 获取种子文件的文件夹名和文件清单
-        _folder_name, _file_list = TorrentHelper().get_fileinfo_from_torrent_content(torrent_content)
+        _folder_name, _file_list = TorrentHelper.get_fileinfo_from_torrent_content(torrent_content)
 
         storage = 'local'
         # 下载目录
@@ -238,7 +238,7 @@ class DownloadChain(ChainBase):
             download_dir = Path(save_path)
         else:
             # 根据媒体信息查询下载目录配置
-            dir_info = DirectoryHelper().get_dir(_media, include_unsorted=True)
+            dir_info = DirectoryHelper.get_dir(_media, include_unsorted=True)
             storage = dir_info.storage if dir_info else storage
             # 拼装子目录
             if dir_info:
@@ -487,7 +487,7 @@ class DownloadChain(ChainBase):
                 contexts = event_data.updated_contexts
 
         # 分组排序
-        contexts = TorrentHelper().sort_group_torrents(contexts)
+        contexts = TorrentHelper.sort_group_torrents(contexts)
 
         # 如果是电影，直接下载
         for context in contexts:
@@ -557,7 +557,7 @@ class DownloadChain(ChainBase):
                                 if isinstance(content, str):
                                     logger.warn(f"{meta.org_string} 下载地址是磁力链，无法确定种子文件集数")
                                     continue
-                                torrent_episodes = TorrentHelper().get_torrent_episodes(torrent_files)
+                                torrent_episodes = TorrentHelper.get_torrent_episodes(torrent_files)
                                 logger.info(f"{meta.org_string} 解析种子文件集数为 {torrent_episodes}")
                                 if not torrent_episodes:
                                     continue
@@ -731,7 +731,7 @@ class DownloadChain(ChainBase):
                                 logger.warn(f"{meta.org_string} 下载地址是磁力链，无法解析种子文件集数")
                                 continue
                             # 种子全部集
-                            torrent_episodes = TorrentHelper().get_torrent_episodes(torrent_files)
+                            torrent_episodes = TorrentHelper.get_torrent_episodes(torrent_files)
                             logger.info(f"{torrent.site_name} - {meta.org_string} 解析种子文件集数：{torrent_episodes}")
                             # 选中的集
                             selected_episodes = set(torrent_episodes).intersection(set(need_episodes))

app/chain/message.py

@@ -274,7 +274,7 @@ class MessageChain(ChainBase):
                                         userid=userid))
                                     return
                                 # 搜索结果排序
-                                contexts = TorrentHelper().sort_torrents(contexts)
+                                contexts = TorrentHelper.sort_torrents(contexts)
                                 try:
                                     # 判断是否设置自动下载
                                     auto_download_user = settings.AUTO_DOWNLOAD_USER

app/chain/search.py

@@ -236,7 +236,7 @@ class SearchChain(ChainBase):
         # 匹配订阅附加参数
         if filter_params:
             logger.info(f'开始附加参数过滤，附加参数：{filter_params} ...')
-            torrents = [torrent for torrent in torrents if TorrentHelper().filter_torrent(torrent, filter_params)]
+            torrents = [torrent for torrent in torrents if TorrentHelper.filter_torrent(torrent, filter_params)]
         # 开始过滤规则过滤
         if rule_groups is None:
             # 取搜索过滤规则
@@ -259,7 +259,6 @@ class SearchChain(ChainBase):
 
         # 开始匹配
         _match_torrents = []
-        torrenthelper = TorrentHelper()
         try:
             # 英文标题应该在别名/原标题中，不需要再匹配
             logger.info(f"开始匹配结果 标题：{mediainfo.title}，原标题：{mediainfo.original_title}，别名：{mediainfo.names}")
@@ -293,7 +292,7 @@ class SearchChain(ChainBase):
                     continue
 
                 # 比对种子
-                if torrenthelper.match_torrent(mediainfo=mediainfo,
+                if TorrentHelper.match_torrent(mediainfo=mediainfo,
                                                torrent_meta=torrent_meta,
                                                torrent=torrent):
                     # 匹配成功
@@ -319,7 +318,7 @@ class SearchChain(ChainBase):
         # 排序
         progress.update(value=99,
                         text=f'正在对 {len(contexts)} 个资源进行排序，请稍候...')
-        contexts = torrenthelper.sort_torrents(contexts)
+        contexts = TorrentHelper.sort_torrents(contexts)
 
         # 结束进度
         logger.info(f'搜索完成，共 {len(contexts)} 个资源')

app/chain/storage.py

@@ -196,7 +196,7 @@ class StorageChain(ChainBase):
         associated_dir = max(
             (
                 Path(p)
-                for d in DirectoryHelper().get_dirs()
+                for d in DirectoryHelper.get_dirs()
                 for p in (d.download_path, d.library_path)
                 if p and fileitem_path.is_relative_to(p)
             ),

app/chain/subscribe.py

@@ -875,7 +875,6 @@ class SubscribeChain(ChainBase):
 
                     # 遍历预识别后的种子
                     _match_context = []
-                    torrenthelper = TorrentHelper()
                     systemconfig = SystemConfigOper()
                     wordsmatcher = WordsMatcher()
                     for domain, contexts in processed_torrents.items():
@@ -926,7 +925,7 @@ class SubscribeChain(ChainBase):
                                     not torrent_mediainfo.tmdb_id and not torrent_mediainfo.douban_id):
                                 logger.debug(
                                     f'{torrent_info.site_name} - {torrent_info.title} 重新识别失败，尝试通过标题匹配...')
-                                if torrenthelper.match_torrent(mediainfo=mediainfo,
+                                if TorrentHelper.match_torrent(mediainfo=mediainfo,
                                                                torrent_meta=torrent_meta,
                                                                torrent=torrent_info):
                                     # 匹配成功
@@ -992,7 +991,7 @@ class SubscribeChain(ChainBase):
                                             continue
 
                             # 匹配订阅附加参数
-                            if not torrenthelper.filter_torrent(torrent_info=torrent_info,
+                            if not TorrentHelper.filter_torrent(torrent_info=torrent_info,
                                                                 filter_params=self.get_params(subscribe)):
                                 continue
 

app/chain/torrents.py

@@ -210,11 +210,12 @@ class TorrentsChain(ChainBase):
 
         # 读取缓存
         torrents_cache = self.get_torrents()
+        torrenthelper = TorrentHelper()
 
         # 缓存过滤掉无效种子
         for _domain, _torrents in torrents_cache.items():
             torrents_cache[_domain] = [_torrent for _torrent in _torrents
-                                       if not TorrentHelper().is_invalid(_torrent.torrent_info.enclosure)]
+                                       if not torrenthelper.is_invalid(_torrent.torrent_info.enclosure)]
 
         # 需要刷新的站点domain
         domains = []

app/chain/transfer.py

@@ -988,12 +988,12 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
             if not task.target_directory:
                 if task.target_path:
                     # 指定目标路径，`手动整理`场景下使用，忽略源目录匹配，使用指定目录匹配
-                    task.target_directory = DirectoryHelper().get_dir(media=task.mediainfo,
+                    task.target_directory = DirectoryHelper.get_dir(media=task.mediainfo,
                                                                       dest_path=task.target_path,
                                                                       target_storage=task.target_storage)
                 else:
                     # 启用源目录匹配时，根据源目录匹配下载目录，否则按源目录同盘优先原则，如无源目录，则根据媒体信息获取目标目录
-                    task.target_directory = DirectoryHelper().get_dir(media=task.mediainfo,
+                    task.target_directory = DirectoryHelper.get_dir(media=task.mediainfo,
                                                                       storage=task.fileitem.storage,
                                                                       src_path=Path(task.fileitem.path),
                                                                       target_storage=task.target_storage)
@@ -1077,7 +1077,7 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
         # 全局锁，避免定时服务重复
         with downloader_lock:
             # 获取下载器监控目录
-            download_dirs = DirectoryHelper().get_download_dirs()
+            download_dirs = DirectoryHelper.get_download_dirs()
 
             # 如果没有下载器监控的目录则不处理
             if not any(dir_info.monitor_type == "downloader" and dir_info.storage == "local"

app/helper/directory.py

@@ -27,31 +27,36 @@ class DirectoryHelper:
             return []
         return [schemas.TransferDirectoryConf(**d) for d in dir_confs]
 
-    def get_download_dirs(self) -> List[schemas.TransferDirectoryConf]:
+    @staticmethod
+    def get_download_dirs() -> List[schemas.TransferDirectoryConf]:
         """
         获取所有下载目录
         """
-        return sorted([d for d in self.get_dirs() if d.download_path], key=lambda x: x.priority)
+        return sorted([d for d in DirectoryHelper.get_dirs() if d.download_path], key=lambda x: x.priority)
 
-    def get_local_download_dirs(self) -> List[schemas.TransferDirectoryConf]:
+    @staticmethod
+    def get_local_download_dirs() -> List[schemas.TransferDirectoryConf]:
         """
         获取所有本地的可下载目录
         """
-        return [d for d in self.get_download_dirs() if d.storage == "local"]
+        return [d for d in DirectoryHelper.get_download_dirs() if d.storage == "local"]
 
-    def get_library_dirs(self) -> List[schemas.TransferDirectoryConf]:
+    @staticmethod
+    def get_library_dirs() -> List[schemas.TransferDirectoryConf]:
         """
         获取所有媒体库目录
         """
-        return sorted([d for d in self.get_dirs() if d.library_path], key=lambda x: x.priority)
+        return sorted([d for d in DirectoryHelper.get_dirs() if d.library_path], key=lambda x: x.priority)
 
-    def get_local_library_dirs(self) -> List[schemas.TransferDirectoryConf]:
+    @staticmethod
+    def get_local_library_dirs() -> List[schemas.TransferDirectoryConf]:
         """
         获取所有本地的媒体库目录
         """
-        return [d for d in self.get_library_dirs() if d.library_storage == "local"]
+        return [d for d in DirectoryHelper.get_library_dirs() if d.library_storage == "local"]
 
-    def get_dir(self, media: Optional[MediaInfo], include_unsorted: Optional[bool] = False,
+    @staticmethod
+    def get_dir(media: Optional[MediaInfo], include_unsorted: Optional[bool] = False,
                 storage: Optional[str] = None, src_path: Path = None,
                 target_storage: Optional[str] = None, dest_path: Path = None
                 ) -> Optional[schemas.TransferDirectoryConf]:
@@ -66,7 +71,7 @@ class DirectoryHelper:
         """
         # 电影/电视剧
         media_type = media.type.value if media else None
-        dirs = self.get_dirs()
+        dirs = DirectoryHelper.get_dirs()
 
         # 如果存在源目录，并源目录为任一下载目录的子目录时，则进行源目录匹配，否则，允许源目录按同盘优先的逻辑匹配
         matching_dirs = [d for d in dirs if src_path.is_relative_to(d.download_path)] if src_path else []
@@ -106,7 +111,7 @@ class DirectoryHelper:
                 # 优先源目录同盘
                 for matched_dir in matched_dirs:
                     matched_path = Path(matched_dir.download_path)
-                    if self._is_same_source((src_path, storage or "local"), (matched_path, matched_dir.library_storage)):
+                    if DirectoryHelper._is_same_source((src_path, storage or "local"), (matched_path, matched_dir.library_storage)):
                         return matched_dir
             return matched_dirs[0]
         return None

app/helper/rule.py

@@ -21,22 +21,24 @@ class RuleHelper:
             return []
         return [FilterRuleGroup(**group) for group in rule_groups]
 
-    def get_rule_group(self, group_name: str) -> Optional[FilterRuleGroup]:
+    @staticmethod
+    def get_rule_group(group_name: str) -> Optional[FilterRuleGroup]:
         """
         获取规则组
         """
-        rule_groups = self.get_rule_groups()
+        rule_groups = RuleHelper.get_rule_groups()
         for group in rule_groups:
             if group.name == group_name:
                 return group
         return None
 
-    def get_rule_group_by_media(self, media: MediaInfo = None, group_names: list = None) -> List[FilterRuleGroup]:
+    @staticmethod
+    def get_rule_group_by_media(media: MediaInfo = None, group_names: list = None) -> List[FilterRuleGroup]:
         """
         根据媒体信息获取规则组
         """
         ret_groups = []
-        rule_groups = self.get_rule_groups()
+        rule_groups = RuleHelper.get_rule_groups()
         if group_names:
             rule_groups = [group for group in rule_groups if group.name in group_names]
         for group in rule_groups:
@@ -58,11 +60,12 @@ class RuleHelper:
             return []
         return [CustomRule(**rule) for rule in rules]
 
-    def get_custom_rule(self, rule_id: str) -> Optional[CustomRule]:
+    @staticmethod
+    def get_custom_rule(rule_id: str) -> Optional[CustomRule]:
         """
         获取自定义规则
         """
-        rules = self.get_custom_rules()
+        rules = RuleHelper.get_custom_rules()
         for rule in rules:
             if rule.id == rule_id:
                 return rule

app/helper/torrent.py

@@ -145,7 +145,8 @@ class TorrentHelper:
             self.add_invalid(url)
             return cache_path, None, "", [], f"下载种子出错，状态码：{req.status_code}"
 
-    def get_torrent_info(self, torrent_path: Path) -> Tuple[str, List[str]]:
+    @staticmethod
+    def get_torrent_info(torrent_path: Path) -> Tuple[str, List[str]]:
         """
         获取种子文件的文件夹名和文件清单
         :param torrent_path: 种子文件路径
@@ -156,7 +157,7 @@ class TorrentHelper:
         try:
             torrentinfo = Torrent.from_file(torrent_path)
             # 获取文件清单
-            return self.get_fileinfo_from_torrent(torrentinfo)
+            return TorrentHelper.get_fileinfo_from_torrent(torrentinfo)
         except Exception as err:
             logger.error(f"种子文件解析失败：{str(err)}")
             return "", []
@@ -192,7 +193,8 @@ class TorrentHelper:
         logger.debug(f"解析种子：{torrent.name} => 目录：{folder_name}，文件清单：{file_list}")
         return folder_name, file_list
 
-    def get_fileinfo_from_torrent_content(self, torrent_content: Union[str, bytes]) -> Tuple[str, List[str]]:
+    @staticmethod
+    def get_fileinfo_from_torrent_content(torrent_content: Union[str, bytes]) -> Tuple[str, List[str]]:
         """
         从种子内容中获取文件夹名和文件清单
         :param torrent_content: 种子内容
@@ -210,7 +212,7 @@ class TorrentHelper:
             # 解析种子内容
             torrentinfo = Torrent.from_string(torrent_content)
             # 获取文件清单
-            return self.get_fileinfo_from_torrent(torrentinfo)
+            return TorrentHelper.get_fileinfo_from_torrent(torrentinfo)
         except Exception as err:
             logger.error(f"种子内容解析失败：{str(err)}")
             return "", []
@@ -292,7 +294,8 @@ class TorrentHelper:
         # 排序
         return sorted(torrent_list, key=lambda x: get_sort_str(x), reverse=True)
 
-    def sort_group_torrents(self, torrent_list: List[Context]) -> List[Context]:
+    @staticmethod
+    def sort_group_torrents(torrent_list: List[Context]) -> List[Context]:
         """
         对媒体信息进行排序、去重
         """
@@ -300,7 +303,7 @@ class TorrentHelper:
             return []
 
         # 排序
-        torrent_list = self.sort_torrents(torrent_list)
+        torrent_list = TorrentHelper.sort_torrents(torrent_list)
 
         # 控重
         result = []

app/modules/filemanager/__init__.py

@@ -533,7 +533,7 @@ class FileManagerModule(_ModuleBase):
         handler = TransHandler()
         ret_fileitems = []
         # 检查本地媒体库
-        dest_dirs = DirectoryHelper().get_library_dirs()
+        dest_dirs = DirectoryHelper.get_library_dirs()
         # 检查每一个媒体库目录
         for dest_dir in dest_dirs:
             # 存储

app/modules/filemanager/storages/local.py

@@ -338,10 +338,9 @@ class LocalStorage(StorageBase):
         """
         存储使用情况
         """
-        directory_helper = DirectoryHelper()
         total_storage, free_storage = SystemUtils.space_usage(
-            [Path(d.download_path) for d in directory_helper.get_local_download_dirs() if d.download_path] +
+            [Path(d.download_path) for d in DirectoryHelper.get_local_download_dirs() if d.download_path] +
-            [Path(d.library_path) for d in directory_helper.get_local_library_dirs() if d.library_path]
+            [Path(d.library_path) for d in DirectoryHelper.get_local_library_dirs() if d.library_path]
         )
         return schemas.StorageUsage(
             total=total_storage,

app/modules/filter/__init__.py

@@ -141,10 +141,6 @@ class FilterModule(_ModuleBase):
         },
     }
 
-    def __init__(self):
-        super().__init__()
-        self.rulehelper = RuleHelper()
-
     def init_module(self) -> None:
         self.parser = RuleParser()
         self.__init_custom_rules()
@@ -153,7 +149,7 @@ class FilterModule(_ModuleBase):
         """
         加载用户自定义规则，如跟内置规则冲突，以用户自定义规则为准
         """
-        custom_rules = self.rulehelper.get_custom_rules()
+        custom_rules = RuleHelper.get_custom_rules()
         for rule in custom_rules:
             logger.info(f"加载自定义规则 {rule.id} - {rule.name}")
             self.rule_set[rule.id] = rule.model_dump()

app/modules/subtitle/__init__.py

@@ -145,7 +145,7 @@ class SubtitleModule(_ModuleBase):
         # 字幕下载目录
         logger.info("开始从站点下载字幕：%s" % torrent.page_url)
         # 获取种子信息
-        folder_name, _ = TorrentHelper().get_fileinfo_from_torrent_content(torrent_content)
+        folder_name, _ = TorrentHelper.get_fileinfo_from_torrent_content(torrent_content)
         # 文件保存目录，如果是单文件种子，则folder_name是空，此时文件保存目录就是下载目录
         storageChain = StorageChain()
         # 等待目录存在

app/monitor.py

@@ -384,7 +384,7 @@ class Monitor(ConfigReloadMixin, metaclass=SingletonClass):
         self.stop()
 
         # 读取目录配置
-        monitor_dirs = DirectoryHelper().get_download_dirs()
+        monitor_dirs = DirectoryHelper.get_download_dirs()
         if not monitor_dirs:
             logger.info("未找到任何目录监控配置")
             return

app/workflow/actions/filter_torrents.py

@@ -58,7 +58,7 @@ class FilterTorrentsAction(BaseAction):
         for torrent in context.torrents:
             if global_vars.is_workflow_stopped(workflow_id):
                 break
-            if TorrentHelper().filter_torrent(
+            if TorrentHelper.filter_torrent(
                     torrent_info=torrent.torrent_info,
                     filter_params={
                         "quality": params.quality,