262 lines
10 KiB
Python
262 lines
10 KiB
Python
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()
|