106 lines
4.1 KiB
Python
106 lines
4.1 KiB
Python
from plexapi.myplex import MyPlexAccount
|
||
from plexapi.server import PlexServer
|
||
from urllib.parse import urlparse
|
||
from app.utils.common import str_is_empty
|
||
from app.utils.logger import logger
|
||
|
||
|
||
def build_plex_url(scheme, url, port="32400"):
|
||
"""Build a full Plex URL from scheme, url, and port."""
|
||
# 如果url不以http://或https://开头，则添加scheme
|
||
full_url = url
|
||
if not full_url.startswith("http://") and not full_url.startswith("https://"):
|
||
full_url = f"{scheme}://{url}"
|
||
parsed = urlparse(full_url)
|
||
|
||
if parsed.scheme in ("http", "https"):
|
||
netloc = parsed.netloc or parsed.path
|
||
if ":" not in netloc and port:
|
||
netloc = f"{netloc}:{port}"
|
||
base_url = f"{parsed.scheme}://{netloc}"
|
||
else:
|
||
base_url = f"http://{url}:{port}"
|
||
return base_url
|
||
|
||
|
||
class PlexClient:
|
||
"""A client for interacting with a Plex server."""
|
||
|
||
def __init__(self) -> None:
|
||
self.server: PlexServer | None = None
|
||
self.token: str | None = None
|
||
self.base_url: str | None = None
|
||
self.connected = False
|
||
|
||
def connect(
|
||
self,
|
||
username: str = "",
|
||
password: str = "",
|
||
token: str = "",
|
||
scheme: str = "https",
|
||
url: str = "",
|
||
port: str = "32400",
|
||
):
|
||
"""Connect to the Plex server using username/password or token."""
|
||
# Connect Server with token if token is provided, otherwise use username/password
|
||
try:
|
||
if not str_is_empty(token):
|
||
self.server, self.token = self._connect_with_token(token, scheme, url, port)
|
||
else:
|
||
self.server, self.token = self._connect_with_pw(
|
||
username, password, scheme, url, port
|
||
)
|
||
# Update the base URL and connection status
|
||
self.base_url = build_plex_url(scheme, url, port)
|
||
self.connected = True
|
||
logger.info(f"Connected to Plex server at {self.base_url} with token: {self.token}")
|
||
return self.server, self.token
|
||
except Exception as e:
|
||
logger.warning(f"Failed to connect to Plex server: {str(e)}")
|
||
self.connected = False
|
||
raise
|
||
|
||
def _connect_with_pw(self, username: str, password: str, scheme: str, url: str, port: str = "32400"):
|
||
"""Return a connected PlexServer instance and update config with token and server info."""
|
||
# url 初始化
|
||
self.base_url = build_plex_url(scheme, url, port)
|
||
# account 初始化
|
||
account = MyPlexAccount(username, password)
|
||
# token 获取
|
||
self.token = account.authenticationToken
|
||
|
||
self.server = PlexServer(self.base_url, self.token)
|
||
logger.debug(f"Connected to Plex server with username: {username}, token: {self.token}")
|
||
return self.server, self.token
|
||
|
||
def _connect_with_token(self, token: str, scheme: str, url: str, port: str = "32400"):
|
||
"""Return a connected PlexServer instance using a token."""
|
||
# URL 初始化
|
||
self.base_url = build_plex_url(scheme, url, port)
|
||
|
||
self.server = PlexServer(self.base_url, token)
|
||
logger.debug(f"Connected to Plex server with token: {token}")
|
||
return self.server, token
|
||
|
||
def get_server(self) -> PlexServer | None:
|
||
"""Return the connected Plex server instance."""
|
||
if not self.connected:
|
||
logger.error("Plex client is not connected.")
|
||
raise RuntimeError("Plex client is not connected.")
|
||
return self.server
|
||
|
||
def get_all_playlist(self) -> list | None:
|
||
"""Return all playlists from the Plex server."""
|
||
if not self.connected or not self.server:
|
||
logger.error("Plex server is not connected.")
|
||
raise ValueError("Plex server is not connected.")
|
||
|
||
try:
|
||
playlists = self.server.playlists()
|
||
logger.info(f"Fetched {len(playlists)} playlists from the server.")
|
||
return playlists
|
||
except Exception as e:
|
||
logger.warning(f"Failed to fetch playlists: {str(e)}")
|
||
raise RuntimeError(f"Failed to fetch playlists: {str(e)}")
|
||
|
||
plex_client = PlexClient() |