feat：Plex 添加公共请求方法

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