diff --git a/app/modules/trimemedia/api.py b/app/modules/trimemedia/api.py index ac44c04d0..f7aa9458e 100644 --- a/app/modules/trimemedia/api.py +++ b/app/modules/trimemedia/api.py @@ -5,6 +5,7 @@ import time from dataclasses import dataclass from enum import Enum from typing import List, Optional, Union +from urllib.parse import quote from app.core.config import settings from app.log import logger @@ -108,6 +109,7 @@ class Api: "_host", "_token", "_apikey", + "_access_code", "_api_path", "_request_utils", "_version", @@ -130,18 +132,51 @@ class Api: def version(self) -> Optional[Version]: return self._version - def __init__(self, host: str, apikey: str): + @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]: """ 飞牛影视版本号 @@ -161,6 +196,9 @@ class Api: :return: 成功返回token 否则返回None """ + # 开启访问码后需先通过访问码校验,否则无法访问登录接口 + if not self.verify_access_code(): + return None if ( res := self.request( "/login", diff --git a/app/modules/trimemedia/trimemedia.py b/app/modules/trimemedia/trimemedia.py index e446a4035..b4d90166f 100644 --- a/app/modules/trimemedia/trimemedia.py +++ b/app/modules/trimemedia/trimemedia.py @@ -12,6 +12,7 @@ from app.utils.url import UrlUtils class TrimeMedia: _username: Optional[str] = None _password: Optional[str] = None + _access_code: Optional[str] = None _userinfo: Optional[fnapi.User] = None _host: Optional[str] = None @@ -28,6 +29,7 @@ class TrimeMedia: host: Optional[str] = None, username: Optional[str] = None, password: Optional[str] = None, + access_code: Optional[str] = None, play_host: Optional[str] = None, sync_libraries: Optional[list] = None, **kwargs, @@ -37,13 +39,14 @@ class TrimeMedia: return self._username = username self._password = password + self._access_code = access_code self._host = host self._sync_libraries = sync_libraries or [] if not self.reconnect(): logger.error(f"请检查服务端地址 {host}") return - if result := self.__create_api(play_host): + if result := self.__create_api(play_host, access_code): self._playhost = result.api.host result.api.close() elif play_host: @@ -69,11 +72,14 @@ class TrimeMedia: version: fnapi.Version @staticmethod - def __create_api(host: Optional[str]) -> Optional["TrimeMedia._ApiCreateResult"]: + def __create_api( + host: Optional[str], access_code: Optional[str] = None + ) -> Optional["TrimeMedia._ApiCreateResult"]: """ 创建一个飞牛API :param host: 服务端地址 + :param access_code: 访问码,未开启时为空 :return: 如果地址无效、不可访问则返回None """ @@ -85,16 +91,19 @@ class TrimeMedia: if not host.endswith("/v"): # 尝试补上结尾的/v 测试能否正常访问 res = TrimeMedia._ApiCreateResult() - res.api = fnapi.Api(host + "/v", api_key) - if fnver := res.api.sys_version(): + res.api = fnapi.Api(host + "/v", api_key, access_code) + # 开启访问码后,需先校验才能访问各应用接口 + if res.api.verify_access_code() and (fnver := res.api.sys_version()): res.version = fnver return res + res.api.close() # 测试用户配置的地址 res = TrimeMedia._ApiCreateResult() - res.api = fnapi.Api(host, api_key) - if fnver := res.api.sys_version(): + res.api = fnapi.Api(host, api_key, access_code) + if res.api.verify_access_code() and (fnver := res.api.sys_version()): res.version = fnver return res + res.api.close() return None def close(self): @@ -130,7 +139,7 @@ class TrimeMedia: if not self.is_configured(): return False self.disconnect() - if result := self.__create_api(self._host): + if result := self.__create_api(self._host, self._access_code): self._api = result.api self._version = result.version # 版本号:0.8.53, 服务版本:0.8.23 @@ -683,4 +692,8 @@ class TrimeMedia: image_url, [self._api.host], strict=True ): return None - return {"Trim-MC-token": self._api.token} + cookies = {"Trim-MC-token": self._api.token} + if self._access_code: + # 开启访问码后,图片请求也需要携带访问码校验凭证 + cookies.update(self._api.cookies) + return cookies