Add docker compose support

cache plex tracks.

.dockerignore

@@ -0,0 +1,15 @@
+__pycache__/
+*.py[cod]
+*.swp
+*.log
+.env
+.venv
+venv
+.git
+.gitignore
+.vscode
+**/.DS_Store
+*.sqlite3
+*.db
+.idea
+node_modules

Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt ./
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app ./app
+
+EXPOSE 8080
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

README.md

@@ -18,16 +18,14 @@ PlexPlaylistSync 是一个用于同步 Plex 播放列表和本地 `.m3u`/`.m3u8`
 默认情况下 Plex 服务器使用 `32400` 端口，可在未修改服务器端口时直接使用该默认值。
 登录页面提供选择 `http` 或 `https` 的下拉框，服务器地址输入框只需填写域名或 IP，默认值会从 `config.json` 读取。
 
-## 安装
+## 开发环境快速启动（Docker Compose）
 
-首先安装依赖：
+项目已内置 Docker 化配置，开发时只需执行一次构建即可运行：
 
 ```bash
-pip install -r requirements.txt
+docker compose up --build
 ```
 
-然后启动服务：
+- 默认会以 `--reload` 模式启动，监听本地 8080 端口，可在浏览器访问 `http://localhost:8080`。
-
+- 通过 `./app/config.json` 保存的 Plex 配置信息会在主机和容器间共享，便于调试时保留登录 token 等数据。
-```bash
+- 如需自定义端口或其他参数，可在 `docker-compose.yml` 中调整。
-uvicorn app.main:app --port 8080
-```

app/utils/local_playlist.py

@@ -0,0 +1,34 @@
+from typing import List
+from app.utils.logger import logger
+
+def load_local_playlist(playlist_path: str) -> List[str]:
+    """
+    Load a local playlist from a m3u or m3u8 file.
+    Skip # comments and empty lines.
+
+    Args:
+        playlist_path (str): The path to the playlist file.
+
+    Returns:
+        list: A list of songs in the playlist.
+    """
+    try:
+        result = []
+        with open(playlist_path, 'r', encoding="utf-8") as file:
+            for line in file:
+                line = line.strip()
+                if not line:
+                    # Skip empty lines
+                    continue
+                if line.startswith('#'):
+                    # Skip comments
+                    continue
+                result.append(line)
+        logger.info(f"Loaded {len(result)} songs from the playlist: {playlist_path}")
+        return result
+    except FileNotFoundError:
+        logger.error(f"Error: The file {playlist_path} does not exist.")
+        return []
+    except Exception as e:
+        logger.error(f"An error occurred while loading the playlist: {e}")
+        return []

app/utils/plex_client.py

@@ -1,9 +1,19 @@
+import os
+import pickle
 from plexapi.myplex import MyPlexAccount
 from plexapi.server import PlexServer
+from plexapi.library import MusicSection
 from urllib.parse import urlparse
 from app.utils.common import str_is_empty
 from app.utils.logger import logger
 
+MUSIC_LIBRARY_TYPE = "artist"
+UNMATCHED_TRACK_RATING_KEY = "unmatched_track"
+CACHE_DIR_ID_LENGTH = 6
+CACHE_TRACKS_BASE_DIR = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), "..", "cache")
+)
+
 
 def build_plex_url(scheme, url, port="32400"):
     """Build a full Plex URL from scheme, url, and port."""
@@ -40,12 +50,27 @@ class PlexClient:
         scheme: str = "https",
         url: str = "",
         port: str = "32400",
-    ):
+    ) -> tuple[PlexServer, str]:
-        """Connect to the Plex server using username/password or token."""
+        """Connect to the Plex server using username/password or token.
+
+        Parameters:
+            username (str): Plex username.
+            password (str): Plex password.
+            token (str): Plex authentication token.
+            scheme (str): URL scheme (http or https).
+            url (str): Plex server URL.
+            port (str): Plex server port, default is 32400.
+
+        Returns:
+            PlexServer: A connected PlexServer instance.
+            str: The authentication token used for the connection.
+        """
         # 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)
+                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
@@ -53,14 +78,18 @@ class PlexClient:
             # 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}")
+            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"):
+    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)
@@ -70,10 +99,14 @@ class PlexClient:
         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}")
