MoviePilot/app/modules/_base/mediaserver.py

"""
媒体服务器模块业务样板基类。

沉淀各媒体服务器模块逐字复制的样板：用户辅助认证、媒体存在性检查、
定时重连与连接测试。服务器差异（认证 API、存在性检查端点、连接探测方式）
通过类属性与钩子方法保留在各模块。
"""
from typing import Any, Iterable, Mapping, Optional, Tuple, cast

from app.application.mediaserver import MusicMediaServerHelper
from app.domain.context import MediaInfo, MusicInfo
from app.modules import TService, _MediaServerBase, _ModuleBase
from app.runtime.events import eventmanager
from app.runtime.log import logger
from app.schemas.event import AuthCredentials as _SchemaAuthCredentials
from app.schemas.event import AuthInterceptCredentials as _SchemaAuthInterceptCredentials
from app.schemas.mediaserver import ExistMediaInfo as _SchemaExistMediaInfo
from app.schemas.mediaserver import MediaServerField, MediaServerQueryOptions
from app.schemas.types import ChainEventType, MediaType


def emby_episode_numbers(item: Mapping[str, Any]) -> range:
    """
    展开 Emby 系分集条目的起止集号，缺少结束集号或倒序时按单集处理。

    :param item: 含 IndexNumber 与可空 IndexNumberEnd 的分集响应
    :return: 包含合并文件所覆盖全部集号的区间，未提供起始集号时为空
    """
    number = item.get("IndexNumber")
    if number is None:
        return range(0)
    start = int(number)
    end = item.get("IndexNumberEnd")
    return range(start, max(start, int(end) if end is not None else start) + 1)


def emby_query_params(query: Optional[MediaServerQueryOptions]) -> dict[str, str]:
    """
    把显式投影翻译为 Emby/Jellyfin 已公开的列表查询控制参数。

    名称、类型由服务端默认返回；年份必须通过 Fields 显式请求，避免精简查询丢失匹配和同步依据。
    Fields 只追加当前投影所需的字段。
    音乐完整性匹配需要 ChildCount，即使未请求额外统计也不能删除它。

    :param query: 调用方投影；None 或 fields=None 时不改变旧请求
    :return: 字段追加、图片信息和用户数据参数；不添加未公开的总量开关
    """
    if query is None or query.fields is None:
        return {}
    fields = {
        MediaServerField.LIBRARY: "ParentId",
        MediaServerField.ORIGINAL_TITLE: "OriginalTitle",
        MediaServerField.YEAR: "ProductionYear",
        MediaServerField.IDENTITY: "ProviderIds",
        MediaServerField.PATH: "Path",
        MediaServerField.MUSIC: "ChildCount",
    }
    return {
        "Fields": ",".join(value for field, value in fields.items() if query.includes(field)),
        "EnableImages": str(query.includes(MediaServerField.IMAGES)).lower(),
        "EnableUserData": str(query.includes(MediaServerField.USER_STATE)).lower(),
    }


