add ModuleBases

app/modules/plex/__init__.py

@@ -4,20 +4,19 @@ from app import schemas
 from app.core.context import MediaInfo
 from app.helper.mediaserver import MediaServerHelper
 from app.log import logger
-from app.modules import _ModuleBase
+from app.modules import _ModuleBase, _MediaServerBase
 from app.modules.plex.plex import Plex
 from app.schemas.types import MediaType
 
 
-class PlexModule(_ModuleBase):
-    _servers: Dict[str, Plex] = {}
+class PlexModule(_ModuleBase, _MediaServerBase):
 
     def init_module(self) -> None:
         """
         初始化模块
         """
         # 读取媒体服务器配置
-        self._servers = {}
+        self._servers: Dict[str, Plex] = {}
         mediaservers = MediaServerHelper().get_mediaservers()
         if not mediaservers:
             return
@@ -29,12 +28,6 @@ class PlexModule(_ModuleBase):
     def get_name() -> str:
         return "Plex"
 
-    def get_server(self, name: str) -> Optional[Plex]:
-        """
-        获取Plex服务器
-        """
-        return self._servers.get(name)
-
     def stop(self):
         pass
 
@@ -74,7 +67,7 @@ class PlexModule(_ModuleBase):
         """
         source = args.get("source")
         if source:
-            server = self.get_server(source)
+            server: Plex = self.get_server(source)
             if not server:
                 return None
             return server.get_webhook_message(body)
@@ -140,10 +133,10 @@ class PlexModule(_ModuleBase):
         媒体数量统计
         """
         if server:
-            server_obj = self.get_server(server)
-            if not server_obj:
+            server: Plex = self.get_server(server)
+            if not server:
                 return None
-            servers = [server_obj]
+            servers = [server]
         else:
             servers = self._servers.values()
         media_statistics = []
@@ -158,27 +151,27 @@ class PlexModule(_ModuleBase):
         """
         媒体库列表
         """
-        server_obj = self.get_server(server)
-        if server_obj:
-            return server_obj.get_librarys()
+        server: Plex = self.get_server(server)
+        if server:
+            return server.get_librarys()
         return None
 
     def mediaserver_items(self, server: str, library_id: str) -> Optional[Generator]:
         """
         媒体库项目列表
         """
-        server_obj = self.get_server(server)
-        if server_obj:
-            return server_obj.get_items(library_id)
+        server: Plex = self.get_server(server)
+        if server:
+            return server.get_items(library_id)
         return None
 
     def mediaserver_iteminfo(self, server: str, item_id: str) -> Optional[schemas.MediaServerItem]:
         """
         媒体库项目详情
         """
-        server_obj = self.get_server(server)
-        if server_obj:
-            return server_obj.get_iteminfo(item_id)
+        server: Plex = self.get_server(server)
+        if server:
+            return server.get_iteminfo(item_id)
         return None
 
     def mediaserver_tv_episodes(self, server: str,
@@ -186,10 +179,10 @@ class PlexModule(_ModuleBase):
         """
         获取剧集信息
         """
-        server_obj = self.get_server(server)
-        if not server_obj:
+        server: Plex = self.get_server(server)
+        if not server:
             return None
-        _, seasoninfo = server_obj.get_tv_episodes(item_id=item_id)
+        _, seasoninfo = server.get_tv_episodes(item_id=item_id)
         if not seasoninfo:
             return []
         return [schemas.MediaServerSeasonInfo(
@@ -201,25 +194,25 @@ class PlexModule(_ModuleBase):
         """
         获取媒体服务器正在播放信息
         """
-        server_obj = self.get_server(server)
-        if not server_obj:
+        server: Plex = self.get_server(server)
+        if not server:
             return []
-        return server_obj.get_resume(num=count)
+        return server.get_resume(num=count)
 
     def mediaserver_latest(self, server: str, count: int = 20, **kwargs) -> List[schemas.MediaServerPlayItem]:
         """
         获取媒体服务器最新入库条目
         """
-        server_obj = self.get_server(server)
-        if not server_obj:
+        server: Plex = self.get_server(server)
+        if not server:
             return []
-        return server_obj.get_latest(num=count)
+        return server.get_latest(num=count)
 
     def mediaserver_play_url(self, server: str, item_id: Union[str, int]) -> Optional[str]:
         """
         获取媒体库播放地址
         """
-        server_obj = self.get_server(server)
-        if not server_obj:
+        server: Plex = self.get_server(server)
+        if not server:
             return None
-        return server_obj.get_play_url(item_id)
+        return server.get_play_url(item_id)