+        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"):
+    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)
@@ -82,25 +115,185 @@ class PlexClient:
         logger.debug(f"Connected to Plex server with token: {token}")
         return self.server, token
 
-    def get_server(self) -> PlexServer | None:
+    def _connect_check(self):
-        """Return the connected Plex server instance."""
+        """Check if the Plex server is connected."""
-        if not self.connected:
+        if not self.connected or not self.server:
             logger.error("Plex client is not connected.")
             raise RuntimeError("Plex client is not connected.")
+
+    def get_server(self) -> PlexServer | None:
+        """Return the connected Plex server instance.
+        returns:
+            PlexServer: A connected PlexServer instance.
+        """
+        self._connect_check()
         return self.server
 
-    def get_all_playlist(self) -> list | None:
+    def get_libs_name_list(self) -> list[str]:
-        """Return all playlists from the Plex server."""
+        """Return a list of library names from the Plex server.
-        if not self.connected or not self.server:
+        returns:
-            logger.error("Plex server is not connected.")
+            list[str]: A list of all library names.
-            raise ValueError("Plex server is not connected.")
+        """
-
+        self._connect_check()
         try:
-            playlists = self.server.playlists()
+            libraries = self.server.library.sections()
+            lib_names = [lib.title for lib in libraries]
+            logger.info(f"Fetched {len(lib_names)} library names from the server.")
+            return lib_names
+        except Exception as e:
+            logger.warning(f"Failed to fetch library names: {str(e)}")
+            raise RuntimeError(f"Failed to fetch library names: {str(e)}")
+
+    def get_lib(self, library_name: str) -> MusicSection | None:
+        """Return a specific library from the Plex server.
+        Args:
+            library_name (str): Name of the library to fetch.
+        returns:
+            MusicSection: A MusicSection representing the specified library.
+        """
+        self._connect_check()
+        try:
+            lib = self.server.library.section(library_name)
+            if not lib:
+                logger.error(f"Library '{library_name}' not found.")
+                raise ValueError(f"Library '{library_name}' not found.")
+            elif lib.type != MUSIC_LIBRARY_TYPE:
+                logger.error(
+                    f"Library '{library_name}' is not a music library. Only music libraries are supported."
+                )
+                raise ValueError(
+                    f"Library '{library_name}' is not a music library. Only music libraries are supported."
+                )
+            logger.debug(f"Fetched library '{library_name}' from the server.")
+            return lib
+        except Exception as e:
+            logger.warning(f"Failed to fetch library '{library_name}': {str(e)}")
+            raise RuntimeError(f"Failed to fetch library '{library_name}': {str(e)}")
+
+    def get_lib_playlists(self, library_name: str) -> list | None:
+        """Return all playlists from the Plex server.
+        Args:
+            library_name (str): Name of the library to fetch playlists from.
+        returns:
+            list: A list of all playlists in the specified library.
+        """
+        self._connect_check()
+        lib = self.get_lib(library_name)
+        # Fetch playlists from the library
+        try:
+            playlists = lib.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)}")
 