class _MediaServerModuleBase(_ModuleBase, _MediaServerBase[TService]):
    """
    媒体服务器模块业务样板基类。
    """

    # 媒体库标识（用于 ExistMediaInfo.server_type，如 "emby"），子类覆写
    _server_type_value: str = ""
    # 仅已实现查询投影的适配器向原生客户端透传新参数。
    _query_options_supported: bool = False

    def user_authenticate(
            self,
            credentials: _SchemaAuthCredentials,
            service_name: Optional[str] = None,
    ) -> Optional[_SchemaAuthCredentials]:
        """
        使用媒体服务器用户辅助完成用户认证。

        :param credentials: 认证数据
        :param service_name: 指定要认证的媒体服务器名称，若为 None 则认证所有服务器
        :return: 成功时返回已写入渠道、服务名和令牌的原凭据；不适用或所有服务认证失败时返回 None
        """
        if not credentials or credentials.grant_type != "password":
            return None
        # 确定要认证的服务器列表
        servers: Iterable[tuple[Optional[str], TService]]
        if service_name:
            # 如果指定了服务名，获取该服务实例
            servers = (
                [(service_name, server)]
                if (server := self.get_instance(service_name))
                else []
            )
        else:
            # 如果没有指定服务名，遍历所有服务
            servers = self.get_instances().items()
        # 遍历要认证的服务器
        for name, server in servers:
            service_name = name or ""
            channel_name = self.get_name() or ""
            # 触发认证拦截事件
            intercept_event = eventmanager.send_event(
                etype=ChainEventType.AuthIntercept,
                data=_SchemaAuthInterceptCredentials(
                    username=credentials.username,
                    channel=channel_name,
                    service=service_name,
                    status="triggered",
                ),
            )
            if intercept_event and intercept_event.event_data:
                intercept_data = (
                    intercept_event.event_data
                    if isinstance(intercept_event.event_data, _SchemaAuthInterceptCredentials)
                    else _SchemaAuthInterceptCredentials.model_validate(intercept_event.event_data)
                )
                if intercept_data.cancel:
                    continue
            token = cast(Any, server).authenticate(credentials.username, credentials.password)
            if token:
                credentials.channel = channel_name
                credentials.service = name
                credentials.token = token
                return credentials
        return None

    def media_exists(
        self,
        mediainfo: MediaInfo,
        itemid: Optional[str] = None,
        server: Optional[str] = None,
        *,
        query: Optional[MediaServerQueryOptions] = None,
    ) -> Optional[_SchemaExistMediaInfo]:
        """
        判断媒体文件是否存在。

        :param mediainfo:  识别的媒体信息
        :param itemid:  媒体服务器ItemID
        :param server:  媒体服务器名称
        :param query: 可选查询内容投影，省略时保留原有请求和返回内容
        :return: 如不存在返回None，存在时返回信息，包括每季已存在所有集{type: movie/tv, seasons: {season: [episodes]}}
        """
        if not self._query_options_supported:
            query = None
        servers: Iterable[tuple[Optional[str], Optional[TService]]]
        if server:
            servers = [(server, self.get_instance(server))]
        else:
            servers = self.get_instances().items()
        for name, s in servers:
            if not s:
                continue
            if mediainfo.type == MediaType.MUSIC:
                # 部分服务器未实现音乐查询，退化为空列表
                if not isinstance(mediainfo, MusicInfo):
                    continue
                matches: list[Any] = getattr(s, "get_music", lambda **_: [])(
                    **MusicMediaServerHelper.search_params(mediainfo),
                    **({"query": query.with_fields(MediaServerField.MUSIC)} if query is not None else {}),
                )
                match = MusicMediaServerHelper.find_match(mediainfo, matches)
                if match:
                    return _SchemaExistMediaInfo(
                        type=MediaType.MUSIC,
                        server_type=self._server_type_value,
                        server=name,
                        itemid=match.item_id,
                    )
                continue
            service = cast(Any, s)
            if mediainfo.type == MediaType.MOVIE:
                if itemid:
                    movie = service.get_iteminfo(itemid, **({"query": query} if query is not None else {}))
                    if movie:
                        logger.info(f"媒体库 {name} 中找到了 {movie}")
                        return _SchemaExistMediaInfo(
                            type=MediaType.MOVIE, server_type=self._server_type_value, server=name, itemid=movie.item_id
                        )
                movies = service.get_movies(
                    title=mediainfo.title,
                    year=mediainfo.year,
                    media_source=mediainfo.media_source,
                    media_id=mediainfo.media_id,
                    **({"query": query} if query is not None else {}),
                )
                if not movies:
                    logger.info(f"{mediainfo.title_year} 没有在媒体库 {name} 中")
                    continue
                else:
                    logger.info(f"媒体库 {name} 中找到了 {movies}")
                    return _SchemaExistMediaInfo(
                        type=MediaType.MOVIE, server_type=self._server_type_value, server=name, itemid=movies[0].item_id
                    )
            else:
                itemid, tvs = service.get_tv_episodes(
                    title=mediainfo.title,
                    year=mediainfo.year,
                    media_source=mediainfo.media_source,
                    media_id=mediainfo.media_id,
                    item_id=itemid,
                    **({"query": query} if query is not None else {}),
                )
                if not tvs:
                    logger.info(f"{mediainfo.title_year} 没有在媒体库 {name} 中")
                    continue
                else:
                    logger.info(f"{mediainfo.title_year} 在媒体库 {name} 中找到 了这些季集：{tvs}")
                    return _SchemaExistMediaInfo(
                        type=MediaType.TV, seasons=tvs, server_type=self._server_type_value, server=name, itemid=itemid
                    )
        return None

    def scheduler_job(self) -> None:
        """
        巡检已配置媒体服务器，对失活连接发起重连。

        由模块调度器每十分钟调用一次。
        """
        # 定时重连
        for name, server in self.get_instances().items():
            if self._is_inactive(server):
                logger.info(f"{self.get_name()}服务器 {name} 连接断开，尝试重连 ...")
                cast(Any, server).reconnect()

    def _is_inactive(self, server: Any) -> bool:
        """
        定时重连的失活判断钩子，子类可覆写（如增加配置完整性检查）。
        """
        return bool(server.is_inactive())

    def test(self) -> Optional[Tuple[bool, str]]:
        """
        检查已配置媒体服务器，遇到首个不可用实例即返回失败。

        :return: 是否就绪与失败说明；没有实例时返回 None
        """
        if not self.get_instances():
            return None
        for name, server in self.get_instances().items():
            error = self._test_server(server, name)
            if error:
                return False, error
        return True, ""

    def _test_server(self, server: Any, name: Optional[str]) -> Optional[str]:
        """
        连接测试钩子，返回失败信息，None 表示就绪，子类可覆写。
        """
        if server.is_inactive():
            server.reconnect()
        if not server.get_user():
            return f"无法连接{self.get_name()}服务器：{name}"
        return None