MoviePilot/app/modules/trimemedia/api.py

import hashlib
import json
import random
import time
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union
from urllib.parse import quote

from app.runtime.settings import get_runtime_setting

from app.runtime.log import logger
from app.adapters.network.http import RequestUtils, requests


@dataclass
class User:
    guid: str
    username: str
    is_admin: int = 0


class Category(Enum):
    MOVIE = "Movie"
    TV = "TV"
    MIX = "Mix"
    OTHERS = "Others"

    @classmethod
    def _missing_(cls, value):
        return cls.OTHERS


class Type(Enum):
    MOVIE = "Movie"
    TV = "TV"
    SEASON = "Season"
    EPISODE = "Episode"
    VIDEO = "Video"
    DIRECTORY = "Directory"

    @classmethod
    def _missing_(cls, value):
        return cls.VIDEO


@dataclass
class MediaDb:
    guid: str
    category: Category
    name: Optional[str] = None
    posters: Optional[list[str]] = None
    dir_list: Optional[list[str]] = None


@dataclass
class MediaDbSummary:
    favorite: int = 0
    movie: int = 0
    tv: int = 0
    video: int = 0
    total: int = 0


@dataclass
class Version:
    # 飞牛影视版本
    frontend: Optional[str] = None
    backend: Optional[str] = None


@dataclass
class Item:
    guid: str
    ancestor_guid: str = ""
    type: Optional[Type] = None
    # 当type为Episode时是剧名，parent_title是季名，title作为分集名称
    tv_title: Optional[str] = None
    parent_title: Optional[str] = None
    title: Optional[str] = None
    original_title: Optional[str] = None
    overview: Optional[str] = None
    poster: Optional[str] = None
    backdrops: Optional[str] = None
    posters: Optional[str] = None
    douban_id: Optional[int] = None
    imdb_id: Optional[str] = None
    trim_id: Optional[str] = None
    release_date: Optional[str] = None
    air_date: Optional[str] = None
    vote_average: Optional[str] = None
    season_number: Optional[int] = None
    episode_number: Optional[int] = None
    duration: Optional[int] = None  # 片长(秒)
    ts: Optional[int] = None  # 已播放(秒)
    watched: Optional[int] = None  # 1:已看完

    @property
    def tmdb_id(self) -> Optional[int]:
        if self.trim_id is None:
            return None
        if self.trim_id.startswith("tt") or self.trim_id.startswith("tm"):
            # 飞牛给tmdbid加了前缀用以区分tv或movie
            return int(self.trim_id[2:])
        return None