+    def get_lib_tracks(self, library_name: str) -> list:
+        """Cache all tracks from the Plex server.
+        Args:
+            library_name (str): Name of the library to fetch tracks from.
+        returns:
+            list: A list of all tracks in the specified library.
+        """
+        self._connect_check()
+        lib = self.get_lib(library_name)
+        try:
+            tracks = lib.all(libtype="track")
+            logger.info(f"Fetched {len(tracks)} tracks from library '{library_name}'.")
+            return tracks
+        except Exception as e:
+            logger.warning(
+                f"Failed to fetch tracks from library '{library_name}': {str(e)}"
+            )
+            raise RuntimeError(
+                f"Failed to fetch tracks from library '{library_name}': {str(e)}"
+            )
+
+    def cache_lib_tracks(self, library_name: str) -> dict:
+        """Cache all tracks from the Plex server.
+            All tracks are cached as a mapping of file paths to rating keys.
+            like: {file_path: ratingKey, ...}
+        Args:
+            library_name (str): Name of the library to cache tracks from.
+        Returns:
+            dict: A mapping of file paths to rating keys.
+        """
+        try:
+            all_tracks = self.get_lib_tracks(library_name)
+            dir_2_ratingKey = {}
+            for track in all_tracks:
+                dir_2_ratingKey[track.media[0].parts[0].file] = track.ratingKey
+            # only use top CACHE_DIR_ID_LENGTH characters of server.machineIdentifier and library key
+            server_id = self.server.machineIdentifier[:CACHE_DIR_ID_LENGTH]
+            lib_id = self.get_lib(library_name).uuid[:CACHE_DIR_ID_LENGTH]
+            # cache directory path
+            cache_dir = os.path.join(CACHE_TRACKS_BASE_DIR, server_id)
+            cache_name = f"{lib_id}.pkl"
+            # Ensure the cache directory exists
+            os.makedirs(cache_dir, exist_ok=True)
+            with open(os.path.join(cache_dir, cache_name), "wb") as f:
+                pickle.dump(dir_2_ratingKey, f, protocol=pickle.HIGHEST_PROTOCOL)
+            logger.info(
+                f"Cached {len(dir_2_ratingKey)} tracks from library '{library_name}' to {cache_dir}."
+            )
+            return dir_2_ratingKey
+        except Exception as e:
+            logger.warning(f"Failed to cache tracks from library '{library_name}': {str(e)}")
+            raise RuntimeError(f"Failed to cache tracks from library '{library_name}': {str(e)}")
+
+    def load_cached_tracks(self, library_name: str) -> dict:
+        """Load cached tracks from the local cache file.
+            Cached tracks will be loaded as a mapping of file paths to rating keys.
+            like: {file_path: ratingKey, ...}
+        Args:
+            library_name (str): Name of the library to load cached tracks from.
+        Returns:
+            dict: A mapping of file paths to rating keys.
+        """
+        self._connect_check()
+        server_id = self.server.machineIdentifier[:CACHE_DIR_ID_LENGTH]
+        lib_id = self.get_lib(library_name).uuid[:CACHE_DIR_ID_LENGTH]
+        # check if cache directory exists
+        cache_dir = os.path.join(CACHE_TRACKS_BASE_DIR, server_id)
+        if not os.path.exists(cache_dir):
+            logger.warning(f"Cache directory {cache_dir} does not exist.")
+            raise FileNotFoundError(f"Cache directory {cache_dir} does not exist. Server may not have cached tracks.")
+        # check if cache file exists
+        cache_name = f"{lib_id}.pkl"
+        cache_path = os.path.join(cache_dir, cache_name)
+        if not os.path.exists(cache_path):
+            logger.warning(f"Cache file {cache_path} does not exist.")
+            raise FileNotFoundError(f"Cache file {cache_path} does not exist. Library  {library_name} may not have cached tracks.")
+        # load cache file
+        with open(cache_path, "rb") as f:
+            return pickle.load(f)
+
+    def match_tracks(self, library_name: str, local_tracks: list) -> dict:
+        """Match local tracks with Plex server tracks ratingKey.
+        Will return a mapping of local file paths to Plex rating keys.
+        like: {local_file_path: plex_ratingKey, ...}
+        unmatched tracks will be assigned a special rating key.
+        like: {local_file_path: 'unmatched_track'}
+
+        Args:
+            library_name (str): Name of the library to match tracks from.
+            tracks (list): List of local tracks to match.
+        Returns:
+            bidict: A bidirectional mapping of local file paths to Plex rating keys.
+        """
+        self._connect_check()
+        cached_tracks = self.load_cached_tracks(library_name)
+        local_2_plex = {}
+        matched_count = 0
+        for track in local_tracks:
+            if track in cached_tracks:
+                local_2_plex[track] = cached_tracks[track]
+                matched_count += 1
+            else:
+                local_2_plex[track] = UNMATCHED_TRACK_RATING_KEY
+        logger.info(
+            f"Matched {matched_count}/{len(local_tracks)} local tracks with Plex server tracks in library '{library_name}'."
+        )
+        return local_2_plex
+
 plex_client = PlexClient()

docker-compose.yml

@@ -0,0 +1,14 @@
+version: "3.9"
+
+services:
+  plex-playlist-sync:
+    build: .
+    command: uvicorn app.main:app --host 0.0.0.0 --port 8080 --reload
+    ports:
+      - "8080:8080"
+    volumes:
+      - ./:/app
+    environment:
+      - PYTHONUNBUFFERED=1
+      - PYTHONDONTWRITEBYTECODE=1
+    restart: unless-stopped