Add pause

src/alembic/versions/20241114_1400_c99239e3445f_add_pause_functionality.py

@@ -0,0 +1,39 @@
+"""add_pause_functionality
+
+Revision ID: [generate a new revision ID]
+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('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', 'paused_at')
+    op.drop_column('MediaItem', 'is_paused')

src/program/media/item.py

@@ -77,6 +77,10 @@ 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)
+
     __mapper_args__ = {
         "polymorphic_identity": "mediaitem",
         "polymorphic_on":"type",
@@ -257,6 +261,8 @@ 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,
         }
 
     def to_extended_dict(self, abbreviated_children=False, with_streams=True):
@@ -407,6 +413,24 @@ def _reset(self):
     def log_string(self):
         return self.title or self.id
 
+    def pause(self) -> None:
+        """Pause processing of this media item"""
+        logger.debug(f"Pausing {self.id}")
+        self.is_paused = True
+        self.paused_at = datetime.now()
+        logger.info(f"{self.log_string} paused, is_paused={self.is_paused}, paused_at={self.paused_at}")
+
+        # Get the session and flush changes
+        session = object_session(self)
+        if session:
+            session.flush()
+
+    def unpause(self) -> None:
+        """Resume processing of this media item"""
+        self.is_paused = False
+        self.paused_at = None
+        logger.info(f"{self.log_string} unpaused")
+
     @property
     def collection(self):
         return self.parent.collection if self.parent else self.id

src/program/state_transition.py

@@ -17,45 +17,62 @@ 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
+    # 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
+
+    # Process new content items or requested items
     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]
 
+    # Process partially completed or ongoing items
     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]:
+                # Skip paused seasons
+                if not season.is_paused and 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:
+                # Skip paused episodes
+                if not episode.is_paused and episode.last_state != States.Completed:
                     _, sub_items = process_event(emitted_by, episode, None)
                     items_to_submit += sub_items
 
+    # Process indexed 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)]
+            # Filter out paused seasons
+            items_to_submit = [s for s in existing_item.seasons 
+                             if not s.is_paused and 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)]
+            # Filter out paused episodes
+            items_to_submit = [e for e in existing_item.episodes 
+                             if not e.is_paused and e.last_state != States.Completed and Scraping.should_submit(e)]
 
+    # Process scraped items
     elif existing_item is not None and existing_item.last_state == States.Scraped:
         next_service = Downloader
         items_to_submit = [existing_item]
 
+    # Process downloaded items
     elif existing_item is not None and existing_item.last_state == States.Downloaded:
         next_service = Symlinker
         items_to_submit = [existing_item]
 
+    # Process symlinked items
     elif existing_item is not None and existing_item.last_state == States.Symlinked:
         next_service = Updater
         items_to_submit = [existing_item]
 
+    # Process completed items
     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]:
@@ -73,11 +90,6 @@ def process_event(emitted_by: Service, existing_item: MediaItem | None = None, c
                 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__}")
-
     return next_service, items_to_submit