From 1a7c3cfff8cc3ea9d6c077e992814b09de654615 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 22 Aug 2026 11:22:59 +0800 Subject: [PATCH] feat(media): add native IMDb source module --- app/locales/en-US.json | 3 + app/locales/zh-CN.json | 3 + app/locales/zh-TW.json | 3 + app/modules/imdb/__init__.py | 677 ++++++++++++++++++ app/modules/imdb/api.py | 635 ++++++++++++++++ app/modules/imdb/capability.toml | 15 + app/runtime/config.py | 10 +- app/schemas/types.py | 2 + .../configuration-debt-baseline.json | 4 +- .../architecture/dependency-baseline.json | 32 +- tests/test_imdb_module.py | 464 ++++++++++++ .../test_module_manager_capability_adapter.py | 12 +- 12 files changed, 1847 insertions(+), 13 deletions(-) create mode 100644 app/modules/imdb/__init__.py create mode 100644 app/modules/imdb/api.py create mode 100644 app/modules/imdb/capability.toml create mode 100644 tests/test_imdb_module.py diff --git a/app/locales/en-US.json b/app/locales/en-US.json index cc1a2209a..24dd339e7 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -25,6 +25,9 @@ "FilterModule": { "name": "Filter" }, + "ImdbModule": { + "name": "IMDb" + }, "IndexerModule": { "name": "Site Indexer" }, diff --git a/app/locales/zh-CN.json b/app/locales/zh-CN.json index 0de92ca86..840b16c51 100644 --- a/app/locales/zh-CN.json +++ b/app/locales/zh-CN.json @@ -25,6 +25,9 @@ "FilterModule": { "name": "过滤器" }, + "ImdbModule": { + "name": "IMDb" + }, "IndexerModule": { "name": "站点索引" }, diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index 4789ea1b3..e2f68cace 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -25,6 +25,9 @@ "FilterModule": { "name": "過濾器" }, + "ImdbModule": { + "name": "IMDb" + }, "IndexerModule": { "name": "站點索引" }, diff --git a/app/modules/imdb/__init__.py b/app/modules/imdb/__init__.py new file mode 100644 index 000000000..ec196f7fd --- /dev/null +++ b/app/modules/imdb/__init__.py @@ -0,0 +1,677 @@ +"""IMDb 原生媒体数据源 Module。""" + +import asyncio +import re +from dataclasses import dataclass +from typing import Any, Optional, Tuple, Union + +from app.domain.context import MediaInfo +from app.domain.media import is_media_source_enabled +from app.domain.meta.metabase import MetaBase +from app.domain.scraper import MediaScraperHelper +from app.foundation.text import convert as zhconv_convert +from app.modules import _ModuleBase +from app.runtime.config import settings +from app.runtime.log import logger +from app.schemas.context import MediaCredit, MediaImageSet +from app.schemas.media import normalize_media_source +from app.schemas.types import ( + MediaRecognizeType, + MediaSource, + MediaSourceSelection, + MediaType, + ModuleType, +) + +from .api import ( + ImdbAka, + ImdbApi, + ImdbCredit, + ImdbEpisode, + ImdbImage, + ImdbSeason, + ImdbTitle, +) + + +@dataclass(frozen=True, slots=True) +class ImdbConfigSnapshot: + """IMDb Module 一次配置 generation 使用的网络快照。""" + + proxy: Any + + +class ImdbModule(_ModuleBase): + """提供 IMDb 搜索、识别、详情补全与刮削能力。""" + + CONFIG_WATCH = {"PROXY_HOST"} + _IMDB_ID_PATTERN = re.compile(r"^tt\d+$", re.IGNORECASE) + _MOVIE_TYPES = frozenset({"movie", "tvMovie"}) + _TV_TYPES = frozenset({"tvSeries", "tvMiniSeries", "tvShort", "tvSpecial"}) + + imdb_api: Optional[ImdbApi] = None + scraper: Optional[MediaScraperHelper] = None + _config = ImdbConfigSnapshot(proxy=None) + + def init_module(self) -> None: + """按当前代理配置初始化 IMDb 客户端和通用刮削器。""" + self._config = ImdbConfigSnapshot(proxy=settings.PROXY) + self.imdb_api = ImdbApi(proxies=self._config.proxy) + self.scraper = MediaScraperHelper() + + def init_setting(self) -> Optional[Tuple[str, Union[str, bool]]]: + """IMDb 随宿主启动，无需独立模块开关。""" + return None + + @staticmethod + def get_name() -> str: + """返回模块展示名称。""" + return "IMDb" + + @staticmethod + def get_type() -> ModuleType: + """返回模块所属的媒体识别类型。""" + return ModuleType.MediaRecognize + + @staticmethod + def get_subtype() -> MediaRecognizeType: + """返回 IMDb 媒体识别子类型。""" + return MediaRecognizeType.IMDb + + @staticmethod + def get_priority() -> int: + """返回模块调度优先级。""" + return 4 + + def stop(self) -> None: + """释放当前 generation 持有的客户端引用。""" + if self.imdb_api: + self.imdb_api.close() + self.imdb_api = None + self.scraper = None + + def test(self) -> Tuple[bool, str]: + """通过最小标题搜索测试 IMDb 数据服务连通性。""" + if not self.imdb_api: + return False, "IMDb 模块未初始化" + results = self.imdb_api.search_titles("The Shawshank Redemption", limit=1) + return (True, "") if results else (False, "IMDb 数据服务未返回搜索结果") + + @classmethod + def _media_type(cls, title_type: str) -> MediaType: + """把 IMDb 标题类型转换为 MoviePilot 媒体类型。""" + if title_type in cls._MOVIE_TYPES: + return MediaType.MOVIE + if title_type in cls._TV_TYPES: + return MediaType.TV + return MediaType.UNKNOWN + + @staticmethod + def _normalize_name(value: Optional[str]) -> str: + """移除标题标点与空白，生成用于精确比较的稳定文本。""" + return re.sub(r"[\W_]+", "", value or "", flags=re.UNICODE).casefold() + + @classmethod + def _name_matches(cls, query: str, names: list[Optional[str]]) -> bool: + """判断查询标题是否与任一 IMDb 标题或别名精确匹配。""" + normalized_query = cls._normalize_name(query) + return bool( + normalized_query + and any( + cls._normalize_name(candidate) == normalized_query + for candidate in names + if candidate + ) + ) + + @classmethod + def _normalize_imdb_id(cls, media_id: object) -> Optional[str]: + """校验并规范化 IMDb 的 ``tt`` 数字 ID。""" + value = str(media_id or "").strip().lower() + return value if cls._IMDB_ID_PATTERN.fullmatch(value) else None + + @classmethod + def _supports_request_type( + cls, meta: Optional[MetaBase], mtype: Optional[MediaType] + ) -> bool: + """限制 IMDb Module 只处理电影和电视剧请求。""" + requested_type = mtype or getattr(meta, "type", None) + return requested_type not in {MediaType.MUSIC, MediaType.MUSIC.value, "music"} + + @staticmethod + def _recognize_names(meta: MetaBase) -> list[str]: + """按中文、简体中文、英文和解析主标题顺序生成识别词。""" + simplified_name = ( + zhconv_convert(meta.cn_name, "zh-hans") if meta.cn_name else None + ) + names = [meta.cn_name, simplified_name, meta.en_name, meta.name] + return list(dict.fromkeys(name for name in names if isinstance(name, str) and name)) + + @classmethod + def _candidate_titles( + cls, + titles: list[ImdbTitle], + mtype: Optional[MediaType], + year: Optional[str], + ) -> list[ImdbTitle]: + """按支持类型、请求类型和年份给 IMDb 候选项排序。""" + candidates = [ + title + for title in titles + if cls._media_type(title.type) in {MediaType.MOVIE, MediaType.TV} + ] + if mtype in {MediaType.MOVIE, MediaType.TV}: + candidates = [ + title for title in candidates if cls._media_type(title.type) == mtype + ] + if year and str(year).isdigit(): + requested_year = int(year) + candidates.sort( + key=lambda title: ( + abs((title.start_year or requested_year + 99) - requested_year), + -(title.start_year or 0), + ) + ) + else: + candidates.sort(key=lambda title: title.start_year or 0, reverse=True) + return candidates + + def _pick_title( + self, + query: str, + titles: list[ImdbTitle], + mtype: Optional[MediaType], + year: Optional[str], + ) -> Optional[ImdbTitle]: + """同步选取标题、原名或别名精确命中的 IMDb 候选项。""" + candidates = self._candidate_titles(titles, mtype, year) + for title in candidates: + if self._name_matches(query, [title.primary_title, title.original_title]): + return title + if not self.imdb_api: + return None + for title in candidates[:10]: + akas = self.imdb_api.list_akas(title.id) + if self._name_matches(query, [aka.text for aka in akas]): + return title + return None + + async def _async_pick_title( + self, + query: str, + titles: list[ImdbTitle], + mtype: Optional[MediaType], + year: Optional[str], + ) -> Optional[ImdbTitle]: + """异步选取标题、原名或别名精确命中的 IMDb 候选项。""" + candidates = self._candidate_titles(titles, mtype, year) + for title in candidates: + if self._name_matches(query, [title.primary_title, title.original_title]): + return title + if not self.imdb_api: + return None + for title in candidates[:10]: + akas = await self.imdb_api.async_list_akas(title.id) + if self._name_matches(query, [aka.text for aka in akas]): + return title + return None + + def _match_by_meta( + self, meta: MetaBase, mtype: Optional[MediaType] + ) -> Optional[ImdbTitle]: + """同步按解析元数据搜索并匹配一个 IMDb 条目。""" + if not self.imdb_api: + return None + requested_type = mtype or meta.type + for name in self._recognize_names(meta): + logger.info("正在使用 IMDb 识别：%s ...", name) + titles = self.imdb_api.search_titles(name) + title = self._pick_title(name, titles, requested_type, meta.year) + if title: + return title + return None + + async def _async_match_by_meta( + self, meta: MetaBase, mtype: Optional[MediaType] + ) -> Optional[ImdbTitle]: + """异步按解析元数据搜索并匹配一个 IMDb 条目。""" + if not self.imdb_api: + return None + requested_type = mtype or meta.type + for name in self._recognize_names(meta): + logger.info("正在使用 IMDb 识别：%s ...", name) + titles = await self.imdb_api.async_search_titles(name) + title = await self._async_pick_title( + name, titles, requested_type, meta.year + ) + if title: + return title + return None + + @staticmethod + def _person_credit( + credit: ImdbCredit, *, director: bool = False + ) -> Optional[dict]: + """把 IMDb 演职员条目转换为统一演职员摘要。""" + if not credit.name or not credit.name.display_name: + return None + image_url = credit.name.primary_image.url if credit.name.primary_image else None + media_credit = MediaCredit( + id=credit.name.id, + name=credit.name.display_name, + character=credit.characters[0] if credit.characters else None, + job="Director" if director else None, + profile_path=image_url, + url=( + f"https://www.imdb.com/name/{credit.name.id}/" + if credit.name.id + else None + ), + avatar=image_url, + images=MediaImageSet(large=image_url) if image_url else None, + ) + return media_credit.model_dump(exclude_none=True) + + @staticmethod + def _backdrop_url(images: list[ImdbImage]) -> Optional[str]: + """按剧照、幕后照顺序选择一张 IMDb 背景图。""" + for image_type in ("still_frame", "behind_the_scenes"): + if image := next( + (item for item in images if item.type == image_type and item.url), + None, + ): + return image.url + return next((item.url for item in images if item.url), None) + + @classmethod + def _category(cls, title: ImdbTitle, mtype: MediaType) -> str: + """按 IMDb 类型、风格、国家和语言生成 MoviePilot 二级分类。""" + genres = {genre.casefold() for genre in title.genres} + countries = {item.code for item in title.origin_countries if item.code} + languages = {item.code for item in title.spoken_languages if item.code} + if mtype == MediaType.MOVIE: + if "animation" in genres: + return "动画电影" + if languages.intersection({"zho", "cmn", "yue", "nan"}): + return "华语电影" + return "外语电影" + if "animation" in genres and countries.intersection({"CN", "TW", "HK"}): + return "国漫" + if "animation" in genres and "JP" in countries: + return "日番" + if "documentary" in genres: + return "纪录片" + if genres.intersection({"reality-tv", "game-show"}): + return "综艺" + if countries.intersection({"CN", "TW", "HK"}): + return "国产剧" + if countries.intersection({"JP", "KP", "KR", "TH", "IN", "SG"}): + return "日韩剧" + if countries.intersection({"US", "FR", "GB", "DE", "ES", "IT", "NL", "PT", "RU", "UK"}): + return "欧美剧" + return "未分类" + + @classmethod + def _to_media_info( + cls, + title: ImdbTitle, + akas: Optional[list[ImdbAka]] = None, + credits: Optional[list[ImdbCredit]] = None, + episodes: Optional[list[ImdbEpisode]] = None, + seasons: Optional[list[ImdbSeason]] = None, + images: Optional[list[ImdbImage]] = None, + ) -> MediaInfo: + """把 IMDb 详情及关联资源转换为统一媒体信息。""" + akas = akas or [] + credits = credits or [] + episodes = episodes or [] + seasons = seasons or [] + images = images or [] + media_type = cls._media_type(title.type) + names = list( + dict.fromkeys( + name + for name in ( + title.primary_title, + title.original_title, + *(aka.text for aka in akas), + ) + if name + ) + ) + media_info = MediaInfo() + media_info.media_source = MediaSource.IMDb + media_info.media_id = title.id + media_info.imdb_id = title.id + media_info.type = media_type + media_info.title = title.primary_title or title.original_title or "" + media_info.en_title = title.primary_title + media_info.original_title = title.original_title + media_info.original_name = title.original_title + media_info.names = names + media_info.year = str(title.start_year) if title.start_year else "" + media_info.overview = title.plot or "" + media_info.adult = bool(title.is_adult) + media_info.poster_path = title.primary_image.url if title.primary_image else None + media_info.backdrop_path = cls._backdrop_url(images) + media_info.genres = [ + {"id": genre, "name": genre} for genre in title.genres + ] + media_info.origin_country = [ + item.code for item in title.origin_countries if item.code + ] + media_info.production_countries = [ + {"name": item.name or item.code} + for item in title.origin_countries + if item.name or item.code + ] + media_info.spoken_languages = [ + {"iso_639_1": item.code, "name": item.name or item.code} + for item in title.spoken_languages + if item.code + ] + if title.spoken_languages: + media_info.original_language = title.spoken_languages[0].code + if title.rating: + media_info.vote_average = title.rating.aggregate_rating + media_info.vote_count = title.rating.vote_count + if title.runtime_seconds: + media_info.runtime = max(1, round(title.runtime_seconds / 60)) + media_info.episode_run_time = [media_info.runtime] + + directors: list[dict] = [] + actors: list[dict] = [] + for credit in credits: + category = (credit.category or "").upper() + if category == "DIRECTOR": + if item := cls._person_credit(credit, director=True): + directors.append(item) + elif category in {"CAST", "ACTOR", "ACTRESS"}: + if item := cls._person_credit(credit): + actors.append(item) + if not directors: + directors = [ + item + for person in title.directors + if (item := cls._person_credit(ImdbCredit(name=person), director=True)) + ] + if not actors: + actors = [ + item + for person in title.stars + if (item := cls._person_credit(ImdbCredit(name=person))) + ] + media_info.directors = directors[:3] + media_info.actors = actors[:12] + + season_info: dict[int, dict] = {} + for season in seasons: + if not season.season or not season.season.isdigit(): + continue + season_number = int(season.season) + season_info[season_number] = { + "season_number": season_number, + "episode_count": season.episode_count, + "name": season.season, + } + for episode in episodes: + if not episode.season or not episode.season.isdigit(): + continue + season_number = int(episode.season) + media_info.seasons.setdefault(season_number, []).append( + episode.episode_number or 0 + ) + if ( + season_number not in media_info.season_years + and episode.release_date + and episode.release_date.year + ): + media_info.season_years[season_number] = str(episode.release_date.year) + season_info.setdefault( + season_number, + { + "season_number": season_number, + "episode_count": None, + "name": str(season_number), + }, + ) + media_info.season_info = list(season_info.values()) + if media_type == MediaType.TV: + media_info.number_of_seasons = len(season_info) + media_info.number_of_episodes = len(episodes) + media_info.set_category(cls._category(title, media_type)) + return media_info + + def _load_media_info(self, title: ImdbTitle) -> MediaInfo: + """同步补齐一个 IMDb 条目的别名、演职员、剧集和图片。""" + if not self.imdb_api: + return self._to_media_info(title) + details = self.imdb_api.get_title(title.id) or title + akas = self.imdb_api.list_akas(title.id) + credits = self.imdb_api.list_credits(title.id) + images = self.imdb_api.list_images(title.id) + if self._media_type(details.type) == MediaType.TV: + episodes = self.imdb_api.list_episodes(title.id) + seasons = self.imdb_api.list_seasons(title.id) + else: + episodes = [] + seasons = [] + return self._to_media_info( + details, + akas=akas, + credits=credits, + episodes=episodes, + seasons=seasons, + images=images, + ) + + async def _async_load_media_info(self, title: ImdbTitle) -> MediaInfo: + """并发补齐一个 IMDb 条目的别名、演职员、剧集和图片。""" + if not self.imdb_api: + return self._to_media_info(title) + is_tv = self._media_type(title.type) == MediaType.TV + results = await asyncio.gather( + self.imdb_api.async_get_title(title.id), + self.imdb_api.async_list_akas(title.id), + self.imdb_api.async_list_credits(title.id), + self.imdb_api.async_list_images(title.id), + self.imdb_api.async_list_episodes(title.id) if is_tv else asyncio.sleep(0, result=[]), + self.imdb_api.async_list_seasons(title.id) if is_tv else asyncio.sleep(0, result=[]), + return_exceptions=True, + ) + details = results[0] if isinstance(results[0], ImdbTitle) else title + return self._to_media_info( + details, + akas=results[1] if isinstance(results[1], list) else [], + credits=results[2] if isinstance(results[2], list) else [], + images=results[3] if isinstance(results[3], list) else [], + episodes=results[4] if isinstance(results[4], list) else [], + seasons=results[5] if isinstance(results[5], list) else [], + ) + + def recognize_media( + self, + meta: Optional[MetaBase] = None, + mtype: Optional[MediaType] = None, + media_source: Optional[MediaSource] = None, + media_id: Optional[str] = None, + episode_group: Optional[str] = None, + cache: bool = True, + **kwargs, + ) -> Optional[MediaInfo]: + """按 IMDb 显式身份或标题元数据同步识别影视信息。""" + del episode_group, cache, kwargs + if not self._supports_request_type(meta, mtype) or not self.imdb_api: + return None + requested_source = normalize_media_source(media_source) + if media_id is not None: + imdb_id = self._normalize_imdb_id(media_id) + if requested_source != MediaSource.IMDb or not imdb_id: + return None + title = self.imdb_api.get_title(imdb_id) + else: + if requested_source not in {None, MediaSource.IMDb}: + return None + selected_source = requested_source or normalize_media_source( + settings.RECOGNIZE_SOURCE + ) + if selected_source != MediaSource.IMDb or not meta or not meta.name: + return None + title = self._match_by_meta(meta, mtype) + if not title: + return None + media_info = self._load_media_info(title) + if meta and meta.begin_season is not None: + media_info.season = meta.begin_season + logger.info( + "IMDb 识别结果：%s %s %s:%s", + media_info.type.value, + media_info.title_year, + media_info.media_source, + media_info.media_id, + ) + return media_info + + async def async_recognize_media( + self, + meta: Optional[MetaBase] = None, + mtype: Optional[MediaType] = None, + media_source: Optional[MediaSource] = None, + media_id: Optional[str] = None, + episode_group: Optional[str] = None, + cache: bool = True, + **kwargs, + ) -> Optional[MediaInfo]: + """按 IMDb 显式身份或标题元数据异步识别影视信息。""" + del episode_group, cache, kwargs + if not self._supports_request_type(meta, mtype) or not self.imdb_api: + return None + requested_source = normalize_media_source(media_source) + if media_id is not None: + imdb_id = self._normalize_imdb_id(media_id) + if requested_source != MediaSource.IMDb or not imdb_id: + return None + title = await self.imdb_api.async_get_title(imdb_id) + else: + if requested_source not in {None, MediaSource.IMDb}: + return None + selected_source = requested_source or normalize_media_source( + settings.RECOGNIZE_SOURCE + ) + if selected_source != MediaSource.IMDb or not meta or not meta.name: + return None + title = await self._async_match_by_meta(meta, mtype) + if not title: + return None + media_info = await self._async_load_media_info(title) + if meta and meta.begin_season is not None: + media_info.season = meta.begin_season + logger.info( + "IMDb 异步识别结果：%s %s %s:%s", + media_info.type.value, + media_info.title_year, + media_info.media_source, + media_info.media_id, + ) + return media_info + + def search_medias( + self, + meta: MetaBase, + media_source: Optional[MediaSourceSelection] = None, + ) -> Optional[list[MediaInfo]]: + """按请求级来源选择同步搜索 IMDb 影视条目。""" + if not is_media_source_enabled(media_source, MediaSource.IMDb): + return None + if not self.imdb_api or not meta.name: + return [] + return [ + self._to_media_info(title) + for title in self._candidate_titles( + self.imdb_api.search_titles(meta.name), meta.type, meta.year + ) + ] + + async def async_search_medias( + self, + meta: MetaBase, + media_source: Optional[MediaSourceSelection] = None, + ) -> Optional[list[MediaInfo]]: + """按请求级来源选择异步搜索 IMDb 影视条目。""" + if not is_media_source_enabled(media_source, MediaSource.IMDb): + return None + if not self.imdb_api or not meta.name: + return [] + titles = await self.imdb_api.async_search_titles(meta.name) + return [ + self._to_media_info(title) + for title in self._candidate_titles(titles, meta.type, meta.year) + ] + + def clear_cache(self) -> None: + """清理 IMDb 请求缓存并记录统一模块日志。""" + if not self.imdb_api: + return + logger.info("开始清除 IMDb 缓存 ...") + self.imdb_api.clear_cache() + logger.info("IMDb 缓存清除完成") + + def obtain_images(self, mediainfo: MediaInfo) -> Optional[MediaInfo]: + """为 IMDb 媒体信息同步补充缺失的背景图片。""" + if mediainfo.media_source != MediaSource.IMDb or not self.imdb_api: + return None + if mediainfo.backdrop_path: + return mediainfo + imdb_id = self._normalize_imdb_id(mediainfo.media_id) + if not imdb_id: + return None + mediainfo.backdrop_path = self._backdrop_url( + self.imdb_api.list_images(imdb_id) + ) + return mediainfo + + async def async_obtain_images(self, mediainfo: MediaInfo) -> Optional[MediaInfo]: + """为 IMDb 媒体信息异步补充缺失的背景图片。""" + if mediainfo.media_source != MediaSource.IMDb or not self.imdb_api: + return None + if mediainfo.backdrop_path: + return mediainfo + imdb_id = self._normalize_imdb_id(mediainfo.media_id) + if not imdb_id: + return None + mediainfo.backdrop_path = self._backdrop_url( + await self.imdb_api.async_list_images(imdb_id) + ) + return mediainfo + + def metadata_nfo( + self, + mediainfo: MediaInfo, + season: Optional[int] = None, + episode: Optional[int] = None, + **kwargs, + ) -> Optional[str]: + """生成 IMDb 来源的 NFO 元数据文本。""" + del kwargs + if (mediainfo.scrape_source or settings.SCRAP_SOURCE) != MediaSource.IMDb.value: + return None + if not self.scraper: + return None + return self.scraper.get_metadata_nfo( + mediainfo, season=season, episode=episode + ) + + def metadata_img( + self, + mediainfo: MediaInfo, + season: Optional[int] = None, + episode: Optional[int] = None, + ) -> Optional[dict]: + """生成 IMDb 来源的图片文件名与下载地址映射。""" + if (mediainfo.scrape_source or settings.SCRAP_SOURCE) != MediaSource.IMDb.value: + return None + if not self.scraper: + return None + return self.scraper.get_metadata_img( + mediainfo, season=season, episode=episode + ) diff --git a/app/modules/imdb/api.py b/app/modules/imdb/api.py new file mode 100644 index 000000000..a500cb929 --- /dev/null +++ b/app/modules/imdb/api.py @@ -0,0 +1,635 @@ +"""IMDb 免 Key 数据接口客户端。""" + +import asyncio +from typing import Any, Optional, TypeVar +from urllib.parse import quote + +import requests +from pydantic import BaseModel, ConfigDict, Field, ValidationError + +from app.adapters.network.http import AsyncRequestUtils, RequestUtils +from app.runtime.cache import cached +from app.runtime.config import settings +from app.runtime.log import logger + +TModel = TypeVar("TModel", bound=BaseModel) + + +def _is_graphql_error(value: Optional[dict]) -> bool: + """避免把 IMDb GraphQL 业务错误写入长期缓存。""" + return bool(value and value.get("errors")) + + +class ImdbModel(BaseModel): + """IMDb 响应模型基类，允许按字段名或外部别名构造。""" + + model_config = ConfigDict(populate_by_name=True) + + +class ImdbImage(ImdbModel): + """IMDb 图片信息。""" + + url: Optional[str] = None + width: Optional[int] = None + height: Optional[int] = None + type: Optional[str] = None + + +class ImdbRating(ImdbModel): + """IMDb 聚合评分信息。""" + + aggregate_rating: Optional[float] = Field(None, alias="aggregateRating") + vote_count: Optional[int] = Field(None, alias="voteCount") + + +class ImdbDate(ImdbModel): + """IMDb 可变精度日期。""" + + year: Optional[int] = None + month: Optional[int] = None + day: Optional[int] = None + + +class ImdbCountry(ImdbModel): + """IMDb 制作国家信息。""" + + code: Optional[str] = None + name: Optional[str] = None + + +class ImdbLanguage(ImdbModel): + """IMDb 语言信息。""" + + code: Optional[str] = None + name: Optional[str] = None + + +class ImdbPerson(ImdbModel): + """IMDb 演职人员信息。""" + + id: Optional[str] = None + display_name: Optional[str] = Field(None, alias="displayName") + primary_image: Optional[ImdbImage] = Field(None, alias="primaryImage") + + +class ImdbTitle(ImdbModel): + """IMDb 影视条目详情。""" + + id: str + type: str + is_adult: Optional[bool] = Field(None, alias="isAdult") + primary_title: Optional[str] = Field(None, alias="primaryTitle") + original_title: Optional[str] = Field(None, alias="originalTitle") + primary_image: Optional[ImdbImage] = Field(None, alias="primaryImage") + start_year: Optional[int] = Field(None, alias="startYear") + end_year: Optional[int] = Field(None, alias="endYear") + runtime_seconds: Optional[int] = Field(None, alias="runtimeSeconds") + genres: list[str] = Field(default_factory=list) + rating: Optional[ImdbRating] = None + plot: Optional[str] = None + directors: list[ImdbPerson] = Field(default_factory=list) + writers: list[ImdbPerson] = Field(default_factory=list) + stars: list[ImdbPerson] = Field(default_factory=list) + origin_countries: list[ImdbCountry] = Field( + default_factory=list, alias="originCountries" + ) + spoken_languages: list[ImdbLanguage] = Field( + default_factory=list, alias="spokenLanguages" + ) + + +class ImdbAka(ImdbModel): + """IMDb 条目别名。""" + + text: Optional[str] = None + + +class ImdbEpisode(ImdbModel): + """IMDb 单集信息。""" + + id: str + title: Optional[str] = None + primary_image: Optional[ImdbImage] = Field(None, alias="primaryImage") + season: Optional[str] = None + episode_number: Optional[int] = Field(None, alias="episodeNumber") + runtime_seconds: Optional[int] = Field(None, alias="runtimeSeconds") + plot: Optional[str] = None + rating: Optional[ImdbRating] = None + release_date: Optional[ImdbDate] = Field(None, alias="releaseDate") + + +class ImdbSeason(ImdbModel): + """IMDb 季信息。""" + + season: Optional[str] = None + episode_count: Optional[int] = Field(None, alias="episodeCount") + + +class ImdbCredit(ImdbModel): + """IMDb 演职员表条目。""" + + name: Optional[ImdbPerson] = None + category: Optional[str] = None + characters: list[str] = Field(default_factory=list) + + +_TITLE_QUERY = """ +query TitleDetails($titles: [ID!]!) { + titles(ids: $titles) { + id + titleText { text } + titleType { id } + releaseYear { year } + originalTitleText { text } + primaryImage { url width height } + ratingsSummary { aggregateRating voteCount } + plot { plotText { plainText } } + runtime { seconds } + titleGenres { genres { genre { text } } } + countriesOfOrigin { countries { id text } } + spokenLanguages { spokenLanguages { id text } } + isAdult + } +} +""" + +_AKAS_QUERY = """ +query TitleAkas($titles: [ID!]!) { + titles(ids: $titles) { + akas(first: 50) { edges { node { text } } } + } +} +""" + +_CREDITS_QUERY = """ +query TitleCredits($titles: [ID!]!) { + titles(ids: $titles) { + directors: credits(first: 5, filter: {categories: ["director"]}) { + edges { node { name { id nameText { text } primaryImage { url } } + category { id } } } + } + cast: credits(first: 20, filter: {categories: ["actor", "actress"]}) { + edges { node { name { id nameText { text } primaryImage { url } } + category { id } ... on Cast { characters { name } } } } + } + } +} +""" + +_IMAGES_QUERY = """ +query TitleImages($titles: [ID!]!) { + titles(ids: $titles) { + images(first: 50) { edges { node { url width height type } } } + } +} +""" + +_SEASONS_QUERY = """ +query TitleSeasons($titles: [ID!]!) { + titles(ids: $titles) { episodes { seasons { number } } } +} +""" + +_EPISODES_QUERY = """ +query TitleEpisodes($titles: [ID!]!, $first: Int!, $after: ID) { + titles(ids: $titles) { + episodes { + episodes(first: $first, after: $after) { + edges { + node { + id + titleText { text } + primaryImage { url width height } + series { episodeNumber { episodeNumber seasonNumber } } + runtime { seconds } + plot { plotText { plainText } } + ratingsSummary { aggregateRating voteCount } + releaseDate { day month year } + } + } + pageInfo { hasNextPage endCursor } + } + } + } +} +""" + + +class ImdbApi: + """封装 IMDb 网页免 Key 接口的同步与异步只读访问。""" + + SEARCH_URL = "https://v2.sg.media-imdb.com/suggestion/x" + GRAPHQL_URL = "https://caching.graphql.imdb.com/" + + def __init__(self, proxies: Optional[dict] = None) -> None: + """按一次模块配置快照创建网络请求适配器。""" + headers = { + "User-Agent": settings.NORMAL_USER_AGENT, + "Accept": "application/graphql+json, application/json", + "Content-Type": "application/json", + "x-imdb-client-name": "imdb-web-next-localized", + } + self._session = requests.Session() + self._request = RequestUtils( + headers=headers, + proxies=proxies, + session=self._session, + timeout=30, + ) + self._async_request = AsyncRequestUtils( + headers=headers, + proxies=proxies, + timeout=30, + ) + + @classmethod + def _freeze_value(cls, value: Any) -> Any: + """递归冻结请求参数，生成跨同步和异步实现一致的缓存键。""" + if isinstance(value, dict): + return tuple( + sorted((key, cls._freeze_value(item)) for key, item in value.items()) + ) + if isinstance(value, (list, tuple)): + return tuple(cls._freeze_value(item) for item in value) + return value + + @classmethod + def _freeze_params(cls, params: Optional[dict]) -> tuple[tuple[str, Any], ...]: + """把请求参数冻结为可用于缓存键的有序元组。""" + return cls._freeze_value(params or {}) + + @cached( + maxsize=settings.CONF.imdb, + ttl=settings.CONF.meta, + skip_none=True, + shared_key="imdb_get", + ) + def _cached_get_json( + self, url: str, params_key: tuple[tuple[str, Any], ...] + ) -> Optional[dict]: + """同步读取并缓存一个 IMDb JSON GET 响应。""" + return self._request.get_json(url, params=dict(params_key)) + + @cached( + maxsize=settings.CONF.imdb, + ttl=settings.CONF.meta, + skip_none=True, + shared_key="imdb_get", + ) + async def _async_cached_get_json( + self, url: str, params_key: tuple[tuple[str, Any], ...] + ) -> Optional[dict]: + """异步读取并缓存一个 IMDb JSON GET 响应。""" + return await self._async_request.get_json(url, params=dict(params_key)) + + @cached( + maxsize=settings.CONF.imdb, + ttl=settings.CONF.meta, + skip_none=True, + shared_key="imdb_graphql", + skip_if=_is_graphql_error, + ) + def _cached_graphql( + self, query: str, variables_key: tuple[tuple[str, Any], ...] + ) -> Optional[dict]: + """同步读取并缓存一个 IMDb GraphQL 响应。""" + return self._request.post_json( + self.GRAPHQL_URL, + json={"query": query, "variables": dict(variables_key)}, + ) + + @cached( + maxsize=settings.CONF.imdb, + ttl=settings.CONF.meta, + skip_none=True, + shared_key="imdb_graphql", + skip_if=_is_graphql_error, + ) + async def _async_cached_graphql( + self, query: str, variables_key: tuple[tuple[str, Any], ...] + ) -> Optional[dict]: + """异步读取并缓存一个 IMDb GraphQL 响应。""" + return await self._async_request.post_json( + self.GRAPHQL_URL, + json={"query": query, "variables": dict(variables_key)}, + ) + + def _graphql(self, query: str, variables: dict) -> Optional[dict]: + """同步执行 GraphQL 查询并提取 data 区域。""" + response = self._cached_graphql(query, self._freeze_params(variables)) + if response and response.get("errors"): + logger.debug("IMDb GraphQL 查询失败：%s", response["errors"]) + return None + return response.get("data") if response else None + + async def _async_graphql(self, query: str, variables: dict) -> Optional[dict]: + """异步执行 GraphQL 查询并提取 data 区域。""" + response = await self._async_cached_graphql( + query, self._freeze_params(variables) + ) + if response and response.get("errors"): + logger.debug("IMDb GraphQL 异步查询失败：%s", response["errors"]) + return None + return response.get("data") if response else None + + def clear_cache(self) -> None: + """清理同步缓存，并在当前或临时事件循环中清理异步缓存。""" + self._cached_get_json.cache_clear() + self._cached_graphql.cache_clear() + try: + loop = asyncio.get_running_loop() + except RuntimeError: + asyncio.run(self.async_clear_cache()) + else: + loop.create_task(self.async_clear_cache()) + + async def async_clear_cache(self) -> None: + """清理 IMDb 异步 GET 与 GraphQL 请求缓存区。""" + await self._async_cached_get_json.cache_clear() + await self._async_cached_graphql.cache_clear() + + def close(self) -> None: + """关闭同步 HTTP Session，异步共享连接池由宿主统一关闭。""" + self._session.close() + + @staticmethod + def _parse( + model: type[TModel], data: Optional[dict], operation: str + ) -> Optional[TModel]: + """把外部响应校验为内部模型，并隔离字段漂移错误。""" + if not data: + return None + try: + return model.model_validate(data) + except (TypeError, ValueError, ValidationError) as err: + logger.debug("解析 IMDb %s 响应失败：%s", operation, str(err)) + return None + + @staticmethod + def _first_title(data: Optional[dict]) -> Optional[dict]: + """从 IMDb GraphQL data 中提取第一个标题对象。""" + titles = data.get("titles") if data else None + return titles[0] if titles else None + + @classmethod + def _parse_search_titles(cls, data: Optional[dict], limit: int) -> list[ImdbTitle]: + """把 IMDb 搜索建议响应转换为稳定标题模型。""" + titles: list[ImdbTitle] = [] + for item in (data or {}).get("d", []): + imdb_id = item.get("id") + title_type = item.get("qid") + if not imdb_id or not imdb_id.startswith("tt") or not title_type: + continue + image = item.get("i") or {} + title = cls._parse( + ImdbTitle, + { + "id": imdb_id, + "type": title_type, + "primaryTitle": item.get("l"), + "primaryImage": { + "url": image.get("imageUrl"), + "width": image.get("width"), + "height": image.get("height"), + }, + "startYear": item.get("y"), + }, + "标题搜索", + ) + if title: + titles.append(title) + if len(titles) >= limit: + break + return titles + + @classmethod + def _parse_title(cls, data: Optional[dict]) -> Optional[ImdbTitle]: + """把 IMDb GraphQL 标题结构展平为内部详情模型。""" + item = cls._first_title(data) + if not item: + return None + genres = item.get("titleGenres", {}).get("genres") or [] + countries = item.get("countriesOfOrigin", {}).get("countries") or [] + languages = item.get("spokenLanguages", {}).get("spokenLanguages") or [] + return cls._parse( + ImdbTitle, + { + "id": item.get("id"), + "type": item.get("titleType", {}).get("id"), + "isAdult": item.get("isAdult"), + "primaryTitle": item.get("titleText", {}).get("text"), + "originalTitle": item.get("originalTitleText", {}).get("text"), + "primaryImage": item.get("primaryImage"), + "startYear": item.get("releaseYear", {}).get("year"), + "runtimeSeconds": item.get("runtime", {}).get("seconds"), + "genres": [ + entry.get("genre", {}).get("text") + for entry in genres + if entry.get("genre", {}).get("text") + ], + "rating": item.get("ratingsSummary"), + "plot": item.get("plot", {}).get("plotText", {}).get("plainText"), + "originCountries": [ + {"code": entry.get("id"), "name": entry.get("text")} + for entry in countries + ], + "spokenLanguages": [ + {"code": entry.get("id"), "name": entry.get("text")} + for entry in languages + ], + }, + "标题详情", + ) + + @classmethod + def _parse_akas(cls, data: Optional[dict]) -> list[ImdbAka]: + """解析 GraphQL 标题别名边。""" + title = cls._first_title(data) or {} + edges = title.get("akas", {}).get("edges") or [] + return [ + aka + for edge in edges + if (aka := cls._parse(ImdbAka, edge.get("node"), "标题别名")) + ] + + @classmethod + def _parse_credits(cls, data: Optional[dict]) -> list[ImdbCredit]: + """解析 GraphQL 导演和主演边。""" + title = cls._first_title(data) or {} + credits: list[ImdbCredit] = [] + for field_name in ("directors", "cast"): + for edge in title.get(field_name, {}).get("edges") or []: + node = edge.get("node") or {} + name = node.get("name") or {} + credit = cls._parse( + ImdbCredit, + { + "name": { + "id": name.get("id"), + "displayName": name.get("nameText", {}).get("text"), + "primaryImage": name.get("primaryImage"), + }, + "category": node.get("category", {}).get("id"), + "characters": [ + item.get("name") + for item in node.get("characters") or [] + if item.get("name") + ], + }, + "演职员", + ) + if credit: + credits.append(credit) + return credits + + @classmethod + def _parse_images(cls, data: Optional[dict]) -> list[ImdbImage]: + """解析 GraphQL 标题图片边。""" + title = cls._first_title(data) or {} + edges = title.get("images", {}).get("edges") or [] + return [ + image + for edge in edges + if (image := cls._parse(ImdbImage, edge.get("node"), "标题图片")) + ] + + @classmethod + def _parse_seasons(cls, data: Optional[dict]) -> list[ImdbSeason]: + """解析 GraphQL 季编号列表。""" + title = cls._first_title(data) or {} + values = title.get("episodes", {}).get("seasons") or [] + return [ImdbSeason(season=str(item["number"])) for item in values] + + @classmethod + def _parse_episode_page( + cls, data: Optional[dict] + ) -> tuple[list[ImdbEpisode], Optional[str]]: + """解析一页 GraphQL 单集边及下一页游标。""" + title = cls._first_title(data) or {} + connection = title.get("episodes", {}).get("episodes") or {} + episodes: list[ImdbEpisode] = [] + for edge in connection.get("edges") or []: + node = edge.get("node") or {} + episode_number = node.get("series", {}).get("episodeNumber") or {} + episode = cls._parse( + ImdbEpisode, + { + "id": node.get("id"), + "title": node.get("titleText", {}).get("text"), + "primaryImage": node.get("primaryImage"), + "season": str(episode_number.get("seasonNumber")) + if episode_number.get("seasonNumber") is not None + else None, + "episodeNumber": episode_number.get("episodeNumber"), + "runtimeSeconds": node.get("runtime", {}).get("seconds"), + "plot": node.get("plot", {}).get("plotText", {}).get("plainText"), + "rating": node.get("ratingsSummary"), + "releaseDate": node.get("releaseDate"), + }, + "单集详情", + ) + if episode: + episodes.append(episode) + page_info = connection.get("pageInfo") or {} + next_cursor = ( + page_info.get("endCursor") if page_info.get("hasNextPage") else None + ) + return episodes, next_cursor + + def search_titles(self, query: str, limit: int = 50) -> list[ImdbTitle]: + """按标题关键字搜索 IMDb 影视条目。""" + url = f"{self.SEARCH_URL}/{quote(query.strip(), safe='')}.json" + data = self._cached_get_json(url, self._freeze_params(None)) + return self._parse_search_titles(data, limit) + + async def async_search_titles( + self, query: str, limit: int = 50 + ) -> list[ImdbTitle]: + """异步按标题关键字搜索 IMDb 影视条目。""" + url = f"{self.SEARCH_URL}/{quote(query.strip(), safe='')}.json" + data = await self._async_cached_get_json(url, self._freeze_params(None)) + return self._parse_search_titles(data, limit) + + def get_title(self, imdb_id: str) -> Optional[ImdbTitle]: + """按 IMDb ID 获取影视条目详情。""" + return self._parse_title(self._graphql(_TITLE_QUERY, {"titles": [imdb_id]})) + + async def async_get_title(self, imdb_id: str) -> Optional[ImdbTitle]: + """异步按 IMDb ID 获取影视条目详情。""" + return self._parse_title( + await self._async_graphql(_TITLE_QUERY, {"titles": [imdb_id]}) + ) + + def list_akas(self, imdb_id: str) -> list[ImdbAka]: + """获取影视条目的可用别名。""" + return self._parse_akas(self._graphql(_AKAS_QUERY, {"titles": [imdb_id]})) + + async def async_list_akas(self, imdb_id: str) -> list[ImdbAka]: + """异步获取影视条目的可用别名。""" + return self._parse_akas( + await self._async_graphql(_AKAS_QUERY, {"titles": [imdb_id]}) + ) + + def list_episodes(self, imdb_id: str) -> list[ImdbEpisode]: + """使用游标分页获取电视剧条目的全部单集。""" + episodes: list[ImdbEpisode] = [] + cursor: Optional[str] = None + while True: + data = self._graphql( + _EPISODES_QUERY, + {"titles": [imdb_id], "first": 100, "after": cursor}, + ) + page, cursor = self._parse_episode_page(data) + episodes.extend(page) + if not cursor: + return episodes + + async def async_list_episodes(self, imdb_id: str) -> list[ImdbEpisode]: + """异步使用游标分页获取电视剧条目的全部单集。""" + episodes: list[ImdbEpisode] = [] + cursor: Optional[str] = None + while True: + data = await self._async_graphql( + _EPISODES_QUERY, + {"titles": [imdb_id], "first": 100, "after": cursor}, + ) + page, cursor = self._parse_episode_page(data) + episodes.extend(page) + if not cursor: + return episodes + + def list_seasons(self, imdb_id: str) -> list[ImdbSeason]: + """获取电视剧条目的季列表。""" + return self._parse_seasons( + self._graphql(_SEASONS_QUERY, {"titles": [imdb_id]}) + ) + + async def async_list_seasons(self, imdb_id: str) -> list[ImdbSeason]: + """异步获取电视剧条目的季列表。""" + return self._parse_seasons( + await self._async_graphql(_SEASONS_QUERY, {"titles": [imdb_id]}) + ) + + def list_credits(self, imdb_id: str) -> list[ImdbCredit]: + """获取影视条目的导演与主要演员。""" + return self._parse_credits( + self._graphql(_CREDITS_QUERY, {"titles": [imdb_id]}) + ) + + async def async_list_credits(self, imdb_id: str) -> list[ImdbCredit]: + """异步获取影视条目的导演与主要演员。""" + return self._parse_credits( + await self._async_graphql(_CREDITS_QUERY, {"titles": [imdb_id]}) + ) + + def list_images(self, imdb_id: str) -> list[ImdbImage]: + """获取影视条目的主要图片。""" + return self._parse_images( + self._graphql(_IMAGES_QUERY, {"titles": [imdb_id]}) + ) + + async def async_list_images(self, imdb_id: str) -> list[ImdbImage]: + """异步获取影视条目的主要图片。""" + return self._parse_images( + await self._async_graphql(_IMAGES_QUERY, {"titles": [imdb_id]}) + ) diff --git a/app/modules/imdb/capability.toml b/app/modules/imdb/capability.toml new file mode 100644 index 000000000..431c588b9 --- /dev/null +++ b/app/modules/imdb/capability.toml @@ -0,0 +1,15 @@ +schema_version = 1 +id = "ImdbModule" +kind = "host_module" +entrypoint = "app.modules.imdb:ImdbModule" +depends_on = [] + +[metadata] +name = "IMDb" +type = "mediarecognize" +subtype = "IMDb" +priority = 4 + +[activation] +policy = "bootstrap" +watch = ["PROXY_HOST"] diff --git a/app/runtime/config.py b/app/runtime/config.py index 70f95d286..fce144102 100644 --- a/app/runtime/config.py +++ b/app/runtime/config.py @@ -53,6 +53,8 @@ class SystemConfModel(BaseModel): bangumi: int = 0 # AniList请求缓存数量 anilist: int = 0 + # IMDb请求缓存数量 + imdb: int = 0 # Fanart请求缓存数量 fanart: int = 0 # MusicBrainz请求缓存数量 @@ -260,11 +262,11 @@ class ConfigModel(BaseModel): DOH_RESOLVERS: str = "1.0.0.1,1.1.1.1,9.9.9.9,149.112.112.112" # ==================== 媒体元数据配置 ==================== - # 媒体搜索来源 themoviedb/douban/bangumi/anilist/musicbrainz/theaudiodb/doubanmusic，多个用,分隔 + # 媒体搜索来源 themoviedb/douban/bangumi/anilist/imdb/musicbrainz/theaudiodb/doubanmusic，多个用,分隔 SEARCH_SOURCE: str = "themoviedb" - # 媒体识别来源 themoviedb/douban/bangumi/anilist/musicbrainz/theaudiodb/doubanmusic + # 媒体识别来源 themoviedb/douban/bangumi/anilist/imdb/musicbrainz/theaudiodb/doubanmusic RECOGNIZE_SOURCE: str = "themoviedb" - # 刮削来源 themoviedb/douban/bangumi/anilist/musicbrainz/theaudiodb/doubanmusic + # 刮削来源 themoviedb/douban/bangumi/anilist/imdb/musicbrainz/theaudiodb/doubanmusic SCRAP_SOURCE: str = "themoviedb" # 电视剧动漫的分类genre_ids ANIME_GENREIDS: List[int] = Field(default=[16]) @@ -1118,6 +1120,7 @@ class Settings(BaseSettings, ConfigModel, LogConfigModel): tmdb=1024, douban=512, bangumi=512, + imdb=512, fanart=512, musicbrainz=512, theaudiodb=512, @@ -1132,6 +1135,7 @@ class Settings(BaseSettings, ConfigModel, LogConfigModel): tmdb=256, douban=256, bangumi=256, + imdb=256, fanart=128, musicbrainz=256, theaudiodb=256, diff --git a/app/schemas/types.py b/app/schemas/types.py index 8b76c9b7b..f988ca3d9 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -605,6 +605,8 @@ class MediaRecognizeType(Enum): Bangumi = "Bangumi" # AniList AniList = "AniList" + # IMDb + IMDb = "IMDb" # MusicBrainz MusicBrainz = "MusicBrainz" # TheAudioDB diff --git a/tests/fixtures/architecture/configuration-debt-baseline.json b/tests/fixtures/architecture/configuration-debt-baseline.json index e5ec2c07d..63c7e531a 100644 --- a/tests/fixtures/architecture/configuration-debt-baseline.json +++ b/tests/fixtures/architecture/configuration-debt-baseline.json @@ -9,7 +9,7 @@ "root": "app" }, "settings_imports": { - "count": 164, + "count": 166, "files": [ "app/adapters/cache/backends.py", "app/adapters/cache/redis.py", @@ -112,6 +112,8 @@ "app/modules/filemanager/storages/smb.py", "app/modules/filemanager/storages/u115.py", "app/modules/filemanager/transhandler.py", + "app/modules/imdb/__init__.py", + "app/modules/imdb/api.py", "app/modules/indexer/parser/__init__.py", "app/modules/indexer/parser/rousi.py", "app/modules/indexer/spider/__init__.py", diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 486645837..a57b1ae9e 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6317, - "edge_sha256": "11978842b2fd9bd6ccec86c01c5f1b9389d959bc53aab5e16e29e3c121b3dab9", + "edge_count": 6341, + "edge_sha256": "073824a6b2c8b3e9baf4755ce4dccf94693fb943d1baba0f621d3a84b17030c0", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -4213,6 +4213,30 @@ "app.modules.filter -> app.runtime.log", "app.modules.filter -> app.schemas", "app.modules.filter -> app.schemas.types", + "app.modules.imdb -> app.domain", + "app.modules.imdb -> app.domain.context", + "app.modules.imdb -> app.domain.media", + "app.modules.imdb -> app.domain.meta", + "app.modules.imdb -> app.domain.meta.metabase", + "app.modules.imdb -> app.domain.scraper", + "app.modules.imdb -> app.foundation", + "app.modules.imdb -> app.foundation.text", + "app.modules.imdb -> app.modules", + "app.modules.imdb -> app.modules.imdb.api", + "app.modules.imdb -> app.runtime", + "app.modules.imdb -> app.runtime.config", + "app.modules.imdb -> app.runtime.log", + "app.modules.imdb -> app.schemas", + "app.modules.imdb -> app.schemas.context", + "app.modules.imdb -> app.schemas.media", + "app.modules.imdb -> app.schemas.types", + "app.modules.imdb.api -> app.adapters", + "app.modules.imdb.api -> app.adapters.network", + "app.modules.imdb.api -> app.adapters.network.http", + "app.modules.imdb.api -> app.runtime", + "app.modules.imdb.api -> app.runtime.cache", + "app.modules.imdb.api -> app.runtime.config", + "app.modules.imdb.api -> app.runtime.log", "app.modules.indexer -> app.application", "app.modules.indexer -> app.application.site", "app.modules.indexer -> app.application.site.health", @@ -6334,7 +6358,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 785, + "module_count": 787, "modules": [ "app", "app.adapters", @@ -6828,6 +6852,8 @@ "app.modules.filemanager.storages.u115", "app.modules.filemanager.transhandler", "app.modules.filter", + "app.modules.imdb", + "app.modules.imdb.api", "app.modules.indexer", "app.modules.indexer.parser", "app.modules.indexer.parser.bitpt", diff --git a/tests/test_imdb_module.py b/tests/test_imdb_module.py new file mode 100644 index 000000000..0a6d9c264 --- /dev/null +++ b/tests/test_imdb_module.py @@ -0,0 +1,464 @@ +"""IMDb 原生媒体数据源 Module 测试。""" + +import asyncio +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from app.domain.context import MediaInfo +from app.domain.meta.metabase import MetaBase +from app.modules.imdb import ImdbModule +from app.modules.imdb.api import ( + ImdbAka, + ImdbApi, + ImdbCredit, + ImdbDate, + ImdbEpisode, + ImdbImage, + ImdbPerson, + ImdbSeason, + ImdbTitle, +) +from app.runtime.config import settings +from app.schemas.types import MediaRecognizeType, MediaSource, MediaType + + +@pytest.fixture +def imdb_title() -> ImdbTitle: + """构造一条包含完整基础字段的 IMDb 电视剧条目。""" + return ImdbTitle( + id="tt0903747", + type="tvSeries", + primaryTitle="Breaking Bad", + originalTitle="Breaking Bad", + primaryImage={"url": "https://image.example/poster.jpg", "type": "poster"}, + startYear=2008, + runtimeSeconds=2820, + genres=["Crime", "Drama"], + rating={"aggregateRating": 9.5, "voteCount": 2200000}, + plot="A chemistry teacher becomes a methamphetamine producer.", + originCountries=[{"code": "US", "name": "United States"}], + spokenLanguages=[{"code": "eng", "name": "English"}], + ) + + +@pytest.fixture +def imdb_api(imdb_title: ImdbTitle) -> Mock: + """构造同步和异步方法行为一致的 IMDb API 替身。""" + api = Mock(spec=ImdbApi) + api.search_titles.return_value = [imdb_title] + api.get_title.return_value = imdb_title + api.list_akas.return_value = [ImdbAka(text="绝命毒师")] + api.list_credits.return_value = [ + ImdbCredit( + name=ImdbPerson( + id="nm0533713", + displayName="Vince Gilligan", + primaryImage=ImdbImage(url="https://image.example/director.jpg"), + ), + category="DIRECTOR", + ), + ImdbCredit( + name=ImdbPerson( + id="nm0186505", + displayName="Bryan Cranston", + primaryImage=ImdbImage(url="https://image.example/actor.jpg"), + ), + category="ACTOR", + characters=["Walter White"], + ), + ] + api.list_images.return_value = [ + ImdbImage(url="https://image.example/backdrop.jpg", type="still_frame") + ] + api.list_episodes.return_value = [ + ImdbEpisode( + id="tt0959621", + season="1", + episodeNumber=1, + releaseDate=ImdbDate(year=2008, month=1, day=20), + ) + ] + api.list_seasons.return_value = [ImdbSeason(season="1", episodeCount=7)] + api.async_search_titles = AsyncMock(return_value=[imdb_title]) + api.async_get_title = AsyncMock(return_value=imdb_title) + api.async_list_akas = AsyncMock(return_value=api.list_akas.return_value) + api.async_list_credits = AsyncMock(return_value=api.list_credits.return_value) + api.async_list_images = AsyncMock(return_value=api.list_images.return_value) + api.async_list_episodes = AsyncMock(return_value=api.list_episodes.return_value) + api.async_list_seasons = AsyncMock(return_value=api.list_seasons.return_value) + return api + + +@pytest.fixture +def imdb_module(imdb_api: Mock) -> ImdbModule: + """构造注入离线 API 替身的 IMDb Module。""" + module = ImdbModule() + module.imdb_api = imdb_api + module.scraper = Mock() + return module + + +def test_imdb_module_declares_independent_media_recognizer() -> None: + """IMDb 应作为独立宿主媒体识别 Module 暴露稳定元数据。""" + assert ImdbModule.get_name() == "IMDb" + assert ImdbModule.get_subtype() == MediaRecognizeType.IMDb + assert ImdbModule.get_priority() == 4 + + +def test_imdb_api_search_uses_keyless_imdb_suggestion_endpoint() -> None: + """IMDb 标题搜索应直接调用 IMDb 免 Key 建议接口并缓存响应。""" + api = ImdbApi() + api.clear_cache() + api._request = Mock() + api._request.get_json.return_value = { + "d": [ + { + "id": "tt0111161", + "qid": "movie", + "l": "The Shawshank Redemption", + "y": 1994, + } + ] + } + + results = api.search_titles("The Shawshank Redemption", limit=1) + cached_results = api.search_titles("The Shawshank Redemption", limit=1) + + assert [item.id for item in results] == ["tt0111161"] + assert [item.id for item in cached_results] == ["tt0111161"] + api._request.get_json.assert_called_once_with( + "https://v2.sg.media-imdb.com/suggestion/x/" + "The%20Shawshank%20Redemption.json", + params={}, + ) + + api.clear_cache() + api.search_titles("The Shawshank Redemption", limit=1) + assert api._request.get_json.call_count == 2 + + +def test_imdb_graphql_error_is_not_cached() -> None: + """GraphQL 错误响应不应污染缓存，后续成功详情仍应正常写入缓存。""" + api = ImdbApi() + api.clear_cache() + api._request = Mock() + api._request.post_json.side_effect = [ + {"errors": [{"message": "temporary failure"}]}, + { + "data": { + "titles": [ + { + "id": "tt0111161", + "titleText": {"text": "The Shawshank Redemption"}, + "titleType": {"id": "movie"}, + "releaseYear": {"year": 1994}, + } + ] + } + }, + ] + + assert api.get_title("tt0111161") is None + detail = api.get_title("tt0111161") + cached_detail = api.get_title("tt0111161") + + assert detail and detail.primary_title == "The Shawshank Redemption" + assert cached_detail and cached_detail.id == "tt0111161" + assert api._request.post_json.call_count == 2 + + +def test_async_imdb_api_merges_paginated_episodes() -> None: + """IMDb 异步剧集查询应使用 GraphQL 游标合并并缓存所有分页。""" + api = ImdbApi() + asyncio.run(api.async_clear_cache()) + api._async_request = Mock() + api._async_request.post_json = AsyncMock( + side_effect=[ + { + "data": { + "titles": [ + { + "episodes": { + "episodes": { + "edges": [ + { + "node": { + "id": "tt-first", + "series": { + "episodeNumber": { + "seasonNumber": 1, + "episodeNumber": 1, + } + }, + } + } + ], + "pageInfo": { + "hasNextPage": True, + "endCursor": "next-page", + }, + } + } + } + ] + } + }, + { + "data": { + "titles": [ + { + "episodes": { + "episodes": { + "edges": [ + { + "node": { + "id": "tt-second", + "series": { + "episodeNumber": { + "seasonNumber": 1, + "episodeNumber": 2, + } + }, + } + } + ], + "pageInfo": { + "hasNextPage": False, + "endCursor": None, + }, + } + } + } + ] + } + }, + { + "data": { + "titles": [ + { + "episodes": { + "episodes": { + "edges": [ + { + "node": { + "id": "tt-first", + "series": { + "episodeNumber": { + "seasonNumber": 1, + "episodeNumber": 1, + } + }, + } + } + ], + "pageInfo": { + "hasNextPage": True, + "endCursor": "next-page", + }, + } + } + } + ] + } + }, + { + "data": { + "titles": [ + { + "episodes": { + "episodes": { + "edges": [ + { + "node": { + "id": "tt-second", + "series": { + "episodeNumber": { + "seasonNumber": 1, + "episodeNumber": 2, + } + }, + } + } + ], + "pageInfo": { + "hasNextPage": False, + "endCursor": None, + }, + } + } + } + ] + } + }, + ] + ) + + results = asyncio.run(api.async_list_episodes("tt-series")) + cached_results = asyncio.run(api.async_list_episodes("tt-series")) + + assert [item.id for item in results] == ["tt-first", "tt-second"] + assert [item.id for item in cached_results] == ["tt-first", "tt-second"] + assert api._async_request.post_json.await_count == 2 + assert api._async_request.post_json.await_args_list[0].kwargs["json"][ + "variables" + ] == {"after": None, "first": 100, "titles": ("tt-series",)} + assert api._async_request.post_json.await_args_list[1].kwargs["json"][ + "variables" + ] == {"after": "next-page", "first": 100, "titles": ("tt-series",)} + + asyncio.run(api.async_clear_cache()) + refreshed_results = asyncio.run(api.async_list_episodes("tt-series")) + assert [item.id for item in refreshed_results] == ["tt-first", "tt-second"] + assert api._async_request.post_json.await_count == 4 + + +def test_imdb_proxy_snapshot_reloads_client_and_closes_old_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """代理变更应关闭旧 IMDb 客户端并按新代理快照重建。""" + module = ImdbModule() + old_client = Mock() + new_client = Mock() + monkeypatch.setattr(settings, "PROXY_HOST", "http://old-proxy:7890") + + with patch("app.modules.imdb.ImdbApi", side_effect=[old_client, new_client]) as api_type: + module.init_module() + old_proxy = settings.PROXY + monkeypatch.setattr(settings, "PROXY_HOST", "http://new-proxy:7890") + new_proxy = settings.PROXY + module.on_config_changed() + + assert api_type.call_args_list[0].kwargs["proxies"] == old_proxy + assert api_type.call_args_list[1].kwargs["proxies"] == new_proxy + old_client.close.assert_called_once_with() + assert module.imdb_api is new_client + + +def test_imdb_module_clear_cache_delegates_to_api( + imdb_module: ImdbModule, +) -> None: + """系统统一清缓存动作应清理 IMDb HTTP 缓存区。""" + imdb_module.clear_cache() + + imdb_module.imdb_api.clear_cache.assert_called_once_with() + + +def test_imdb_search_respects_source_and_returns_generic_identity( + imdb_module: ImdbModule, +) -> None: + """IMDb 搜索只响应选中的来源，并返回来源与原生 ID 对。""" + meta = MetaBase("Breaking Bad") + meta.name = "Breaking Bad" + meta.type = MediaType.TV + + assert imdb_module.search_medias(meta, media_source=MediaSource.Douban) is None + + results = imdb_module.search_medias(meta, media_source=MediaSource.IMDb) + + assert results and len(results) == 1 + assert results[0].media_source == MediaSource.IMDb + assert results[0].media_id == "tt0903747" + assert results[0].imdb_id == "tt0903747" + assert results[0].type == MediaType.TV + + +def test_imdb_explicit_identity_recognition_enriches_media_info( + imdb_module: ImdbModule, +) -> None: + """显式 IMDb ID 识别应补齐别名、演职员、剧集和图片。""" + result = imdb_module.recognize_media( + media_source=MediaSource.IMDb, + media_id="TT0903747", + ) + + assert isinstance(result, MediaInfo) + assert result.media_source == MediaSource.IMDb + assert result.media_id == "tt0903747" + assert result.title == "Breaking Bad" + assert result.names == ["Breaking Bad", "绝命毒师"] + assert result.backdrop_path == "https://image.example/backdrop.jpg" + assert result.directors[0]["name"] == "Vince Gilligan" + assert result.actors[0]["character"] == "Walter White" + assert result.seasons == {1: [1]} + assert result.season_years == {1: "2008"} + assert result.number_of_episodes == 1 + assert result.runtime == 47 + assert result.category == "欧美剧" + + +def test_imdb_name_recognition_requires_imdb_selection( + imdb_module: ImdbModule, monkeypatch: pytest.MonkeyPatch +) -> None: + """名称识别仅在请求级或全局来源选中 IMDb 时执行。""" + meta = MetaBase("绝命毒师") + meta.name = "绝命毒师" + meta.cn_name = "绝命毒师" + meta.type = MediaType.TV + monkeypatch.setattr(settings, "RECOGNIZE_SOURCE", MediaSource.TMDB.value) + + assert imdb_module.recognize_media(meta=meta) is None + + result = imdb_module.recognize_media( + meta=meta, + media_source=MediaSource.IMDb, + ) + + assert result and result.media_id == "tt0903747" + imdb_module.imdb_api.search_titles.assert_called_with("绝命毒师") + + +def test_imdb_recognition_rejects_invalid_or_music_identity( + imdb_module: ImdbModule, +) -> None: + """IMDb Module 不应接管非法 ID 或音乐识别请求。""" + assert ( + imdb_module.recognize_media( + media_source=MediaSource.IMDb, + media_id="0903747", + ) + is None + ) + assert ( + imdb_module.recognize_media( + mtype=MediaType.MUSIC, + media_source=MediaSource.IMDb, + media_id="tt0903747", + ) + is None + ) + + +def test_async_imdb_recognition_uses_async_detail_pipeline( + imdb_module: ImdbModule, +) -> None: + """异步显式识别应使用异步详情管线并保持统一身份。""" + result = asyncio.run( + imdb_module.async_recognize_media( + media_source=MediaSource.IMDb, + media_id="tt0903747", + ) + ) + + assert result and result.media_source == MediaSource.IMDb + assert result.media_id == "tt0903747" + assert result.backdrop_path == "https://image.example/backdrop.jpg" + imdb_module.imdb_api.async_get_title.assert_awaited() + imdb_module.imdb_api.async_list_credits.assert_awaited_once_with("tt0903747") + + +def test_imdb_scraping_only_handles_imdb_source( + imdb_module: ImdbModule, +) -> None: + """IMDb NFO 和图片刮削只响应显式 IMDb 刮削来源。""" + media_info = MediaInfo() + media_info.scrape_source = MediaSource.Douban.value + assert imdb_module.metadata_nfo(media_info) is None + assert imdb_module.metadata_img(media_info) is None + + media_info.scrape_source = MediaSource.IMDb.value + imdb_module.scraper.get_metadata_nfo.return_value = "" + imdb_module.scraper.get_metadata_img.return_value = {"poster.jpg": "url"} + + assert imdb_module.metadata_nfo(media_info) == "" + assert imdb_module.metadata_img(media_info) == {"poster.jpg": "url"} diff --git a/tests/test_module_manager_capability_adapter.py b/tests/test_module_manager_capability_adapter.py index 5cc183eac..2e443e396 100644 --- a/tests/test_module_manager_capability_adapter.py +++ b/tests/test_module_manager_capability_adapter.py @@ -490,7 +490,7 @@ from app.runtime.extensions.host_module_adapter import ( registry = build_host_module_registry() specs = registry.list_specs() -assert len(specs) == 37 +assert len(specs) == 38 adapter = HostModuleAdapter() lifecycle_events = [] @@ -537,7 +537,7 @@ from app.schemas.types import EventType registry = build_host_module_registry() specs = registry.list_specs() -assert len(specs) == 37 +assert len(specs) == 38 spec_by_id = {spec.id: spec for spec in specs} events = {spec.id: [] for spec in specs} @@ -704,7 +704,7 @@ from app.runtime.extensions.host_module_adapter import ( registry = build_host_module_registry() specs = registry.list_specs() -assert len(specs) == 37 +assert len(specs) == 38 configured_specs = tuple( spec for spec in specs if spec.activation is ActivationPolicy.WHEN_CONFIGURED @@ -800,12 +800,12 @@ from app.application.module import configure_module_runtime configure_module_runtime(lambda: ModuleManager()) manager = ModuleManager() -assert len(manager.list_specs()) == 37 +assert len(manager.list_specs()) == 38 assert manager.get_specs() == manager.list_specs() from app.api.endpoints.system import modulelist response = modulelist(None) -assert len(response.data["modules"]) == 37 +assert len(response.data["modules"]) == 38 heavy_prefixes = ( "lark_oapi", @@ -913,7 +913,7 @@ from app.runtime.extensions.module_manager import ModuleManager manager = ModuleManager() modules = manager.get_modules() -assert len(modules) == len(manager.list_specs()) == 37 +assert len(modules) == len(manager.list_specs()) == 38 for spec in manager.list_specs(): implementation = modules[spec.id] assert implementation.get_name() == spec.metadata["name"]