MoviePilot-Plugins/docs/V2_Plugin_Development.md

MoviePilot V2 插件开发指南

本指南详细介绍了如何开发适用于MoviePilot V2版本的插件，并实现插件的多版本兼容性，同时包括了服务封装类的使用示例，帮助开发者快速升级插件至V2版本。

1. 多版本插件开发与兼容性

1.1 开发V2版本的插件

要开发适用于MoviePilot V2版本的插件，并实现多版本兼容性，请按照以下步骤操作：

1. 目录结构调整：

   • 将插件代码放置在plugins.v2文件夹中。
   • 将插件的定义放置在package.v2.json中，以实现该插件仅在MoviePilot V2版本中可见。

2. 插件定义示例：

   {
     "CustomSites": {
         "name": "自定义站点",
         "description": "增加自定义站点为签到和统计使用。",
         "labels": "站点",
         "version": "1.0",
         "icon": "world.png",
         "author": "lightolly",
         "level": 2
     }
   }
   

3. 版本判断：

   • MoviePilot V2中 Settings 模块新增了VERSION_FLAG属性，V2版本值为v2，可通过以下代码判断当前的版本，以便在插件中兼容处理：

   from app.core.config import settings
   
   if hasattr(settings, 'VERSION_FLAG'):
     version = settings.VERSION_FLAG # V2
   else:
     version = "v1"
   

1.2 实现插件多版本兼容

如果V1版本插件在V2版本中实际可用，或在插件中主动兼容了V1和V2版本，则可以在package.json中定义 "v2": true属性，以便在MoviePilot V2版本插件市场中显示。

{
  "CustomSites": {
      "name": "自定义站点",
      "description": "增加自定义站点为签到和统计使用。",
      "labels": "站点",
      "version": "1.0",
      "icon": "world.png",
      "author": "lightolly",
      "level": 2,
      "v2": true
  }
}

• 目录结构示例：

  plugins/
    ├── customsites/
    │   ├── __init__.py
    │   └── ...
  plugins.v2/
    ├── customsites/
    │   ├── __init__.py
    │   └── ...
  package.json
  package.v2.json
  

• 插件代码中实现版本兼容：

  在插件代码中，可以根据version变量执行不同的逻辑，以适应不同的MoviePilot版本。

  from app.core.config import settings
  
  class MyPlugin:
      def init_plugin(self, config: dict = None):
          if hasattr(settings, 'VERSION_FLAG'):
              version = settings.VERSION_FLAG  # V2
          else:
              version = "v1"
  
          if version == "v2":
              self.setup_v2()
          else:
              self.setup_v1()
  
      def setup_v2(self):
          # V2版本特有的初始化逻辑
          pass
  
      def setup_v1(self):
          # V1版本特有的初始化逻辑
          pass
  

2. 服务封装与使用示例

为了插件调用并共享实例，主程序针对几种服务进行了封装。以下是相关实现及如何在插件中使用这些封装的详细说明，帮助开发者快速将插件从 V1 升级到 V2。

2.1 服务封装类介绍

ServiceInfo

ServiceInfo 是一个数据类，用于封装服务的相关信息。

from dataclasses import dataclass
from typing import Optional, Any

@dataclass
class ServiceInfo:
    """
    封装服务相关信息的数据类
    """
    # 名称
    name: Optional[str] = None
    # 实例
    instance: Optional[Any] = None
    # 模块
    module: Optional[Any] = None
    # 类型
    type: Optional[str] = None
    # 配置
    config: Optional[Any] = None

ServiceBaseHelper

ServiceBaseHelper 是一个通用的服务帮助类，提供了获取配置和服务实例的通用逻辑。

from typing import Dict, List, Optional, Type, TypeVar, Generic, Iterator
from app.core.module import ModuleManager
from app.helper.serviceconfig import ServiceConfigHelper
from app.schemas import ServiceInfo
from app.schemas.types import SystemConfigKey

TConf = TypeVar("TConf")

