Add pause feature to backend

src/alembic/versions/20241114_1400_c99239e3445f_add_pause_functionality.py

@@ -0,0 +1,43 @@
+"""add_pause_functionality
+
+Revision ID: c99239e3445f
+revision: str = 'c99239e3445f'
+Revises: c99709e3648f
+Create Date: 2024-11-14 16:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = '[generate a new revision ID]'
+down_revision: Union[str, None] = 'c99709e3648f'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Add pause-related columns to MediaItem table
+    op.add_column('MediaItem',
+        sa.Column('is_paused', sa.Boolean(), nullable=True, default=False))
+    op.add_column('MediaItem',
+        sa.Column('paused_at', sa.DateTime(), nullable=True))
+    op.add_column('MediaItem',
+        sa.Column('unpaused_at', sa.DateTime(), nullable=True))
+    op.add_column('MediaItem',
+        sa.Column('paused_by', sa.String(), nullable=True))
+
+    # Add index for is_paused column
+    op.create_index(op.f('ix_mediaitem_is_paused'), 'MediaItem', ['is_paused'])
+
+
+def downgrade() -> None:
+    # Remove pause-related columns from MediaItem table
+    op.drop_index(op.f('ix_mediaitem_is_paused'), table_name='MediaItem')
+    op.drop_column('MediaItem', 'paused_by')
+    op.drop_column('MediaItem', 'unpaused_at')
+    op.drop_column('MediaItem', 'paused_at')
+    op.drop_column('MediaItem', 'is_paused')

src/program/media/item.py

@@ -61,6 +61,12 @@ class MediaItem(db.Model):
     last_state: Mapped[Optional[States]] = mapped_column(sqlalchemy.Enum(States), default=States.Unknown)
     subtitles: Mapped[list[Subtitle]] = relationship(Subtitle, back_populates="parent", lazy="selectin", cascade="all, delete-orphan")
 
+    # Pause related fields
+    is_paused: Mapped[Optional[bool]] = mapped_column(sqlalchemy.Boolean, default=False)
+    paused_at: Mapped[Optional[datetime]] = mapped_column(sqlalchemy.DateTime, nullable=True)
+    unpaused_at: Mapped[Optional[datetime]] = mapped_column(sqlalchemy.DateTime, nullable=True)
+    paused_by: Mapped[Optional[str]] = mapped_column(sqlalchemy.String, nullable=True)
+
     __mapper_args__ = {
         "polymorphic_identity": "mediaitem",
         "polymorphic_on":"type",
@@ -70,6 +76,7 @@ class MediaItem(db.Model):
     __table_args__ = (
         Index("ix_mediaitem_type", "type"),
         Index("ix_mediaitem_requested_by", "requested_by"),
+        Index("ix_mediaitem_is_paused", "is_paused"),
         Index("ix_mediaitem_title", "title"),
         Index("ix_mediaitem_imdb_id", "imdb_id"),
         Index("ix_mediaitem_tvdb_id", "tvdb_id"),
@@ -241,6 +248,9 @@ def to_dict(self):
             "requested_by": self.requested_by,
             "scraped_at": str(self.scraped_at),
             "scraped_times": self.scraped_times,
+            "is_paused": self.is_paused,
+            "paused_at": str(self.paused_at) if self.paused_at else None,
+            "unpaused_at": str(self.unpaused_at) if self.unpaused_at else None,
         }
 
     def to_extended_dict(self, abbreviated_children=False, with_streams=True):
@@ -391,6 +401,49 @@ def _reset(self):
     def log_string(self):
         return self.title or self.id
 
+    def pause(self) -> None:
+        """Pause processing of this media item"""
+        if self.is_paused:
+            logger.debug(f"{self.log_string} is already paused")
+            return
+
+        logger.debug(f"Pausing {self.id}")
+        try:
+            self.is_paused = True
+            self.paused_at = datetime.now()
+            self.unpaused_at = None
+
+            session = object_session(self)
+            if session:
+                session.flush()
+
+            logger.info(f"{self.log_string} paused, is_paused={self.is_paused}, paused_at={self.paused_at}")
+        except Exception as e:
+            logger.error(f"Failed to pause {self.log_string}: {str(e)}")
+            raise
+
+    def unpause(self) -> None:
+        """Resume processing of this media item"""
+
+        if not self.is_paused:
+            logger.debug(f"{self.log_string} is not paused")
+            return
+
+        logger.debug(f"Unpausing {self.id}")
+        try:
+            self.is_paused = False
+            self.unpaused_at = datetime.now()
+            # Keep paused_at for history
+
+            session = object_session(self)
+            if session:
+                session.flush()
+
+            logger.info(f"{self.log_string} unpaused, is_paused={self.is_paused}, unpaused_at={self.unpaused_at}")
+        except Exception as e:
+            logger.error(f"Failed to unpause {self.log_string}: {str(e)}")
+            raise
+
     @property
     def collection(self):
         return self.parent.collection if self.parent else self.id
@@ -714,4 +767,4 @@ def copy_item(item):
     elif isinstance(item, MediaItem):
         return MediaItem(item={}).copy(item)
     else:
-        raise ValueError(f"Cannot copy item of type {type(item)}")
+        raise ValueError(f"Cannot copy item of type {type(item)}")

src/program/state_transition.py

@@ -18,67 +18,10 @@ def process_event(emitted_by: Service, existing_item: MediaItem | None = None, c
     no_further_processing: ProcessedEvent = (None, [])
     items_to_submit = []
 
-#TODO - Reindex non-released badly indexed items here
-    if content_item or (existing_item is not None and existing_item.last_state == States.Requested):
-        next_service = TraktIndexer
-        logger.debug(f"Submitting {content_item.log_string if content_item else existing_item.log_string} to trakt indexer")
-        return next_service, [content_item or existing_item]
-
-    elif existing_item is not None and existing_item.last_state in [States.PartiallyCompleted, States.Ongoing]:
-        if existing_item.type == "show":
-            for season in existing_item.seasons:
-                if season.last_state not in [States.Completed, States.Unreleased]:
-                    _, sub_items = process_event(emitted_by, season, None)
-                    items_to_submit += sub_items
-        elif existing_item.type == "season":
-            for episode in existing_item.episodes:
-                if episode.last_state != States.Completed:
-                    _, sub_items = process_event(emitted_by, episode, None)
-                    items_to_submit += sub_items
-
-    elif existing_item is not None and existing_item.last_state == States.Indexed:
-        next_service = Scraping
-        if emitted_by != Scraping and Scraping.should_submit(existing_item):
-            items_to_submit = [existing_item]
-        elif existing_item.type == "show":
-            items_to_submit = [s for s in existing_item.seasons if s.last_state != States.Completed and Scraping.should_submit(s)]
-        elif existing_item.type == "season":
-            items_to_submit = [e for e in existing_item.episodes if e.last_state != States.Completed and Scraping.should_submit(e)]
-
-    elif existing_item is not None and existing_item.last_state == States.Scraped:
-        next_service = Downloader
-        items_to_submit = [existing_item]
-
-    elif existing_item is not None and existing_item.last_state == States.Downloaded:
-        next_service = Symlinker
-        items_to_submit = [existing_item]
-
-    elif existing_item is not None and existing_item.last_state == States.Symlinked:
-        next_service = Updater
-        items_to_submit = [existing_item]
-
-    elif existing_item is not None and existing_item.last_state == States.Completed:
-        # If a user manually retries an item, lets not notify them again
-        if emitted_by not in ["RetryItem", PostProcessing]:
-            notify(existing_item)
-        # Avoid multiple post-processing runs
-        if emitted_by != PostProcessing:
-            if settings_manager.settings.post_processing.subliminal.enabled:
-                next_service = PostProcessing
-                if existing_item.type in ["movie", "episode"] and Subliminal.should_submit(existing_item):
-                    items_to_submit = [existing_item]
-                elif existing_item.type == "show":
-                    items_to_submit = [e for s in existing_item.seasons for e in s.episodes if e.last_state == States.Completed and Subliminal.should_submit(e)]
-                elif existing_item.type == "season":
-                    items_to_submit = [e for e in existing_item.episodes if e.last_state == States.Completed and Subliminal.should_submit(e)]
-                if not items_to_submit:
-                    return no_further_processing
-        else:
-
-            return no_further_processing
-
-    # if items_to_submit and next_service:
-    #     for item in items_to_submit:
-    #         logger.debug(f"Submitting {item.log_string} ({item.id}) to {next_service if isinstance(next_service, str) else next_service.__name__}")
-
+    # Skip processing if item is paused
+    if existing_item and existing_item.is_paused:
+        logger.debug(f"Skipping {existing_item.log_string} - item is paused")
+        return no_further_processing
+
+    #not sure if i need to remove this
     return next_service, items_to_submit