refactor plex_client,config

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,
+            "scheme": server_config.scheme,
-            "token": token,
+            "token": server_config.token,
-            "server_url": host,
+            "server_url": server_config.url,
-            "port": port,
+            "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,
+            "token": server_config.token,
-            "scheme": scheme,
+            "scheme": server_config.scheme,
-            "server_url": host,
+            "server_url": server_config.url,
-            "port": port,
+            "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({
+        if plex_client.connected:
-            "server_url": url,
+            server_config.set_config(
-            "server_scheme": scheme,
+                token=token_success, scheme=scheme, url=url, port=port
-            "server_port": port,
+            )
-            "token": token_success,
-        })
-        save_config(config)
-        token, scheme, host, port = get_server_settings(config)
         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()
+    return templates.TemplateResponse(
-    theme = config.get("theme", "auto")
+        "playlist.html",
-    return templates.TemplateResponse("playlist.html", {"request": request, "theme": theme, "path": "/playlist"})
+        {"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(...)):
+async def set_playlist(
-    config = load_config()
+    request: Request, address: str = Form(...), interval: str = Form(...)
-    theme = config.get("theme", "auto")
+):
     # demo：返回提交的设置
-    return templates.TemplateResponse("playlist.html", {
+    return templates.TemplateResponse(
-        "request": request,
+        "playlist.html",
-        "message": f"设置成功：地址 {address}，间隔 {interval} 分钟",
+        {
-        "theme": theme,
+            "request": request,
-        "path": "/playlist"
+            "message": f"设置成功：地址 {address}，间隔 {interval} 分钟",
-    })
+            "theme": server_config.theme,
+            "path": "/playlist",
+        },
+    )
+
 
 @app.post("/set-theme")
 async def set_theme(theme: str = Form(...)):
-    config = load_config()
+    server_config.set_config(theme=theme)
-    config["theme"] = theme
+    return RedirectResponse("/login", status_code=303)
-    save_config(config)
-    return RedirectResponse("/login", status_code=303)

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")
+CONFIG_PATH = os.path.abspath(
-
+    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)
 
 
-def get_server_settings(config):
+class ServerConfig:
-    """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"
 
-    url = config.get("server_url", "")
+    def __init__(self):
-    if url:
+        self.theme = "auto"
-        parsed = urlparse(url)
+        self.token = ""
-        if parsed.scheme:
+        self.url = ""
-            scheme = parsed.scheme
+        self.scheme = "https"
-            host = parsed.hostname or parsed.netloc
+        self.port = "32400"
-            if parsed.port:
+        self.load()
-                port = str(parsed.port)
+
-        else:
+    def load(self) -> None:
-            host = url
+        try:
-    print(f"server_scheme: {scheme}, host: {host}, port: {port}")
+            with open(CONFIG_PATH, "r", encoding="utf-8") as f:
-    return token, scheme, host, port
+                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()

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"):
+    def __init__(self) -> None:
-    """Return a connected PlexServer instance and update config with token and server info."""
+        self.server: PlexServer | None = None
-    # url 初始化
+        self.token: str | None = None
-    base_url = build_plex_url(scheme, url, port)
+        self.base_url: str | None = None
-    # account 初始化
+        self.connected = False
-    account = MyPlexAccount(username, password)
-    # token 获取
-    token = account.authenticationToken
 
-    server = PlexServer(base_url, token)
+    def connect(
-    return server, token
+        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"):
+    def _connect_with_pw(self, username: str, password: str, scheme: str, url: str, port: str = "32400"):
-    """Return a connected PlexServer instance using a token."""
+        """Return a connected PlexServer instance and update config with token and server info."""
-    # URL 初始化
+        # url 初始化
-    base_url = build_plex_url(scheme, url, port)
+        self.base_url = build_plex_url(scheme, url, port)
-    
+        # account 初始化
-    server = PlexServer(base_url, token)
+        account = MyPlexAccount(username, password)
-    return server, token
+        # 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()