class ServiceBaseHelper(Generic[TConf]):
    """
    通用服务帮助类，抽象获取配置和服务实例的通用逻辑
    """

    def __init__(self, config_key: SystemConfigKey, conf_type: Type[TConf], modules: List[str]):
        self.modulemanager = ModuleManager()
        self.config_key = config_key
        self.conf_type = conf_type
        self.modules = modules

    def get_configs(self, include_disabled: bool = False) -> Dict[str, TConf]:
        """
        获取配置列表
        """
        configs: List[TConf] = ServiceConfigHelper.get_configs(self.config_key, self.conf_type)
        return {
            config.name: config
            for config in configs
            if (config.name and config.type and config.enabled) or include_disabled
        } if configs else {}

    def get_config(self, name: str) -> Optional[TConf]:
        """
        获取指定名称配置
        """
        if not name:
            return None
        configs = self.get_configs()
        return configs.get(name)

    def iterate_module_instances(self) -> Iterator[ServiceInfo]:
        """
        迭代所有模块的实例及其对应的配置，返回 ServiceInfo 实例
        """
        configs = self.get_configs()
        for module_name in self.modules:
            module = self.modulemanager.get_running_module(module_name)
            if not module:
                continue
            module_instances = module.get_instances()
            if not isinstance(module_instances, dict):
                continue
            for name, instance in module_instances.items():
                if not instance:
                    continue
                config = configs.get(name)
                service_info = ServiceInfo(
                    name=name,
                    instance=instance,
                    module=module,
                    type=config.type if config else None,
                    config=config
                )
                yield service_info

    def get_services(self, type_filter: Optional[str] = None) -> Dict[str, ServiceInfo]:
        """
        获取服务信息列表，并根据类型过滤
        """
        return {
            service_info.name: service_info
            for service_info in self.iterate_module_instances()
            if service_info.config and (type_filter is None or service_info.type == type_filter)
        }

    def get_service(self, name: str, type_filter: Optional[str] = None) -> Optional[ServiceInfo]:
        """
        获取指定名称的服务信息，并根据类型过滤
        """
        if not name:
            return None
        for service_info in self.iterate_module_instances():
            if service_info.name == name:
                if service_info.config and (type_filter is None or service_info.type == type_filter):
                    return service_info
        return None

2.2 特定服务的帮助类

以下是针对不同服务类型的帮助类，这些类继承自 ServiceBaseHelper，并预设了特定的配置。同时，为了简化类型检查，新增了相应的方法来判断服务类型。

DownloaderHelper

用于管理下载器服务。

from typing import Optional

from app.helper.servicebase import ServiceBaseHelper
from app.schemas import DownloaderConf, ServiceInfo
from app.schemas.types import SystemConfigKey


