diff --git a/buildPy2exe.py b/buildPy2exe.py index 048fcd8c..69cef6df 100755 --- a/buildPy2exe.py +++ b/buildPy2exe.py @@ -750,7 +750,7 @@ def run(self): options={ 'py2exe': { 'dist_dir': OUT_DIR, - 'packages': '{}, cffi, OpenSSL, certifi'.format(QT_PACKAGES), + 'packages': '{}, cffi, OpenSSL, certifi, watchdog'.format(QT_PACKAGES), 'includes': 'twisted, sys, encodings, datetime, os, time, math, urllib, ast, unicodedata, _ssl, win32pipe, win32file, sqlite3', 'excludes': 'venv, doctest, pdb, unittest, win32clipboard, win32pdh, win32security, win32trace, win32ui, winxpgui, win32process, tcl, tkinter', 'dll_excludes': 'msvcr71.dll, MSVCP90.dll, POWRPROF.dll', diff --git a/requirements.txt b/requirements.txt index 0d61f2fd..01b38a3a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,7 @@ certifi>=2018.11.29 pem>=21.2.0 twisted[tls]>=16.4.0 +watchdog>=2.1.0,<4.0.0 appnope>=0.1.0; sys_platform == 'darwin' pypiwin32>=223; sys_platform == 'win32' zope.interface>=4.4.0; sys_platform == 'win32' diff --git a/syncplay/client.py b/syncplay/client.py index 379e4b86..dfe5bc59 100755 --- a/syncplay/client.py +++ b/syncplay/client.py @@ -42,8 +42,9 @@ PRIVACY_HIDDENFILENAME from syncplay.messages import getMissingStrings, getMessage, isNoOSDMessage from syncplay.protocols import SyncClientProtocol -from syncplay.watched import WatchedManager -from syncplay.utils import isMacOS +from syncplay.watched import WatchedManager, getRelatedEpisodeDirectories +from syncplay.filemonitor import FileMonitor, isWindowsNetworkPath +from syncplay.utils import isMacOS, isURL class SyncClientFactory(ClientFactory): def __init__(self, client, retry=constants.RECONNECT_RETRIES): self._client = client @@ -2497,16 +2498,48 @@ def __init__(self, client): self.filenameWatchlist = [] self.currentDirectory = None self.mediaDirectories = client.getConfig().get('mediaSearchDirectories') - self.lock = threading.Lock() self.folderSearchEnabled = True self.directorySearchError = None self.newInfo = False - self.currentlyUpdating = False self.newWatchlist = [] - self.fileSwitchTimer = task.LoopingCall(self.updateInfo) - self.fileSwitchTimer.start(constants.FOLDER_SEARCH_DOUBLE_CHECK_INTERVAL, True) self.mediaDirectoriesNotFound = [] + # Scan concurrency: only one full/targeted worker in flight; further + # requests while it runs collapse into a single pending reconciliation. + self._scanInFlight = False + self._pendingReconciliation = False + self.currentlyUpdating = False # external-compat mirror of _scanInFlight + + # Operational degraded state (cleared after N consecutive clean full + # scans) is kept separate from the once-per-configuration user warning. + self._degraded = False + self._consecutiveSuccessfulScans = 0 + self._slowWarningShown = False + self._missingDirectoryNotified = False + + # Coalescing of the downstream "new local file info" notification and of + # the adaptive reconciliation timer. + self._infoNotifyCall = None + self._reconciliationCall = None + + # Filesystem notifications (playlist-blind helper). Polling/reconciliation + # remains the correctness authority; notifications only accelerate it. + self._fileMonitor = FileMonitor(self._onFileMonitorEvent, self._debug) + self._priorityDirectories = [] + + reactor.addSystemEventTrigger("before", "shutdown", self._shutdown) + self._fileMonitor.setMediaDirectories(self.mediaDirectories or []) + self.updateInfo() + self._scheduleReconciliation() + + # -- small helpers ---------------------------------------------------- + + def _debug(self, message): + try: + self._client.ui.showDebugMessage(message) + except Exception: + pass + def setClient(self, newClient): self._client = newClient @@ -2520,6 +2553,12 @@ def changeMediaDirectories(self, mediaDirs): self._client.ui.showMessage(getMessage("media-directory-list-updated-notification")) self.mediaDirectoriesNotFound = [] self.folderSearchEnabled = True + # Reaffirming media directories resets the once-per-configuration warning + # latch and clears any degraded state. + self._slowWarningShown = False + self._missingDirectoryNotified = False + self._degraded = False + self._consecutiveSuccessfulScans = 0 self.setMediaDirectories(mediaDirs) if mediaDirs == "": self._client.ui.showErrorMessage(getMessage("no-media-directories-error")) @@ -2529,8 +2568,12 @@ def changeMediaDirectories(self, mediaDirs): def setMediaDirectories(self, mediaDirs): self.mediaDirectories = mediaDirs + self._fileMonitor.setMediaDirectories(mediaDirs or []) self.updateInfo() + def setFilenameWatchlist(self, unfoundFilenames): + self.filenameWatchlist = unfoundFilenames + def checkForFileSwitchUpdate(self): if self.newInfo: self.newInfo = False @@ -2540,67 +2583,442 @@ def checkForFileSwitchUpdate(self): self.directorySearchError = None self._client.playlist.doubleCheckForWatchedPreviousFile() + def infoUpdated(self): + self._client.fileSwitchFoundFiles() + + # -- authoritative full reconciliation scan --------------------------- + def updateInfo(self): - if not self.currentlyUpdating and self.mediaDirectories: - threads.deferToThread(self._updateInfoThread).addCallback(lambda x: self.checkForFileSwitchUpdate()) + # Request an authoritative full reconciliation scan. Single-flight: if a + # scan is already running, record one pending reconciliation instead of + # dispatching a second worker. + if not self.mediaDirectories: + return + if self._scanInFlight: + self._pendingReconciliation = True + return + self._startFullReconciliation() + + def _startFullReconciliation(self): + self._scanInFlight = True + self.currentlyUpdating = True + deferred = threads.deferToThread(self._fullReconciliationWorker) + deferred.addCallback(self._commitFullReconciliation) + deferred.addErrback(self._scanErrback) + + def _fullReconciliationWorker(self): + # Worker thread: read the filesystem and return a complete result. It must + # not mutate mediaFilesCache; the reactor thread commits the result. + dirsToSearch = self.mediaDirectories + result = {"cache": {}, "fileCount": 0, "timedOut": False, "firstFileTimedOut": False, + "missingDirectory": None, "timedOutDirectory": None} + if not dirsToSearch: + return result + randomFilename = "RandomFile" + str(random.randrange(10000, 99999)) + ".txt" + for directory in dirsToSearch: + if not os.path.isdir(directory): + result["missingDirectory"] = directory + startTime = time.time() + try: + if os.path.isfile(os.path.join(directory, randomFilename)): + randomFilename = "RandomFile" + str(random.randrange(10000, 99999)) + ".txt" + except OSError: + pass + if time.time() - startTime > constants.FOLDER_SEARCH_FIRST_FILE_TIMEOUT: + result["firstFileTimedOut"] = True + result["timedOutDirectory"] = directory + return result + + newCache = {} + startTime = time.time() + fileCount = 0 + effectiveWarningDelay = self._effectiveWarningDelay() + for directory in dirsToSearch: + for root, dirs, files in os.walk(directory): + fileCount += 1 + newCache[root] = files + elapsed = time.time() - startTime + if elapsed > constants.FOLDER_SEARCH_TIMEOUT: + result["cache"] = newCache + result["fileCount"] = fileCount + result["timedOut"] = True + result["timedOutDirectory"] = directory + return result + if elapsed > effectiveWarningDelay: + reactor.callFromThread(self._maybeShowSlowWarning, int(elapsed), fileCount, directory) + result["cache"] = newCache + result["fileCount"] = fileCount + return result + + def _commitFullReconciliation(self, result): + self._scanInFlight = False + self.currentlyUpdating = False + try: + if result.get("missingDirectory") and not self._missingDirectoryNotified: + self._missingDirectoryNotified = True + self.directorySearchError = getMessage("cannot-find-directory-error").format(result["missingDirectory"]) + + failed = result.get("timedOut") or result.get("firstFileTimedOut") + if failed: + self._enterDegraded() + self._maybeShowSlowWarning( + int(constants.FOLDER_SEARCH_TIMEOUT), + result.get("fileCount", 0), + result.get("timedOutDirectory") or "") + else: + newCache = result.get("cache", {}) + if newCache != self.mediaFilesCache: + self.mediaFilesCache = newCache + self.newInfo = True + self._scheduleInfoNotify() + self._registerSuccessfulScan() + self._refreshPriorityDirectories() + self.checkForFileSwitchUpdate() + finally: + self._afterScan() - def setFilenameWatchlist(self, unfoundFilenames): - self.filenameWatchlist = unfoundFilenames + def _scanErrback(self, failure): + self._scanInFlight = False + self.currentlyUpdating = False + self._debug("Media folder reconciliation scan failed: {}".format(failure.getErrorMessage())) + self._enterDegraded() + self._afterScan() + + def _afterScan(self): + # Consume a pending reconciliation, otherwise reschedule the adaptive timer. + if self._pendingReconciliation and not self._scanInFlight: + self._pendingReconciliation = False + self._startFullReconciliation() + else: + self._scheduleReconciliation() + + # -- degraded state and warning latch (spec 14) ----------------------- + + def _effectiveWarningDelay(self): + return constants.FOLDER_SEARCH_WARNING_BASE_DELAY + constants.FOLDER_SEARCH_WARNING_THRESHOLD + + def _maybeShowSlowWarning(self, seconds, fileCount, directory): + # Reactor thread. At most one slow/failure warning per media-directory + # configuration per session; successful scans do not clear the latch. + if self._slowWarningShown: + return + self._slowWarningShown = True + self._client.ui.showErrorMessage( + getMessage("folder-search-timeout-warning").format(int(seconds), fileCount, directory)) - def _updateInfoThread(self): - with self.lock: + def _enterDegraded(self): + self._degraded = True + self._consecutiveSuccessfulScans = 0 + + def _registerSuccessfulScan(self): + if not self._degraded: + return + self._consecutiveSuccessfulScans += 1 + if self._consecutiveSuccessfulScans >= constants.FOLDER_SEARCH_DEGRADED_RECOVERY_SCANS: + self._degraded = False + self._consecutiveSuccessfulScans = 0 + + # -- adaptive reconciliation cadence (spec 10) ------------------------ + + def _scheduleReconciliation(self): + self._cancelReconciliation() + if self._degraded or self._isUrgent(): + interval = constants.FOLDER_SEARCH_DOUBLE_CHECK_INTERVAL + else: + interval = constants.FOLDER_SEARCH_RECONCILIATION_INTERVAL + try: + self._reconciliationCall = reactor.callLater(interval, self._reconciliationTick) + except Exception: + self._reconciliationCall = None + + def _cancelReconciliation(self): + if self._reconciliationCall is not None: try: - self.currentlyUpdating = True - dirsToSearch = self.mediaDirectories + if self._reconciliationCall.active(): + self._reconciliationCall.cancel() + except Exception: + pass + self._reconciliationCall = None - if not self.folderSearchEnabled: - return + def _reconciliationTick(self): + self._reconciliationCall = None + self.updateInfo() - if dirsToSearch: - # Spin up hard drives to prevent premature timeout - randomFilename = "RandomFile"+str(random.randrange(10000, 99999))+".txt" - for directory in dirsToSearch: - if not os.path.isdir(directory): - self.directorySearchError = getMessage("cannot-find-directory-error").format(directory) - - startTime = time.time() - if os.path.isfile(os.path.join(directory, randomFilename)): - randomFilename = "RandomFile"+str(random.randrange(10000, 99999))+".txt" - print("Found random file (?)") - if time.time() - startTime > constants.FOLDER_SEARCH_FIRST_FILE_TIMEOUT: - self.folderSearchEnabled = False - self.directorySearchError = getMessage("folder-search-first-file-timeout-error").format(directory) - return - - # Actual directory search - newMediaFilesCache = {} - startTime = time.time() - fileCount = 0 - lastWarningTime = None - for directory in dirsToSearch: - for root, dirs, files in os.walk(directory): - fileCount += 1 - newMediaFilesCache[root] = files - timeTakenSoFar = time.time() - startTime - if timeTakenSoFar > constants.FOLDER_SEARCH_TIMEOUT: - reactor.callLater(0.1, self._client.ui.showErrorMessage, getMessage("folder-search-timeout-error").format(directory, fileCount),False) - self.folderSearchEnabled = False - return - if timeTakenSoFar > constants.FOLDER_SEARCH_WARNING_THRESHOLD: - if not lastWarningTime or timeTakenSoFar - lastWarningTime >= 1: - reactor.callLater(0.1, self._client.ui.showErrorMessage, getMessage("folder-search-timeout-warning").format(int(timeTakenSoFar), fileCount, directory),False) - lastWarningTime = timeTakenSoFar - - if self.mediaFilesCache != newMediaFilesCache: - self.mediaFilesCache = newMediaFilesCache - self.newInfo = True - except Exception as e: - self._client.ui.showDebugMessage(str(e)) - finally: - self.currentlyUpdating = False + def _isUrgent(self): + # Urgent while a needed current/next local playlist file cannot be + # resolved. URLs and absent targets do not force urgent local polling. + for filename in self._neededLocalFilenames(): + if filename and self._resolveExistingPath(filename) is None: + return True + return False - def infoUpdated(self): - self._client.fileSwitchFoundFiles() + def _neededLocalFilenames(self): + filenames = [] + try: + playlist = self._client.playlist + entries = playlist._playlist + index = playlist._playlistIndex + except Exception: + return filenames + if not entries or index is None: + return filenames + for candidateIndex in (index, index + 1): + if 0 <= candidateIndex < len(entries): + candidate = entries[candidateIndex] + if candidate and not isURL(candidate): + filenames.append(candidate) + return filenames + + def _resolveExistingPath(self, filename): + # Non-mutating check that a filename currently resolves to a real file, + # using the same semantics as findFilepath (cache path + on-disk check). + if filename is None: + return None + currentFile = self._client.userlist.currentUser.file + if currentFile and utils.sameFilename(filename, currentFile['name']): + return utils.getCorrectedPathForFile(currentFile['path']) + if self.mediaFilesCache is not None: + for directory in self.mediaFilesCache: + files = self.mediaFilesCache[directory] + if files and filename in files: + filepath = utils.getCorrectedPathForFile(os.path.join(directory, filename)) + if os.path.isfile(filepath): + return filepath + return None + + # -- filesystem-event driven cache updates (spec 11, 12, 13) ---------- + + def _onFileMonitorEvent(self, event): + # Reactor thread (marshalled by FileMonitor). Apply clear events directly; + # never confirm a clear event with a blocking scan. + if self._scanInFlight: + # Do not mutate the cache while a scan result is pending; collapse into + # a single follow-up reconciliation. + self._pendingReconciliation = True + return + try: + if event.isDirectory: + self._applyDirectoryEvent(event) + else: + self._applyFileEvent(event) + except Exception as e: + self._debug("Error applying filesystem event {}: {}".format(event, e)) + + def _cacheDirKey(self, path): + # Return the existing cache key matching path (component/case aware), else + # the normalised path itself. + target = os.path.normcase(os.path.normpath(path)) + for directory in self.mediaFilesCache: + if os.path.normcase(os.path.normpath(directory)) == target: + return directory + return path + + def _applyFileEvent(self, event): + eventType = event.eventType + if eventType == "moved": + self._removeFileFromCache(event.sourcePath) + if event.destinationPath and self._isInsideMediaRoots(event.destinationPath): + self._addFileToCache(event.destinationPath) + elif eventType == "deleted": + self._removeFileFromCache(event.sourcePath) + elif eventType in ("created", "modified"): + if self._isInsideMediaRoots(event.sourcePath): + self._addFileToCache(event.sourcePath) + # other activity-only event types (opened/closed) do not affect membership + + def _addFileToCache(self, path): + directory = self._cacheDirKey(os.path.dirname(path)) + basename = os.path.basename(path) + files = self.mediaFilesCache.get(directory) + if files is None: + self.mediaFilesCache[directory] = [basename] + self._markCacheChanged() + elif basename not in files: + files.append(basename) + self._markCacheChanged() + + def _removeFileFromCache(self, path): + directory = self._cacheDirKey(os.path.dirname(path)) + basename = os.path.basename(path) + files = self.mediaFilesCache.get(directory) + if files is not None and basename in files: + files.remove(basename) + self._markCacheChanged() + + def _applyDirectoryEvent(self, event): + eventType = event.eventType + if eventType == "deleted": + self._removeSubtreeFromCache(event.sourcePath) + elif eventType == "created": + if self._isInsideMediaRoots(event.sourcePath): + self._requestTargetedScan(event.sourcePath) + elif eventType == "moved": + self._removeSubtreeFromCache(event.sourcePath) + if event.destinationPath and self._isInsideMediaRoots(event.destinationPath): + self._requestTargetedScan(event.destinationPath) + + def _removeSubtreeFromCache(self, path): + target = os.path.normcase(os.path.normpath(path)) + removed = False + for directory in list(self.mediaFilesCache.keys()): + normed = os.path.normcase(os.path.normpath(directory)) + if normed == target or normed.startswith(target + os.sep): + del self.mediaFilesCache[directory] + removed = True + if removed: + self._markCacheChanged() + + def _isInsideMediaRoots(self, path): + if not path or not self.mediaDirectories: + return False + normed = os.path.normcase(os.path.normpath(path)) + for root in self.mediaDirectories: + normedRoot = os.path.normcase(os.path.normpath(root)) + if normed == normedRoot or normed.startswith(normedRoot + os.sep): + return True + return False + + def _markCacheChanged(self): + self.newInfo = True + self._scheduleInfoNotify() + + def _scheduleInfoNotify(self): + # Coalesce the downstream fileSwitchFoundFiles() notification so a burst of + # events does not invoke file-switch discovery once per low-level event. + if self._infoNotifyCall is not None and self._infoNotifyCall.active(): + return + try: + self._infoNotifyCall = reactor.callLater(constants.FOLDER_SEARCH_EVENT_COALESCE_INTERVAL, self._flushInfoNotify) + except Exception: + self._infoNotifyCall = None + self.checkForFileSwitchUpdate() + + def _flushInfoNotify(self): + self._infoNotifyCall = None + self.checkForFileSwitchUpdate() + + # -- targeted subtree scan (spec 11, 12) ------------------------------ + + def _requestTargetedScan(self, directory): + if self._scanInFlight: + self._pendingReconciliation = True + return + self._scanInFlight = True + self.currentlyUpdating = True + deferred = threads.deferToThread(self._targetedScanWorker, directory) + deferred.addCallback(self._commitTargetedScan) + deferred.addErrback(self._scanErrback) + + def _targetedScanWorker(self, directory): + result = {"directory": directory, "entries": {}, "missing": False} + if not os.path.isdir(directory): + result["missing"] = True + return result + entries = {} + try: + for root, dirs, files in os.walk(directory): + entries[root] = files + except OSError: + result["missing"] = True + return result + result["entries"] = entries + return result + + def _commitTargetedScan(self, result): + self._scanInFlight = False + self.currentlyUpdating = False + try: + directory = result.get("directory") + self._removeSubtreeFromCache(directory) + entries = result.get("entries") or {} + if entries: + self.mediaFilesCache.update(entries) + self._markCacheChanged() + elif not result.get("missing"): + self._markCacheChanged() + if entries: + self._refreshPriorityDirectories() + self.checkForFileSwitchUpdate() + finally: + self._afterScan() + + # -- priority directory selection (spec 7, 8) ------------------------- + + def _refreshPriorityDirectories(self): + # Only Windows network roots benefit from supplemental direct watches; skip + # the work entirely when no configured root is a network path. + if not self.mediaDirectories or not any(isWindowsNetworkPath(root) for root in self.mediaDirectories): + if self._priorityDirectories: + self._priorityDirectories = [] + self._fileMonitor.setPriorityDirectories([]) + return + ranked = self._rankPriorityDirectories() + if ranked != self._priorityDirectories: + self._priorityDirectories = ranked + self._fileMonitor.setPriorityDirectories(ranked) + + def _rankPriorityDirectories(self): + ordered = [] + seen = set() + + def add(directory): + if not directory: + return + normed = os.path.normcase(os.path.normpath(directory)) + if normed in seen: + return + if not self._isInsideMediaRoots(directory): + return + if not isWindowsNetworkPath(directory): + return + seen.add(normed) + ordered.append(directory) + + # 1. Directory of the currently open local file. + currentFile = self._client.userlist.currentUser.file + if currentFile and currentFile.get('path'): + add(os.path.dirname(currentFile['path'])) + + # 2-3. Cached directories of the current and next playlist filenames. + neededFilenames = self._neededLocalFilenames() + for filename in neededFilenames: + directory = self.getDirectoryOfFilenameInCache(filename) + add(directory) + + # 4. Same-series/season directories for the current/next entries (Phase 6 + # episode-parser hook; a false positive only affects watch ranking). + for directory in self._relatedEpisodeDirectories(neededFilenames): + add(directory) + + # 5-7. Exact matches for other playlist entries, then remaining known + # directories, so small trees end up fully watched within budget. + for filename in self._allPlaylistFilenames(): + add(self.getDirectoryOfFilenameInCache(filename)) + for directory in self.mediaFilesCache: + add(directory) + + return ordered + + def _relatedEpisodeDirectories(self, filenames): + # Reuse the watched EpisodeFilenameParser to spot same-series/season + # directories. Ranking hint only; a false positive cannot affect switching. + try: + return getRelatedEpisodeDirectories(filenames, self.mediaFilesCache) + except Exception: + return [] + + def _allPlaylistFilenames(self): + try: + entries = self._client.playlist._playlist + index = self._client.playlist._playlistIndex or 0 + except Exception: + return [] + if not entries: + return [] + ordered = entries[index:] + entries[:index] + return [entry for entry in ordered if entry and not isURL(entry)] + + # -- resolution and lookup helpers ------------------------------------ def findFilepath(self, filename, highPriority=False): if filename is None: @@ -2626,6 +3044,11 @@ def findFilepath(self, filename, highPriority=False): if os.path.isfile(filepath): return filepath + # A needed file that cannot currently be resolved should hurry the next + # authoritative scan rather than waiting for the slow timer. + if highPriority: + self.updateInfo() + def areWatchedFilenamesInCache(self): if self.filenameWatchlist is not None: for filename in self.filenameWatchlist: @@ -2680,3 +3103,19 @@ def notifyUserIfFileNotInMediaDirectory(self, filenameToFind, path): directoryToFind = str(directoryToFind) self._client.ui.showErrorMessage(getMessage("added-file-not-in-media-directory-error").format(directoryToFind)) self.mediaDirectoriesNotFound.append(directoryToFind) + + # -- shutdown --------------------------------------------------------- + + def _shutdown(self): + self._cancelReconciliation() + if self._infoNotifyCall is not None: + try: + if self._infoNotifyCall.active(): + self._infoNotifyCall.cancel() + except Exception: + pass + self._infoNotifyCall = None + try: + self._fileMonitor.stop() + except Exception: + pass diff --git a/syncplay/constants.py b/syncplay/constants.py index 57d5c091..3ecba7ae 100755 --- a/syncplay/constants.py +++ b/syncplay/constants.py @@ -116,8 +116,13 @@ def getValueForOS(constantDict): # Options for the File Switch feature: FOLDER_SEARCH_FIRST_FILE_TIMEOUT = 25.0 # Secs - How long to wait to find the first file in folder search (to take account of HDD spin up) FOLDER_SEARCH_TIMEOUT = 20.0 # Secs - How long to wait until searches in folder to update cache are aborted (after first file is found) -FOLDER_SEARCH_WARNING_THRESHOLD = 2.0 # Secs - how long until a warning saying how many files have been scanned -FOLDER_SEARCH_DOUBLE_CHECK_INTERVAL = 30.0 # Secs - Frequency of updating cache +FOLDER_SEARCH_WARNING_THRESHOLD = 2.0 # Secs - additional grace (on top of FOLDER_SEARCH_WARNING_BASE_DELAY) before the slow-scan warning may appear +FOLDER_SEARCH_WARNING_BASE_DELAY = 3.0 # Secs - base grace before a slow scan may warn; effective point is this + folderSearchWarningThreshold +FOLDER_SEARCH_DOUBLE_CHECK_INTERVAL = 30.0 # Secs - Urgent polling cadence used while a needed current/next local file is unresolved +FOLDER_SEARCH_RECONCILIATION_INTERVAL = 300.0 # Secs - Slow safety reconciliation cadence while needed files are resolvable (internal, not user-configurable) +FOLDER_SEARCH_EVENT_COALESCE_INTERVAL = 1.0 # Secs - Coalescing window for the downstream fileSwitchFoundFiles() notification after direct cache events +FOLDER_SEARCH_DEGRADED_RECOVERY_SCANS = 3 # Consecutive successful full reconciliations required to clear the degraded operational state +FOLDER_SEARCH_NETWORK_WATCH_LIMIT = 32 # Process-wide budget for Windows network-directory watches (recursive roots + supplemental direct watches) # Changable values for watched features (you usually don't need to change these) WATCHED_CHECKQUEUE_INTERVAL = 1.0 # Secs diff --git a/syncplay/filemonitor.py b/syncplay/filemonitor.py new file mode 100644 index 00000000..c64b8446 --- /dev/null +++ b/syncplay/filemonitor.py @@ -0,0 +1,264 @@ +# coding:utf8 + +""" +Client-side filesystem-watch helper for media directory monitoring. + +FileMonitor is deliberately playlist-blind: it knows about paths and native +filesystem watch characteristics only. It maintains the most useful native +watchdog watches within a budget and reports structured filesystem events back +to its owner (FileSwitchManager) on the Twisted reactor thread. It never touches +mediaFilesCache, playlist state, episode parsing or file-switching decisions. + +watchdog is an optional dependency: if it is unavailable, or a watch cannot be +established, FileMonitor simply provides no events and Syncplay falls back to its +own reconciliation scanning. +""" + +import os +import sys +from collections import namedtuple + +from twisted.internet import reactor + +from syncplay import constants + +try: + from watchdog.observers import Observer as _Observer + from watchdog.events import FileSystemEventHandler as _FileSystemEventHandler + watchdogAvailable = True +except ImportError: + watchdogAvailable = False + _FileSystemEventHandler = object + +# Simple immutable value handed to the owner's callback. It carries filesystem +# facts only - never playlist or file-matching information. +FileMonitorEvent = namedtuple("FileMonitorEvent", ["root", "eventType", "sourcePath", "destinationPath", "isDirectory"]) + + +def isWindowsNetworkPath(path): + # A Windows UNC path, or a mapped drive the OS reports as remote. Everywhere + # else this is False and a single recursive native watch is trusted at depth. + if not path or sys.platform != "win32": + return False + if path.startswith("\\\\") or path.startswith("//"): + return True + try: + import ctypes + drive = os.path.splitdrive(os.path.abspath(path))[0] + if not drive: + return False + DRIVE_REMOTE = 4 + return ctypes.windll.kernel32.GetDriveTypeW(drive + "\\") == DRIVE_REMOTE + except Exception: + return False + + +def _normPath(path): + return os.path.normcase(os.path.normpath(path)) + + +def _pathIsWithin(path, root): + # Component-aware containment so that, for example, "P:\\TV2" is not treated + # as being inside "P:\\TV". + normedPath = _normPath(path) + normedRoot = _normPath(root) + if normedPath == normedRoot: + return True + return normedPath.startswith(normedRoot + os.sep) + + +if watchdogAvailable: + class _WatchdogAdapter(_FileSystemEventHandler): + def __init__(self, monitor): + _FileSystemEventHandler.__init__(self) + self._monitor = monitor + + def on_any_event(self, event): + self._monitor._onWatchdogEvent(event) + + +class FileMonitor(object): + def __init__(self, eventCallback, debugCallback=None): + self._eventCallback = eventCallback + self._debugCallback = debugCallback + self._observer = None + self._adapter = None + self._mediaDirectories = [] + self._rootWatches = {} # normed root -> (watch, path, isNetwork) + self._priorityWatches = {} # normed dir -> (watch, path) + + def isAvailable(self): + return watchdogAvailable + + def _debug(self, message): + if self._debugCallback: + try: + self._debugCallback(message) + except Exception: + pass + + def _ensureObserver(self): + if self._observer is None: + observer = _Observer() + self._adapter = _WatchdogAdapter(self) + observer.start() + self._observer = observer + return self._observer + + # -- public API ------------------------------------------------------- + + def setMediaDirectories(self, mediaDirectories): + if not watchdogAvailable: + return + newRoots = [directory for directory in (mediaDirectories or []) if directory] + self._mediaDirectories = list(newRoots) + newNormedRoots = set(_normPath(root) for root in newRoots) + + for normedRoot in list(self._rootWatches.keys()): + if normedRoot not in newNormedRoots: + self._removeRootWatch(normedRoot) + + for normedDir in list(self._priorityWatches.keys()): + watchedPath = self._priorityWatches[normedDir][1] + if not any(_pathIsWithin(watchedPath, root) for root in newRoots): + self._removePriorityWatch(normedDir) + + for root in newRoots: + normedRoot = _normPath(root) + if normedRoot in self._rootWatches: + continue + isNetwork = isWindowsNetworkPath(root) + if isNetwork and self._networkWatchCount() >= constants.FOLDER_SEARCH_NETWORK_WATCH_LIMIT: + self._debug("Network watch budget reached; media root not natively watched: {}".format(root)) + continue + self._scheduleRoot(root, isNetwork) + + def setPriorityDirectories(self, directories): + # Supplemental non-recursive watches are only useful for Windows network + # roots; a local recursive root watch already reports changes at depth. + if not watchdogAvailable: + return + wanted = [] + seen = set() + for directory in (directories or []): + if not directory: + continue + normedDir = _normPath(directory) + if normedDir in seen or normedDir in self._rootWatches: + continue + if self._findNetworkRoot(directory) is None: + continue + seen.add(normedDir) + wanted.append((normedDir, directory)) + + wantedNormed = set(normedDir for normedDir, _ in wanted) + for normedDir in list(self._priorityWatches.keys()): + if normedDir not in wantedNormed: + self._removePriorityWatch(normedDir) + + for normedDir, directory in wanted: + if normedDir in self._priorityWatches: + continue + if self._networkWatchCount() >= constants.FOLDER_SEARCH_NETWORK_WATCH_LIMIT: + self._debug("Network watch budget reached; priority directory not watched: {}".format(directory)) + break + self._schedulePriority(directory) + + def stop(self): + observer = self._observer + self._observer = None + self._adapter = None + self._rootWatches = {} + self._priorityWatches = {} + if observer is not None: + try: + observer.unschedule_all() + observer.stop() + observer.join(5.0) + except Exception: + pass + + # -- watch bookkeeping ------------------------------------------------ + + def _networkWatchCount(self): + count = sum(1 for entry in self._rootWatches.values() if entry[2]) + count += len(self._priorityWatches) + return count + + def _scheduleRoot(self, root, isNetwork): + try: + observer = self._ensureObserver() + watch = observer.schedule(self._adapter, root, recursive=True) + self._rootWatches[_normPath(root)] = (watch, root, isNetwork) + except Exception as e: + self._debug("Could not watch media root {}: {}: {}".format(root, type(e).__name__, e)) + + def _schedulePriority(self, directory): + try: + observer = self._ensureObserver() + watch = observer.schedule(self._adapter, directory, recursive=False) + self._priorityWatches[_normPath(directory)] = (watch, directory) + except Exception as e: + self._debug("Could not watch priority directory {}: {}: {}".format(directory, type(e).__name__, e)) + + def _removeRootWatch(self, normedRoot): + entry = self._rootWatches.pop(normedRoot, None) + if entry is not None and self._observer is not None: + try: + self._observer.unschedule(entry[0]) + except Exception: + pass + + def _removePriorityWatch(self, normedDir): + entry = self._priorityWatches.pop(normedDir, None) + if entry is not None and self._observer is not None: + try: + self._observer.unschedule(entry[0]) + except Exception: + pass + + # -- event handling --------------------------------------------------- + + def _findRoot(self, path): + if not path: + return None + best = None + bestLength = -1 + for root in self._mediaDirectories: + if _pathIsWithin(path, root): + length = len(_normPath(root)) + if length > bestLength: + best = root + bestLength = length + return best + + def _findNetworkRoot(self, path): + root = self._findRoot(path) + if root is not None and isWindowsNetworkPath(root): + return root + return None + + def _onWatchdogEvent(self, event): + # Runs on the watchdog worker thread. Copy the useful fields and hand off + # to the reactor thread; do not touch shared Syncplay state here. + try: + sourcePath = getattr(event, "src_path", None) + destinationPath = getattr(event, "dest_path", None) or None + root = self._findRoot(sourcePath) + if root is None and destinationPath: + root = self._findRoot(destinationPath) + if root is None: + return + payload = FileMonitorEvent( + root, + getattr(event, "event_type", None), + sourcePath, + destinationPath, + bool(getattr(event, "is_directory", False)), + ) + except Exception: + return + try: + reactor.callFromThread(self._eventCallback, payload) + except Exception: + pass diff --git a/syncplay/messages_en.py b/syncplay/messages_en.py index b92e0a7a..81c3ef1a 100644 --- a/syncplay/messages_en.py +++ b/syncplay/messages_en.py @@ -179,7 +179,7 @@ "switch-file-not-found-error": "Could not switch to file '{0}'. Syncplay looks in specified media directories.", # File not found "folder-search-timeout-error": "The search for media in media directories was aborted as it took too long to search through '{}' after having processed the first {:,} files. This will occur if you select a folder with too many subfolders in your list of media folders to search through or if there are too many files to process. For automatic file switching to work again please select File->Set Media Directories in the menu bar and remove this directory or replace it with an appropriate subfolder. If the folder is actually fine then you can re-enable it by selecting File->Set Media Directories and pressing 'OK'.", # Folder, Files processed. Note: {:,} is {} but with added commas seprators. - "folder-search-timeout-warning": "Warning: It has taken {} seconds to scan {:,} files in the folder '{}'. This will occur if you select a folder with too many subfolders in your list of media folders to search through or if there are too many files to process.", # Folder, Files processed. Note: {:,} is {} but with added commas seprators. + "folder-search-timeout-warning": "Note: Syncplay is taking a while to scan your media directories (about {} seconds and {:,} files so far, currently in '{}'). This can happen with very large folders or slow/network drives. Syncplay will keep trying automatically in the background, so no action is needed.", # Seconds, Files processed, Folder. Note: {:,} is {} but with added commas separators. "folder-search-first-file-timeout-error": "The search for media in '{}' was aborted as it took too long to access the directory. This could happen if it is a network drive or if you configure your drive to spin down after a period of inactivity. For automatic file switching to work again please go to File->Set Media Directories and either remove the directory or resolve the issue (e.g. by changing power saving settings).", # Folder "added-file-not-in-media-directory-error": "You loaded a file in '{}' which is not a known media directory. You can add this as a media directory by selecting File->Set Media Directories in the menu bar.", # Folder "no-media-directories-error": "No media directories have been set. For shared playlist and file switching features to work properly please select File->Set Media Directories and specify where Syncplay should look to find media files.", diff --git a/syncplay/watched.py b/syncplay/watched.py index 0917faac..daf1a515 100644 --- a/syncplay/watched.py +++ b/syncplay/watched.py @@ -716,6 +716,49 @@ def _normaliseEpisodeNumber(self, episodeNumber): return episodeNumber +def getRelatedEpisodeDirectories(filenames, mediaFilesCache): + """Directories in mediaFilesCache holding the same series/season as any of the + target filenames, using the existing episode parser. Ranking hint only: it does + no filesystem access, must not affect playlist behaviour, and a false positive + is harmless. When a match lives in a Watched subfolder, the parent (the likely + location for a newly-arriving unwatched episode) is returned instead. + """ + if not filenames or not mediaFilesCache: + return [] + parser = EpisodeFilenameParser() + targetKeys = set() + targetContext = parser.getContext(filenames) + for filename in filenames: + info = parser.parse(filename, context=targetContext) + if info and info.get("seriesKey") is not None: + targetKeys.add((info.get("seriesKey"), info.get("season"))) + if not targetKeys: + return [] + + related = [] + seen = set() + watchedName = constants.WATCHED_SUBFOLDER.lower() if constants.WATCHED_SUBFOLDER else None + for directory, files in mediaFilesCache.items(): + if not files: + continue + directoryContext = parser.getContext(files) + matched = False + for filename in files: + info = parser.parse(filename, context=directoryContext) + if info and (info.get("seriesKey"), info.get("season")) in targetKeys: + matched = True + break + if not matched: + continue + resultDirectory = directory + if watchedName and os.path.basename(os.path.normpath(directory)).lower() == watchedName: + resultDirectory = os.path.dirname(os.path.normpath(directory)) + key = os.path.normcase(os.path.normpath(resultDirectory)) + if key not in seen: + seen.add(key) + related.append(resultDirectory) + return related + class WatchedManager(object): """