import threading import asyncio import json import os import hashlib import re import time 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 from app.utils.backup import perform_backup_before_sync from app.utils.local_playlist import load_local_playlist, write_local_playlist, delete_local_playlist from app.utils.plex_client import plex_client 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 # Suppress watcher events briefly after we write/delete local playlists. # This prevents feedback loops where a sync triggers another sync. self._watcher_suppress_until = 0.0 # Tunable defaults (seconds). Keep short to avoid missing real user edits. self._watcher_suppress_after_write_seconds = 2.5 self._watcher_suppress_after_sync_seconds = 2.5 def suppress_watcher_events(self, seconds: float): now = time.monotonic() with self._lock: self._watcher_suppress_until = max(self._watcher_suppress_until, now + float(seconds)) @property def is_watcher_suppressed(self) -> bool: with self._lock: return time.monotonic() < self._watcher_suppress_until 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 # Preemptively suppress watcher in case the poller notices changes right after. self._watcher_suppress_until = max( self._watcher_suppress_until, time.monotonic() + self._watcher_suppress_after_sync_seconds, ) 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) # Perform backup before sync if enabled local_dir = kwargs.get("local_dir", server_config.local_path) perform_backup_before_sync(local_dir, server_config.library_name) # Execute sync results = sync_all_playlists(**kwargs) # Apply results (write to local and remote) self._apply_sync_results(results) return results def _apply_sync_results(self, results): logger.info("Applying sync results to local and remote...") for result in results: playlist_name = result.name action = result.action output_dir = result.output_dir try: if action == "synced": # 1. Write Local local_result_path = os.path.join(output_dir, "outputs", "local_result.m3u8") if os.path.exists(local_result_path): tracks = load_local_playlist(local_result_path) dest_path = self._safe_local_playlist_path(server_config.local_path, playlist_name, ".m3u8") # Ensure directory exists os.makedirs(os.path.dirname(dest_path), exist_ok=True) self.suppress_watcher_events(self._watcher_suppress_after_write_seconds) write_local_playlist(dest_path, tracks) # 2. Write Remote (Plex) remote_result_path = os.path.join(output_dir, "outputs", "remote_result.m3u8") if os.path.exists(remote_result_path): tracks = load_local_playlist(remote_result_path) if server_config.library_name: items = plex_client.get_items_by_paths(server_config.library_name, tracks) plex_client.update_playlist(playlist_name, items) else: logger.warning("Library name not configured, skipping Plex update.") elif action == "deleted": # Delete Local dest_path = self._safe_local_playlist_path(server_config.local_path, playlist_name, ".m3u8") self.suppress_watcher_events(self._watcher_suppress_after_write_seconds) delete_local_playlist(dest_path) # Also check for .m3u dest_path_m3u = self._safe_local_playlist_path(server_config.local_path, playlist_name, ".m3u") self.suppress_watcher_events(self._watcher_suppress_after_write_seconds) delete_local_playlist(dest_path_m3u) # Delete Remote plex_client.delete_playlist(playlist_name) except Exception as e: logger.error(f"Error applying sync result for playlist {playlist_name}: {e}") @staticmethod def _safe_local_playlist_path(local_dir: str, playlist_name: str, extension: str) -> str: base_dir = os.path.abspath(local_dir or "") if not base_dir: raise ValueError("Local playlist directory is not configured") original = (playlist_name or "").strip() # Drop any path components. name = os.path.basename(original) # Remove control chars. name = re.sub(r"[\x00-\x1f\x7f]", "_", name) # Replace path separators and Windows-invalid characters. invalid = set('<>:"/\\|?*') cleaned = "".join(("_" if ch in invalid else ch) for ch in name).strip().strip(". ") windows_reserved = { "CON", "PRN", "AUX", "NUL", *(f"COM{i}" for i in range(1, 10)), *(f"LPT{i}" for i in range(1, 10)), } needs_hash = False if not cleaned: cleaned = "playlist" needs_hash = True if cleaned.upper() in windows_reserved: needs_hash = True if cleaned != original: needs_hash = True cleaned = cleaned[:160].rstrip().strip(". ") if not cleaned: cleaned = "playlist" needs_hash = True if needs_hash: digest = hashlib.sha1(original.encode("utf-8", errors="ignore")).hexdigest()[:8] cleaned = f"{cleaned}__{digest}" filename = f"{cleaned}{extension}" candidate = os.path.abspath(os.path.join(base_dir, filename)) # Ensure the final path stays within base_dir. if os.path.commonpath([base_dir, candidate]) != base_dir: raise ValueError("Refusing to write outside local playlist directory") return candidate def _complete_sync(self, status, error=None): now = time.monotonic() with self._lock: # Keep watcher suppression a bit after sync finishes, since PollingObserver # may detect file changes on the next polling tick. self._watcher_suppress_until = max( self._watcher_suppress_until, now + self._watcher_suppress_after_sync_seconds, ) self._last_status = status self._last_error = error self._last_sync_time = datetime.now() self._is_syncing = False self._notify_listeners() sync_manager = SyncManager()