feat(subscribe): merge mediaserver library entries into files info (#6131)

app/modules/jellyfin/__init__.py

@@ -1,4 +1,4 @@
-from typing import Any, Generator, List, Optional, Tuple, Union
+from typing import Any, Dict, Generator, List, Optional, Tuple, Union
 
 from app import schemas
 from app.core.context import MediaInfo
@@ -299,6 +299,21 @@ class JellyfinModule(_ModuleBase, _MediaServerBase[Jellyfin]):
             return None
         return server_obj.get_play_url(item_id)
 
+    def mediaserver_season_episode_ids(self, server: str, item_id: Union[str, int],
+                                       season: int) -> Optional[Dict[int, str]]:
+        """
+        获取指定季的集号到条目 ID 映射
+
+        :param server: Jellyfin 媒体服务器名称
+        :param item_id: 剧集在 Jellyfin 中的条目 ID
+        :param season: 季号
+        :return: 集号到条目 ID 的映射，服务器不可用或无数据时返回 None
+        """
+        server_obj: Jellyfin = self.get_instance(server)
+        if not server_obj:
+            return None
+        return server_obj.get_season_episode_ids(str(item_id), season)
+
     def mediaserver_latest(self, server: Optional[str] = None, count: Optional[int] = 20,
                            username: Optional[str] = None) -> List[schemas.MediaServerPlayItem]:
         """

app/modules/jellyfin/jellyfin.py

@@ -523,6 +523,38 @@ class Jellyfin:
             return None, None
         return None, {}
 
+    def get_season_episode_ids(self, item_id: str, season: int) -> Dict[int, str]:
+        """
+        获取指定季的集号到媒体服务器条目 ID 映射
+        :param item_id: 剧集在 Jellyfin 中的 ID
+        :param season: 季号
+        :return: {集号: episode_item_id}
+        """
+        if not item_id or not self._host or not self._apikey or not self.user:
+            return {}
+        try:
+            url = f"{self._host}Shows/{item_id}/Episodes"
+            params = {
+                "season": season,
+                "userId": self.user,
+                "isMissing": "false",
+                "api_key": self._apikey
+            }
+            res_json = RequestUtils().get_res(url, params)
+            if not res_json:
+                return {}
+            episode_ids: Dict[int, str] = {}
+            for res_item in res_json.json().get("Items") or []:
+                episode_index = res_item.get("IndexNumber")
+                episode_id = res_item.get("Id")
+                if episode_index is None or not episode_id:
+                    continue
+                episode_ids[int(episode_index)] = str(episode_id)
+            return episode_ids
+        except Exception as e:
+            logger.error(f"获取 Jellyfin 季集条目 ID 出错：{str(e)}")
+            return {}
+
     def get_remote_image_by_id(self, item_id: str, image_type: str) -> Optional[str]:
         """
         根据ItemId从Jellyfin查询TMDB图片地址