Koha9/PlexPlaylistSync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix frontend entrypoint for Vite build

Browse Source
This commit is contained in:
Koha9
2025-11-29 00:25:45 +09:00
parent e5ba790b44
commit 832dbc11d5
21 changed files with 3817 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/config.json
+5 -2
View File
@@ -2,7 +2,10 @@
"theme": "auto",
"token": "",
"server_url": "",
"server_port": "",
"server_scheme": "",
"server_port": "32400",
"server_scheme": "https",
"library_name": "",
"sync_mode": "merge_local_primary",
"local_path": "playlist",
"path_rules": []
}
app/main.py
+321 -9
View File
@@ -1,19 +1,37 @@
import os
from app.utils.config import server_config
from app.utils.playlist_merge import SyncMode, sync_all_playlists, TEST_PLAYLIST_DIR
from fastapi import FastAPI, Request, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from datetime import datetime
from typing import Sequence
from fastapi import FastAPI, Form, HTTPException, Query, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from app.utils.plex_client import plex_client
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel, Field
from app.utils.config import server_config
from app.utils.local_playlist import load_local_playlist, scan_local_playlists
from app.utils.logger import logger
from app.utils.local_playlist import scan_local_playlists
from app.utils.playlist_merge import SyncMode, TEST_PLAYLIST_DIR, sync_all_playlists
from app.utils.plex_client import plex_client
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
templates = Jinja2Templates(
directory=os.path.join(os.path.dirname(__file__), "templates")
)
FRONTEND_DIST_PATH = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "frontend", "dist")
)
# mount static files
# 这里的路径是相对于 main.py 文件所在的目录
app.mount(
@@ -22,6 +40,13 @@ app.mount(
name="static",
)
if os.path.isdir(os.path.join(FRONTEND_DIST_PATH, "assets")):
app.mount(
"/assets",
StaticFiles(directory=os.path.join(FRONTEND_DIST_PATH, "assets")),
name="frontend-assets",
)
SYNC_MODE_OPTIONS = [
{
@@ -47,6 +72,45 @@ SYNC_MODE_OPTIONS = [
]
class PlaylistItem(BaseModel):
id: str
title: str
trackCount: int = Field(..., ge=0)
lastUpdated: str | None = None
class RegexRule(BaseModel):
pattern: str
replacement: str = ""
class SyncSettingsResponse(BaseModel):
sync_mode: str
path_rules: list[RegexRule]
local_path: str
library_name: str | None = None
server_url: str | None = None
scheme: str | None = None
port: str | None = None
token: str | None = None
class ConnectRequest(BaseModel):
protocol: str = Field("https", pattern="https?", description="HTTP or HTTPS")
address: str
port: str = "32400"
token: str = ""
username: str | None = None
password: str | None = None
timeout: int | None = None
library_name: str | None = None
class ConnectResponse(BaseModel):
token: str
serverInfo: dict
def _get_cloud_playlists() -> tuple[list[dict], str, dict, str, list[str]]:
"""Fetch playlists and connection state from the remote Plex server."""
@@ -110,6 +174,251 @@ def _get_cloud_playlists() -> tuple[list[dict], str, dict, str, list[str]]:
return playlists, status, server_info, selected_library, music_libraries
def _library_dicts(names: Sequence[str]) -> list[dict]:
return [{"id": name, "title": name, "type": "artist"} for name in names]
def _playlist_item(name: str, track_count: int, prefix: str, last_updated: float | None = None) -> PlaylistItem:
updated_value = (
datetime.utcfromtimestamp(last_updated).isoformat() + "Z"
if last_updated
else datetime.utcnow().isoformat() + "Z"
)
return PlaylistItem(
id=f"{prefix}-{name}",
title=name,
trackCount=track_count,
lastUpdated=updated_value,
)
def _scan_local_playlists_with_meta(local_path: str) -> list[PlaylistItem]:
items: list[PlaylistItem] = []
base_path = local_path or server_config.local_path
if not base_path:
return items
absolute_path = os.path.abspath(base_path)
if not os.path.isdir(absolute_path):
logger.warning(f"Playlist path does not exist or is not a directory: {absolute_path}")
return items
for entry in os.scandir(absolute_path):
if not entry.is_file():
continue
if not entry.name.lower().endswith((".m3u", ".m3u8")):
continue
tracks = load_local_playlist(entry.path)
stat_info = entry.stat()
items.append(
_playlist_item(
name=entry.name,
track_count=len(tracks),
prefix="local",
last_updated=stat_info.st_mtime,
)
)
items.sort(key=lambda playlist: playlist.title.lower())
return items
def _get_server_status() -> tuple[dict, str, list[dict]]:
"""Return server connection status and available libraries."""
server_config.load()
if not server_config.url:
return {"isConnected": False}, "unset", []
connection_status = "failed"
libraries: list[dict] = []
server_info = {
"isConnected": False,
"name": "未设置",
"ip": server_config.url,
"port": server_config.port,
"libraryName": server_config.library_name,
}
try:
plex_client.connect(
token=server_config.token,
scheme=server_config.scheme,
url=server_config.url,
port=server_config.port,
)
connection_status = "connected" if plex_client.connected else "failed"
server_info.update(
{
"isConnected": plex_client.connected,
"name": getattr(plex_client.server, "friendlyName", "未命名服务器"),
"ip": server_config.url,
"port": server_config.port,
}
)
if plex_client.connected:
lib_names = plex_client.get_libs_name_list()
libraries = _library_dicts(lib_names)
if lib_names:
selected_library = server_config.library_name or lib_names[0]
if selected_library not in lib_names:
selected_library = lib_names[0]
server_config.set_and_save_config(library_name=selected_library)
server_info["libraryName"] = selected_library
except Exception as exc:
logger.warning(f"Failed to connect to Plex server: {exc}")
return server_info, connection_status, libraries
class SyncModePayload(BaseModel):
mode: SyncMode
class RegexRulePayload(BaseModel):
rules: list[RegexRule]
class LibrarySelection(BaseModel):
library_name: str
class SyncRequest(BaseModel):
mode: SyncMode | None = None
local_path: str | None = None
@app.get("/api/settings", response_model=SyncSettingsResponse)
async def get_settings():
server_config.load()
rules = [
RegexRule(pattern=rule.get("pattern", ""), replacement=rule.get("replacement", ""))
for rule in server_config.path_rules
]
return SyncSettingsResponse(
sync_mode=server_config.sync_mode,
path_rules=rules,
local_path=server_config.local_path,
library_name=server_config.library_name,
server_url=server_config.url,
scheme=server_config.scheme,
port=server_config.port,
token=server_config.token,
)
@app.put("/api/settings/sync-mode")
async def update_sync_mode(payload: SyncModePayload):
server_config.set_and_save_config(sync_mode=payload.mode.value)
return {"sync_mode": payload.mode.value}
@app.get("/api/settings/regex-rules")
async def get_regex_rules():
server_config.load()
return {"rules": server_config.path_rules}
@app.put("/api/settings/regex-rules")
async def update_regex_rules(payload: RegexRulePayload):
server_config.set_and_save_config(path_rules=[rule.model_dump() for rule in payload.rules])
return {"rules": payload.rules}
@app.put("/api/settings/library")
async def update_library(payload: LibrarySelection):
server_config.set_and_save_config(library_name=payload.library_name)
return {"library_name": server_config.library_name}
@app.get("/api/server")
async def api_server_status():
server_info, status, libraries = _get_server_status()
return {"status": status, "serverInfo": server_info, "libraries": libraries}
@app.post("/api/connect", response_model=ConnectResponse)
async def api_connect(payload: ConnectRequest):
try:
_, token = plex_client.connect(
username=payload.username or "",
password=payload.password or "",
token=payload.token or "",
scheme=payload.protocol,
url=payload.address,
port=payload.port,
)
libraries = []
selected_library = payload.library_name or server_config.library_name
if plex_client.connected:
lib_names = plex_client.get_libs_name_list()
libraries = _library_dicts(lib_names)
if lib_names:
if not selected_library or selected_library not in lib_names:
selected_library = lib_names[0]
server_config.set_and_save_config(
token=token,
scheme=payload.protocol,
url=payload.address,
port=payload.port,
library_name=selected_library or "",
)
server_info = {
"isConnected": plex_client.connected,
"name": getattr(plex_client.server, "friendlyName", "未命名服务器") if plex_client.connected else "未命名服务器",
"ip": payload.address,
"port": payload.port,
"libraryName": selected_library or "",
"libraries": libraries,
}
return ConnectResponse(token=token, serverInfo=server_info)
except Exception as exc:
logger.warning(f"Failed to connect via API: {exc}")
raise HTTPException(status_code=400, detail=str(exc))
@app.get("/api/playlists")
async def api_playlists(server: str = Query(..., pattern="(?i)^(local|cloud)$"), local_path: str | None = None):
server_type = server.lower()
if server_type == "local":
resolved_path = local_path or server_config.local_path
server_config.set_and_save_config(local_path=resolved_path)
playlists = _scan_local_playlists_with_meta(resolved_path)
return {"playlists": [item.model_dump() for item in playlists]}
playlists, connection_status, server_info, selected_library, libraries = _get_cloud_playlists()
items = [_playlist_item(item["name"], item.get("track_count", 0), "cloud") for item in playlists]
return {
"playlists": [item.model_dump() for item in items],
"connection_status": connection_status,
"server_info": server_info,
"library": selected_library,
"libraries": _library_dicts(libraries),
}
@app.post("/api/sync")
async def api_sync(payload: SyncRequest):
server_config.load()
try:
sync_mode = payload.mode or SyncMode(server_config.sync_mode)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
local_dir = payload.local_path or server_config.local_path
results = sync_all_playlists(local_dir=local_dir, mode=sync_mode, test_folder=TEST_PLAYLIST_DIR)
merged_count = sum(len(item.merged_paths) for item in results)
conflict_count = sum(len(item.conflicts) for item in results)
deleted_count = sum(1 for item in results if item.action == "deleted")
return {
"mode": sync_mode.value,
"merged_count": merged_count,
"conflict_count": conflict_count,
"delete_count": deleted_count,
"playlist_count": len(results),
"output_dir": TEST_PLAYLIST_DIR,
}
def _build_home_context(
request: Request,
local_path: str,
@@ -140,7 +449,7 @@ def _build_home_context(
"selected_library": selected_library,
"music_libraries": music_libraries,
"sync_modes": SYNC_MODE_OPTIONS,
"selected_mode": selected_mode,
"selected_mode": selected_mode or server_config.sync_mode,
"message": message,
"message_type": message_type,
"sync_result": sync_result,
@@ -151,8 +460,11 @@ def _build_home_context(
# 显示主页
@app.get("/", response_class=HTMLResponse)
async def home(request: Request, local_path: str = "playlist"):
context = _build_home_context(request, local_path)
index_path = os.path.join(FRONTEND_DIST_PATH, "index.html")
if os.path.exists(index_path):
return FileResponse(index_path)
context = _build_home_context(request, local_path or server_config.local_path)
return templates.TemplateResponse("home.html", context)
app/utils/config.py
+20
View File
@@ -2,6 +2,8 @@ import json
import os
from app.utils.logger import logger
DEFAULT_SYNC_MODE = "merge_local_primary"
CONFIG_PATH = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "config.json")
)
@@ -16,6 +18,8 @@ class ServerConfig:
self.scheme = "https"
self.port = "32400"
self.library_name = ""
self.sync_mode = DEFAULT_SYNC_MODE
self.local_path = "playlist"
self.path_rules: list[dict[str, str]] = []
self.load()
@@ -40,6 +44,8 @@ class ServerConfig:
self.scheme = config.get("server_scheme", "https")
self.port = config.get("server_port", "32400")
self.library_name = config.get("library_name", "")
self.sync_mode = config.get("sync_mode", DEFAULT_SYNC_MODE)
self.local_path = config.get("local_path", "playlist")
self.path_rules = config.get("path_rules", []) or []
logger.info(f"Server config loaded: {self.__dict__}")
@@ -51,6 +57,8 @@ class ServerConfig:
"server_scheme": self.scheme,
"server_port": self.port,
"library_name": self.library_name,
"sync_mode": self.sync_mode,
"local_path": self.local_path,
"path_rules": self.path_rules,
}
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
@@ -72,6 +80,12 @@ class ServerConfig:
def set_library(self, library_name: str) -> None:
self.library_name = library_name or ""
def set_sync_mode(self, sync_mode: str) -> None:
self.sync_mode = sync_mode
def set_local_path(self, local_path: str) -> None:
self.local_path = local_path or "playlist"
def set_theme(self, theme: str) -> None:
# check theme is valid
if theme not in ["auto", "dark", "light"]:
@@ -90,6 +104,8 @@ class ServerConfig:
scheme: str = None,
port: str = None,
library_name: str | None = None,
sync_mode: str | None = None,
local_path: str | None = None,
path_rules: list[dict[str, str]] | None = None,
) -> None:
if theme is not None:
@@ -104,6 +120,10 @@ class ServerConfig:
self.set_port(port)
if library_name is not None:
self.set_library(library_name)
if sync_mode is not None:
self.set_sync_mode(sync_mode)
if local_path is not None:
self.set_local_path(local_path)
if path_rules is not None:
self.set_path_rules(path_rules)
self.save()