diff --git a/app/main.py b/app/main.py index a80f7ee..f800ce5 100644 --- a/app/main.py +++ b/app/main.py @@ -1,57 +1,59 @@ import os -from app.utils.config import load_config, save_config, get_server_settings +from app.utils.config import server_config from fastapi import FastAPI, Request, Form from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from fastapi.staticfiles import StaticFiles -from app.utils.plex_client import connect_plex +from app.utils.plex_client import plex_client app = FastAPI() -templates = Jinja2Templates(directory=os.path.join(os.path.dirname(__file__), "templates")) +templates = Jinja2Templates( + directory=os.path.join(os.path.dirname(__file__), "templates") +) # mount static files # 这里的路径是相对于 main.py 文件所在的目录 -app.mount("/static", StaticFiles(directory=os.path.join(os.path.dirname(__file__), "static")), name="static") +app.mount( + "/static", + StaticFiles(directory=os.path.join(os.path.dirname(__file__), "static")), + name="static", +) # 显示主页 @app.get("/", response_class=HTMLResponse) async def home(request: Request): - config = load_config() - theme = config.get("theme", "auto") - token,scheme, host, port = get_server_settings(config) return templates.TemplateResponse( "login.html", { "request": request, - "theme": theme, + "theme": server_config.theme, "path": "/login", - "scheme": scheme, - "token": token, - "server_url": host, - "port": port, + "scheme": server_config.scheme, + "token": server_config.token, + "server_url": server_config.url, + "port": server_config.port, }, ) + # 登录页面和处理 @app.get("/login", response_class=HTMLResponse) async def login_page(request: Request): - config = load_config() - theme = config.get("theme", "auto") - token, scheme, host, port = get_server_settings(config) return templates.TemplateResponse( "login.html", { "request": request, - "theme": theme, + "theme": server_config.theme, "path": "/login", - "token": token, - "scheme": scheme, - "server_url": host, - "port": port, + "token": server_config.token, + "scheme": server_config.scheme, + "server_url": server_config.url, + "port": server_config.port, }, ) + @app.post("/login", response_class=HTMLResponse) async def login( request: Request, @@ -62,32 +64,33 @@ async def login( url: str = Form(...), port: str = Form("32400"), ): - config = load_config() - theme = config.get("theme", "auto") # 尝试连接到 Plex 服务器 try: # 优先使用 token 连接,如果 token 为空则使用用户名和密码连接 - _, token_success = connect_plex(user, pw, token, scheme, url, port) + _, token_success = plex_client.connect( + username=user, + password=pw, + token=token, + scheme=scheme, + url=url, + port=port, + ) # 成功连接后保存配置到配置文件 - config.update({ - "server_url": url, - "server_scheme": scheme, - "server_port": port, - "token": token_success, - }) - save_config(config) - token, scheme, host, port = get_server_settings(config) + if plex_client.connected: + server_config.set_config( + token=token_success, scheme=scheme, url=url, port=port + ) return templates.TemplateResponse( "login.html", { "request": request, "message": "连接成功", "success": True, - "theme": theme, + "theme": server_config.theme, "path": "/login", "token": token, "scheme": scheme, - "server_url": host, + "server_url": server_config.url, "port": port, }, ) @@ -98,7 +101,7 @@ async def login( "request": request, "message": f"连接失败:{str(e)}", "success": False, - "theme": theme, + "theme": server_config.theme, "path": "/login", "scheme": scheme, "server_url": url, @@ -106,27 +109,32 @@ async def login( }, ) + @app.get("/playlist", response_class=HTMLResponse) async def get_playlist(request: Request): - config = load_config() - theme = config.get("theme", "auto") - return templates.TemplateResponse("playlist.html", {"request": request, "theme": theme, "path": "/playlist"}) + return templates.TemplateResponse( + "playlist.html", + {"request": request, "theme": server_config.theme, "path": "/playlist"}, + ) + @app.post("/playlist", response_class=HTMLResponse) -async def set_playlist(request: Request, address: str = Form(...), interval: str = Form(...)): - config = load_config() - theme = config.get("theme", "auto") +async def set_playlist( + request: Request, address: str = Form(...), interval: str = Form(...) +): # demo:返回提交的设置 - return templates.TemplateResponse("playlist.html", { - "request": request, - "message": f"设置成功:地址 {address},间隔 {interval} 分钟", - "theme": theme, - "path": "/playlist" - }) + return templates.TemplateResponse( + "playlist.html", + { + "request": request, + "message": f"设置成功:地址 {address},间隔 {interval} 分钟", + "theme": server_config.theme, + "path": "/playlist", + }, + ) + @app.post("/set-theme") async def set_theme(theme: str = Form(...)): - config = load_config() - config["theme"] = theme - save_config(config) - return RedirectResponse("/login", status_code=303) \ No newline at end of file + server_config.set_config(theme=theme) + return RedirectResponse("/login", status_code=303) diff --git a/app/utils/config.py b/app/utils/config.py index 5c635cb..72046ab 100644 --- a/app/utils/config.py +++ b/app/utils/config.py @@ -1,36 +1,88 @@ import json import os -from urllib.parse import urlparse -CONFIG_PATH = os.path.join(os.path.dirname(__file__), "..", "config.json") - -def load_config(): - if not os.path.exists(CONFIG_PATH): - return {} - with open(CONFIG_PATH, "r", encoding="utf-8") as f: - return json.load(f) - -def save_config(new_config): - with open(CONFIG_PATH, "w", encoding="utf-8") as f: - json.dump(new_config, f, indent=4, ensure_ascii=False) +CONFIG_PATH = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "config.json") +) -def get_server_settings(config): - """Return (scheme, host, port) using defaults when not configured.""" - scheme = config.get("server_scheme", "https") - token = config.get("token", "") or "" - host = "" - port = config.get("server_port", "32400") or "32400" +class ServerConfig: - url = config.get("server_url", "") - if url: - parsed = urlparse(url) - if parsed.scheme: - scheme = parsed.scheme - host = parsed.hostname or parsed.netloc - if parsed.port: - port = str(parsed.port) - else: - host = url - print(f"server_scheme: {scheme}, host: {host}, port: {port}") - return token, scheme, host, port + def __init__(self): + self.theme = "auto" + self.token = "" + self.url = "" + self.scheme = "https" + self.port = "32400" + self.load() + + def load(self) -> None: + try: + with open(CONFIG_PATH, "r", encoding="utf-8") as f: + config = json.load(f) + + except FileNotFoundError: + # 如果配置文件不存在,使用默认值 + self.save() + return + except json.JSONDecodeError: + # 如果配置文件格式错误,使用默认值 + self.save() + return + self.theme = config.get("theme", "auto") + self.token = config.get("token", "") + self.url = config.get("server_url", "") + self.scheme = config.get("server_scheme", "https") + self.port = config.get("server_port", "32400") + + def save(self): + config = { + "theme": self.theme, + "token": self.token, + "server_url": self.url, + "server_scheme": self.scheme, + "server_port": self.port, + } + with open(CONFIG_PATH, "w", encoding="utf-8") as f: + json.dump(config, f, indent=4, ensure_ascii=False) + + def set_url(self, url: str) -> None: + self.url = url + + def set_scheme(self, scheme: str) -> None: + self.scheme = scheme + + def set_port(self, port: str) -> None: + self.port = port + + def set_token(self, token: str) -> None: + self.token = token + + def set_theme(self, theme: str) -> None: + # check theme is valid + if theme not in ["auto", "dark", "light"]: + raise ValueError("Invalid theme. Must be 'auto', 'dark', or 'light'.") + self.theme = theme + + def set_config( + self, + theme: str = None, + token: str = None, + url: str = None, + scheme: str = None, + port: str = None, + ) -> None: + if theme is not None: + self.set_theme(theme) + if token is not None: + self.set_token(token) + if url is not None: + self.set_url(url) + if scheme is not None: + self.set_scheme(scheme) + if port is not None: + self.set_port(port) + self.save() + + +server_config = ServerConfig() diff --git a/app/utils/plex_client.py b/app/utils/plex_client.py index da05dd2..b1cadf6 100644 --- a/app/utils/plex_client.py +++ b/app/utils/plex_client.py @@ -3,6 +3,7 @@ from plexapi.server import PlexServer from urllib.parse import urlparse from app.utils.common import str_is_empty + def build_plex_url(scheme, url, port="32400"): """Build a full Plex URL from scheme, url, and port.""" # 如果url不以http://或https://开头,则添加scheme @@ -20,31 +21,77 @@ def build_plex_url(scheme, url, port="32400"): base_url = f"http://{url}:{port}" return base_url -def connect_plex(username, password, token, scheme, url, port="32400"): - """Return a connected PlexServer instance and update config with token and server info.""" - # 如果token存在且不为空,则使用token连接 - if not str_is_empty(token): - return connect_plex_with_token(token, scheme, url, port) - else: - return connect_plex_with_pw(username, password, scheme, url, port) +class PlexClient: + """A client for interacting with a Plex server.""" -def connect_plex_with_pw(username, password, scheme, url, port="32400"): - """Return a connected PlexServer instance and update config with token and server info.""" - # url 初始化 - base_url = build_plex_url(scheme, url, port) - # account 初始化 - account = MyPlexAccount(username, password) - # token 获取 - token = account.authenticationToken + def __init__(self) -> None: + self.server: PlexServer | None = None + self.token: str | None = None + self.base_url: str | None = None + self.connected = False - server = PlexServer(base_url, token) - return server, token + 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 + return self.server, self.token + except Exception as e: + self.connected = False + raise -def connect_plex_with_token(token, scheme, url, port="32400"): - """Return a connected PlexServer instance using a token.""" - # URL 初始化 - base_url = build_plex_url(scheme, url, port) - - server = PlexServer(base_url, token) - return server, token \ No newline at end of file + 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) + 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) + return self.server, self.token + + def get_server(self) -> PlexServer | None: + """Return the connected Plex server instance.""" + if not self.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: + raise ValueError("Plex server is not connected.") + + try: + playlists = self.server.playlists() + return playlists + except Exception as e: + raise RuntimeError(f"Failed to fetch playlists: {str(e)}") + +plex_client = PlexClient() \ No newline at end of file