diff --git a/app/modules/plex/plex.py b/app/modules/plex/plex.py index edd5b22b..d7a7f7b5 100644 --- a/app/modules/plex/plex.py +++ b/app/modules/plex/plex.py @@ -6,6 +6,7 @@ from urllib.parse import quote_plus from cachetools import TTLCache, cached from plexapi import media from plexapi.server import PlexServer +from requests import Response, Session from app import schemas from app.core.config import settings @@ -16,6 +17,7 @@ from app.utils.http import RequestUtils class Plex: _plex = None + _session = None def __init__(self): self._host = settings.PLEX_HOST @@ -32,6 +34,7 @@ class Plex: except Exception as e: self._plex = None logger.error(f"Plex服务器连接失败:{str(e)}") + self._session = self.__adapt_plex_session() def is_inactive(self) -> bool: """ @@ -722,3 +725,71 @@ class Plex: )) offset += num return ret_resume[:num] + + def get_data(self, endpoint: str, **kwargs) -> Optional[Response]: + """ + 自定义从媒体服务器获取数据 + :param endpoint: 端点 + :param kwargs: 其他请求参数,如headers, cookies, proxies等 + """ + return self.__request(method="get", endpoint=endpoint, **kwargs) + + def post_data(self, endpoint: str, **kwargs) -> Optional[Response]: + """ + 自定义从媒体服务器获取数据 + :param endpoint: 端点 + :param kwargs: 其他请求参数,如headers, cookies, proxies等 + """ + return self.__request(method="post", endpoint=endpoint, **kwargs) + + def put_data(self, endpoint: str, **kwargs) -> Optional[Response]: + """ + 自定义从媒体服务器获取数据 + :param endpoint: 端点 + :param kwargs: 其他请求参数,如headers, cookies, proxies等 + """ + return self.__request(method="put", endpoint=endpoint, **kwargs) + + def __request(self, method: str, endpoint: str, **kwargs) -> Optional[Response]: + """ + 自定义从媒体服务器获取数据 + :param method: HTTP方法,如 get, post, put 等 + :param endpoint: 端点 + :param kwargs: 其他请求参数,如headers, cookies, proxies等 + """ + if not self._session: + return + try: + url = RequestUtils.adapt_request_url(host=self._host, endpoint=endpoint) + kwargs.setdefault("headers", self.__get_request_headers()) + kwargs.setdefault("raise_exception", True) + request_method = getattr(RequestUtils(session=self._session), f"{method}_res", None) + if request_method: + return request_method(url=url, **kwargs) + else: + logger.error(f"方法 {method} 不存在") + return None + except Exception as e: + logger.error(f"连接Plex出错:" + str(e)) + return None + + @staticmethod + def __get_request_headers() -> dict: + """获取请求头""" + return { + "X-Plex-Token": settings.PLEX_TOKEN, + "Accept": "application/json", + "Content-Type": "application/json" + } + + @staticmethod + def __adapt_plex_session() -> Session: + """ + 创建并配置一个针对Plex服务的requests.Session实例 + 这个会话包括特定的头部信息,用于处理所有的Plex请求 + """ + # 设置请求头部,通常包括验证令牌和接受/内容类型头部 + headers = Plex.__get_request_headers() + session = Session() + session.headers = headers + return session