class DownloaderHelper(ServiceBaseHelper[DownloaderConf]):
    """
    下载器帮助类
    """

    def __init__(self, config: dict = None):
        super().__init__(
            config_key=SystemConfigKey.Downloaders,
            conf_type=DownloaderConf,
            modules=["QbittorrentModule", "TransmissionModule"]
        )

    def is_qbittorrent(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
        """
        判断指定的下载器是否为 qbittorrent 类型，需要传入 `service` 或 `name` 中的任一参数

        :param service: 要判断的服务信息
        :param name: 服务的名称
        :return: 如果服务类型为 qbittorrent，返回 True；否则返回 False。
        """
        if not service:
            service = self.get_service(name=name)
        return service.type == "qbittorrent" if service else False

    def is_transmission(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
        """
        判断指定的下载器是否为 transmission 类型，需要传入 `service` 或 `name` 中的任一参数

        :param service: 要判断的服务信息
        :param name: 服务的名称
        :return: 如果服务类型为 transmission，返回 True；否则返回 False。
        """
        if not service:
            service = self.get_service(name=name)
        return service.type == "transmission" if service else False

MediaServerHelper

用于管理媒体服务器服务。

from app.helper.servicebase import ServiceBaseHelper
from app.schemas import MediaServerConf
from app.schemas.types import SystemConfigKey

class MediaServerHelper(ServiceBaseHelper[MediaServerConf]):
    """
    媒体服务器帮助类
    """

    def __init__(self, config: dict = None):
        super().__init__(
            config_key=SystemConfigKey.MediaServers,
            conf_type=MediaServerConf,
            modules=["PlexModule", "EmbyModule", "JellyfinModule"]
        )
    
    ...

NotificationHelper

用于管理消息通知服务。

from app.helper.servicebase import ServiceBaseHelper
from app.schemas import NotificationConf
from app.schemas.types import SystemConfigKey

class NotificationHelper(ServiceBaseHelper[NotificationConf]):
    """
    消息通知帮助类
    """

    def __init__(self, config: dict = None):
        super().__init__(
            config_key=SystemConfigKey.Notifications,
            conf_type=NotificationConf,
            modules=["WechatModule", "WebPushModule", "VoceChatModule", "TelegramModule", "SynologyChatModule", "SlackModule"]
        )
    
    ...

2.3 在插件中使用服务帮助类

通过这些帮助类，插件可以方便地获取和管理各种服务。以下是 DownloaderHelper 的使用示例，包括类型检查服务和监听模块重载事件的两种方法。

获取下载器选项

插件可以通过 DownloaderHelper 获取所有可用的下载器配置，并生成选项列表供用户选择。

from app.helper.downloader import DownloaderHelper

class MyPlugin:
    def init_plugin(self, config: dict = None):
        self.downloaderhelper = DownloaderHelper(config)
        self.downloader_options = [
            {"title": config.name, "value": config.name}
            for config in self.downloaderhelper.get_configs().values()
        ]

获取特定下载器服务

根据用户选择的下载器名称，插件可以获取对应的服务实例，并执行相应的操作。以下展示了两种方法：

1. 使用事件监听进行模块重载，从而保持服务实例共享

   如果外部模块进行了重载，需要监听模块重载事件以重置下载器服务。

   from typing import Optional, Union
   from app.helper import DownloaderHelper
   from app.modules.qbittorrent import Qbittorrent
   from app.modules.transmission import Transmission
   from app.events import EventType, eventmanager
   
   class MyPlugin:
       def init_plugin(self, config: dict = None):
           self.downloaderhelper = DownloaderHelper(config)
           self._downloader = None
           self.__setup_downloader(config.get("downloader_name"))
   
       def __setup_downloader(self, downloader_name: str):
           self._downloader = self.downloaderhelper.get_service(name=downloader_name)
   
       def __get_downloader(self) -> Optional[Union[Transmission, Qbittorrent]]:
           """
           获取下载器实例
           """
           if not self._downloader:
               return None
           return self._downloader.instance
   
       @eventmanager.register(EventType.ModuleReload)
       def module_reload(self, event: Event):
           """
           模块重载事件
           """
           if not event:
               return
           event_data = event.event_data or {}
           module_id = event_data.get("module_id")
           # 如果模块标识不存在，则说明所有模块均发生重载
           if not module_id:
               self.__setup_downloader()
   
       def check_downloader_type(self) -> bool:
           """
           检查下载器类型是否为 qbittorrent 或 transmission
           """
           downloader = self.__get_downloader()
           if self.downloaderhelper.is_qbittorrent(service=downloader):
               # 处理 qbittorrent 类型
               return True
           elif self.downloaderhelper.is_transmission(service=downloader):
               # 处理 transmission 类型
               return True
           return False
   

2. 使用 Property 实现服务实例共享

   通过 Property 方法，从而保持服务实例共享，而无需通过事件监听。

   from typing import Optional, Union
   from app.helper import DownloaderHelper
   from app.modules.qbittorrent import Qbittorrent
   from app.modules.transmission import Transmission
   
   class MyPlugin:
       def init_plugin(self, config: dict = None):
           self.downloaderhelper = DownloaderHelper(config)
   
       @property
       def service_info(self) -> Optional[ServiceInfo]:
           """
           服务信息
           """
           service = self.downloaderhelper.get_service(name=self.downloader_name)
           if not service:
               return None
   
           if service.instance.is_inactive():
               return None
   
           return service
   
       @property
       def downloader(self) -> Optional[Union[Qbittorrent, Transmission]]:
           """
           下载器实例
           """
           return self.service_info.instance if self.service_info else None
   
       def check_downloader_type(self) -> bool:
           """
           检查下载器类型是否为 qbittorrent 或 transmission
           """
           if self.downloaderhelper.is_qbittorrent(service=self.service_info):
               # 处理 qbittorrent 类型
               return True
           elif self.downloaderhelper.is_transmission(service=self.service_info):
               # 处理 transmission 类型
               return True
           return False
   

2.4 服务封装的优势

• 统一管理：通过 ServiceBaseHelper，不同类型的服务配置和实例管理变得统一和简洁。
• 灵活扩展：新增服务类型时，只需创建相应的帮助类，无需修改现有逻辑。
• 便捷调用：插件可以轻松获取所需的服务实例，简化了服务的调用过程。

2.5 从 V1 升级到 V2 的注意事项

• 使用帮助类：确保插件中使用了新的服务帮助类，如 DownloaderHelper、MediaServerHelper、NotificationHelper 等，而不是直接操作服务实例。
• 更新依赖：检查并更新 requirements.txt 中的依赖，确保与 V2 的服务封装兼容。
• 测试插件：在 V2 环境中全面测试插件，确保所有服务调用正常工作。