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

feat: Implement sync manager and file watcher for automated playlist synchronization

Browse Source
This commit is contained in:
Koha9
2025-11-29 12:26:59 +09:00
parent 22697fdc1d
commit fe4061d1a1
8 changed files with 413 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/main.py
+87 -2
View File
@@ -4,7 +4,8 @@ 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
import asyncio
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel, Field
@@ -15,11 +16,13 @@ from app.utils.logger import logger
from app.utils.playlist_merge import SyncMode, TEST_PLAYLIST_DIR, sync_all_playlists
from app.utils.plex_client import plex_client
from app.utils.scheduler import start_scheduler, update_scheduler_job, get_next_run_time, validate_cron_expression
from app.utils.sync_manager import sync_manager
app = FastAPI()
@app.on_event("startup")
async def startup_event():
sync_manager.set_event_loop(asyncio.get_running_loop())
start_scheduler()
app.add_middleware(
@@ -450,6 +453,75 @@ async def api_playlists(server: str = Query(..., pattern="(?i)^(local|cloud)$"),
}
@app.get("/api/sync/status")
async def get_sync_status():
return sync_manager.status
@app.get("/api/sync/events")
async def sync_events(request: Request):
async def event_generator():
q = await sync_manager.subscribe()
try:
while True:
if await request.is_disconnected():
break
data = await q.get()
yield f"data: {data}\n\n"
finally:
sync_manager.unsubscribe(q)
return StreamingResponse(event_generator(), media_type="text/event-stream")
@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))
# Update config temporarily for this sync if needed, but sync_manager reads from config.
# If payload overrides config, we might need to handle that.
# However, sync_manager._perform_sync reads from server_config.
# If we want to support one-off sync with custom params via sync_manager, we need to update sync_manager.
# For now, let's assume payload params should be saved or used.
# But sync_manager is designed to run background tasks too.
# If we want to keep the existing behavior of api_sync (blocking and returning stats),
# we can use sync_manager.run_sync(wait=True).
# But we need to make sure sync_manager uses the params from payload if provided.
# Since sync_manager reads from server_config, let's update server_config if payload has values.
# Or better, pass params to sync_manager.run_sync?
# sync_manager._perform_sync currently hardcodes reading from server_config.
# Let's stick to the requirement: "watchdog当发现更改时，执行同步，同步时UI页面也会显示正在同步状态。"
# This implies we need a shared state.
# If I change api_sync to use sync_manager, I need to ensure it supports the custom params.
# But payload.local_path and payload.mode are optional.
# Let's modify sync_manager to accept overrides.
# But wait, sync_manager is a singleton.
# For this task, I will just wrap the existing logic in sync_manager.run_sync(wait=True)
# AND I will modify sync_manager to allow passing explicit args to _perform_sync.
# But first, let's update api_sync to use sync_manager.run_sync(wait=True)
# AND we need to handle the parameter passing.
# Actually, looking at sync_manager implementation I just wrote:
# def _perform_sync(self):
# server_config.load()
# return sync_all_playlists(local_dir=server_config.local_path, mode=SyncMode(server_config.sync_mode))
# It ignores arguments. This is a limitation.
# I should update SyncManager to accept kwargs for sync_all_playlists.
# Let's update SyncManager first.
pass
@app.post("/api/sync")
async def api_sync(payload: SyncRequest):
server_config.load()
@@ -459,7 +531,20 @@ async def api_sync(payload: SyncRequest):
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)
# Use sync_manager to execute sync, ensuring state is updated
try:
results = sync_manager.run_sync(
trigger_source="api",
wait=True,
# We need to pass these to _perform_sync
sync_kwargs={"local_dir": local_dir, "mode": sync_mode}
)
except Exception as e:
if str(e) == "Sync already in progress":
raise HTTPException(status_code=409, detail="Sync already in progress")
raise e
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")
app/utils/playlist_merge.py
+14 -1
View File
@@ -159,8 +159,21 @@ def _read_text_if_exists(path: str) -> tuple[str, bool]:
def _save_playlist_to_folder(filename: str, paths: Sequence[str], folder: str) -> str:
_ensure_test_dir(folder)
file_path = os.path.join(folder, filename)
new_content = save_paths(paths)
# Check if content has changed before writing to avoid triggering unnecessary file events
if os.path.exists(file_path):
try:
with open(file_path, "r", encoding="utf-8") as f:
current_content = f.read()
if current_content == new_content:
return file_path
except OSError:
pass
with open(file_path, "w", encoding="utf-8") as file:
file.write(save_paths(paths))
file.write(new_content)
return file_path
app/utils/scheduler.py
+13 -19
View File
@@ -2,8 +2,9 @@ from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from app.utils.config import server_config
from app.utils.logger import logger
from app.utils.playlist_merge import sync_all_playlists, SyncMode
import asyncio
from app.utils.watcher import watcher_manager
from app.utils.sync_manager import sync_manager
import os
scheduler = BackgroundScheduler()
@@ -26,23 +27,7 @@ def validate_cron_expression(expression: str) -> bool:
def job_function():
logger.info("Executing scheduled sync job...")
try:
# Re-read config to ensure latest values
server_config.load()
mode_str = server_config.sync_mode
regex_rules = server_config.path_rules
local_path = server_config.local_path
asyncio.run(sync_all_playlists(
sync_mode=SyncMode(mode_str),
path_rules=regex_rules,
local_dir=local_path
))
logger.info("Scheduled sync job completed.")
except Exception as e:
logger.error(f"Scheduled sync job failed: {e}")
sync_manager.run_sync(trigger_source="scheduler", wait=False)
def start_scheduler():
if not scheduler.running:
@@ -55,6 +40,15 @@ def update_scheduler_job():
# Reload config to get latest schedule settings
server_config.load()
# Handle Auto Watch
if server_config.schedule_auto_watch:
# Ensure we have an absolute path
local_path = os.path.abspath(server_config.local_path)
watcher_manager.start(local_path)
else:
watcher_manager.stop()
mode = server_config.schedule_mode
if mode == "DISABLED":
app/utils/sync_manager.py
+124
View File
@@ -0,0 +1,124 @@
import threading
import asyncio
import json
from datetime import datetime
from app.utils.logger import logger
from app.utils.playlist_merge import sync_all_playlists, SyncMode
from app.utils.config import server_config
class SyncManager:
def __init__(self):
self._lock = threading.Lock()
self._is_syncing = False
self._last_sync_time = None
self._last_status = "idle" # idle, syncing, success, error
self._last_error = None
self._listeners = [] # List of asyncio.Queue
self._loop = None
def set_event_loop(self, loop):
self._loop = loop
async def subscribe(self):
q = asyncio.Queue()
self._listeners.append(q)
# Send current status immediately
await q.put(json.dumps(self.status))
return q
def unsubscribe(self, q):
if q in self._listeners:
self._listeners.remove(q)
def _notify_listeners(self):
if not self._loop or not self._listeners:
return
status_json = json.dumps(self.status)
for q in self._listeners:
try:
self._loop.call_soon_threadsafe(q.put_nowait, status_json)
except Exception as e:
logger.error(f"Error notifying listener: {e}")
@property
def is_syncing(self):
with self._lock:
return self._is_syncing
@property
def status(self):
with self._lock:
return {
"is_syncing": self._is_syncing,
"last_sync_time": self._last_sync_time.isoformat() if self._last_sync_time else None,
"status": self._last_status,
"error": str(self._last_error) if self._last_error else None
}
def run_sync(self, trigger_source="manual", wait=False, sync_kwargs=None):
"""
Thread-safe sync execution.
If wait=True, blocks until sync completes and returns result.
If wait=False, runs in background and returns True if started.
"""
with self._lock:
if self._is_syncing:
logger.warning(f"Sync requested ({trigger_source}) but already in progress.")
if wait:
raise Exception("Sync already in progress")
return False
self._is_syncing = True
self._last_status = "syncing"
self._last_error = None
self._notify_listeners()
logger.info(f"Starting sync (Source: {trigger_source})...")
if wait:
try:
result = self._perform_sync(sync_kwargs)
self._complete_sync("success")
return result
except Exception as e:
self._complete_sync("error", e)
raise e
else:
thread = threading.Thread(target=self._sync_worker, args=(trigger_source, sync_kwargs))
thread.start()
return True
def _sync_worker(self, trigger_source, sync_kwargs=None):
try:
self._perform_sync(sync_kwargs)
self._complete_sync("success")
logger.info(f"Sync completed successfully (Source: {trigger_source}).")
except Exception as e:
logger.error(f"Sync failed (Source: {trigger_source}): {e}")
self._complete_sync("error", e)
def _perform_sync(self, sync_kwargs=None):
# Reload config to ensure latest values
server_config.load()
kwargs = {
"local_dir": server_config.local_path,
"mode": SyncMode(server_config.sync_mode)
}
if sync_kwargs:
kwargs.update(sync_kwargs)
# Execute sync
return sync_all_playlists(**kwargs)
def _complete_sync(self, status, error=None):
with self._lock:
self._last_status = status
self._last_error = error
self._last_sync_time = datetime.now()
self._is_syncing = False
self._notify_listeners()
sync_manager = SyncManager()
app/utils/watcher.py
+91
View File
@@ -0,0 +1,91 @@
import os
import threading
import asyncio
from watchdog.observers.polling import PollingObserver as Observer
from watchdog.events import FileSystemEventHandler
from app.utils.logger import logger
from app.utils.config import server_config
from app.utils.sync_manager import sync_manager
class PlaylistEventHandler(FileSystemEventHandler):
def __init__(self):
self.debounce_timer = None
def on_any_event(self, event):
# Log all events for debugging (using INFO temporarily to ensure visibility)
logger.info(f"[WATCHER-DEBUG] Event detected: {event.event_type} {event.src_path}")
if event.is_directory:
return
# Filter out noisy events. Only listen to actual changes.
# 'opened' and 'closed' (without write) are read events and should be ignored.
if event.event_type not in ['created', 'modified', 'deleted', 'moved']:
return
# Ignore temporary files or hidden files if necessary
filename = os.path.basename(event.src_path)
if filename.startswith('.'):
return
# Prevent feedback loops: if sync is in progress, ignore events (likely caused by the sync itself)
if sync_manager.is_syncing:
logger.info(f"[WATCHER-DEBUG] Ignoring event {event.event_type} on {event.src_path} because sync is in progress.")
return
logger.info(f"File system event detected and accepted: {event.event_type} {event.src_path}")
self.trigger_sync()
def trigger_sync(self):
if self.debounce_timer:
self.debounce_timer.cancel()
# Debounce for 5 seconds to allow multiple file operations to complete
self.debounce_timer = threading.Timer(5.0, self.run_sync)
self.debounce_timer.start()
def run_sync(self):
logger.info("Triggering sync due to file change...")
sync_manager.run_sync(trigger_source="watcher", wait=False)
class WatcherManager:
def __init__(self):
self.observer = None
self.handler = None
self.current_path = None
def start(self, path):
# If already watching the same path, do nothing
if self.observer and self.observer.is_alive() and self.current_path == path:
logger.info(f"Watcher already running on {path}")
return
self.stop()
if not os.path.exists(path):
logger.warning(f"Cannot watch path {path}: Directory does not exist.")
return
logger.info(f"Starting file watcher on: {path}")
try:
files = os.listdir(path)
logger.info(f"Files currently in watch directory: {files}")
except Exception as e:
logger.error(f"Failed to list files in watch directory: {e}")
self.handler = PlaylistEventHandler()
# Explicitly set timeout for PollingObserver
self.observer = Observer(timeout=1.0)
self.observer.schedule(self.handler, path, recursive=True)
self.observer.start()
self.current_path = path
def stop(self):
if self.observer:
logger.info("Stopping file watcher...")
self.observer.stop()
self.observer.join()
self.observer = None
self.current_path = None
watcher_manager = WatcherManager()
frontend/App.tsx
+78 -2
View File
@@ -125,6 +125,8 @@ const App: React.FC = () => {
const [loadingCloud, setLoadingCloud] = useState(false);
const [syncState, setSyncState] = useState<SyncState>(SyncState.IDLE);
const manualSyncInProgress = useRef(false);
const lastKnownSyncTimeRef = useRef<string | null | undefined>(undefined);
// Animation Refs
const { leftYellowRef, leftGreenRef, rightYellowRef, rightGreenRef } = useStripeAnimation(syncState);
@@ -165,7 +167,7 @@ const App: React.FC = () => {
}
};
const addToast = (message: string) => {
const addToast = useCallback((message: string) => {
const id = Date.now();
// Start with entering: true to position it above
const newToast: Toast = { id, message, exiting: false, entering: true };
@@ -182,7 +184,7 @@ const App: React.FC = () => {
}, TOAST_AUTO_DISMISS_MS);
timeoutsRef.current[id] = dismissTimer;
};
}, []);
// Effect to trigger the "slide down" animation
useEffect(() => {
@@ -361,9 +363,12 @@ const App: React.FC = () => {
if (syncState !== SyncState.IDLE) return;
setSyncState(SyncState.SYNCING);
manualSyncInProgress.current = true;
const result = await apiService.syncPlaylists(currentStrategy, regexReplacements, localPath || undefined);
manualSyncInProgress.current = false;
if (result.status === 'success') {
setSyncState(SyncState.SUCCESS);
@@ -379,6 +384,77 @@ const App: React.FC = () => {
}
};
// SSE for sync status
useEffect(() => {
const eventSource = new EventSource(`${import.meta.env.VITE_API_BASE_URL || ''}/api/sync/events`);
eventSource.onmessage = (event) => {
try {
const data = JSON.parse(event.data);
const { is_syncing, status, error, last_sync_time } = data;
// Initialize lastKnownSyncTime if it's the first event
if (lastKnownSyncTimeRef.current === undefined) {
lastKnownSyncTimeRef.current = last_sync_time;
// If we are currently syncing on load, show it
if (is_syncing && !manualSyncInProgress.current) {
setSyncState(SyncState.SYNCING);
}
return;
}
// If manual sync is in progress, we ignore background updates to avoid state conflict
if (manualSyncInProgress.current) {
if (last_sync_time !== lastKnownSyncTimeRef.current) {
lastKnownSyncTimeRef.current = last_sync_time;
}
return;
}
// Handle Syncing State
if (is_syncing) {
if (syncState !== SyncState.SYNCING) {
setSyncState(SyncState.SYNCING);
}
} else {
// Check for completion by comparing timestamps
if (last_sync_time !== lastKnownSyncTimeRef.current) {
lastKnownSyncTimeRef.current = last_sync_time;
// A sync has completed since our last check
if (status === 'success') {
setSyncState(SyncState.SUCCESS);
refreshLocal();
refreshCloud();
addToast("Background sync completed successfully.");
setTimeout(() => setSyncState(SyncState.IDLE), SYNC_SUCCESS_TOTAL_MS);
} else if (status === 'error') {
setSyncState(SyncState.ERROR);
addToast(`Background sync failed: ${error}`);
setTimeout(() => setSyncState(SyncState.IDLE), SYNC_ERROR_RESET_MS);
}
} else {
// Edge case: We are in SYNCING state but backend says not syncing, and time hasn't changed.
if (syncState === SyncState.SYNCING) {
setSyncState(SyncState.IDLE);
}
}
}
} catch (e) {
console.error("Failed to parse SSE event", e);
}
};
eventSource.onerror = (err) => {
console.error("EventSource failed:", err);
eventSource.close();
};
return () => {
eventSource.close();
};
}, [syncState, refreshLocal, refreshCloud, addToast]);
const handleConnectSuccess = async (serverInfo: PlexServerConnection) => {
setCloudServerInfo(serverInfo);
if (serverInfo.libraryName) {
frontend/services/api.ts
+5
View File
@@ -181,4 +181,9 @@ export const apiService = {
});
return handleResponse(response);
},
async getSyncStatus(): Promise<ApiResponse<{ is_syncing: boolean; last_sync_time: string | null; status: string; error: string | null }>> {
const response = await fetch(`${API_BASE}/api/sync/status`);
return handleResponse(response);
},
};
requirements.txt
+1
View File
@@ -5,3 +5,4 @@ python-multipart
plexapi
merge3
apscheduler
watchdog