fix: minor tweaks and validation handling (#1009)

* fix: fix state check on paused/failed

* fix: tidy validation and logging

* fix: switched to singular container on session response

* fix: minor tweaks

* fix: minor tweaks to logging and some validation

* chore: pin min/max version for RTN

Author: dreulavelle

Dockerfile

@@ -38,6 +38,7 @@ RUN apk add --no-cache \
     rclone \
     unzip \
     gcc \
+    ffmpeg \
     musl-dev \
     libffi-dev \
     python3-dev \

makefile

@@ -106,6 +106,10 @@ hard_reset: clean
 install:
 	@poetry install --with dev
 
+update:
+	@poetry cache clear PyPI --all
+	@poetry update
+
 # Run the application
 run:
 	@poetry run python src/main.py

poetry.lock

@@ -504,7 +504,7 @@ def run_thread_with_db_item(fn, service, program, event: Event, cancellation_eve
             if event.content_item:
                 indexed_item = next(fn(event.content_item), None)
                 if indexed_item is None:
-                    logger.debug(f"Unable to index {event.content_item.log_string}")
+                    logger.debug(f"Unable to index {event.content_item.log_string if event.content_item.log_string is not None else event.content_item.imdb_id}")
                     return None
                 indexed_item.store_state()
                 session.add(indexed_item)

src/program/db/db_functions.py

@@ -206,95 +206,6 @@ def submit_job(self, service, program, event=None):
         websocket_manager.publish("event_update", self.get_event_updates())
         future.add_done_callback(lambda f:self._process_future(f, service))
 
-    # For debugging purposes we can monitor the execution time of the service. (comment out above and uncomment below)
-    # def submit_job(self, service, program, event=None):
-    #     """
-    #     Submits a job to be executed by the service.
-
-    #     Args:
-    #         service (type): The service class to execute.
-    #         program (Program): The program containing the service.
-    #         item (Event, optional): The event item to process. Defaults to None.
-    #     """
-    #     log_message = f"Submitting service {service.__name__} to be executed"
-    #     if event:
-    #         log_message += f" with {event.log_message}"
-    #     logger.debug(log_message)
-
-    #     cancellation_event = threading.Event()
-    #     executor = self._find_or_create_executor(service)
-        
-    #     # Add start time to track execution duration
-    #     start_time = datetime.now()
-        
-    #     def _monitor_execution(future):
-    #         """Monitor execution time and log if taking too long"""
-    #         while not future.done():
-    #             execution_time = (datetime.now() - start_time).total_seconds()
-    #             if execution_time > 180:  # 3 minutes
-    #                 current_thread = None
-    #                 for thread in threading.enumerate():
-    #                     if thread.name.startswith(service.__name__) and not thread.name.endswith('_monitor'):
-    #                         current_thread = thread
-    #                         break
-                            
-    #                 if current_thread:
-    #                     # Get stack frames for the worker thread
-    #                     frames = sys._current_frames()
-    #                     thread_frame = None
-    #                     for thread_id, frame in frames.items():
-    #                         if thread_id == current_thread.ident:
-    #                             thread_frame = frame
-    #                             break
-                        
-    #                     if thread_frame:
-    #                         stack_trace = ''.join(traceback.format_stack(thread_frame))
-    #                     else:
-    #                         stack_trace = "Could not get stack trace for worker thread"
-    #                 else:
-    #                     stack_trace = "Could not find worker thread"
-                    
-    #                 logger.warning(
-    #                     f"Service {service.__name__} execution taking longer than 3 minutes!\n"
-    #                     f"Event: {event.log_message if event else 'No event'}\n"
-    #                     f"Execution time: {execution_time:.1f} seconds\n"
-    #                     f"Thread name: {current_thread.name if current_thread else 'Unknown'}\n"
-    #                     f"Thread alive: {current_thread.is_alive() if current_thread else 'Unknown'}\n"
-    #                     f"Stack trace:\n{stack_trace}"
-    #                 )
-                    
-    #                 # Cancel the future and kill the thread
-    #                 future.cancellation_event.set()
-    #                 future.cancel()
-    #                 if current_thread:
-    #                     logger.warning(f"Killing thread {current_thread.name} due to timeout")
-    #                     self._futures.remove(future)
-    #                     if event:
-    #                         self.remove_event_from_running(event)
-    #                 return  # Exit the monitoring thread
-                    
-    #             time.sleep(60)  # Check every minute
-
-    #     future = executor.submit(db_functions.run_thread_with_db_item, 
-    #                         program.all_services[service].run, 
-    #                         service, program, event, cancellation_event)
-        
-    #     # Start monitoring thread
-    #     monitor_thread = threading.Thread(
-    #         target=_monitor_execution, 
-    #         args=(future,),
-    #         name=f"{service.__name__}_monitor",
-    #         daemon=True
-    #     )
-    #     monitor_thread.start()
-        
-    #     future.cancellation_event = cancellation_event
-    #     if event:
-    #         future.event = event
-    #     self._futures.append(future)
-    #     sse_manager.publish("event_update", self.get_event_updates())
-    #     future.add_done_callback(lambda f: self._process_future(f, service))
-
     def cancel_job(self, item_id: str, suppress_logs=False):
         """
         Cancels a job associated with the given item.
@@ -343,7 +254,7 @@ def next(self) -> Event:
             if self._queued_events:
                 with self.mutex:
                     self._queued_events.sort(key=lambda event: event.run_at)
-                    if datetime.now() >= self._queued_events[0].run_at:
+                    if self._queued_events and datetime.now() >= self._queued_events[0].run_at:
                         event = self._queued_events.pop(0)
                         return event
             raise Empty

src/program/managers/event_manager.py

@@ -486,6 +486,10 @@ def get_season_index_by_id(self, item_id):
         return None
 
     def _determine_state(self):
+        if all(season.state == States.Paused for season in self.seasons):
+            return States.Paused
+        if all(season.state == States.Failed for season in self.seasons):
+            return States.Failed
         if all(season.state == States.Completed for season in self.seasons):
             return States.Completed
         if any(season.state in [States.Ongoing, States.Unreleased] for season in self.seasons):
@@ -595,6 +599,10 @@ def __init__(self, item):
 
     def _determine_state(self):
         if len(self.episodes) > 0:
+            if all(episode.state == States.Paused for episode in self.episodes):
+                return States.Paused
+            if all(episode.state == States.Failed for episode in self.episodes):
+                return States.Failed
             if all(episode.state == States.Completed for episode in self.episodes):
                 return States.Completed
             if any(episode.state == States.Unreleased for episode in self.episodes):

src/program/media/item.py

@@ -247,18 +247,18 @@ def _update_ongoing(self) -> None:
     def _schedule_functions(self) -> None:
         """Schedule each service based on its update interval."""
         scheduled_functions = {
-            self._update_ongoing: {"interval": 60 * 60 * 24},
+            self._update_ongoing: {"interval": 60 * 60 * 4},
             self._retry_library: {"interval": 60 * 60 * 24},
             log_cleaner: {"interval": 60 * 60},
             vacuum_and_analyze_index_maintenance: {"interval": 60 * 60 * 24},
         }
 
         if settings_manager.settings.symlink.repair_symlinks:
-            # scheduled_functions[fix_broken_symlinks] = {
-            #     "interval": 60 * 60 * settings_manager.settings.symlink.repair_interval,
-            #     "args": [settings_manager.settings.symlink.library_path, settings_manager.settings.symlink.rclone_path]
-            # }
-            logger.warning("Symlink repair is disabled, this will be re-enabled in the future.")
+            scheduled_functions[fix_broken_symlinks] = {
+                "interval": 60 * 60 * settings_manager.settings.symlink.repair_interval,
+                "args": [settings_manager.settings.symlink.library_path, settings_manager.settings.symlink.rclone_path]
+            }
+            # logger.warning("Symlink repair is disabled, this will be re-enabled in the future.")
 
         for func, config in scheduled_functions.items():
             self.scheduler.add_job(

src/program/program.py

@@ -43,14 +43,14 @@ def validate(self):
     def run(self, item: MediaItem):
         logger.debug(f"Starting download process for {item.log_string} ({item.id})")
 
-        if item.is_parent_blocked():
-            logger.debug(f"Skipping {item.log_string} ({item.id}) as it has a blocked parent, or is a blocked item")
-            yield item
-
         if item.active_stream or item.last_state in [States.Completed, States.Symlinked, States.Downloaded]:
             logger.debug(f"Skipping {item.log_string} ({item.id}) as it has already been downloaded by another download session")
             yield item
 
+        if item.is_parent_blocked():
+            logger.debug(f"Skipping {item.log_string} ({item.id}) as it has a blocked parent, or is a blocked item")
+            yield item
+
         if not item.streams:
             logger.debug(f"No streams available for {item.log_string} ({item.id})")
             yield item
@@ -96,18 +96,23 @@ def validate_stream(self, stream: Stream, item: MediaItem) -> Optional[TorrentCo
 
         valid_files = []
         for file in container.files or []:
+            if isinstance(file, DebridFile):
+                valid_files.append(file)
+                continue
+
             try:
                 debrid_file = DebridFile.create(
                     filename=file.filename,
                     filesize_bytes=file.filesize,
                     filetype=item.type,
                     file_id=file.file_id
                 )
+
+                if isinstance(debrid_file, DebridFile):
+                    valid_files.append(debrid_file)
             except InvalidDebridFileException as e:
                 logger.debug(f"{stream.infohash}: {e}")
                 continue
-            if debrid_file:
-                valid_files.append(debrid_file)
 
         if valid_files:
             container.files = valid_files

src/program/services/downloaders/__init__.py

@@ -77,6 +77,14 @@ def create(
 
         return cls(filename=filename, filesize=filesize_bytes, file_id=file_id)
 
+    def to_dict(self) -> Dict[str, Union[int, str]]:
+        """Convert the DebridFile to a dictionary"""
+        return {
+            "filename": self.filename,
+            "filesize": self.filesize,
+            "file_id": self.file_id
+        }
+
 
 class ParsedFileData(BaseModel):
     """Represents a parsed file from a filename"""
@@ -100,6 +108,13 @@ def file_ids(self) -> List[int]:
         """Get the file ids of the cached files"""
         return [file.file_id for file in self.files if file.file_id is not None]
 
+    def to_dict(self) -> Dict[str, Union[str, Dict]]:
+        """Convert the TorrentContainer to a dictionary including the infohash"""
+        return {
+            "infohash": self.infohash,
+            "files": {file.file_id: file.to_dict() for file in self.files if file}
+        }
+
 
 class TorrentInfo(BaseModel):
     """Torrent information from a debrid service"""

src/program/services/downloaders/models.py

@@ -132,17 +132,15 @@ def _validate_premium(self) -> bool:
 
     def get_instant_availability(self, infohash: str, item_type: str) -> Optional[TorrentContainer]:
         """
-        Get instant availability for multiple infohashes.
+        Get instant availability for a single infohash.
         Creates a makeshift availability check since Real-Debrid no longer supports instant availability.
         """
-        valid_container: Optional[TorrentContainer] = None
+        container: Optional[TorrentContainer] = None
         torrent_id = None
 
         try:
             torrent_id = self.add_torrent(infohash)
             container = self._process_torrent(torrent_id, infohash, item_type)
-            if container:
-                valid_container = container
         except InvalidDebridFileException as e:
             logger.debug(f"{infohash}: {e}")
         except Exception as e:
@@ -151,12 +149,23 @@ def get_instant_availability(self, infohash: str, item_type: str) -> Optional[To
             if torrent_id is not None:
                 self.delete_torrent(torrent_id)
 
-        return valid_container
+        return container
 
     def _process_torrent(self, torrent_id: str, infohash: str, item_type: str) -> Optional[TorrentContainer]:
         """Process a single torrent and return a TorrentContainer if valid."""
         torrent_info = self.get_torrent_info(torrent_id)
-        
+        if not torrent_info:
+            logger.debug(f"No torrent info found for {torrent_id} with infohash {infohash}")
+            return None
+
+        if not torrent_info.files:
+            logger.debug(f"No files found in torrent {torrent_id} with infohash {infohash}")
+            return None
+
+        if torrent_info.status in ("magnet_error", "error", "virus", "dead"):
+            logger.debug(f"Torrent {torrent_id} with infohash {infohash} is invalid. Torrent status on Real-Debrid: {torrent_info.status}")
+            return None
+
         if torrent_info.status == "waiting_files_selection":
             video_file_ids = [
                 file_id for file_id, file_info in torrent_info.files.items()
@@ -171,25 +180,30 @@ def _process_torrent(self, torrent_id: str, infohash: str, item_type: str) -> Op
             torrent_info = self.get_torrent_info(torrent_id)
 
             if torrent_info.status != "downloaded":
-                logger.debug(f"Torrent {torrent_id} with infohash {infohash} is not cached")
+                logger.debug(f"Torrent {torrent_id} with infohash {infohash} is not cached.")
                 return None
 
-        if not torrent_info.files:
-            return None
-
-        torrent_files = [
-            file for file in (
-                DebridFile.create(
-                    file_info["filename"],
-                    file_info["bytes"],
-                    item_type,
-                    file_id
+        torrent_files = []
+        for file_id, file_info in torrent_info.files.items():
+            try:
+                debrid_file = DebridFile.create(
+                    filename=file_info["filename"],
+                    filesize_bytes=file_info["bytes"],
+                    filetype=item_type,
+                    file_id=file_id
                 )
-                for file_id, file_info in torrent_info.files.items()
-            ) if file is not None
-        ]
 
-        return TorrentContainer(infohash=infohash, files=torrent_files) if torrent_files else None
+                if isinstance(debrid_file, DebridFile):
+                    torrent_files.append(debrid_file)
+            except InvalidDebridFileException as e:
+                logger.debug(f"{infohash}: {e}")
+                continue
+
+        if not torrent_files:
+            logger.debug(f"No valid files found after validating files in torrent {torrent_id} with infohash {infohash}")
+            return None
+
+        return TorrentContainer(infohash=infohash, files=torrent_files)
 
     def add_torrent(self, infohash: str) -> str:
         """Add a torrent by infohash"""
@@ -223,7 +237,7 @@ def get_torrent_info(self, torrent_id: str) -> TorrentInfo:
         """Get information about a torrent"""
         try:
             data = self.api.request_handler.execute(HttpMethod.GET, f"torrents/info/{torrent_id}")
-            files = {file["id"]: {"filename": file["path"].split("/")[-1], "bytes": file["bytes"]} for file in data["files"]}
+            files = {file["id"]: {"filename": file["path"].split("/")[-1], "bytes": file["bytes"], "selected": file["selected"]} for file in data["files"]}
             return TorrentInfo(
                 id=data["id"],
                 name=data["filename"],

src/program/services/downloaders/realdebrid.py

@@ -8,6 +8,7 @@
 from program.media.item import MediaItem, Movie, Show
 from program.media.state import States
 from program.services.post_processing.subliminal import Subliminal
+# from program.services.post_processing.ffmpeg import VideoValidator
 from program.settings.manager import settings_manager
 from program.utils.notifications import notify_on_complete
 
@@ -18,17 +19,20 @@ def __init__(self):
         self.initialized = False
         self.settings = settings_manager.settings.post_processing
         self.services = {
-            Subliminal: Subliminal()
+            Subliminal: Subliminal(),
+            # VideoValidator: VideoValidator()
         }
         self.initialized = True
 
     def run(self, item: MediaItem):
-        if Subliminal.should_submit(item):
-            self.services[Subliminal].run(item)
+        for service in self.services.values():
+            if service.initialized and service.should_submit(item):
+                service.run(item)
         if item.last_state == States.Completed:
             clear_streams(item)
         yield item
 
+
 def notify(item: MediaItem):
     show = None
     if item.type in ["show", "movie"]:
@@ -49,4 +53,4 @@ def _notify(_item: Show | Movie):
     duration = round((datetime.now() - _item.requested_at).total_seconds())
     logger.success(f"{_item.log_string} has been completed in {duration} seconds.")
     if settings_manager.settings.notifications.enabled:
-        notify_on_complete(_item)
+        notify_on_complete(_item)