class Api:
    __slots__ = (
        "_host",
        "_token",
        "_apikey",
        "_access_code",
        "_api_path",
        "_request_utils",
        "_version",
        "_session",
    )

    @property
    def token(self) -> Optional[str]:
        return self._token

    @property
    def host(self) -> str:
        return self._host

    @property
    def apikey(self) -> str:
        return self._apikey

    @property
    def version(self) -> Optional[Version]:
        return self._version

    @property
    def cookies(self) -> dict:
        """
        当前会话的Cookies，开启访问码后包含访问码校验凭证
        """
        return self._session.cookies.get_dict()

    def __init__(self, host: str, apikey: str, access_code: Optional[str] = None):
        """
        :param host: 飞牛服务端地址，如http://127.0.0.1:5666/v
        :param access_code: 访问码，未开启时为空
        """
        self._api_path = "/api/v1"
        self._host = host.rstrip("/")
        self._apikey = apikey
        self._access_code = access_code
        self._token: Optional[str] = None
        self._version: Optional[Version] = None
        self._session = requests.Session()
        self._request_utils = RequestUtils(session=self._session, timeout=10)

    def verify_access_code(self) -> bool:
        """
        校验访问码，通过后会话获得访问凭证，否则无法访问登录页和各应用接口

        :return: 未配置访问码或校验通过返回True
        """
        if not self._access_code:
            return True
        # 访问码校验地址位于设备根路径，不在/v下
        root = self._host[: -len("/v")] if self._host.endswith("/v") else self._host
        url = f"{root}/c/{quote(self._access_code, safe='')}"
        res = self._request_utils.get_res(url, allow_redirects=True)
        if res is None:
            logger.error(f"校验飞牛访问码失败，无法访问 {url}")
            return False
        if res.status_code == 404:
            # 访问码错误或校验失败时返回404
            logger.error("飞牛访问码校验失败，请检查访问码是否正确")
            return False
        if not res.ok:
            logger.error(f"飞牛访问码校验失败，状态码：{res.status_code}")
            return False
        return True

    def sys_version(self) -> Optional[Version]:
        """
        飞牛影视版本号
        """
        if (res := self.request("/sys/version")) and res.success:
            if res.data:
                self._version = Version(
                    frontend=res.data.get("version"),
                    backend=res.data.get("mediasrvVersion"),
                )
                return self._version
        return None

    def login(self, username, password) -> Optional[str]:
        """
        登录飞牛影视

        新版服务端已废弃 v1 明文登录接口，优先使用 v2 协议登录（密码传输 SHA256 摘要），
        v2 接口不存在时回退旧版 v1 明文登录

        :return: 成功返回token 否则返回None
        """
        # 开启访问码后需先通过访问码校验，否则无法访问登录接口
        if not self.verify_access_code():
            return None
        # v2 协议要求密码为明文密码的 SHA256 十六进制小写摘要
        password_hash = hashlib.sha256(password.encode()).hexdigest()
        res = self.request(
            "/user/loginByPassword",
            data={
                "username": username,
                "password": password_hash,
                "app_name": "trimemedia-web",
            },
            base_path="/api/v2",
            suppress_log=True,
        )
        if res and res.success:
            self._token = res.data.get("token")
            return self._token
        if res:
            # v2 接口存在但登录失败（如账号密码错误），回退 v1 也无法成功
            logger.error(f"飞牛影视登录失败，错误码：{res.code} {res.msg}")
            return None
        # v2 接口不可用（旧版服务端），回退 v1 明文登录
        if (
            res := self.request(
                "/login",
                data={
                    "username": username,
                    "password": password,
                    "app_name": "trimemedia-web",
                },
            )
        ) and res.success:
            self._token = res.data.get("token")
        return self._token

    def logout(self) -> bool:
        """
        退出账号
        """
        if not self._token:
            return True
        if (res := self.request("/user/logout", method="post")) and res.success:
            if res.data:
                self._token = None
                return True
        return False

    def user_list(self) -> Optional[list[User]]:
        """
        用户列表(仅管理员有权访问)
        """
        if (res := self.request("/manager/user/list")) and res.success:
            if not res.data:
                return []
            return [
                User(
                    guid=info.get("guid"),
                    username=info.get("username"),
                    is_admin=info.get("is_admin", 0),
                )
                for info in res.data
            ]
        return None

    def user_info(self) -> Optional[User]:
        """
        当前用户信息
        """
        if (res := self.request("/user/info")) and res.success:
            _user = User("", "")
            _user.__dict__.update(res.data)
            return _user
        return None

    def mediadb_sum(self) -> Optional[MediaDbSummary]:
        """
        媒体数量统计
        """
        if (res := self.request("/mediadb/sum")) and res.success:
            sums = MediaDbSummary()
            sums.__dict__.update(res.data)
            return sums
        return None

    def mediadb_list(self) -> Optional[List[MediaDb]]:
        """
        媒体库列表(普通用户)
        """
        if (res := self.request("/mediadb/list")) and res.success:
            _items = []
            for info in res.data or []:
                mdb = MediaDb(
                    guid=info.get("guid"),
                    category=Category(info.get("category")),
                    name=info.get("title", ""),
                    posters=[
                        self.__build_img_api_url(poster)
                        for poster in info.get("posters", [])
                    ],
                )
                _items.append(mdb)
            return _items
        return None

    def __build_img_api_url(self, img_path: Optional[str]) -> Optional[str]:
        if not img_path:
            return None
        if img_path[0] != "/":
            img_path = "/" + img_path
        return f"{self._api_path}/sys/img{img_path}"

    def mdb_list(self) -> Optional[list[MediaDb]]:
        """
        媒体库列表(管理员)
        """
        if (res := self.request("/mdb/list")) and res.success:
            _items = []
            for info in res.data or []:
                mdb = MediaDb(
                    guid=info.get("guid"),
                    category=Category(info.get("category")),
                    name=info.get("name", ""),
                    posters=[
                        self.__build_img_api_url(poster)
                        for poster in info.get("posters", [])
                    ],
                    dir_list=info.get("dir_list"),
                )
                _items.append(mdb)
            return _items
        return None

    def mdb_scanall(self) -> bool:
        """
        扫描所有媒体库
        """
        if (res := self.request("/mdb/scanall", method="post")) and res.success:
            if res.data:
                return True
        return False

    def mdb_scan(self, mdb: MediaDb) -> bool:
        """
        扫描指定媒体库
        """
        if (res := self.request(f"/mdb/scan/{mdb.guid}", data={})) and res.success:
            if res.data:
                return True
        return False

    def task_running(self):
        """
        当前正在运行的任务
        """
        if (res := self.request("/task/running")) and res.success:
            if res.data:
                # TODO 具体正在运行的任务
                return True
        return False

    def __build_item(self, info: dict) -> Item:
        """
        构造媒体Item
        """
        item = Item(guid="")
        item.__dict__.update(info)
        item.type = Type(info.get("type"))
        # Item详情接口才有posters和backdrops
        item.posters = self.__build_img_api_url(item.posters)
        item.backdrops = self.__build_img_api_url(item.backdrops)
        item.poster = (
            self.__build_img_api_url(item.poster) if item.poster else item.posters
        )
        return item

    def item_list(
        self,
        guid: Optional[str] = None,
        types=None,
        exclude_grouped_video=True,
        page=1,
        page_size=20,
        sort_by="create_time",
        sort="DESC",
    ) -> Optional[list[Item]]:
        """
        媒体列表
        """
        if types is None:
            types = [Type.MOVIE, Type.TV, Type.DIRECTORY, Type.VIDEO]
        post = {
            "tags": {"type": types} if types else {},
            "sort_type": sort,
            "sort_column": sort_by,
            "page": page,
            "page_size": page_size,
        }
        if guid:
            post["ancestor_guid"] = guid
        if exclude_grouped_video:
            post["exclude_grouped_video"] = 1

        if (res := self.request("/item/list", data=post)) and res.success:
            if not res.data:
                return []
            return [self.__build_item(info) for info in res.data.get("list", [])]
        return None

    def item_count(self, guid: str, types=None) -> Optional[int]:
        """
        获取指定媒体库的媒体条目总数

        :param guid: 媒体库GUID
        :param types: 需要统计的媒体类型
        :return: 媒体条目总数，查询失败时返回None
        """
        if types is None:
            types = [Type.MOVIE, Type.TV]
        post = {
            "ancestor_guid": guid,
            "tags": {"type": types},
            "exclude_grouped_video": 1,
            "page": 1,
            "page_size": 1,
        }
        if (res := self.request("/item/list", data=post)) and res.success:
            if not res.data:
                return 0
            total_count = res.data.get("total")
            if total_count is None:
                total_count = res.data.get("total_count")
            return int(total_count) if total_count is not None else None
        return None

    def search_list(self, keywords: str) -> Optional[list[Item]]:
        """
        搜索影片、演员
        """
        if (
            res := self.request("/search/list", params={"q": keywords})
        ) and res.success:
            if not res.data:
                return []
            return [self.__build_item(info) for info in res.data]
        return None

    def item(self, guid: str) -> Optional[Item]:
        """
        查询媒体详情
        """
        if (res := self.request(f"/item/{guid}")) and res.success:
            return self.__build_item(res.data)
        return None

    def del_item(self, guid: str, delete_file: bool) -> bool:
        """
        删除媒体
        :param guid: 媒体GUID
        :param delete_file: True删除媒体文件，False仅从媒体库移除
        """
        if (
            res := self.request(
                f"/item/{guid}",
                method="delete",
                data={"delete_file": 1 if delete_file else 0, "media_guids": []},
            )
        ) and res.success:
            if res.data:
                return True
        return False

    def season_list(self, tv_guid: str) -> Optional[list[Item]]:
        """
        查询季列表
        """
        if (res := self.request(f"/season/list/{tv_guid}")) and res.success:
            if not res.data:
                return []
            return [self.__build_item(info) for info in res.data]
        return None

    def episode_list(self, season_guid: str) -> Optional[list[Item]]:
        """
        查询剧集列表
        """
        if (res := self.request(f"/episode/list/{season_guid}")) and res.success:
            if not res.data:
                return []
            return [self.__build_item(info) for info in res.data]
        return None

    def play_list(self) -> Optional[list[Item]]:
        """
        继续观看列表
        """
        if (res := self.request("/play/list")) and res.success:
            if not res.data:
                return []
            return [self.__build_item(info) for info in res.data]
        return None

    def __get_authx(self, api_path: str, body: Optional[str]):
        """
        计算消息签名
        """
        if not api_path.startswith("/v"):
            api_path = "/v" + api_path
        nonce = str(random.randint(100000, 999999))
        ts = str(int(time.time() * 1000))
        md5 = hashlib.md5()
        md5.update((body or "").encode())
        data_hash = md5.hexdigest()
        md5 = hashlib.md5()
        md5.update(
            "_".join(
                [
                    "NDzZTVxnRKP8Z0jXg1VAMonaG8akvh",
                    api_path,
                    nonce,
                    ts,
                    data_hash,
                    self._apikey,
                ]
            ).encode()
        )
        sign = md5.hexdigest()
        return f"nonce={nonce}&timestamp={ts}&sign={sign}"

    def request(
        self,
        api: str,
        method: Optional[str] = None,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
        base_path: Optional[str] = None,
        suppress_log=False,
    ):
        """
        请求飞牛影视API

        :param base_path: 接口路径前缀（如 /api/v2），默认使用 v1 路径
        :param suppress_log: 是否禁止日志
        """

        @dataclass
        class Result:
            @property
            def success(self) -> bool:
                return code == 0

            code: int
            msg: Optional[str] = None
            data: Optional[Union[dict, list, str, bool]] = None

        class JsonEncoder(json.JSONEncoder):
            def default(self, obj):
                if isinstance(obj, Type):
                    return obj.value
                return super().default(obj)

        if not self._host or not api:
            return None
        prefix = base_path if base_path is not None else self._api_path
        if not api.startswith("/"):
            api_path = f"{prefix}/{api}"
        else:
            api_path = prefix + api
        url = self._host + api_path
        if method is None:
            method = "get" if data is None else "post"
        if method != "get":
            json_body = (
                json.dumps(data, allow_nan=False, cls=JsonEncoder) if data else ""
            )
        else:
            json_body = None
        if params:
            queries_unquoted = "&".join([f"{k}={v}" for k, v in params.items()])
        else:
            queries_unquoted = None
        headers = {
            "User-Agent": get_runtime_setting('USER_AGENT'),
            "Accept": "application/json",
            "Referer": self._host,
            "Authorization": self._token,
            "authx": self.__get_authx(api_path, json_body or queries_unquoted),
        }
        if json_body is not None:
            headers["Content-Type"] = "application/json"
        try:
            res = self._request_utils.request(
                method=method, url=url, headers=headers, params=params, data=json_body
            )
            if res:
                resp = res.json()
                msg = resp.get("msg")
                if code := int(resp.get("code", -1)):
                    if not suppress_log:
                        logger.error(f"请求接口 {url} 失败，错误码：{code} {msg}")
                    return Result(code, msg)
                return Result(0, msg, resp.get("data"))
            elif not suppress_log:
                logger.error(f"请求接口 {url} 失败")
        except Exception as e:
            if not suppress_log:
                logger.error(f"请求接口 {url} 异常：" + str(e))
        return None

    def close(self):
        """
        关闭API会话
        """
        if self._session:
            self._session.close()