diff --git a/src/plugin_subscriber.py b/src/plugin_subscriber.py deleted file mode 100644 index d8fbde0..0000000 --- a/src/plugin_subscriber.py +++ /dev/null @@ -1,307 +0,0 @@ -""" -Plugin Completion Subscriber for meta-stremio. - -Subscribes to Redis pub/sub channel 'meta-sort:plugin:complete' to be notified -when specific plugins finish processing files. This allows Stremio to: -1. Log metadata updates for debugging -2. Invalidate any local caches (if implemented) -3. Track which files have complete metadata - -Plugins of interest to Stremio: -- filename-parser: Provides season/episode info for TV vs Movie categorization -- tmdb: Provides rich metadata (title, year, poster, description, etc.) -""" -from __future__ import annotations - -import os -import json -import threading -from typing import Callable, Optional, List, Set - -try: - import redis - REDIS_AVAILABLE = True -except ImportError: - REDIS_AVAILABLE = False - redis = None - - -# Redis configuration -REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379') - -# Plugins that Stremio cares about -WATCHED_PLUGINS: Set[str] = {'filename-parser', 'tmdb', 'jellyfin-nfo'} - -# Channel for plugin completion events -PLUGIN_COMPLETE_CHANNEL = 'meta-sort:plugin:complete' - -# Channel for scan reset events -RESET_CHANNEL = 'meta-sort:scan:reset' - - -class PluginSubscriber: - """ - Subscribes to plugin completion events from meta-sort. - - When filename-parser or tmdb plugins complete, the subscriber - receives notifications that can be used to update Stremio's view - of the metadata. - """ - - def __init__( - self, - url: str = None, - watched_plugins: Set[str] = None, - on_plugin_complete: Callable[[str, str, str], None] = None, - on_scan_reset: Callable[[], None] = None - ): - """ - Initialize the plugin subscriber. - - Args: - url: Redis connection URL - watched_plugins: Set of plugin IDs to watch (default: filename-parser, tmdb, jellyfin-nfo) - on_plugin_complete: Callback when a watched plugin completes - Signature: (file_hash, plugin_id, file_path) -> None - on_scan_reset: Callback when a scan reset event is received - Signature: () -> None - """ - if not REDIS_AVAILABLE: - raise ImportError("redis package not installed. Run: pip install redis") - - self._url = url or REDIS_URL - self._watched_plugins = watched_plugins or WATCHED_PLUGINS - self._on_plugin_complete = on_plugin_complete - self._on_scan_reset = on_scan_reset - self._pubsub: Optional[redis.client.PubSub] = None - self._thread: Optional[threading.Thread] = None - self._running = False - self._client: Optional[redis.Redis] = None - - # Track files with complete metadata (all watched plugins finished) - self._file_plugin_status: dict[str, Set[str]] = {} - - def start(self) -> bool: - """Start the subscriber in a background thread.""" - if self._running: - return True - - try: - self._client = redis.from_url(self._url, decode_responses=True) - self._client.ping() # Test connection - - self._pubsub = self._client.pubsub() - self._pubsub.subscribe(PLUGIN_COMPLETE_CHANNEL, RESET_CHANNEL) - - self._running = True - self._thread = threading.Thread( - target=self._listen_loop, - daemon=True, - name='PluginSubscriber' - ) - self._thread.start() - - print(f"[PluginSubscriber] Started, watching plugins: {', '.join(self._watched_plugins)}") - print(f"[PluginSubscriber] Subscribed to: {PLUGIN_COMPLETE_CHANNEL}, {RESET_CHANNEL}") - return True - - except Exception as e: - print(f"[PluginSubscriber] Failed to start: {e}") - self._running = False - return False - - def stop(self) -> None: - """Stop the subscriber.""" - self._running = False - - if self._pubsub: - try: - self._pubsub.unsubscribe() - self._pubsub.close() - except Exception: - pass - self._pubsub = None - - if self._client: - try: - self._client.close() - except Exception: - pass - self._client = None - - print("[PluginSubscriber] Stopped") - - def _listen_loop(self) -> None: - """Background loop that listens for plugin completion and reset events.""" - while self._running and self._pubsub: - try: - message = self._pubsub.get_message(timeout=1.0) - if message and message['type'] == 'message': - channel = message.get('channel', '') - if channel == RESET_CHANNEL: - self._handle_reset_message(message['data']) - else: - self._handle_message(message['data']) - except redis.ConnectionError: - print("[PluginSubscriber] Connection lost, attempting reconnect...") - self._reconnect() - except Exception as e: - if self._running: - print(f"[PluginSubscriber] Error in listen loop: {e}") - - def _reconnect(self) -> None: - """Attempt to reconnect to Redis.""" - import time - - for attempt in range(5): - if not self._running: - return - - try: - time.sleep(2 ** attempt) # Exponential backoff - - if self._pubsub: - self._pubsub.close() - if self._client: - self._client.close() - - self._client = redis.from_url(self._url, decode_responses=True) - self._client.ping() - - self._pubsub = self._client.pubsub() - self._pubsub.subscribe(PLUGIN_COMPLETE_CHANNEL, RESET_CHANNEL) - - print("[PluginSubscriber] Reconnected successfully") - return - - except Exception as e: - print(f"[PluginSubscriber] Reconnect attempt {attempt + 1} failed: {e}") - - print("[PluginSubscriber] Failed to reconnect after 5 attempts") - self._running = False - - def _handle_reset_message(self, data: str) -> None: - """Handle a scan reset message.""" - try: - payload = json.loads(data) - action = payload.get('action', '') - - print(f"[PluginSubscriber] Received scan reset event: {action}") - - # Clear plugin completion tracking - fresh scan starting - files_cleared = len(self._file_plugin_status) - self._file_plugin_status.clear() - print(f"[PluginSubscriber] Cleared plugin status tracking for {files_cleared} files") - - # Call the callback if provided - if self._on_scan_reset: - try: - self._on_scan_reset() - except Exception as e: - print(f"[PluginSubscriber] Reset callback error: {e}") - - except json.JSONDecodeError: - print(f"[PluginSubscriber] Invalid JSON in reset message: {data[:100]}") - except Exception as e: - print(f"[PluginSubscriber] Error handling reset message: {e}") - - def _handle_message(self, data: str) -> None: - """Handle a plugin completion message.""" - try: - payload = json.loads(data) - file_hash = payload.get('fileHash', '') - plugin_id = payload.get('pluginId', '') - file_path = payload.get('filePath', '') - - # Only process plugins we're watching - if plugin_id not in self._watched_plugins: - return - - # Track plugin completion for this file - if file_hash not in self._file_plugin_status: - self._file_plugin_status[file_hash] = set() - self._file_plugin_status[file_hash].add(plugin_id) - - # Log the completion - print(f"[PluginSubscriber] Plugin '{plugin_id}' completed for {file_hash[:12]}...") - - # Check if all watched plugins have completed for this file - completed_plugins = self._file_plugin_status[file_hash] - if self._watched_plugins.issubset(completed_plugins): - print(f"[PluginSubscriber] All watched plugins complete for {file_hash[:12]}...") - # Could trigger a cache refresh here if we had local caching - - # Call the callback if provided - if self._on_plugin_complete: - try: - self._on_plugin_complete(file_hash, plugin_id, file_path) - except Exception as e: - print(f"[PluginSubscriber] Callback error: {e}") - - except json.JSONDecodeError: - print(f"[PluginSubscriber] Invalid JSON message: {data[:100]}") - except Exception as e: - print(f"[PluginSubscriber] Error handling message: {e}") - - def is_running(self) -> bool: - """Check if the subscriber is running.""" - return self._running - - def get_status(self) -> dict: - """Get subscriber status for debugging.""" - return { - 'running': self._running, - 'watched_plugins': list(self._watched_plugins), - 'tracked_files': len(self._file_plugin_status), - } - - -# Global subscriber instance -_subscriber: Optional[PluginSubscriber] = None - - -def init_subscriber( - on_plugin_complete: Callable[[str, str, str], None] = None, - on_scan_reset: Callable[[], None] = None -) -> Optional[PluginSubscriber]: - """ - Initialize and start the global plugin subscriber. - - Args: - on_plugin_complete: Optional callback when watched plugins complete - on_scan_reset: Optional callback when a scan reset event is received - - Returns: - The subscriber instance, or None if Redis is unavailable - """ - global _subscriber - - if not REDIS_AVAILABLE: - print("[PluginSubscriber] Redis not available, skipping subscriber init") - return None - - if _subscriber and _subscriber.is_running(): - return _subscriber - - _subscriber = PluginSubscriber( - on_plugin_complete=on_plugin_complete, - on_scan_reset=on_scan_reset - ) - if _subscriber.start(): - return _subscriber - - _subscriber = None - return None - - -def get_subscriber() -> Optional[PluginSubscriber]: - """Get the global subscriber instance.""" - return _subscriber - - -def stop_subscriber() -> None: - """Stop the global subscriber.""" - global _subscriber - if _subscriber: - _subscriber.stop() - _subscriber = None diff --git a/src/server.py b/src/server.py index c6310fd..38f71a5 100644 --- a/src/server.py +++ b/src/server.py @@ -36,7 +36,6 @@ import stremio import transcoder from storage import init_service_discovery, get_service_discovery import fileserver -import plugin_subscriber import webdav_client # Configuration @@ -238,13 +237,6 @@ class Handler(BaseHTTPRequestHandler): if path == '/api/services': return self.handle_services_api() - # Plugin subscriber status - if path == '/api/plugin-subscriber': - subscriber = plugin_subscriber.get_subscriber() - if subscriber: - return self.send_json(subscriber.get_status()) - return self.send_json({'running': False, 'reason': 'Not initialized'}) - # === File API === # /file/{cid} or /poster/{cid} (backward compat) - serve file by CID # /file/{cid}/w{width} - serve resized image @@ -880,12 +872,6 @@ def shutdown_handler(signum, frame): """Handle graceful shutdown.""" print("\n[Server] Shutting down...") - # Stop plugin subscriber - try: - plugin_subscriber.stop_subscriber() - except Exception as e: - print(f"[Server] Error stopping plugin subscriber: {e}") - # Stop service discovery try: sd = get_service_discovery() @@ -909,10 +895,6 @@ def main(): # Initialize fileserver with storage reference fileserver.init(storage) - # Initialize plugin subscriber to receive notifications when - # filename-parser and tmdb plugins complete - subscriber = plugin_subscriber.init_subscriber() - print(f"Meta-Stremio starting on port {PORT}") print(f"Storage: {storage_type} | Videos: {video_count}") print(f"Media: {transcoder.MEDIA_DIR} | Cache: {transcoder.CACHE_DIR}") @@ -926,11 +908,6 @@ def main(): else: print(f"Service discovery: disabled (META_CORE_PATH not found)") - if subscriber: - print(f"Plugin subscriber: enabled (watching: filename-parser, tmdb)") - else: - print(f"Plugin subscriber: disabled (Redis not available)") - ThreadedServer(('0.0.0.0', PORT), Handler).serve_forever()